strategy.symbol

SymbolStrategy: the single-instrument chassis you subclass. Owns indicator updates, the warmup gate, and position bookkeeping.

symbol

Single-contract ergonomic strategy base (the backtesting.py-style dialect).

SymbolStrategy narrows the raw Strategy seam to one contract: events for other instruments never reach the on_* hooks, use()-registered indicators are auto-updated once per matching bar (registration order == update order), and on_bar is gated until every indicator is ready. The position/working-order views fold ONLY the SDK models both brokers emit (plus a submit-time latch of each sugar-placed order id, superseded by its first real event), and the order sugar targets the ctx protocol surfaces with rejections routed to on_reject — a rejection is normal combine control flow (position cap, no-trade window), not an exception. The raw raising path remains self.ctx.orders.

IndicatorScope

Bases: Struct

A registered indicator's data scope and how far it has advanced.

The diagnostic for "why is nothing happening?" — a scoped indicator that has advanced far less than the bar count is warming slowly by design, not misbehaving. session is None for an indicator on the continuous tape.

indicator instance-attribute

indicator: Indicator

session instance-attribute

session: Session | None

updates instance-attribute

updates: int

needs instance-attribute

needs: int

Updates required before this indicator is warm, in ITS OWN cadence.

warm property

warm: bool

SymbolStrategy

SymbolStrategy(contract_id: str, *, require_ready: bool = True, warmup: int | None = None, trade_sessions: Sequence[Session] | None = None)

Bases: Strategy

Base for strategies trading exactly one contract.

Subclasses register indicators with use() in __init__ and override on_bar — invoked only for contract_id, only once every registered indicator is ready (and, when warmup is explicitly overridden, at least that many matching bars have been seen; an explicit warmup gates even under require_ready=False), with every indicator already updated for that bar. position and working_orders are folded from the same user events live emits, and buy/sell latch the returned order id into working_orders at submit time — so entry guards like position.flat and not self.working_orders hold even when live hub confirmations lag the REST return past the next bar.

A bracket is not an attachment to the position: the venue turns it into two real reduce-only orders when the entry fills, so move_stop / move_target amend them mid-trade (stop_orders / target_orders are the same children, unfiltered by intent). An amended level is live from the NEXT bar — this bar's fill walk ran before on_bar was called.

Sessions are two INDEPENDENT switches, and keeping them independent is the point:

• use(indicator, session=...) scopes an indicator's INPUT DATA — which bars it is computed from.
• trade_sessions= scopes DECISIONS — when on_bar may fire.

So a strategy can hold a continuous 24h Ema beside an NY-only Atr and still trade only the NY session. Indicators advance regardless of trade_sessions: starving one outside the tradable window would leave it with gaps and a different value than the same indicator on the same tape.

trade_sessions narrows only when this strategy chooses to act. It never widens what the venue permits — the engine's 16:10 ET flatten and the 16:10-18:00 no-trade window apply either way.

One known hook-cadence gap: the sim emits no PositionModel event on a full close (live sends a closed/size-0 snapshot), so detect flatness from position.flat in on_fill — never by overriding on_position. The views themselves stay correct on both sides.

Source code in src/topstep_backtest/strategy/symbol.py

def __init__(
    self,
    contract_id: str,
    *,
    require_ready: bool = True,
    warmup: int | None = None,
    trade_sessions: Sequence[Session] | None = None,
) -> None:
    self.contract_id = contract_id
    self._require_ready = require_ready
    self._warmup_override = warmup
    self._trade_sessions = tuple(trade_sessions) if trade_sessions is not None else None
    self._slots: list[_Slot] = []
    self._position = NetPosition()
    self._orders = OrderTracker()
    self._bars_seen = 0
    self._bars_gated = 0
    self._bars_out_of_session = 0

contract_id instance-attribute

contract_id = contract_id

warmup property

warmup: int

Declared warmup: the explicit override, else the max indicator lookback.

registered_indicators property

registered_indicators: tuple[Indicator, ...]

The use()-registered indicators, in registration (= update) order.

Read-only view; the replay recorder introspects it to capture per-bar indicator values without the strategy having to do anything.

position property

position: NetPosition

This contract's net position, folded from fills and snapshots.

working_orders property

working_orders: tuple[OrderModel, ...]

Non-terminal orders on this contract, ascending order id.

spec property

spec: InstrumentSpec

Instrument economics for this contract.

history_bars property

history_bars: int

Bars needed before every registered indicator is WARM.

warmup is where values first EXIST; this is where they stop depending on where the run started. Parity with a live account begins here (AGENTS.md §5.2), and so does comparability between two windows of one long tape — a strategy started cold in the middle of a tape computes different values from the same bars than one that has been running.

Indicators holding no history of their own (Cross) contribute their lookback: their warmth is their inputs', and those are registered separately.

COUNTED IN EACH INDICATOR'S OWN CADENCE. A session-scoped indicator advances only on its session's bars, so it needs this many bars OF THAT SESSION — which is several times more tape. On 5-minute bars an NY-scoped Sma(30) sees 78 bars per session, so its 1920 updates span ~25 trading days, against ~7 for the same indicator on a 24h feed. Size a preload from history_bars_by_session when anything is scoped, and read warm rather than comparing counts by hand.

history_bars_by_session property

history_bars_by_session: dict[Session | None, int]

history_bars split by data scope; None keys the continuous tape.

Each entry is a requirement in that scope's own cadence, which is what makes it actionable: a preload must contain at least that many bars of each session, not that many bars in total.

warm property

warm: bool

Has every registered indicator received its history_bars updates?

Counted per indicator from what it actually consumed, so this stays correct when scoped and continuous indicators advance at different rates — a total bar count cannot express that.

registrations property

registrations: tuple[IndicatorScope, ...]

Every use()-d indicator with its scope and progress, in registration order.

bars_seen property

bars_seen: int

Matching bars delivered to this strategy, gated or not.

bars_gated property

bars_gated: int

Matching bars swallowed by the ready-gate so far.

bars_out_of_session property

bars_out_of_session: int

Matching bars that passed the ready-gate but fell outside trade_sessions.

Separate from bars_gated because the two answer different questions when a strategy never trades: still warming up, or never in session.

trade_sessions property

trade_sessions: tuple[Session, ...] | None

Sessions in which on_bar may fire; None means every bar.

stop_orders property

stop_orders: tuple[OrderModel, ...]

The working bracket STOPs protecting this contract's position.

Bracket children only — an order carrying parent_order_id, the gateway's own shape for "this exists because that entry filled". A stop you placed yourself is not one of these, and that is the distinction that matters: a breakout strategy resting a stop-ENTRY above the market while flat must never have it mistaken for protection and moved.

target_orders property

target_orders: tuple[OrderModel, ...]

The working bracket take-profits, by the same rule as stop_orders.

use

use(indicator: T, *, session: Session | None = None) -> T

Register an indicator: auto-updated on every matching bar and

counted toward the ready-gate. Registration order == update order — use() a Cross's inputs BEFORE the Cross that reads them (enforced: a Cross with unregistered inputs would read one-bar- stale values with no error, so it is refused here instead). A TalibLine input counts as registered once the indicator that OWNS it is, since one owner computes all of its lines together.

session scopes the indicator's INPUT DATA. The default None feeds it every matching bar — the continuous tape, including Asia and London on a 24h feed. Naming a Session advances it only on that session's bars::

self.trend = self.use(Ema(50))                     # continuous
self.atr   = self.use(Atr(14), session=NEW_YORK)   # NY bars only

This is independent of trade_sessions: scoping an indicator does not restrict when the strategy trades, and restricting trading does not starve an indicator. Choosing between them is a modelling decision — dispersion measures (Atr, StdDev, Rsi) describe how much price moves per bar and that is session-dependent, while level measures (Sma, Ema) answer where price is and the overnight move is real.

A scoped indicator warms in ITS OWN cadence, so it needs history_bars bars OF ITS SESSION — several times more tape than an unscoped one (see history_bars).

A Cross INHERITS its inputs' scope and may not be given a conflicting one: comparing values sampled on different cadences is meaningless, and inheriting removes the trap of scoping the inputs but forgetting the Cross.

Source code in src/topstep_backtest/strategy/symbol.py

def use[T: Indicator](self, indicator: T, *, session: Session | None = None) -> T:
    """Register an indicator: auto-updated on every matching bar and

    counted toward the ready-gate. Registration order == update order —
    ``use()`` a ``Cross``'s inputs BEFORE the ``Cross`` that reads them
    (enforced: a ``Cross`` with unregistered inputs would read one-bar-
    stale values with no error, so it is refused here instead). A
    ``TalibLine`` input counts as registered once the indicator that OWNS
    it is, since one owner computes all of its lines together.

    ``session`` scopes the indicator's INPUT DATA. The default ``None``
    feeds it every matching bar — the continuous tape, including Asia and
    London on a 24h feed. Naming a ``Session`` advances it only on that
    session's bars::

        self.trend = self.use(Ema(50))                     # continuous
        self.atr   = self.use(Atr(14), session=NEW_YORK)   # NY bars only

    This is independent of ``trade_sessions``: scoping an indicator does
    not restrict when the strategy trades, and restricting trading does not
    starve an indicator. Choosing between them is a modelling decision —
    dispersion measures (``Atr``, ``StdDev``, ``Rsi``) describe how much
    price moves per bar and that is session-dependent, while level measures
    (``Sma``, ``Ema``) answer where price is and the overnight move is real.

    A scoped indicator warms in ITS OWN cadence, so it needs
    ``history_bars`` bars OF ITS SESSION — several times more tape than an
    unscoped one (see ``history_bars``).

    A ``Cross`` INHERITS its inputs' scope and may not be given a
    conflicting one: comparing values sampled on different cadences is
    meaningless, and inheriting removes the trap of scoping the inputs but
    forgetting the ``Cross``.
    """
    if isinstance(indicator, TalibLine):
        raise ValueError(
            "a TalibLine cannot be use()-registered — it has no state of "
            "its own and would never advance; register the indicator that "
            "owns it and keep the line only as a Cross input"
        )
    _require_indicator(indicator)
    if isinstance(indicator, Cross):
        _check_cross_inputs(indicator, {id(slot.indicator) for slot in self._slots})
        session = self._cross_scope(indicator, session)
        indicator.mark_registered()
    self._slots.append(_Slot(indicator=indicator, session=session))
    return indicator

in_trade_session

in_trade_session(bar: Bar) -> bool

Is bar inside a tradable session? Always True when unrestricted.

Source code in src/topstep_backtest/strategy/symbol.py

def in_trade_session(self, bar: Bar) -> bool:
    """Is ``bar`` inside a tradable session? Always True when unrestricted."""
    if self._trade_sessions is None:
        return True
    return any(session.contains(bar) for session in self._trade_sessions)

prewarm

prewarm(bars: Iterable[Bar]) -> int

Advance the registered indicators over bars WITHOUT trading.

Returns the count of matching bars consumed. Bars for other contracts are skipped, exactly as handle_bar skips them.

This exists for running one strategy over successive windows of a long tape. The window's Backtest must see ONLY that window's bars — hand it the preceding history too and those days land in day_records as flat days, which moves closed_days, every daily percentile, stdev, sortino and the drawdown durations while leaving P&L untouched. So the history is driven through here instead: indicators advance, on_bar is never called, no order can exist, and the run that follows starts warm on its first real bar with a clean set of statistics.

Call this BEFORE handing the strategy to a Backtest. Afterwards it would interleave history with live bars and corrupt the indicator state it is meant to establish — which is why it raises once bound.

Raises:

Type	Description
RuntimeError	if the strategy has already been bound to a context.

Source code in src/topstep_backtest/strategy/symbol.py

def prewarm(self, bars: Iterable[Bar]) -> int:
    """Advance the registered indicators over ``bars`` WITHOUT trading.

    Returns the count of matching bars consumed. Bars for other contracts
    are skipped, exactly as ``handle_bar`` skips them.

    This exists for running one strategy over successive windows of a long
    tape. The window's ``Backtest`` must see ONLY that window's bars — hand
    it the preceding history too and those days land in ``day_records`` as
    flat days, which moves `closed_days`, every daily percentile, `stdev`,
    `sortino` and the drawdown durations while leaving P&L untouched. So the
    history is driven through here instead: indicators advance, ``on_bar``
    is never called, no order can exist, and the run that follows starts
    warm on its first real bar with a clean set of statistics.

    Call this BEFORE handing the strategy to a ``Backtest``. Afterwards it
    would interleave history with live bars and corrupt the indicator state
    it is meant to establish — which is why it raises once bound.

    Raises:
        RuntimeError: if the strategy has already been bound to a context.
    """
    if hasattr(self, "ctx"):
        raise RuntimeError(
            "prewarm() after the strategy was bound would interleave history "
            "with live bars; prewarm before constructing the Backtest"
        )
    consumed = 0
    for bar in bars:
        if bar.bar_type.contract_id != self.contract_id:
            continue
        # Deliberately mirrors handle_bar minus the gate and on_bar: same
        # indicators, same order, same scope filter, same bars_seen
        # accounting, no decision. The scope filter MUST match handle_bar's
        # exactly — a divergence corrupts the very warm state this exists to
        # establish, and would do so silently.
        self._advance(bar)
        self._bars_seen += 1
        consumed += 1
    return consumed

handle_bar async

handle_bar(bar: Bar) -> None

Source code in src/topstep_backtest/strategy/symbol.py

async def handle_bar(self, bar: Bar) -> None:
    if bar.bar_type.contract_id != self.contract_id:
        return
    # Indicators advance FIRST and independently of trade_sessions: an
    # indicator starved outside the tradable window would develop gaps and
    # compute different values than the same indicator on the same tape.
    # Scoping data and scoping decisions are separate switches on purpose.
    self._advance(bar)
    self._bars_seen += 1
    if self._gated():
        self._bars_gated += 1
        return
    if not self.in_trade_session(bar):
        self._bars_out_of_session += 1
        return
    await self.on_bar(bar)

handle_order async

handle_order(order: OrderModel) -> None

Source code in src/topstep_backtest/strategy/symbol.py

async def handle_order(self, order: OrderModel) -> None:
    if order.contract_id != self.contract_id:
        return
    self._orders.apply(order)
    await self.on_order(order)

handle_fill async

handle_fill(trade: HalfTradeModel) -> None

Source code in src/topstep_backtest/strategy/symbol.py

async def handle_fill(self, trade: HalfTradeModel) -> None:
    if trade.contract_id != self.contract_id:
        return
    self._position.apply_fill(trade)
    await self.on_fill(trade)

handle_position async

handle_position(position: PositionModel) -> None

Source code in src/topstep_backtest/strategy/symbol.py

async def handle_position(self, position: PositionModel) -> None:
    if position.contract_id != self.contract_id:
        return
    self._position.apply_snapshot(position)
    await self.on_position(position)

on_reject async

on_reject(error: APIError) -> None

Called with the APIError when a sugar order call is rejected.

Default: ignore (the sugar call returns None).

Source code in src/topstep_backtest/strategy/symbol.py

async def on_reject(self, error: APIError) -> None:
    """Called with the ``APIError`` when a sugar order call is rejected.

    Default: ignore (the sugar call returns ``None``).
    """

buy async

buy(size: int, *, stop_loss_ticks: int | None = None, take_profit_ticks: int | None = None, limit_price: Decimal | None = None, stop_price: Decimal | None = None, custom_tag: str | None = None) -> int | None

Buy this contract (market unless a price kwarg implies otherwise);

returns the order id — latched into working_orders at submit time — or None on rejection (see on_reject).

Source code in src/topstep_backtest/strategy/symbol.py

async def buy(
    self,
    size: int,
    *,
    stop_loss_ticks: int | None = None,
    take_profit_ticks: int | None = None,
    limit_price: Decimal | None = None,
    stop_price: Decimal | None = None,
    custom_tag: str | None = None,
) -> int | None:
    """Buy this contract (market unless a price kwarg implies otherwise);

    returns the order id — latched into ``working_orders`` at submit time —
    or ``None`` on rejection (see ``on_reject``).
    """
    order_type = _implied_type(limit_price, stop_price)
    try:
        order_id = await self.ctx.orders.buy(
            self.ctx.account_id,
            self.contract_id,
            size,
            type=order_type,
            limit_price=limit_price,
            stop_price=stop_price,
            custom_tag=custom_tag,
            stop_loss_ticks=stop_loss_ticks,
            take_profit_ticks=take_profit_ticks,
        )
    except APIError as error:
        await self.on_reject(error)
        return None
    self._latch(order_id, OrderSide.BUY, order_type, size, limit_price, stop_price, custom_tag)
    return order_id

sell async

sell(size: int, *, stop_loss_ticks: int | None = None, take_profit_ticks: int | None = None, limit_price: Decimal | None = None, stop_price: Decimal | None = None, custom_tag: str | None = None) -> int | None

Sell this contract (market unless a price kwarg implies otherwise);

returns the order id — latched into working_orders at submit time — or None on rejection (see on_reject).

Source code in src/topstep_backtest/strategy/symbol.py

async def sell(
    self,
    size: int,
    *,
    stop_loss_ticks: int | None = None,
    take_profit_ticks: int | None = None,
    limit_price: Decimal | None = None,
    stop_price: Decimal | None = None,
    custom_tag: str | None = None,
) -> int | None:
    """Sell this contract (market unless a price kwarg implies otherwise);

    returns the order id — latched into ``working_orders`` at submit time —
    or ``None`` on rejection (see ``on_reject``).
    """
    order_type = _implied_type(limit_price, stop_price)
    try:
        order_id = await self.ctx.orders.sell(
            self.ctx.account_id,
            self.contract_id,
            size,
            type=order_type,
            limit_price=limit_price,
            stop_price=stop_price,
            custom_tag=custom_tag,
            stop_loss_ticks=stop_loss_ticks,
            take_profit_ticks=take_profit_ticks,
        )
    except APIError as error:
        await self.on_reject(error)
        return None
    self._latch(order_id, OrderSide.SELL, order_type, size, limit_price, stop_price, custom_tag)
    return order_id

move_stop async

move_stop(*, price: Decimal | None = None, ticks: int | None = None) -> int

Move every bracket stop on this contract; returns how many moved.

Pass exactly one of price (an absolute level, used as given — an off-grid price is the broker's rejection to make, not this method's to paper over) or ticks (an offset from the average entry, signed IN THE POSITION'S FAVOUR, so ticks=0 is breakeven whether long or short and ticks=10 is ten ticks of locked profit either way).

Rejections go to on_reject and the remaining orders still move, as cancel_working behaves. Returns 0 when there is nothing to move — an entry placed without stop_loss_ticks has no bracket stop — so check it if "the position is protected" is load-bearing.

The amended price reaches working_orders with the broker's order event, which is the next bar: this returns what the venue accepted, not a mutated local view.

Source code in src/topstep_backtest/strategy/symbol.py

async def move_stop(self, *, price: Decimal | None = None, ticks: int | None = None) -> int:
    """Move every bracket stop on this contract; returns how many moved.

    Pass exactly one of ``price`` (an absolute level, used as given — an
    off-grid price is the broker's rejection to make, not this method's to
    paper over) or ``ticks`` (an offset from the average entry, signed IN
    THE POSITION'S FAVOUR, so ``ticks=0`` is breakeven whether long or
    short and ``ticks=10`` is ten ticks of locked profit either way).

    Rejections go to ``on_reject`` and the remaining orders still move, as
    ``cancel_working`` behaves. Returns 0 when there is nothing to move —
    an entry placed without ``stop_loss_ticks`` has no bracket stop — so
    check it if "the position is protected" is load-bearing.

    The amended price reaches ``working_orders`` with the broker's order
    event, which is the next bar: this returns what the venue accepted,
    not a mutated local view.
    """
    return await self._amend(self.stop_orders, stop_price=self._level(price, ticks, "stop"))

move_target async

move_target(*, price: Decimal | None = None, ticks: int | None = None) -> int

Move every bracket take-profit on this contract; returns how many moved.

price / ticks and the rejection handling are move_stop's (ticks again measured from the average entry in the position's favour, so ticks=80 is an 80-tick target on either side).

Source code in src/topstep_backtest/strategy/symbol.py

async def move_target(self, *, price: Decimal | None = None, ticks: int | None = None) -> int:
    """Move every bracket take-profit on this contract; returns how many moved.

    ``price`` / ``ticks`` and the rejection handling are ``move_stop``'s
    (``ticks`` again measured from the average entry in the position's
    favour, so ``ticks=80`` is an 80-tick target on either side).
    """
    return await self._amend(
        self.target_orders, limit_price=self._level(price, ticks, "target")
    )

close async

close() -> None

Flatten this contract's position; a rejection goes to on_reject.

Source code in src/topstep_backtest/strategy/symbol.py

async def close(self) -> None:
    """Flatten this contract's position; a rejection goes to ``on_reject``."""
    try:
        await self.ctx.positions.close(self.ctx.account_id, self.contract_id)
    except APIError as error:
        await self.on_reject(error)

cancel_working async

cancel_working() -> None

Cancel every working order on this contract, one cancel per order;

each rejection goes to on_reject and the remaining cancels proceed.

Source code in src/topstep_backtest/strategy/symbol.py

async def cancel_working(self) -> None:
    """Cancel every working order on this contract, one cancel per order;

    each rejection goes to ``on_reject`` and the remaining cancels proceed.
    """
    for order in self.working_orders:
        try:
            await self.ctx.orders.cancel(self.ctx.account_id, order.id)
        except APIError as error:
            await self.on_reject(error)