trading_agent.risk.kill_switch

The kill switch: stops every desk and closes every position in the account.

It calls tradingmachine's Account.flatten, which halts UBI's order engine, cancels every open order at every broker and closes every position. tradingmachine warns that a flatten that times out may still have happened, and that calling it twice can close positions twice, so the halt is written to the ledger before UBI is called, and the switch never fires again while that halt is in force. In dry_run mode the flatten is a dry run: UBI reports what it would cancel and close, and sends nothing.

Typical usage example:

kill_switch = KillSwitch(reader, ledger, event_bus, 'dry_run') await kill_switch.trigger('Daily loss limit reached', until_date, now)

KillSwitch

KillSwitch(reader, ledger, event_bus, mode)

Halts trading and flattens the account.

Creates the switch.

Parameters:

Name	Type	Description	Default
reader	BrokerReader	The shared reader, whose client the flatten goes through.	required
ledger	RiskLedger	Where the halt is recorded.	required
event_bus	EventBus	Where the switch's firing and its outcome are published.	required
mode	str	"dry_run" for a flatten that sends nothing, "live" for a real one.	required

Source code in src/trading_agent/risk/kill_switch.py

def __init__(
    self,
    reader: broker_reader_module.BrokerReader,
    ledger: risk_ledger_module.RiskLedger,
    event_bus: event_bus_module.EventBus,
    mode: str,
):
    """Creates the switch.

    Args:
        reader (broker_reader_module.BrokerReader): The shared reader, whose client the flatten goes through.
        ledger (risk_ledger_module.RiskLedger): Where the halt is recorded.
        event_bus (event_bus_module.EventBus): Where the switch's firing and its outcome are published.
        mode (str): "dry_run" for a flatten that sends nothing, "live" for a real one.
    """
    self._reader = reader
    self._ledger = ledger
    self._event_bus = event_bus
    self._mode = mode

trigger async

trigger(reason, until_date, now)

Halts the account and flattens it, unless a halt is already in force.

Parameters:

Name	Type	Description	Default
reason	str	Why, for the ledger, the journal and the desks.	required
until_date	date	The first day trading is allowed again.	required
now	datetime	The current time.	required

Returns:

Type	Description
dict[str, Any]	UBI's answer to the flatten, an error description, or a note that the switch had already fired.

Source code in src/trading_agent/risk/kill_switch.py

async def trigger(
    self,
    reason: str,
    until_date: datetime.date,
    now: datetime.datetime,
) -> dict[str, Any]:
    """Halts the account and flattens it, unless a halt is already in force.

    Args:
        reason (str): Why, for the ledger, the journal and the desks.
        until_date (datetime.date): The first day trading is allowed again.
        now (datetime.datetime): The current time.

    Returns:
        dict[str, Any]: UBI's answer to the flatten, an error description, or a note that the switch had already fired.
    """
    today = now.date()
    existing = self._ledger.active_halt(
        risk_ledger_module.ACCOUNT_SCOPE, today
    )
    if existing is not None:
        return {
            'already_halted': existing,
        }
    self._ledger.halt(
        risk_ledger_module.ACCOUNT_SCOPE, reason, until_date, now
    )
    self._publish(
        event_module.EventKind.KILL_SWITCH,
        {
            'stage': 'fired',
            'reason': reason,
            'until_date': until_date.isoformat(),
            'dry_run': self._mode != 'live',
        },
    )
    try:
        outcome = await self._reader.call(self._flatten)
    except ubi_exceptions.UnifiedBrokerInterfaceError as error:
        _LOGGER.exception('The flatten failed')
        outcome = {
            'error': f'{type(error).__name__}: {error}',
        }
    self._publish(
        event_module.EventKind.KILL_SWITCH,
        {
            'stage': 'flatten_finished',
            'outcome': outcome,
        },
    )
    return outcome