trading_agent.runtime.event_bus

Delivers every published event to every subscriber.

Each subscriber gets its own bounded queue. A subscriber that falls behind loses its oldest events rather than slowing the desks down, and the number lost is counted.

Typical usage example:

event_bus = EventBus() with event_bus.subscribe() as subscription: event_bus.publish(Event.now(EventKind.TEXT, desk='main')) event = await subscription.next_event()

EventSubscription

EventSubscription(event_bus, maximum_queued)

One subscriber's queue of events.

Attributes:

Name	Type	Description
dropped_count		How many events were thrown away because the queue was full.

Creates an empty subscription.

Parameters:

Name	Type	Description	Default
event_bus	EventBus	The bus this subscription belongs to.	required
maximum_queued	int	How many unread events to keep before dropping the oldest.	required

Source code in src/trading_agent/runtime/event_bus.py

def __init__(self, event_bus: EventBus, maximum_queued: int):
    """Creates an empty subscription.

    Args:
        event_bus (EventBus): The bus this subscription belongs to.
        maximum_queued (int): How many unread events to keep before dropping the oldest.
    """
    self._event_bus = event_bus
    self._queue = asyncio.Queue(maxsize=maximum_queued)
    self.dropped_count = 0

deliver

deliver(published_event)

Adds an event to the queue, dropping the oldest one if it is full.

Parameters:

Name	Type	Description	Default
published_event	Event	The event to deliver.	required

Source code in src/trading_agent/runtime/event_bus.py

def deliver(self, published_event: event_module.Event) -> None:
    """Adds an event to the queue, dropping the oldest one if it is full.

    Args:
        published_event (event_module.Event): The event to deliver.
    """
    if self._queue.full():
        self._queue.get_nowait()
        self.dropped_count += 1
    self._queue.put_nowait(published_event)

next_event async

next_event()

Waits for and returns the next event.

Returns:

Type	Description
Event	The oldest unread event.

Source code in src/trading_agent/runtime/event_bus.py

async def next_event(self) -> event_module.Event:
    """Waits for and returns the next event.

    Returns:
        event_module.Event: The oldest unread event.
    """
    return await self._queue.get()

close

close()

Stops delivery to this subscription.

Source code in src/trading_agent/runtime/event_bus.py

def close(self) -> None:
    """Stops delivery to this subscription."""
    self._event_bus.unsubscribe(self)

__enter__

__enter__()

Returns the subscription for use in a with statement.

Returns:

Type	Description
EventSubscription	This subscription.

Source code in src/trading_agent/runtime/event_bus.py

def __enter__(self) -> Self:
    """Returns the subscription for use in a with statement.

    Returns:
        EventSubscription: This subscription.
    """
    return self

__exit__

__exit__(exception_type, exception, traceback)

Closes the subscription when the with statement ends.

Parameters:

Name	Type	Description	Default
exception_type	type[BaseException] | None	The type of exception raised in the block, if any.	required
exception	BaseException | None	The exception raised in the block, if any.	required
traceback	TracebackType | None	The traceback of that exception, if any.	required

Source code in src/trading_agent/runtime/event_bus.py

def __exit__(
    self,
    exception_type: type[BaseException] | None,
    exception: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    """Closes the subscription when the with statement ends.

    Args:
        exception_type (type[BaseException] | None): The type of exception raised in the block, if any.
        exception (BaseException | None): The exception raised in the block, if any.
        traceback (TracebackType | None): The traceback of that exception, if any.
    """
    self.close()

EventBus

EventBus()

The publish and subscribe hub that every component shares.

Creates a bus with no subscribers.

Source code in src/trading_agent/runtime/event_bus.py

def __init__(self):
    """Creates a bus with no subscribers."""
    self._subscriptions = set()

publish

publish(published_event)

Delivers an event to every current subscriber.

Parameters:

Name	Type	Description	Default
published_event	Event	The event to deliver.	required

Source code in src/trading_agent/runtime/event_bus.py

def publish(self, published_event: event_module.Event) -> None:
    """Delivers an event to every current subscriber.

    Args:
        published_event (event_module.Event): The event to deliver.
    """
    for subscription in list(self._subscriptions):
        subscription.deliver(published_event)

subscribe

subscribe(maximum_queued=_DEFAULT_MAXIMUM_QUEUED)

Starts a new subscription that receives every later event.

Parameters:

Name	Type	Description	Default
maximum_queued	int	How many unread events to keep before dropping the oldest.	_DEFAULT_MAXIMUM_QUEUED

Returns:

Type	Description
EventSubscription	The new subscription.

Source code in src/trading_agent/runtime/event_bus.py

def subscribe(
    self,
    maximum_queued: int = _DEFAULT_MAXIMUM_QUEUED,
) -> EventSubscription:
    """Starts a new subscription that receives every later event.

    Args:
        maximum_queued (int): How many unread events to keep before dropping the oldest.

    Returns:
        EventSubscription: The new subscription.
    """
    subscription = EventSubscription(self, maximum_queued)
    self._subscriptions.add(subscription)
    return subscription

unsubscribe

unsubscribe(subscription)

Stops delivering events to a subscription.

Parameters:

Name	Type	Description	Default
subscription	EventSubscription	The subscription to stop.	required

Source code in src/trading_agent/runtime/event_bus.py

def unsubscribe(self, subscription: EventSubscription) -> None:
    """Stops delivering events to a subscription.

    Args:
        subscription (EventSubscription): The subscription to stop.
    """
    self._subscriptions.discard(subscription)

subscriber_count property

subscriber_count

How many subscriptions are currently receiving events.