strategy.base

The lower-level Strategy protocol and StrategyContext — the seam that lets one class run against the simulator or the live gateway.

base

The write-once Strategy base class and its injected context.

A strategy subclasses :class:Strategy and talks ONLY to self.ctx — protocol-typed edges (OrderApi/PositionApi/HistoryApi/Clock). The SAME subclass runs in backtest (wired to a SimBroker) and live (wired to topstep_sdk.AsyncTopstepClient), because both satisfy the identical structural protocols. Never import a concrete broker or fill model here.

StrategyContext dataclass

StrategyContext(orders: OrderApi, positions: PositionApi, history: HistoryApi, clock: Clock, account_id: int, instruments: Mapping[str, InstrumentSpec] = dict[str, InstrumentSpec]())

Everything a strategy may touch. Protocol-typed: sim and live inject

different concretions behind the same names.

orders instance-attribute

orders: OrderApi

positions instance-attribute

positions: PositionApi

history instance-attribute

history: HistoryApi

clock instance-attribute

clock: Clock

account_id instance-attribute

account_id: int

instruments class-attribute instance-attribute

instruments: Mapping[str, InstrumentSpec] = field(default_factory=dict[str, InstrumentSpec])

from_broker classmethod

from_broker(broker: Broker, *, clock: Clock, account_id: int, instruments: Mapping[str, InstrumentSpec] | None = None) -> StrategyContext

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

@classmethod
def from_broker(
    cls,
    broker: Broker,
    *,
    clock: Clock,
    account_id: int,
    instruments: Mapping[str, InstrumentSpec] | None = None,
) -> StrategyContext:
    return cls(
        orders=broker.orders,
        positions=broker.positions,
        history=broker.history,
        clock=clock,
        account_id=account_id,
        instruments=dict(instruments or {}),
    )

instrument

instrument(contract_id: str) -> InstrumentSpec

Spec for a contract id (registered mapping first, table fallback).

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

def instrument(self, contract_id: str) -> InstrumentSpec:
    """Spec for a contract id (registered mapping first, table fallback)."""
    found = self.instruments.get(contract_id)
    if found is not None:
        return found
    return spec_for_symbol(symbol_of_contract_id(contract_id))

Strategy

Base strategy. Override the hooks you need; all are optional.

Lifecycle: on_start -> (on_bar / on_order / on_fill / on_position)* -> on_stop. Order-state changes arrive via on_order; executions via on_fill (the parity-safe pattern in both sim and live — never busy-poll wait_for_fill in a backtest).

Drivers (the BacktestEngine, the future live runner) deliver events through handle_* — pure pass-throughs here. User strategies override on_*; framework base classes interpose in handle_*, so overriding a user hook can never sever framework bookkeeping.

ctx instance-attribute

ctx: StrategyContext

bind

bind(ctx: StrategyContext) -> None

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

def bind(self, ctx: StrategyContext) -> None:
    self.ctx = ctx

note

note(text: str) -> None

Attach a free-text breadcrumb to the current bar's replay frame.

This is how a strategy explains WHY, which no recorder can infer from the order flow: self.note(f"cross up, {self.fast.value} > ..."). A pure sink — with no recorder attached (Backtest(record=False), the default) it is a no-op, and with one attached it writes to the recording only. It can never influence the run: results are byte-identical with and without notes (pinned by a golden test).

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

def note(self, text: str) -> None:
    """Attach a free-text breadcrumb to the current bar's replay frame.

    This is how a strategy explains WHY, which no recorder can infer from
    the order flow: ``self.note(f"cross up, {self.fast.value} > ...")``.
    A pure sink — with no recorder attached (``Backtest(record=False)``,
    the default) it is a no-op, and with one attached it writes to the
    recording only. It can never influence the run: results are
    byte-identical with and without notes (pinned by a golden test).
    """
    sink = self._note_sink
    if sink is not None:
        sink(text)

on_start

on_start() -> None

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

def on_start(self) -> None:
    pass

on_bar async

on_bar(bar: Bar) -> None

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

async def on_bar(self, bar: Bar) -> None:
    pass

on_order async

on_order(order: OrderModel) -> None

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

async def on_order(self, order: OrderModel) -> None:
    pass

on_fill async

on_fill(trade: HalfTradeModel) -> None

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

async def on_fill(self, trade: HalfTradeModel) -> None:
    pass

on_position async

on_position(position: PositionModel) -> None

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

async def on_position(self, position: PositionModel) -> None:
    pass

on_stop

on_stop() -> None

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

def on_stop(self) -> None:
    pass

handle_bar async

handle_bar(bar: Bar) -> None

Driver entry point for a completed bar (drivers call handle_*, never on_*); the default simply awaits the user hook.

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

async def handle_bar(self, bar: Bar) -> None:
    """Driver entry point for a completed bar (drivers call ``handle_*``,
    never ``on_*``); the default simply awaits the user hook."""
    await self.on_bar(bar)

handle_order async

handle_order(order: OrderModel) -> None

Driver entry point for an order-state event (see handle_bar).

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

async def handle_order(self, order: OrderModel) -> None:
    """Driver entry point for an order-state event (see ``handle_bar``)."""
    await self.on_order(order)

handle_fill async

handle_fill(trade: HalfTradeModel) -> None

Driver entry point for an execution event (see handle_bar).

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

async def handle_fill(self, trade: HalfTradeModel) -> None:
    """Driver entry point for an execution event (see ``handle_bar``)."""
    await self.on_fill(trade)

handle_position async

handle_position(position: PositionModel) -> None

Driver entry point for a position event (see handle_bar).

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

async def handle_position(self, position: PositionModel) -> None:
    """Driver entry point for a position event (see ``handle_bar``)."""
    await self.on_position(position)