Skip to content

trading.execution.engine.events.event_bus

trading.execution.engine.events.event_bus

EventBus 

EventBus()

Allows for methods to subscribe whenever a specific event is published.

Initializes EventBus with empty subscriptions.

Source code in src\contango\trading\execution\engine\events\event_bus.py 
30
31
32
33
34
35
def __init__(self) -> None:
    """
    Initializes `EventBus` with empty subscriptions.
    """
    self._handlers: dict[type, list[_handler_type]] = defaultdict(list)
    self._priorities: dict[type, list[int]] = defaultdict(list)

publish 

publish(event: object) -> None

Publishes an event to all handlers registered for its type.

Parameters:

Name Type Description Default
event object

The underlying event to publish.

 required
Source code in src\contango\trading\execution\engine\events\event_bus.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def publish(self, event: object) -> None:
    """
    Publishes an event to all handlers registered for its type.

    Args:
        event: The underlying event to publish.
    """
    handlers = self._handlers.get(type(event))

    if handlers is None:
        return

    for handler in handlers:
        handler(event)

subscribe 

subscribe(event_type: type, handler: _handler_type, priority: int) -> None

Subscribes a handler to an event type. Higher priority handlers execute first.

Parameters:

Name Type Description Default
event_type type

The type for the event to be subscribed to.

 required
handler _handler_type

The method that recieves the event.

 required
priority int

The priority (highest first) for the subscription.

 required
Source code in src\contango\trading\execution\engine\events\event_bus.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def subscribe(
    self,
    event_type: type,
    handler: _handler_type,
    priority: int,
) -> None:
    """
    Subscribes a handler to an event type. Higher priority handlers execute first.

    Args:
        event_type: The type for the event to be subscribed to.
        handler: The method that recieves the event.
        priority: The priority (highest first) for the subscription.
    """
    if priority in self._priorities[event_type]:
        raise ValueError("Two subscriptions must not have the same priority!")

    self._priorities[event_type].append(priority)
    self._handlers[event_type].append(handler)

    # Sort both lists together by priority descending
    paired = sorted(
        zip(self._priorities[event_type], self._handlers[event_type]),
        reverse=True
    )
    self._priorities[event_type] = [p for p, _ in paired]
    self._handlers[event_type] = [h for _, h in paired]