topstep_backtest

The public surface. Five names cover almost every script: Backtest, SymbolStrategy, AccountSize, Report and SummaryStats.

topstep_backtest

topstep-backtest: event-driven Topstep Combine backtesting with live parity.

Strategies are written once against the Broker/Clock protocols in :mod:topstep_backtest.protocols and run unchanged against the SimBroker (backtest) or topstep_sdk.AsyncTopstepClient (live).

Quick start (AGENTS.md §2)::

from topstep_backtest import AccountSize, Backtest

report = Backtest(bars, MyStrategy("CON.F.US.MNQ.U26"),
                  account=AccountSize.S50K).run()
print(report)          # verdict + balance path + day trail + summary stats
report.result          # the unchanged frozen BacktestResult

ASIA module-attribute

ASIA = Session('ASIA', ZoneInfo('Asia/Tokyo'), time(9, 0), time(15, 0))

LONDON module-attribute

LONDON = Session('LONDON', ZoneInfo('Europe/London'), time(8, 0), time(16, 30))

NEW_YORK module-attribute

NEW_YORK = Session('NEW_YORK', ET, time(9, 30), time(16, 0))

SESSIONS module-attribute

SESSIONS: dict[str, Session] = {session.name: session for session in (ASIA, LONDON, NEW_YORK)}

Session

Bases: Struct

A named intraday window, defined in its own local timezone.

The window is half-open — [start, end) — so a bar opening exactly at end belongs to the next window, and two adjacent sessions sharing a boundary never both claim the same bar.

start > end denotes a window that wraps midnight (an Asia session expressed in ET, say); membership is then "at/after start OR before end". start == end is rejected: it reads equally as an empty window and a 24-hour one.

name instance-attribute

name: str

tz instance-attribute

tz: ZoneInfo

start instance-attribute

start: time

end instance-attribute

end: time

wraps_midnight property

wraps_midnight: bool

True when the window runs past local midnight (start > end).

contains_ns

contains_ns(ns: int) -> bool

Is UTC-nanosecond instant ns inside this session's window?

The instant is converted to the session's OWN timezone before the comparison, so the answer follows that region's daylight-saving schedule rather than ET's.

Source code in src/topstep_backtest/core/sessions.py

def contains_ns(self, ns: int) -> bool:
    """Is UTC-nanosecond instant ``ns`` inside this session's window?

    The instant is converted to the session's OWN timezone before the
    comparison, so the answer follows that region's daylight-saving
    schedule rather than ET's.
    """
    local = ns_to_dt(ns).astimezone(self.tz).time()
    if self.wraps_midnight:
        return local >= self.start or local < self.end
    return self.start <= local < self.end

contains

contains(bar: Bar) -> bool

Does bar belong to this session?

Tested on ts_event — the bar's OPEN — not ts_init. A bar's session is a property of the window it covers, and ts_init is the close: testing it would pull the 09:29->09:30 bar, whose every print is pre-market, into the New York session. Both stamps are already in the past when a decision is made, so this introduces no look-ahead either way; it is purely about classifying the bar correctly.

Source code in src/topstep_backtest/core/sessions.py

def contains(self, bar: Bar) -> bool:
    """Does ``bar`` belong to this session?

    Tested on ``ts_event`` — the bar's OPEN — not ``ts_init``. A bar's
    session is a property of the window it covers, and ``ts_init`` is the
    close: testing it would pull the 09:29->09:30 bar, whose every print is
    pre-market, into the New York session. Both stamps are already in the
    past when a decision is made, so this introduces no look-ahead either
    way; it is purely about classifying the bar correctly.
    """
    return self.contains_ns(bar.ts_event)

Backtest

Backtest(data: Sequence[Bar], strategy: Strategy, *, account: AccountSize = S50K, dll_enabled: bool = False, validate: bool = True, record: bool = False, account_id: int = 1, fill_config: BarFillConfig | None = None, broker_config: SimBrokerConfig | None = None, fee_model: TopstepFees | None = None, **rejected: object)

The two-line runner: assemble the sim stack correctly and run it once.

data is a time-ordered Bar sequence (feed ordering is enforced at construction); strategy is a bound-ready instance with typed constructor parameters — never a class (AGENTS.md §2). Knobs are only things that exist in this project: the account size (and its optional Personal DLL), validation strictness, record (bar-by-bar replay capture onto Report.replay — observation only, results are byte-identical either way), the sim account id, and the fill/broker fidelity configs. Economics are never knobs.

Source code in src/topstep_backtest/harness.py

def __init__(
    self,
    data: Sequence[Bar],
    strategy: Strategy,
    *,
    account: AccountSize = AccountSize.S50K,
    dll_enabled: bool = False,
    validate: bool = True,
    record: bool = False,
    account_id: int = 1,
    fill_config: BarFillConfig | None = None,
    broker_config: SimBrokerConfig | None = None,
    fee_model: TopstepFees | None = None,
    **rejected: object,
) -> None:
    _refuse_strategy_class(strategy)
    _refuse_unknown_kwargs(rejected)
    self._bars: tuple[Bar, ...] = tuple(data)
    self._strategy = strategy
    self._account = account
    self._dll_enabled = dll_enabled
    self._record = record
    self._account_id = account_id
    self._fill_config = fill_config
    self._broker_config = broker_config
    self._fee_model = TopstepFees() if fee_model is None else fee_model
    self._ran = False

    # Instruments derive from the feed's contract ids — never hand-passed;
    # each contract validates against ITS spec (validate_bars is single-spec).
    groups: dict[str, list[Bar]] = {}
    for bar in self._bars:
        groups.setdefault(bar.bar_type.contract_id, []).append(bar)
    self._instruments = {cid: _spec_for_contract_id(cid) for cid in groups}
    if validate:
        errors: list[ValidationIssue] = []
        for cid, group in groups.items():
            report = validate_bars(group, self._instruments[cid])
            # The refuse/proceed decision is the validator's (report.ok:
            # INFO-only findings proceed); the filter below only strips
            # INFO findings from the error listing.
            if not report.ok:
                errors.extend(i for i in report.issues if i.code not in INFO_CODES)
        if errors:
            raise DataValidationError(tuple(errors))
    # Construction enforces the ordering invariant even under validate=False:
    # the engine trusts the DataFeed contract unconditionally.
    self._feed = ListBarFeed(self._bars)

from_dataframe classmethod

from_dataframe(df: Any, strategy: Strategy, *, contract_id: str, stamp: Literal['open', 'close'], unit: AggregateBarUnit, unit_number: int, account: AccountSize = S50K, dll_enabled: bool = False, validate: bool = True, record: bool = False, account_id: int = 1, fill_config: BarFillConfig | None = None, broker_config: SimBrokerConfig | None = None, fee_model: TopstepFees | None = None, **rejected: object) -> Backtest

Wrangle a pandas OHLCV DataFrame, then assemble as usual.

stamp, unit, and unit_number stay REQUIRED (no defaults), exactly as on the wrangler: the caller must declare whether source timestamps are bar opens or closes AND the bar span — guessing the stamp is the classic silent one-bar look-ahead, and defaulting the span would mis-stamp every non-1-minute bar's close (a 5-minute bar stamped with a 60s span acts 4 minutes early against the session clock).

Source code in src/topstep_backtest/harness.py

@classmethod
def from_dataframe(
    cls,
    df: Any,
    strategy: Strategy,
    *,
    contract_id: str,
    stamp: Literal["open", "close"],
    unit: AggregateBarUnit,
    unit_number: int,
    account: AccountSize = AccountSize.S50K,
    dll_enabled: bool = False,
    validate: bool = True,
    record: bool = False,
    account_id: int = 1,
    fill_config: BarFillConfig | None = None,
    broker_config: SimBrokerConfig | None = None,
    fee_model: TopstepFees | None = None,
    **rejected: object,
) -> Backtest:
    """Wrangle a pandas OHLCV DataFrame, then assemble as usual.

    ``stamp``, ``unit``, and ``unit_number`` stay REQUIRED (no defaults),
    exactly as on the wrangler: the caller must declare whether source
    timestamps are bar opens or closes AND the bar span — guessing the
    stamp is the classic silent one-bar look-ahead, and defaulting the
    span would mis-stamp every non-1-minute bar's close (a 5-minute bar
    stamped with a 60s span acts 4 minutes early against the session
    clock).
    """
    # Refusals come BEFORE the wrangle: a wrangler error (naive
    # timestamps, off-grid row) must never mask a §4 rejection.
    _refuse_strategy_class(strategy)
    _refuse_unknown_kwargs(rejected)
    bars = bars_from_dataframe(
        df,
        contract_id=contract_id,
        spec=_spec_for_contract_id(contract_id),
        unit=unit,
        unit_number=unit_number,
        stamp=stamp,
    )
    return cls(
        bars,
        strategy,
        account=account,
        dll_enabled=dll_enabled,
        validate=validate,
        record=record,
        account_id=account_id,
        fill_config=fill_config,
        broker_config=broker_config,
        fee_model=fee_model,
        **rejected,
    )

run

run() -> Report

Run to completion synchronously (wraps asyncio.run).

Raises:

Type	Description
RuntimeError	If called from inside a running event loop — use await backtest.arun() there instead.

Source code in src/topstep_backtest/harness.py

def run(self) -> Report:
    """Run to completion synchronously (wraps ``asyncio.run``).

    Raises:
        RuntimeError: If called from inside a running event loop — use
            ``await backtest.arun()`` there instead.
    """
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(self.arun())
    raise RuntimeError(
        "Backtest.run() was called from a running event loop; "
        "use `await backtest.arun()` instead"
    )

run_with_tearsheet

run_with_tearsheet(directory: str | PathLike[str] = '.', *, prefix: str = 'tearsheet', replay: ReplaySpec = 'auto') -> tuple[Report, Path]

Run, then ALWAYS write a timestamped tearsheet: (report, path).

For "give me the HTML on every run" — exactly :meth:run followed by :meth:Report.to_timestamped_html, so nothing about the run changes. Inside a running event loop, compose those two yourself::

report = await backtest.arun()
written = report.to_timestamped_html("runs")

Raises:

Type	Description
RuntimeError	If called from inside a running event loop (from :meth:run), or if this Backtest has already run.

Source code in src/topstep_backtest/harness.py

def run_with_tearsheet(
    self,
    directory: str | os.PathLike[str] = ".",
    *,
    prefix: str = "tearsheet",
    replay: ReplaySpec = "auto",
) -> tuple[Report, Path]:
    """Run, then ALWAYS write a timestamped tearsheet: ``(report, path)``.

    For "give me the HTML on every run" — exactly :meth:`run` followed by
    :meth:`Report.to_timestamped_html`, so nothing about the run changes.
    Inside a running event loop, compose those two yourself::

        report = await backtest.arun()
        written = report.to_timestamped_html("runs")

    Raises:
        RuntimeError: If called from inside a running event loop (from
            :meth:`run`), or if this ``Backtest`` has already run.
    """
    report = self.run()
    return report, report.to_timestamped_html(directory, prefix=prefix, replay=replay)

arun async

arun() -> Report

Assemble fresh engine state, run once, and report.

Clock, kernel, broker, engine, and the strategy instance are all stateful and single-use, so a Backtest refuses to run twice.

Source code in src/topstep_backtest/harness.py

async def arun(self) -> Report:
    """Assemble fresh engine state, run once, and report.

    Clock, kernel, broker, engine, and the strategy instance are all
    stateful and single-use, so a ``Backtest`` refuses to run twice.
    """
    if self._ran:
        raise RuntimeError(
            "this Backtest has already run — engine, broker, and strategy "
            "state are single-use; construct a new Backtest (with a fresh "
            "strategy instance) for another run"
        )
    self._ran = True
    params = combine_params(self._account, dll_enabled=self._dll_enabled)
    # ONE clock for broker AND engine — the assembly invariant this facade
    # exists to enforce (a broker on its own clock stamps accepted_ts=0,
    # making every order eligible for the current bar: silent look-ahead).
    clock = TestClock()
    broker = SimBroker(
        account_id=self._account_id,
        instruments=self._instruments,
        fill_model=BarFillModel(self._fill_config),
        fee_model=self._fee_model,
        kernel=CombineKernel(params),
        clock=clock,
        config=self._broker_config,
    )
    recorder = None
    if self._record:
        from .replay import Recorder  # deferred: keeps the default path slim

        recorder = Recorder()
    engine = BacktestEngine(
        feed=self._feed,
        broker=broker,
        strategy=self._strategy,
        clock=clock,
        recorder=recorder,
    )
    result = await engine.run()
    trades = broker.trades
    stats = compute_summary(result, trades=trades, consistency_pct=params.consistency_pct)
    bars_gated = (
        self._strategy.bars_gated if isinstance(self._strategy, SymbolStrategy) else None
    )
    return Report(
        result=result,
        stats=stats,
        trades=trades,
        params=params,
        bars_gated=bars_gated,
        bars=self._bars,
        instruments=self._instruments,
        replay=None if recorder is None else recorder.replay,
    )

DataValidationError

DataValidationError(issues: tuple[ValidationIssue, ...])

Bases: ValueError

Bar data failed validation; issues carries every ERROR finding.

Source code in src/topstep_backtest/harness.py

def __init__(self, issues: tuple[ValidationIssue, ...]) -> None:
    self.issues = issues
    shown = [f"[{issue.code}] {issue.message}" for issue in issues[:20]]
    if len(issues) > 20:
        shown.append(f"... and {len(issues) - 20} more")
    super().__init__(
        "bar data failed validation (pass validate=False to run anyway):\n  "
        + "\n  ".join(shown)
    )

issues instance-attribute

issues = issues

Report

Report(*, result: BacktestResult, stats: SummaryStats, trades: tuple[HalfTradeModel, ...], params: CombineParams, bars_gated: int | None = None, bars: tuple[Bar, ...] = (), instruments: dict[str, InstrumentSpec] | None = None, replay: Replay | None = None)

One run's full report: the untouched frozen BacktestResult, derived

SummaryStats, the broker's half-turn trade list, and the CombineParams the run used. str(report) renders the combine verdict, balance path, day-by-day trail, and summary stats as plain aligned text — deterministically (pure function of the held frozen data; no wall clock, no unordered iteration). :meth:to_html renders the same report as a self-contained interactive tearsheet under the same determinism contract, :meth:to_timestamped_html names that file for you, and :meth:show opens it in the default browser.

Source code in src/topstep_backtest/harness.py

def __init__(
    self,
    *,
    result: BacktestResult,
    stats: SummaryStats,
    trades: tuple[HalfTradeModel, ...],
    params: CombineParams,
    bars_gated: int | None = None,
    bars: tuple[Bar, ...] = (),
    instruments: dict[str, InstrumentSpec] | None = None,
    replay: Replay | None = None,
) -> None:
    self.result = result
    self.stats = stats
    self.trades = trades
    self.params = params
    self.bars_gated = bars_gated
    self.bars = bars
    self.instruments = {} if instruments is None else dict(instruments)
    self.replay = replay

result instance-attribute

result: BacktestResult = result

stats instance-attribute

stats: SummaryStats = stats

trades instance-attribute

trades: tuple[HalfTradeModel, ...] = trades

params instance-attribute

params: CombineParams = params

bars_gated instance-attribute

bars_gated: int | None = bars_gated

Warmup-gated bars before the strategy's first decision (None when the strategy is not a SymbolStrategy).

bars instance-attribute

bars: tuple[Bar, ...] = bars

The OHLCV tape the run consumed (empty on hand-built reports) — what the HTML tearsheet's candlestick panes draw.

instruments instance-attribute

instruments: dict[str, InstrumentSpec] = {} if instruments is None else dict(instruments)

Specs keyed by contract id, for the tearsheet's price-axis formatting.

replay instance-attribute

replay: Replay | None = replay

The bar-by-bar recording (Backtest(..., record=True)); None otherwise. With one present the HTML tearsheet grows the replay scrubber.

to_html

to_html(path: str | PathLike[str], *, replay: ReplaySpec = 'auto', confidence: MonteCarloConfidence | None = None, crosscheck: CrossCheck | None = None) -> Path

Write the interactive HTML tearsheet to path and return it.

The document is SELF-CONTAINED — charts, styling and data are inlined, so it opens offline and can be archived next to a run. Rendering is a pure function of the held frozen data (byte-identical across calls); the named file is the only disk write.

replay controls the bar-by-bar scrubber when this report carries a recording (Backtest(..., record=True)): "auto" embeds every frame up to the documented limit and falls back to a loudly-labelled window (breach-centred, else the tail) beyond it; "full" forces every frame regardless of size; (start, end) embeds exactly that frame range; "off" omits the scrubber. Without a recording the knob is inert and the page says how to record one.

confidence and crosscheck add the Monte-Carlo cards (estimate with CI, block sensitivity, year strata, bootstrap-vs-windows) — pass the results of :func:~topstep_backtest.metrics.confidence.mc_confidence and :func:~topstep_backtest.metrics.confidence.crosscheck. They are caller-computed on purpose: writing a file must never trigger thousands of simulations as a side effect.

Source code in src/topstep_backtest/harness.py

def to_html(
    self,
    path: str | os.PathLike[str],
    *,
    replay: ReplaySpec = "auto",
    confidence: MonteCarloConfidence | None = None,
    crosscheck: CrossCheck | None = None,
) -> Path:
    """Write the interactive HTML tearsheet to ``path`` and return it.

    The document is SELF-CONTAINED — charts, styling and data are inlined,
    so it opens offline and can be archived next to a run. Rendering is a
    pure function of the held frozen data (byte-identical across calls);
    the named file is the only disk write.

    ``replay`` controls the bar-by-bar scrubber when this report carries a
    recording (``Backtest(..., record=True)``): ``"auto"`` embeds every
    frame up to the documented limit and falls back to a loudly-labelled
    window (breach-centred, else the tail) beyond it; ``"full"`` forces
    every frame regardless of size; ``(start, end)`` embeds exactly that
    frame range; ``"off"`` omits the scrubber. Without a recording the
    knob is inert and the page says how to record one.

    ``confidence`` and ``crosscheck`` add the Monte-Carlo cards (estimate
    with CI, block sensitivity, year strata, bootstrap-vs-windows) — pass
    the results of
    :func:`~topstep_backtest.metrics.confidence.mc_confidence` and
    :func:`~topstep_backtest.metrics.confidence.crosscheck`. They are
    caller-computed on purpose: writing a file must never trigger
    thousands of simulations as a side effect.
    """
    from .tearsheet import render_html  # deferred: keeps import graphs slim

    target = Path(path)
    target.write_text(
        render_html(self, replay=replay, confidence=confidence, crosscheck=crosscheck),
        encoding="utf-8",
        newline="\n",
    )
    return target

to_timestamped_html

to_timestamped_html(directory: str | PathLike[str] = '.', *, prefix: str = 'tearsheet', replay: ReplaySpec = 'auto', confidence: MonteCarloConfidence | None = None, crosscheck: CrossCheck | None = None) -> Path

Write the tearsheet to <directory>/<prefix>-<UTC stamp>.html.

The wall clock is read for the FILENAME ONLY — the document is still the pure function of frozen run data that :meth:to_html renders, so determinism holds where it is checked (the bytes). Missing directories are created. A name already taken — two runs inside the same second — gains a -2, -3, ... suffix instead of overwriting the sheet that is already there. replay, confidence and crosscheck are :meth:to_html's knobs, forwarded.

Source code in src/topstep_backtest/harness.py

def to_timestamped_html(
    self,
    directory: str | os.PathLike[str] = ".",
    *,
    prefix: str = "tearsheet",
    replay: ReplaySpec = "auto",
    confidence: MonteCarloConfidence | None = None,
    crosscheck: CrossCheck | None = None,
) -> Path:
    """Write the tearsheet to ``<directory>/<prefix>-<UTC stamp>.html``.

    The wall clock is read for the FILENAME ONLY — the document is still
    the pure function of frozen run data that :meth:`to_html` renders, so
    determinism holds where it is checked (the bytes). Missing directories
    are created. A name already taken — two runs inside the same second —
    gains a ``-2``, ``-3``, ... suffix instead of overwriting the sheet
    that is already there. ``replay``, ``confidence`` and ``crosscheck``
    are :meth:`to_html`'s knobs, forwarded.
    """
    target_dir = Path(directory)
    target_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    target = target_dir / f"{prefix}-{stamp}.html"
    bump = 2
    while target.exists():
        target = target_dir / f"{prefix}-{stamp}-{bump}.html"
        bump += 1
    return self.to_html(target, replay=replay, confidence=confidence, crosscheck=crosscheck)

show

show(*, replay: ReplaySpec = 'auto', confidence: MonteCarloConfidence | None = None, crosscheck: CrossCheck | None = None) -> Path

Open the tearsheet in the default browser; return the file written.

This is the one API that writes without being handed a path: the document goes to a NEW topstep-tearsheet-*.html temp file (never overwriting anything), which is left in place so the tab survives — delete it, or use :meth:to_html, when you want control of the path. replay, confidence and crosscheck are :meth:to_html's knobs, forwarded.

Source code in src/topstep_backtest/harness.py

def show(
    self,
    *,
    replay: ReplaySpec = "auto",
    confidence: MonteCarloConfidence | None = None,
    crosscheck: CrossCheck | None = None,
) -> Path:
    """Open the tearsheet in the default browser; return the file written.

    This is the one API that writes without being handed a path: the
    document goes to a NEW ``topstep-tearsheet-*.html`` temp file (never
    overwriting anything), which is left in place so the tab survives —
    delete it, or use :meth:`to_html`, when you want control of the path.
    ``replay``, ``confidence`` and ``crosscheck`` are :meth:`to_html`'s
    knobs, forwarded.
    """
    from .tearsheet import render_html  # deferred: keeps import graphs slim

    fd, name = tempfile.mkstemp(prefix="topstep-tearsheet-", suffix=".html")
    with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
        handle.write(
            render_html(self, replay=replay, confidence=confidence, crosscheck=crosscheck)
        )
    target = Path(name)
    webbrowser.open(target.as_uri())
    return target

replay_json

replay_json(path: str | PathLike[str]) -> Path

Dump the raw :class:~topstep_backtest.replay.Replay as JSON.

The debugging tap under the scrubber's floorboards: every frame, event, and snapshot exactly as recorded, uninterpreted by any UI — for diffing two runs, or verifying what was captured before trusting a rendering of it.

Raises:

Type	Description
ValueError	if this report carries no recording (run with Backtest(..., record=True)).

Source code in src/topstep_backtest/harness.py

def replay_json(self, path: str | os.PathLike[str]) -> Path:
    """Dump the raw :class:`~topstep_backtest.replay.Replay` as JSON.

    The debugging tap under the scrubber's floorboards: every frame,
    event, and snapshot exactly as recorded, uninterpreted by any UI —
    for diffing two runs, or verifying what was captured before trusting
    a rendering of it.

    Raises:
        ValueError: if this report carries no recording (run with
            ``Backtest(..., record=True)``).
    """
    replay = self.replay
    if replay is None:
        raise ValueError("this report has no recording — run with Backtest(..., record=True)")
    target = Path(path)
    target.write_bytes(msgspec.json.encode(replay))
    return target

SummaryStats

Bases: Struct

Combine-centric summary metrics for one finished backtest run.

Empty-run conventions: with zero closing half-turns, win_rate, expectancy and profit_factor are None (undefined, not 0); profit_factor is also None when there are no losing closes (the ratio would be infinite). max_drawdown over an empty equity curve reduces to the terminal mark alone: max(0, starting - ending), which is 0 for a run with no bars (the balance never moved).

verdict instance-attribute

verdict: Verdict

Did the strategy pass the evaluation: PASSED, FAILED, or IN_PROGRESS. Mirrors BacktestResult.verdict so a serialized SummaryStats carries the outcome it describes.

THREE states, not two. IN_PROGRESS means the tape ran out before the combine resolved — the strategy neither hit the profit target nor breached — and it is the OUTCOME OF MOST RUNS. It is not a bad result; it is an unfinished one. Never collapse this to a boolean by testing verdict != PASSED: that reports every unfinished run as a failure. Use :attr:passed / :attr:failed (both False while in progress), or branch on all three.

closed_trades instance-attribute

closed_trades: int

Closing half-turns: trade records with profit_and_loss set and not voided. NOT round trips — a flip's single half-turn closes one position and opens the next.

win_rate instance-attribute

win_rate: Decimal | None

Fraction of closing half-turns with gross profit_and_loss > 0 (fees are charged per half-turn separately, so this is a GROSS stat). None when there are no closing half-turns.

expectancy instance-attribute

expectancy: Decimal | None

Mean gross profit_and_loss per closing half-turn. None when there are no closing half-turns; the aggregate NET counterpart is net_pnl / closed_trades.

profit_factor instance-attribute

profit_factor: Decimal | None

Sum of gross winning closes / |sum of gross losing closes|. None when undefined: no closing half-turns, or no losing closes.

max_drawdown instance-attribute

max_drawdown: Decimal

Largest peak-to-trough decline of the per-bar CLOSE equity curve plus one terminal mark at ending_balance (the final session roll's flatten costs land after the last curve point), with the peak seeded at the starting balance (always >= 0, and never below -net_pnl). Close-basis only: intrabar excursions are not in BacktestResult.equity_curve.

equity_peak instance-attribute

equity_peak: Decimal

Highest equity mark the run reached, close-basis, seeded at the starting balance (so it never reports below it).

Not decoration in a prop account: the peak is what a trailing MLL floor is ANCHORED to, so this is the number that set the floor you then had to stay above. Close-basis to match max_drawdown — the two are the opposite ends of one curve, and equity_peak - max_drawdown is the trough that produced it.

The real Topstep MLL ratchets on END-OF-DAY closed balances, so the floor actually in force followed the EOD peak, which is at or below this one. Where that distinction matters, read drawdown.eod_trailing, which is measured against that basis.

final_balance instance-attribute

final_balance: Decimal

Ending realized balance (BacktestResult.ending_balance).

net_pnl instance-attribute

net_pnl: Decimal

ending_balance - starting_balance, net of ALL fees and commissions (the engine's session roll flattens at end of run, so nothing is open).

distance_to_floor instance-attribute

distance_to_floor: Decimal

ending_balance - floor: dollars of room above the trailing MLL floor at end of run.

consistency_headroom instance-attribute

consistency_headroom: Decimal

consistency_pct x total_profit - best_day — dollar slack in the consistency rule (docs/topstep-rules.md §4). Negative means the best day is currently too large: the effective target inflates until best_day <= consistency_pct x total_profit holds.

days_traded instance-attribute

days_traded: int

Closed trading days with trade activity (BacktestResult.days_traded).

exposure instance-attribute

exposure: Decimal | None

Fraction of the run's bars during which ANY position was open — a FRACTION in [0, 1] like win_rate, NOT a percentage. None for a run with no bars.

This is the figure that tells you how to read every other figure here. Two strategies with identical drawdowns, one at 0.05 exposure and one at 0.95, are not the same risk: the first got that result while off the tape nineteen bars in twenty, and the second has been holding through everything and merely has not met its bad day yet.

A bar counts as exposed when a round trip was open at any point strictly inside it, with the boundary resolved FORWARD — a position opened exactly at a bar's close belongs to the next bar, and an excursion whose open and close carry the same stamp therefore contributes nothing. Multi-instrument runs count each timestamp once: any open contract makes that slice exposed, so this is time-with-risk-on and not a sum of per-symbol exposures. The denominator is every bar the run saw, including bars outside tradable hours.

start_ts_ns instance-attribute

start_ts_ns: int | None

end_ts_ns instance-attribute

end_ts_ns: int | None

First and last bar-CLOSE stamp of the run (Bar.ts_init), in epoch nanoseconds; None for a run with no bars. Provenance — a summary carrying no window cannot honestly be compared against another one — and the two ends of :attr:duration_ns.

provisional instance-attribute

provisional: bool

closed_trades < PROVISIONAL_TRADE_FLOOR: this run is too thin for any distributional metric on it to mean anything.

When True, win_rate, expectancy, profit_factor, payoff_ratio, sortino, calmar and every daily percentile are still COMPUTED and still arithmetically correct — they are simply estimates with a standard error large enough to swamp the effect being measured. The renderer says so out loud. Treat them as provisional, not as findings.

avg_win instance-attribute

avg_win: Decimal | None

Mean GROSS P&L of winning closes. None with no winners.

avg_loss instance-attribute

avg_loss: Decimal | None

Mean GROSS loss of losing closes, as a POSITIVE magnitude (so payoff_ratio is a plain ratio). None with no losers.

payoff_ratio instance-attribute

payoff_ratio: Decimal | None

avg_win / avg_loss — the size asymmetry that win_rate alone cannot show. Read the two together, never either alone: a 70% win rate at a 0.3 payoff ratio is a negative-expectancy strategy waiting for its sequence. None when either side has no closes.

expectancy_r instance-attribute

expectancy_r: Decimal | None

Expectancy expressed in R-multiples, where R is defined as the average losing close — NOT as per-trade initial risk.

This is the approximation available from what the framework records today. True R-multiples need the stop distance at entry attached to each round trip; SymbolStrategy.buy/sell accept stop_loss_ticks but do not retain it, and a strategy that exits on signal has no defined R at all. Until that lands, read this as "expectancy in units of a typical loss" — useful for comparing two strategies in this framework, NOT comparable to an R-multiple quoted anywhere else. None when there are no losing closes.

longest_losing_streak instance-attribute

longest_losing_streak: int

Longest unbroken run of losing closes. A scratch close (exactly 0) is not a loss and BREAKS the streak.

breakeven_cost_per_half_turn instance-attribute

breakeven_cost_per_half_turn: Decimal | None

net_pnl / trade_count: the ADDITIONAL cost per half-turn, on top of the fees already charged, that would drive this run to exactly zero.

Positive is the slack you have; negative means the run is already underwater and the figure is how much per half-turn you would have to SAVE to break even. Per half-turn, not per round trip, because that is how fees are actually charged (see the module docstring). None with no half-turns.

sortino instance-attribute

sortino: Decimal | None

Mean daily P&L / downside deviation of daily P&L, target 0.

Daily-dollar basis and NOT annualized. Sortino rather than Sharpe because prop rules punish the downside path specifically and are wholly indifferent to upside variance. Downside deviation divides by the count of ALL closed days (the standard convention), not just losing ones. None with no closed days or no downside deviation (nothing to be punished for).

calmar instance-attribute

calmar: Decimal | None

total_profit / max_drawdown over the run.

Combine-horizon basis and NOT annualized, which is a deliberate deviation from the conventional annualized-return form: annualizing a twenty-day sample produces a number with no defensible meaning. Read it as "profit earned per dollar of worst decline." None when max_drawdown is zero.

daily instance-attribute

daily: DailyStats

Per-day P&L distribution — see :class:DailyStats.

drawdown instance-attribute

drawdown: DrawdownStats

Drawdown under all three prop-firm conventions — see :class:DrawdownStats.

round_trips instance-attribute

round_trips: RoundTripStats

Flat-to-flat trade statistics, NET basis, with true R-multiples where the entry carried a bracket stop — see :class:RoundTripStats. Note the basis flip: these are net, the half-turn figures above are gross.

duration_ns property

duration_ns: int | None

end_ts_ns - start_ts_ns: the run's wall-clock span, in nanoseconds. None for a run with no bars.

CALENDAR time, including every night, weekend and holiday the market was shut — it says how far apart the run's ends were, not how much trading it contains. days_traded is the figure to judge a run's length by, and the two diverge sharply on any sparse feed.

passed property

passed: bool

The combine was actually cleared. False for IN_PROGRESS — see :attr:verdict, and do not read not passed as "failed".

failed property

failed: bool

The combine was actually blown (an MLL/DLL breach, or a rule violation the kernel treats as terminal). False for IN_PROGRESS: missing the profit target is not a failure, and a run that simply ended is neither passed nor failed.

AccountSize

Bases: StrEnum

The three Trading Combine account sizes Topstep offers.

S50K class-attribute instance-attribute

S50K = '50K'

S100K class-attribute instance-attribute

S100K = '100K'

S150K class-attribute instance-attribute

S150K = '150K'

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)

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)