trading_agent.runtime.trigger

A reason to wake a desk and run one turn.

Typical usage example:

trigger = Trigger.now(TriggerSource.USER_MESSAGE, desk='main', text='How is the market?')

TriggerSource

Bases: StrEnum

Where a trigger came from.

Trigger dataclass

Trigger(trigger_id, source, desk, text, created_at, attachments=(), request_id=None)

One reason to run a turn.

Attributes:

Name	Type	Description
trigger_id	str	A unique identifier, used to match events to the trigger that caused them.
source	TriggerSource	Where the trigger came from.
desk	str	The desk that should handle it.
text	str	What the desk is asked or told.
created_at	datetime	When the trigger was created, in UTC.
attachments	tuple[Attachment, ...]	The pictures and documents the user sent with a message.
request_id	str | None	The desk's question, plan or permission request this message answers, or None.

now classmethod

now(source, desk, text, attachments=(), request_id=None)

Creates a trigger with a new identifier and the current time.

Parameters:

Name	Type	Description	Default
source	TriggerSource	Where the trigger came from.	required
desk	str	The desk that should handle it.	required
text	str	What the desk is asked or told.	required
attachments	tuple[Attachment, ...]	The pictures and documents sent with it.	()
request_id	str | None	The request this message answers, or None.	None

Returns:

Type	Description
Trigger	The new trigger.

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

@classmethod
def now(
    cls,
    source: TriggerSource,
    desk: str,
    text: str,
    attachments: tuple[attachment_module.Attachment, ...] = (),
    request_id: str | None = None,
) -> Self:
    """Creates a trigger with a new identifier and the current time.

    Args:
        source (TriggerSource): Where the trigger came from.
        desk (str): The desk that should handle it.
        text (str): What the desk is asked or told.
        attachments (tuple[attachment_module.Attachment, ...]): The pictures and documents sent with it.
        request_id (str | None): The request this message answers, or None.

    Returns:
        Trigger: The new trigger.
    """
    return cls(
        trigger_id=uuid.uuid4().hex,
        source=source,
        desk=desk,
        text=text,
        created_at=datetime.datetime.now(datetime.UTC),
        attachments=attachments,
        request_id=request_id,
    )

to_json_dictionary

to_json_dictionary()

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

Returns:

Type	Description
dict[str, Any]	The trigger's fields, with the time as an ISO 8601 string; attachments only when there are some.

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

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

    Returns:
        dict[str, Any]: The trigger's fields, with the time as an ISO 8601 string; attachments only when there are some.
    """
    dictionary = {
        'trigger_id': self.trigger_id,
        'source': str(self.source),
        'desk': self.desk,
        'text': self.text,
        'created_at': self.created_at.isoformat(),
    }
    if self.attachments:
        described = []
        for attachment in self.attachments:
            described.append(attachment.to_json_dictionary())
        dictionary['attachments'] = described
    if self.request_id is not None:
        dictionary['request_id'] = self.request_id
    return dictionary

images property

images

The attachments that are pictures.