trading_agent.scheduling.market_watcher

Watches the market for the desks, with no model: fills, price alerts and the account's loss limits.

While NSE equities or MCX trade, every cycle (15 seconds by default) it:

1. reads UBI's order book once and finds the finished broker orders (complete, cancelled, rejected or expired) that belong to a desk's own orders in the desk order book, matched by parent_id or order_id; each desk gets one message listing its updates;
2. reads the last price of every instrument a desk has an alert on, once each, and wakes the desk for each alert crossed, removing the alert;
3. once a minute, asks the risk engine to check the account, so the kill switch fires at the daily or weekly loss limit, and a desk is halted at its own daily loss limit, even when no desk is placing orders; the account's and each desk's profit and loss are then published as an account_status event.

The ids of orders already reported are kept in a small file, so a restart neither repeats a fill nor forgets one that finished while the gateway was down. Dry-run orders reach no broker, so they never appear in the order book and are skipped.

Typical usage example:

watcher = MarketWatcher(desks, alert_stores, order_book, reader, risk_engine, hours, event_bus, state_file) task = asyncio.create_task(watcher.run())

MarketWatcher

MarketWatcher(desks, alert_stores, order_book, reader, risk_engine, hours, event_bus, state_file, interval_seconds=15.0, account_check_seconds=60.0, clock=None)

The model-free loop that wakes desks when something happens in the market.

Creates the watcher.

Parameters:

Name	Type	Description	Default
desks	Mapping[str, DeskRuntime]	The running desks, keyed by name.	required
alert_stores	Mapping[str, PriceAlertStore]	Each desk's alerts.	required
order_book	DeskOrderBook	Which desk placed which order.	required
reader	BrokerReader	The shared UBI reader.	required
risk_engine	RiskEngine	Checks the account's loss limits.	required
hours	ExchangeHours	Decides when the market is open.	required
event_bus	EventBus	Where fills and alerts are published.	required
state_file	Path	Where the ids of reported orders are kept.	required
interval_seconds	float	The time between cycles while the market is open.	15.0
account_check_seconds	float	The time between account checks.	60.0
clock	Callable[[], datetime] | None	Returns the current time; None uses the system clock.	None

Source code in src/trading_agent/scheduling/market_watcher.py

def __init__(
    self,
    desks: Mapping[str, desk_runtime_module.DeskRuntime],
    alert_stores: Mapping[str, price_alert_store_module.PriceAlertStore],
    order_book: desk_order_book_module.DeskOrderBook,
    reader: broker_reader_module.BrokerReader,
    risk_engine: risk_engine_module.RiskEngine,
    hours: exchange_hours_module.ExchangeHours,
    event_bus: event_bus_module.EventBus,
    state_file: Path,
    interval_seconds: float = 15.0,
    account_check_seconds: float = 60.0,
    clock: Callable[[], datetime.datetime] | None = None,
):
    """Creates the watcher.

    Args:
        desks (Mapping[str, desk_runtime_module.DeskRuntime]): The running desks, keyed by name.
        alert_stores (Mapping[str, price_alert_store_module.PriceAlertStore]): Each desk's alerts.
        order_book (desk_order_book_module.DeskOrderBook): Which desk placed which order.
        reader (broker_reader_module.BrokerReader): The shared UBI reader.
        risk_engine (risk_engine_module.RiskEngine): Checks the account's loss limits.
        hours (exchange_hours_module.ExchangeHours): Decides when the market is open.
        event_bus (event_bus_module.EventBus): Where fills and alerts are published.
        state_file (Path): Where the ids of reported orders are kept.
        interval_seconds (float): The time between cycles while the market is open.
        account_check_seconds (float): The time between account checks.
        clock (Callable[[], datetime.datetime] | None): Returns the current time; None uses the system clock.
    """
    self._desks = desks
    self._alert_stores = alert_stores
    self._order_book = order_book
    self._reader = reader
    self._risk_engine = risk_engine
    self._hours = hours
    self._event_bus = event_bus
    self._state_file = state_file
    self._interval_seconds = interval_seconds
    self._account_check_seconds = account_check_seconds
    self._clock = clock
    self._last_account_check: datetime.datetime | None = None
    self._reported_order_ids = self._read_reported_order_ids()

run async

run()

Runs a cycle whenever the market is open, until cancelled.

Source code in src/trading_agent/scheduling/market_watcher.py

async def run(self) -> None:
    """Runs a cycle whenever the market is open, until cancelled."""
    while True:
        now = self._now()
        if not self.market_open(now):
            await asyncio.sleep(_CLOSED_MARKET_SLEEP_SECONDS)
            continue
        try:
            await self.cycle(now)
        except Exception:
            _LOGGER.exception('A market watcher cycle failed')
        await asyncio.sleep(self._interval_seconds)

market_open

market_open(moment)

Says whether NSE equities or MCX trade at a moment.

Parameters:

Name	Type	Description	Default
moment	datetime	The moment.	required

Returns:

Type	Description
bool	True when either is open.

Source code in src/trading_agent/scheduling/market_watcher.py

def market_open(self, moment: datetime.datetime) -> bool:
    """Says whether NSE equities or MCX trade at a moment.

    Args:
        moment (datetime.datetime): The moment.

    Returns:
        bool: True when either is open.
    """
    nse_open, _ = self._hours.is_open(
        'nse', 'nse_equities', moment, 'exchange_hours'
    )
    mcx_open, _ = self._hours.is_open(
        'mcx', 'mcx_commodities', moment, 'exchange_hours'
    )
    return nse_open or mcx_open

cycle async

cycle(now)

Checks fills, alerts and, when due, the account.

Parameters:

Name	Type	Description	Default
now	datetime	The current time.	required

Source code in src/trading_agent/scheduling/market_watcher.py

async def cycle(self, now: datetime.datetime) -> None:
    """Checks fills, alerts and, when due, the account.

    Args:
        now (datetime.datetime): The current time.
    """
    await self.check_fills()
    await self.check_alerts()
    due = (
        self._last_account_check is None
        or (now - self._last_account_check).total_seconds()
        >= self._account_check_seconds
    )
    if due:
        self._last_account_check = now
        snapshot = await self._risk_engine.check_account()
        if snapshot is not None:
            self._publish_account_status(snapshot)

check_fills async

check_fills()

Reports newly finished broker orders to the desks that placed them.

Returns:

Type	Description
int	How many order updates were reported.

Source code in src/trading_agent/scheduling/market_watcher.py

async def check_fills(self) -> int:
    """Reports newly finished broker orders to the desks that placed them.

    Returns:
        int: How many order updates were reported.
    """
    records = []
    for desk_name in self._desks:
        for record in self._order_book.open_orders(desk_name):
            if not record.dry_run:
                records.append(record)
    if not records:
        return 0
    try:
        book = await self._reader.call(
            functools.partial(self._reader.get, _ORDER_BOOK_PATH)
        )
    except ubi_exceptions.UnifiedBrokerInterfaceError as error:
        _LOGGER.warning('The order book could not be read: %s', error)
        return 0
    updates: dict[str, list[str]] = {}
    orders = []
    if isinstance(book, dict):
        orders = book.get('orders') or []
    for entry in orders:
        order_id = str(entry.get('order_id', ''))
        status = str(entry.get('status') or '').upper()
        if (
            status not in _FINISHED_STATUSES
            or order_id in self._reported_order_ids
        ):
            continue
        record = self._matching_record(entry, records)
        if record is None:
            continue
        self._reported_order_ids.append(order_id)
        line = self._fill_line(entry, record, status)
        updates.setdefault(record.desk, []).append(line)
        self._publish(
            event_module.EventKind.FILL,
            record.desk,
            {
                'order_id': order_id,
                'parent_id': record.parent_id,
                'record_id': record.record_id,
                'symbol': record.symbol,
                'transaction_type': record.transaction_type,
                'quantity': record.quantity,
                'status': status,
                'filled_quantity': entry.get('filled_quantity'),
                'average_price': entry.get('average_price'),
                'message': entry.get('status_message'),
            },
        )
    if updates:
        self._write_reported_order_ids()
    for desk_name, lines in updates.items():
        self._wake(
            desk_name, '[Order update from the market watcher]', lines
        )
    return sum(len(lines) for lines in updates.values())

check_alerts async

check_alerts()

Wakes desks whose price alerts were crossed, and removes those alerts.

Returns:

Type	Description
int	How many alerts fired.

Source code in src/trading_agent/scheduling/market_watcher.py

async def check_alerts(self) -> int:
    """Wakes desks whose price alerts were crossed, and removes those alerts.

    Returns:
        int: How many alerts fired.
    """
    prices: dict[str, float | None] = {}
    fired = 0
    for desk_name, store in self._alert_stores.items():
        lines = []
        for alert in store.alerts():
            if alert.instrument_id not in prices:
                prices[alert.instrument_id] = await self._last_price(
                    alert.instrument_id
                )
            last_price = prices[alert.instrument_id]
            if last_price is None or not alert.is_crossed(last_price):
                continue
            store.remove(alert.alert_id)
            fired += 1
            note = f' Your note: {alert.note}' if alert.note else ''
            lines.append(
                f'- {alert.symbol} traded at {last_price:,.2f}, {alert.condition} your alert at {alert.price:,.2f}.{note}'
            )
            self._publish(
                event_module.EventKind.PRICE_ALERT,
                desk_name,
                {
                    'alert_id': alert.alert_id,
                    'symbol': alert.symbol,
                    'condition': alert.condition,
                    'price': alert.price,
                    'last_price': last_price,
                    'note': alert.note,
                },
            )
        if lines:
            self._wake(
                desk_name, '[Price alert from the market watcher]', lines
            )
    return fired