trading_agent.runtime.event

One thing that happened inside the agent, as published on the event bus.

Typical usage example:

event = Event.now(EventKind.TURN_STARTED, desk='main', payload={'trigger_id': trigger_id})

EventKind

Bases: StrEnum

The kinds of event the agent publishes.

Event dataclass

Event(kind, desk, payload, created_at)

One published event.

Attributes:

Name	Type	Description
kind	EventKind	What happened.
desk	str | None	The desk it happened on, or None for the whole gateway.
payload	dict[str, Any]	Details that depend on the kind, kept JSON-serialisable.
created_at	datetime	When it happened, in UTC.

now classmethod

now(kind, desk, payload=None)

Creates an event stamped with the current time.

Parameters:

Name	Type	Description	Default
kind	EventKind	What happened.	required
desk	str | None	The desk it happened on, or None for the whole gateway.	required
payload	dict[str, Any] | None	Details of the event, or None for no details.	None

Returns:

Type	Description
Event	The new event.

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

@classmethod
def now(
    cls,
    kind: EventKind,
    desk: str | None,
    payload: dict[str, Any] | None = None,
) -> Self:
    """Creates an event stamped with the current time.

    Args:
        kind (EventKind): What happened.
        desk (str | None): The desk it happened on, or None for the whole gateway.
        payload (dict[str, Any] | None): Details of the event, or None for no details.

    Returns:
        Event: The new event.
    """
    if payload is None:
        payload = {}
    return cls(
        kind=kind,
        desk=desk,
        payload=payload,
        created_at=datetime.datetime.now(datetime.UTC),
    )

to_json_dictionary

to_json_dictionary()

Converts the event to a dictionary that json.dumps can write.

Returns:

Type	Description
dict[str, Any]	The event's fields, with the time as an ISO 8601 string.

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

def to_json_dictionary(self) -> dict[str, Any]:
    """Converts the event to a dictionary that json.dumps can write.

    Returns:
        dict[str, Any]: The event's fields, with the time as an ISO 8601 string.
    """
    return {
        'kind': str(self.kind),
        'desk': self.desk,
        'payload': self.payload,
        'created_at': self.created_at.isoformat(),
    }