Skip to content

Market data

Phase 2 risk checks (stop-losses, notional cap, price thresholds) need a live price source. Strix gets that through a MarketDataAdapter — a small Protocol with three methods. Strix ships one concrete impl (StaticMarketData, a dict-backed fixture); real adapters wrap your broker or vendor feed.

This guide covers when you need the adapter, what shape it has to be, and how marks flow through the system for stop-loss evaluation.

When you need an adapter

You need a MarketDataAdapter when:

  • You configure max_open_notional and use market orders (limit orders don't need an adapter — limit_price is the worst-case execution price).
  • You configure any price threshold (max_share_price, min_share_price, min_share_price_short) and use market orders.
  • You configure SessionStopLoss(mode="total") or any CostBasedStopLoss and want them to fire on unrealized drawdown (which requires marks).

You don't need one when:

  • All your risk checks are Phase 1 (max_qty_per_order, max_orders, position limits, etc.).
  • You only use limit orders and have no stop-losses (or SessionStopLoss(mode="realized") only).

The port

from typing import Protocol
from decimal import Decimal
from strix import Quote

class MarketDataAdapter(Protocol):
    def last(self, symbol: str) -> Decimal | None: ...
    def quote(self, symbol: str) -> Quote | None: ...
    def mark(self, symbol: str) -> Decimal | None: ...

All three return None when no data is available. Strix interprets None according to:

  • Pre-trade checks (notional cap, price thresholds): SessionConfig.on_missing_market_data policy. "reject" (default) raises StrixRiskError; "allow" logs a WARNING and proceeds.
  • Continuous risk (stop-loss evaluation): always logs a WARNING and skips the affected position — there's no order to reject mid-evaluation.

Sync contract

The adapter is synchronous. Strix calls last() / quote() on the order-placement hot path and on every continuous-risk evaluation. If your adapter does a blocking HTTP call, every order placement blocks for that long.

The expected pattern: your adapter is a thin wrapper over an in-memory dict your code maintains. A separate WebSocket/polling thread updates the dict; the adapter's accessors read from it without I/O.

The shipped fixture: StaticMarketData

For tests, replays, or single-snapshot workflows:

from decimal import Decimal
from strix import StaticMarketData, Quote

md = StaticMarketData(
    prices={"AAPL": Decimal("150.10"), "MSFT": Decimal("420.50")},
    quotes={"AAPL": Quote(bid=Decimal("150.05"), ask=Decimal("150.15"))},
)

You can mutate the snapshot:

md.set_last("AAPL", Decimal("151.00"))
md.set_quote("MSFT", Quote(bid=Decimal("420.30"), ask=Decimal("420.70")))
md.set_mark("AAPL", Decimal("150.85"))  # mark distinct from last

mark() falls back to last() if no explicit mark has been set — useful when your feed gives trade prices but not a separate end-of-day mark.

Quote validates bid <= ask at construction. A crossed market raises ValueError.

Building your own adapter

Anything that satisfies the Protocol works. A simple example wrapping a dict your feeder thread updates:

from decimal import Decimal
import threading
from strix import Quote

class LiveFeed:
    """Thin adapter; a separate thread writes; Strix reads."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._last: dict[str, Decimal] = {}
        self._quotes: dict[str, Quote] = {}

    # ── written by your feeder thread ──
    def update_last(self, symbol: str, price: Decimal) -> None:
        with self._lock:
            self._last[symbol] = price

    def update_quote(self, symbol: str, bid: Decimal, ask: Decimal) -> None:
        with self._lock:
            self._quotes[symbol] = Quote(bid=bid, ask=ask)

    # ── MarketDataAdapter Protocol ──
    def last(self, symbol: str) -> Decimal | None:
        with self._lock:
            return self._last.get(symbol)

    def quote(self, symbol: str) -> Quote | None:
        with self._lock:
            return self._quotes.get(symbol)

    def mark(self, symbol: str) -> Decimal | None:
        # Often == last; override if you have a distinct mark price
        return self.last(symbol)

feed = LiveFeed()
# Start your WebSocket subscriber thread that calls feed.update_last / update_quote
strix.init(market_data=feed, risk=...)

Wiring the adapter

Pass market_data to strix.init (and to strix.resume on restart — it's a live source, not persisted state):

strix.init(
    transport=strix.LocalTransport(data_dir="./strix_data"),
    market_data=feed,
    risk=...,
)

# After a crash:
strix.resume(transport=strix.LocalTransport(data_dir="./strix_data"), market_data=feed)

Forgetting to re-supply market_data on resume means any subsequent market order check will follow the missing-data policy (reject by default).

Marks vs. quotes — two different paths

Strix uses market data in two distinct ways:

  1. Quotes (live) — MarketDataAdapter.quote() and .last() are queried on demand during pre-trade checks (notional, price thresholds). Whatever the adapter has right now.

  2. Marks (cached) — Session._marks is a separate cache, populated only by explicit strix.mark_to_market(prices) calls. Used for unrealized-P&L computation during stop-loss evaluation.

Why a separate cache? Two reasons:

  • Ephemeral, not replayed. The marks cache is pure runtime state — it is not persisted and not rebuilt on resume. On strix.resume, the cache starts empty and repopulates as the caller resumes feeding marks (the live MarketDataAdapter must also be re-supplied). Periodic PnLSnapshot events keep the audit trail of P&L over time without writing every tick.
  • User control of cadence. Stop-losses fire on mark updates; the user decides when to push marks rather than Strix polling.

In practice, your tick loop does both:

# Your strategy's tick loop
def on_tick():
    # ... your usual logic ...

    # Feed marks for symbols Strix cares about so unrealized P&L is current.
    syms = strix.active_symbols()
    marks = {s: feed.last(s) for s in syms if feed.last(s) is not None}
    strix.mark_to_market(marks)

strix.active_symbols()

Returns the set of symbols Strix needs marks for right now:

  • Non-zero net position, OR
  • At least one order in an open status (PENDING_NEW, NEW, PARTIALLY_FILLED, PENDING_CANCEL).

Not a watchlist API — symbols are "active" by virtue of position or open order; there's no strix.watch(...). Filter your mark payload to active symbols to keep mark_to_market traffic minimal and let Strix warn you when you forget one.

Stale-mark warnings

When RiskSettings.stale_mark_threshold is set (default timedelta(seconds=5)), Strix logs a WARNING in two cases:

  • mark_to_market is called and an active symbol is missing from the payload — its cached mark may be used at evaluation time without refresh.
  • Stop-loss evaluation uses a cached mark whose age exceeds the threshold — the eval still runs, but the data is old.

Both warnings go through the standard logging module (logger strix._session). Set stale_mark_threshold=None to disable both.

import logging
logging.basicConfig(level=logging.WARNING)
# Strix warnings now visible.

In production, route these to your monitoring rather than stdout — repeated warnings often signal a feeder thread that's stuck or a list of held symbols that drifted from what your feed subscribes to.

What's not in v1

  • A real broker / vendor adapter. Strix doesn't ship integrations — every user's broker is different. You write the adapter once and reuse it.
  • Async API. MarketDataAdapter is sync. Async users either run their feeder thread in a run_in_executor or write a sync wrapper over their async client. If demand for strix.aio.* emerges, it'll ship as a sibling, not a replacement.
  • Strix-side polling. Strix never calls your adapter on its own timer. Marks flow through mark_to_market only; pre-trade quotes are pulled on order entry. This keeps Strix simple and lets you control I/O.