Bar-by-bar run recording: what the strategy saw, decided, and was told.
Backtest(..., record=True) hooks a :class:Recorder into the engine loop.
After the run, Report.replay holds a :class:Replay — one frame per feed
bar, carrying everything the tearsheet's replay scrubber steps through:
- Decisions at the
ctx seam: every orders.* / positions.* call
the strategy made, with its full intent (side/size/type/prices/brackets) and
its outcome — the accepted order id, or the gateway rejection code and
message. Captured by wrapping the context's API objects in recording
proxies, so raw ctx.orders calls are seen exactly like the sugar's.
- Events the broker emitted: order-state changes, fills (with gross P&L
and costs), and position snapshots — the same
UserEvent stream the
strategy's on_* hooks received, in delivery order.
- Indicator values per bar, named by the attribute the strategy stored
them under (
self.fast = self.use(Sma(20)) records as fast), with
multi-output indicators recorded line by line and Cross fires recorded
as events. Warmup-gated bars are recorded as gated, not skipped: "could not
act yet" is a state the scrubber must show, not a hole in the timeline.
- State after each bar settles: net position, working orders, and realized
balance — stored only on change, folded from the same SDK events the
strategy's own views fold (
strategy.tracker), never from broker
internals.
- Running statistics as sparse :class:
StatsSnapshot\ s (one per closing
trade, day close, and breach — not per bar). Each snapshot is a full
SummaryStats computed over the run's PREFIX by the same
metrics.stats helpers compute_summary uses, so a snapshot cannot
drift from the final report's definitions. The one deliberate exception is
exposure, which is recomputed only at day closes (it is the single
prefix stat with no O(1) incremental form) and carried forward between
them; the terminal snapshot recomputes it in full, and a golden test pins
the terminal snapshot equal to compute_summary field for field.
- Session enforcement between bars: the 16:10 ET flatten, the 17:00 ET
session close, MLL/DLL trips — stamped as events on the frame they precede
(events after the final bar land on an epilogue frame
frame_count).
Recording changes nothing. The recorder only observes: results are
byte-identical with record=True and record=False (pinned by a golden
test), and Strategy.note() — the strategy's own breadcrumb channel into
the recording — is a pure sink that cannot influence the run.
Cost: recording holds per-bar indicator values and per-event structs in
memory, so a multi-year record=True run costs real RAM and the day-close
exposure recomputation costs one pass over the stamps per closed day. Record
the runs you intend to step through, not every sweep trial.
Bases: Struct
One named indicator line, one value per bar of the strategy's contract.
values[i] belongs to that contract's i-th bar (for a
single-contract feed that is simply frame i); None means the
indicator was not ready. Values are the exact decimal strings the
strategy compared (str of the adapter's Decimal), not re-rounded.
instance-attribute
values: tuple[str | None, ...]
Bases: Struct
A Cross's fires: (contract_bar_index, +1 up / -1 down) pairs.
instance-attribute
fires: tuple[tuple[int, int], ...]
Bases: Struct
One strategy decision at the ctx seam, with its outcome.
error_code/error are set when the call was rejected (the sugar
routes that to on_reject; the raw path raises) — a rejected intent is
exactly the "intent diverged from execution" moment the report's REJECTED
line counts, kept here with the full context that count discards.
instance-attribute
Which call: orders.place/buy/sell/modify/cancel/
cancel_all or positions.close/partial_close/close_all.
class-attribute
instance-attribute
contract_id: str | None = None
class-attribute
instance-attribute
class-attribute
instance-attribute
class-attribute
instance-attribute
class-attribute
instance-attribute
limit_price: str | None = None
class-attribute
instance-attribute
stop_price: str | None = None
class-attribute
instance-attribute
trail_price: str | None = None
class-attribute
instance-attribute
stop_loss_ticks: int | None = None
class-attribute
instance-attribute
take_profit_ticks: int | None = None
class-attribute
instance-attribute
custom_tag: str | None = None
class-attribute
instance-attribute
order_id: int | None = None
The accepted order id (place/buy/sell), or the targeted id
(modify/cancel).
class-attribute
instance-attribute
error_code: int | None = None
class-attribute
instance-attribute
Bases: Struct
An order-state change the broker emitted (the OrderModel stream).
instance-attribute
instance-attribute
instance-attribute
instance-attribute
class-attribute
instance-attribute
limit_price: str | None = None
class-attribute
instance-attribute
stop_price: str | None = None
class-attribute
instance-attribute
fill_price: str | None = None
class-attribute
instance-attribute
custom_tag: str | None = None
Bases: Struct
An execution (the HalfTradeModel stream). pnl is the GROSS
realized P&L of the closing portion (None for a pure opening fill);
costs is this half-turn's fees + commissions.
instance-attribute
instance-attribute
instance-attribute
Bases: Struct
A position snapshot the broker emitted (the PositionModel stream).
instance-attribute
instance-attribute
Bases: Struct
A Strategy.note() breadcrumb, tagged with the hook that emitted it.
A note from inside on_reject carries the ENCLOSING hook (the sugar
calls on_reject from within the order call), which is where the
decision it explains was made.
Bases: Struct
Session enforcement between bars: flatten, session close, MLL/DLL.
frame is the frame the event PRECEDES (its fills land on that frame's
event stream); events after the final bar carry frame == frame_count.
instance-attribute
eod_flatten | session_close | mll_breach | dll_lock.
class-attribute
instance-attribute
Bases: Struct
Net position for one contract AFTER a frame settled; recorded on change.
average_price is the display average from the broker's last position
snapshot (None when flat — and, live-parity caveat, a full close emits
no snapshot, so flatness is detected from the fill fold exactly as
SymbolStrategy.position detects it).
instance-attribute
instance-attribute
average_price: str | None
Bases: Struct
One working order in a :class:WorkingSnap.
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
Bases: Struct
The COMPLETE working-order set after a frame settled; on change only.
instance-attribute
Bases: Struct
Realized balance after a frame settled; recorded on change.
instance-attribute
Bases: Struct
Running statistics as of a frame's settle — a full SummaryStats
over the run's prefix, computed by the same metrics.stats code the
final report uses.
Prefix semantics: day-level figures (daily distribution, EOD drawdown,
sortino, total-profit-derived fields) cover CLOSED days only, exactly as
the final report's do; exposure is as of the last day close (None
before the first). The terminal snapshot (frame == frame_count)
additionally applies the ending-balance mark, matching
compute_summary exactly.
Bases: Struct
One recorded run: everything the tearsheet's replay scrubber shows.
frame_count is the number of feed bars processed (== the engine's
equity-curve length); event frame fields of frame_count mark the
epilogue (the final session roll). All lists are in capture order and the
seq fields give the single total order across event kinds.
instance-attribute
instance-attribute
strategy_contract_id: str | None
instance-attribute
bars_gated: tuple[int, ...]
Frames swallowed by the warmup/ready gate (no on_bar fired).
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
instance-attribute
The terminal snapshot's stats — built by the recorder's accumulators
and pinned equal to compute_summary by a golden test. Report.stats
remains the canonical figure; this exists so the equality is testable.
OrderApi proxy: records every call's intent + outcome, then defers.
Structurally satisfies protocols.OrderApi (same signatures as
SimOrderApi), so a strategy cannot tell it is being observed.
Read-only calls (search_open/get/wait_for_fill) pass through
unrecorded — they are questions, not decisions.
Source code in src/topstep_backtest/replay.py
| def __init__(self, inner: OrderApi, recorder: Recorder) -> None:
self._inner = inner
self._rec = recorder
|
async
place(account_id: int, contract_id: str, *, side: OrderSide | int, type: OrderType | int, size: int, limit_price: float | Decimal | None = None, stop_price: float | Decimal | None = None, trail_price: float | Decimal | None = None, custom_tag: str | None = None, stop_loss_bracket: PlaceOrderBracket | dict[str, int] | None = None, take_profit_bracket: PlaceOrderBracket | dict[str, int] | None = None, stop_loss_ticks: int | None = None, take_profit_ticks: int | None = None) -> int
Source code in src/topstep_backtest/replay.py
| async def place(
self,
account_id: int,
contract_id: str,
*,
side: OrderSide | int,
type: OrderType | int,
size: int,
limit_price: float | Decimal | None = None,
stop_price: float | Decimal | None = None,
trail_price: float | Decimal | None = None,
custom_tag: str | None = None,
stop_loss_bracket: PlaceOrderBracket | dict[str, int] | None = None,
take_profit_bracket: PlaceOrderBracket | dict[str, int] | None = None,
stop_loss_ticks: int | None = None,
take_profit_ticks: int | None = None,
) -> int:
return await self._record_place(
"orders.place",
account_id,
contract_id,
side=side,
type=type,
size=size,
limit_price=limit_price,
stop_price=stop_price,
trail_price=trail_price,
custom_tag=custom_tag,
stop_loss_bracket=stop_loss_bracket,
take_profit_bracket=take_profit_bracket,
stop_loss_ticks=stop_loss_ticks,
take_profit_ticks=take_profit_ticks,
)
|
async
buy(account_id: int, contract_id: str, size: int, *, type: OrderType | int = MARKET, limit_price: float | Decimal | None = None, stop_price: float | Decimal | None = None, trail_price: float | Decimal | None = None, custom_tag: str | None = None, stop_loss_bracket: PlaceOrderBracket | dict[str, int] | None = None, take_profit_bracket: PlaceOrderBracket | dict[str, int] | None = None, stop_loss_ticks: int | None = None, take_profit_ticks: int | None = None) -> int
Source code in src/topstep_backtest/replay.py
| async def buy(
self,
account_id: int,
contract_id: str,
size: int,
*,
type: OrderType | int = OrderType.MARKET,
limit_price: float | Decimal | None = None,
stop_price: float | Decimal | None = None,
trail_price: float | Decimal | None = None,
custom_tag: str | None = None,
stop_loss_bracket: PlaceOrderBracket | dict[str, int] | None = None,
take_profit_bracket: PlaceOrderBracket | dict[str, int] | None = None,
stop_loss_ticks: int | None = None,
take_profit_ticks: int | None = None,
) -> int:
return await self._record_place(
"orders.buy",
account_id,
contract_id,
side=OrderSide.BUY,
type=type,
size=size,
limit_price=limit_price,
stop_price=stop_price,
trail_price=trail_price,
custom_tag=custom_tag,
stop_loss_bracket=stop_loss_bracket,
take_profit_bracket=take_profit_bracket,
stop_loss_ticks=stop_loss_ticks,
take_profit_ticks=take_profit_ticks,
)
|
async
sell(account_id: int, contract_id: str, size: int, *, type: OrderType | int = MARKET, limit_price: float | Decimal | None = None, stop_price: float | Decimal | None = None, trail_price: float | Decimal | None = None, custom_tag: str | None = None, stop_loss_bracket: PlaceOrderBracket | dict[str, int] | None = None, take_profit_bracket: PlaceOrderBracket | dict[str, int] | None = None, stop_loss_ticks: int | None = None, take_profit_ticks: int | None = None) -> int
Source code in src/topstep_backtest/replay.py
| async def sell(
self,
account_id: int,
contract_id: str,
size: int,
*,
type: OrderType | int = OrderType.MARKET,
limit_price: float | Decimal | None = None,
stop_price: float | Decimal | None = None,
trail_price: float | Decimal | None = None,
custom_tag: str | None = None,
stop_loss_bracket: PlaceOrderBracket | dict[str, int] | None = None,
take_profit_bracket: PlaceOrderBracket | dict[str, int] | None = None,
stop_loss_ticks: int | None = None,
take_profit_ticks: int | None = None,
) -> int:
return await self._record_place(
"orders.sell",
account_id,
contract_id,
side=OrderSide.SELL,
type=type,
size=size,
limit_price=limit_price,
stop_price=stop_price,
trail_price=trail_price,
custom_tag=custom_tag,
stop_loss_bracket=stop_loss_bracket,
take_profit_bracket=take_profit_bracket,
stop_loss_ticks=stop_loss_ticks,
take_profit_ticks=take_profit_ticks,
)
|
async
modify(account_id: int, order_id: int, *, size: int | None = None, limit_price: float | Decimal | None = None, stop_price: float | Decimal | None = None, trail_price: float | Decimal | None = None) -> None
Source code in src/topstep_backtest/replay.py
| async def modify(
self,
account_id: int,
order_id: int,
*,
size: int | None = None,
limit_price: float | Decimal | None = None,
stop_price: float | Decimal | None = None,
trail_price: float | Decimal | None = None,
) -> None:
intent = OrderIntent(
seq=self._rec.next_seq(),
frame=self._rec.current_frame,
api="orders.modify",
size=size,
limit_price=_odec(limit_price),
stop_price=_odec(stop_price),
trail_price=_odec(trail_price),
order_id=order_id,
)
try:
await self._inner.modify(
account_id,
order_id,
size=size,
limit_price=limit_price,
stop_price=stop_price,
trail_price=trail_price,
)
except APIError as error:
self._rec.record_intent(
msgspec.structs.replace(intent, error_code=error.error_code, error=str(error))
)
raise
self._rec.record_intent(intent)
|
async
cancel(account_id: int, order_id: int) -> None
Source code in src/topstep_backtest/replay.py
| async def cancel(self, account_id: int, order_id: int) -> None:
intent = OrderIntent(
seq=self._rec.next_seq(),
frame=self._rec.current_frame,
api="orders.cancel",
order_id=order_id,
)
try:
await self._inner.cancel(account_id, order_id)
except APIError as error:
self._rec.record_intent(
msgspec.structs.replace(intent, error_code=error.error_code, error=str(error))
)
raise
self._rec.record_intent(intent)
|
async
cancel_all(account_id: int) -> list[int]
Source code in src/topstep_backtest/replay.py
| async def cancel_all(self, account_id: int) -> list[int]:
intent = OrderIntent(
seq=self._rec.next_seq(),
frame=self._rec.current_frame,
api="orders.cancel_all",
)
try:
cancelled = await self._inner.cancel_all(account_id)
except APIError as error:
self._rec.record_intent(
msgspec.structs.replace(intent, error_code=error.error_code, error=str(error))
)
raise
self._rec.record_intent(msgspec.structs.replace(intent, size=len(cancelled)))
return cancelled
|
async
search_open(account_id: int) -> list[OrderModel]
Source code in src/topstep_backtest/replay.py
| async def search_open(self, account_id: int) -> list[OrderModel]:
return await self._inner.search_open(account_id)
|
async
get(account_id: int, order_id: int) -> OrderModel | None
Source code in src/topstep_backtest/replay.py
| async def get(self, account_id: int, order_id: int) -> OrderModel | None:
return await self._inner.get(account_id, order_id)
|
async
wait_for_fill(account_id: int, order_id: int, *, timeout: float = 30.0, poll_interval: float = 1.0) -> OrderModel
Source code in src/topstep_backtest/replay.py
| async def wait_for_fill(
self,
account_id: int,
order_id: int,
*,
timeout: float = 30.0, # noqa: ASYNC109 - mirrors the SDK signature
poll_interval: float = 1.0,
) -> OrderModel:
return await self._inner.wait_for_fill(
account_id, order_id, timeout=timeout, poll_interval=poll_interval
)
|
PositionApi proxy: records close intents + outcomes, then defers.
Source code in src/topstep_backtest/replay.py
| def __init__(self, inner: PositionApi, recorder: Recorder) -> None:
self._inner = inner
self._rec = recorder
|
async
search_open(account_id: int) -> list[PositionModel]
Source code in src/topstep_backtest/replay.py
| async def search_open(self, account_id: int) -> list[PositionModel]:
return await self._inner.search_open(account_id)
|
async
close(account_id: int, contract_id: str) -> None
Source code in src/topstep_backtest/replay.py
| async def close(self, account_id: int, contract_id: str) -> None:
intent = OrderIntent(
seq=self._rec.next_seq(),
frame=self._rec.current_frame,
api="positions.close",
contract_id=contract_id,
)
try:
await self._inner.close(account_id, contract_id)
except APIError as error:
self._rec.record_intent(
msgspec.structs.replace(intent, error_code=error.error_code, error=str(error))
)
raise
self._rec.record_intent(intent)
|
async
partial_close(account_id: int, contract_id: str, size: int) -> None
Source code in src/topstep_backtest/replay.py
| async def partial_close(self, account_id: int, contract_id: str, size: int) -> None:
intent = OrderIntent(
seq=self._rec.next_seq(),
frame=self._rec.current_frame,
api="positions.partial_close",
contract_id=contract_id,
size=size,
)
try:
await self._inner.partial_close(account_id, contract_id, size)
except APIError as error:
self._rec.record_intent(
msgspec.structs.replace(intent, error_code=error.error_code, error=str(error))
)
raise
self._rec.record_intent(intent)
|
async
close_all(account_id: int) -> list[str]
Source code in src/topstep_backtest/replay.py
| async def close_all(self, account_id: int) -> list[str]:
intent = OrderIntent(
seq=self._rec.next_seq(),
frame=self._rec.current_frame,
api="positions.close_all",
)
try:
closed = await self._inner.close_all(account_id)
except APIError as error:
self._rec.record_intent(
msgspec.structs.replace(intent, error_code=error.error_code, error=str(error))
)
raise
self._rec.record_intent(msgspec.structs.replace(intent, size=len(closed)))
return closed
|
Engine-side run recorder. Observes; never influences.
Wiring (all done by BacktestEngine when constructed with one):
attach() before the strategy binds, wrap_context() around the
strategy's ctx, set_hook() around each strategy hook delivery,
on_user_event() per drained broker event, on_session() /
on_day_closed() at session boundaries, commit_bar() once per feed
bar after it fully settles, and finalize() with the frozen result.
replay is available after finalize().
Source code in src/topstep_backtest/replay.py
| def __init__(self) -> None:
self._broker: SimBroker | None = None
self._strategy: Strategy | None = None
self._frame = 0
self._seq = 0
self._hook: str | None = None
self._contract_bars = 0 # bars of the strategy's contract seen
self._last_gated = 0
self._gated: list[int] = []
# Indicator capture (SymbolStrategy only).
self._named: list[_NamedIndicator] = []
self._crosses: list[_NamedCross] = []
self._series: dict[str, list[str | None]] = {}
self._strategy_contract: str | None = None
# Event streams.
self._intents: list[OrderIntent] = []
self._order_events: list[OrderEvent] = []
self._fill_events: list[FillEvent] = []
self._position_events: list[PositionEvent] = []
self._notes: list[NoteEvent] = []
self._session_events: list[SessionEvent] = []
# State tracks (on change).
self._positions: list[PositionSnap] = []
self._working: list[WorkingSnap] = []
self._balances: list[BalanceSnap] = []
self._net: dict[str, NetPosition] = {}
self._avg: dict[str, str | None] = {}
self._last_pos: dict[str, tuple[int, str | None]] = {}
self._orders = OrderTracker()
self._last_working: tuple[WorkingRow, ...] = ()
self._last_balance: Decimal | None = None
# Running-stats accumulators (mirrors of compute_summary's folds).
# _closes exists for CADENCE only (did a close land this frame?);
# snapshot figures re-read broker.trades, the authoritative ledger.
self._closes: list[Decimal] = []
self._bar_stamps: list[int] = []
self._min_mark: Decimal | None = None
self._peak: Decimal | None = None # seeded at starting balance on attach
self._max_dd = _ZERO
self._intraday_peak: Decimal | None = None
self._intraday_dd: Decimal | None = None
self._min_headroom: Decimal | None = None
self._min_headroom_ts: int | None = None
self._exposure_last: Decimal | None = None
self._snapshots: list[StatsSnapshot] = []
self._closes_at_snapshot = 0
self._breach_recorded = False
self._day_locked_seen = False
self._replay: Replay | None = None
|
property
The frame in progress (== frames committed so far). Events between
bars land on the frame they precede; after the last commit this is
frame_count, the epilogue.
property
The sealed recording. Raises until finalize() has run.
Source code in src/topstep_backtest/replay.py
| def next_seq(self) -> int:
self._seq += 1
return self._seq
|
Bind to the run's broker and strategy; introspect the indicators.
Source code in src/topstep_backtest/replay.py
| def attach(self, *, broker: SimBroker, strategy: Strategy) -> None:
"""Bind to the run's broker and strategy; introspect the indicators."""
self._broker = broker
self._strategy = strategy
self._peak = broker.kernel.params.starting_balance
strategy._note_sink = self._on_note # pyright: ignore[reportPrivateUsage]
if isinstance(strategy, SymbolStrategy):
self._strategy_contract = strategy.contract_id
registered = strategy.registered_indicators
names = _attr_names(strategy, registered)
for indicator in registered:
attr = names[id(indicator)]
if isinstance(indicator, Cross):
self._crosses.append(_NamedCross(indicator, attr))
else:
named = _NamedIndicator(indicator, attr)
self._named.append(named)
for column in named.columns:
self._series[column] = []
|
The same context with orders/positions behind recording
proxies; history/clock are read-only and pass through bare.
Source code in src/topstep_backtest/replay.py
| def wrap_context(self, ctx: StrategyContext) -> StrategyContext:
"""The same context with ``orders``/``positions`` behind recording
proxies; ``history``/``clock`` are read-only and pass through bare."""
return _dc_replace(
ctx,
orders=RecordingOrderApi(ctx.orders, self),
positions=RecordingPositionApi(ctx.positions, self),
)
|
set_hook(hook: str | None) -> None
Name the strategy hook now executing, for note attribution.
Source code in src/topstep_backtest/replay.py
| def set_hook(self, hook: str | None) -> None:
"""Name the strategy hook now executing, for note attribution."""
self._hook = hook
|
Source code in src/topstep_backtest/replay.py
| def record_intent(self, intent: OrderIntent) -> None:
self._intents.append(intent)
|
Fold one drained broker event into the streams and state trackers.
Source code in src/topstep_backtest/replay.py
| def on_user_event(self, event: UserEvent) -> None:
"""Fold one drained broker event into the streams and state trackers."""
frame = self._frame
if isinstance(event, OrderModel):
self._order_events.append(
OrderEvent(
seq=self.next_seq(),
frame=frame,
order_id=event.id,
contract_id=event.contract_id,
status=event.status.name,
type=event.type.name,
side=_side_name(event.side),
size=event.size,
filled=event.fill_volume,
limit_price=_odec(event.limit_price),
stop_price=_odec(event.stop_price),
fill_price=_odec(event.filled_price),
custom_tag=event.custom_tag,
)
)
self._orders.apply(event)
elif isinstance(event, HalfTradeModel):
costs = event.fees + (event.commissions if event.commissions is not None else _ZERO)
self._fill_events.append(
FillEvent(
seq=self.next_seq(),
frame=frame,
order_id=event.order_id,
contract_id=event.contract_id,
side=_side_name(event.side),
size=event.size,
price=str(event.price),
pnl=_odec(event.profit_and_loss),
costs=str(costs),
voided=event.voided,
)
)
if event.profit_and_loss is not None and not event.voided:
self._closes.append(event.profit_and_loss)
net = self._net.setdefault(event.contract_id, NetPosition())
net.apply_fill(event)
if net.net == 0:
self._avg[event.contract_id] = None
else:
if event.type == PositionType.LONG:
net_size = event.size
elif event.type == PositionType.SHORT:
net_size = -event.size
else: # closed/undefined snapshot: flat (mirrors NetPosition)
net_size = 0
self._position_events.append(
PositionEvent(
seq=self.next_seq(),
frame=frame,
contract_id=event.contract_id,
net=net_size,
average_price=str(event.average_price),
)
)
self._net.setdefault(event.contract_id, NetPosition()).apply_snapshot(event)
self._avg[event.contract_id] = str(event.average_price) if net_size else None
|
on_session(kind: str, ts_ns: int, detail: str = '') -> None
Stamp a session-boundary enforcement action on the current frame.
Source code in src/topstep_backtest/replay.py
| def on_session(self, kind: str, ts_ns: int, detail: str = "") -> None:
"""Stamp a session-boundary enforcement action on the current frame."""
self._session_events.append(
SessionEvent(
seq=self.next_seq(), frame=self._frame, kind=kind, ts_ns=ts_ns, detail=detail
)
)
|
on_day_closed(ts_ns: int) -> None
A trading day closed (kernel ratchet applied): snapshot, with the
one per-day-only recomputation — exposure — refreshed here.
Source code in src/topstep_backtest/replay.py
| def on_day_closed(self, ts_ns: int) -> None:
"""A trading day closed (kernel ratchet applied): snapshot, with the
one per-day-only recomputation — exposure — refreshed here."""
broker = self._require_broker()
self._exposure_last = exposure_fraction(self._bar_stamps, broker.round_trips)
self._append_snapshot()
|
commit_bar(bar: Bar, equity: Decimal) -> None
Seal one frame: the bar fully settled (fills applied, hooks run).
Source code in src/topstep_backtest/replay.py
| def commit_bar(self, bar: Bar, equity: Decimal) -> None:
"""Seal one frame: the bar fully settled (fills applied, hooks run)."""
broker = self._require_broker()
strategy = self._strategy
# Indicator values — only bars of the strategy's contract advance them.
if (
self._strategy_contract is not None
and bar.bar_type.contract_id == self._strategy_contract
):
contract_index = self._contract_bars
self._contract_bars += 1
for named in self._named:
for column, value in zip(named.columns, named.read(), strict=True):
self._series[column].append(value)
for cross in self._crosses:
cross.capture(contract_index)
# Warmup gate: a swallowed bar is a state, not a hole.
if isinstance(strategy, SymbolStrategy) and strategy.bars_gated > self._last_gated:
self._last_gated = strategy.bars_gated
self._gated.append(self._frame)
# State tracks, on change only.
balance = broker.balance
if balance != self._last_balance:
self._last_balance = balance
self._balances.append(BalanceSnap(frame=self._frame, balance=str(balance)))
for cid in sorted(self._net):
net = self._net[cid].net
avg = self._avg.get(cid) if net != 0 else None
state = (net, avg)
if self._last_pos.get(cid) != state:
self._last_pos[cid] = state
self._positions.append(
PositionSnap(frame=self._frame, contract_id=cid, net=net, average_price=avg)
)
working = tuple(
WorkingRow(
order_id=order.id,
contract_id=order.contract_id,
side=_side_name(order.side),
type=order.type.name,
size=order.size,
limit_price=_odec(order.limit_price),
stop_price=_odec(order.stop_price),
custom_tag=order.custom_tag,
)
for order in self._orders.working
)
if working != self._last_working:
self._last_working = working
self._working.append(WorkingSnap(frame=self._frame, orders=working))
# Rule-state transitions (kernel breach is stored; DLL is a flag flip).
kernel = broker.kernel
breach = kernel.breach
breach_new = breach is not None and not self._breach_recorded
if breach is not None and breach_new:
self._breach_recorded = True
self._session_events.append(
SessionEvent(
seq=self.next_seq(),
frame=self._frame,
kind="mll_breach",
ts_ns=breach.ts_ns,
detail=f"equity {breach.equity} <= floor {breach.limit}",
)
)
if kernel.day_locked and not self._day_locked_seen:
self._day_locked_seen = True
self._session_events.append(
SessionEvent(
seq=self.next_seq(),
frame=self._frame,
kind="dll_lock",
ts_ns=bar.ts_init,
detail="daily loss limit reached: trading locked until 18:00 ET",
)
)
elif not kernel.day_locked:
self._day_locked_seen = False
# Equity folds — the exact loop bodies compute_summary runs at the end.
self._bar_stamps.append(bar.ts_init)
self._min_mark = equity if self._min_mark is None else min(self._min_mark, equity)
peak = self._require_peak()
if equity > peak:
self._peak = equity
elif peak - equity > self._max_dd:
self._max_dd = peak - equity
sample = broker.last_bar_equity
if sample is not None:
ts_ns, high, low, floor = sample
intraday_peak = self._intraday_peak
intraday_dd = self._intraday_dd
if intraday_peak is None or intraday_dd is None:
intraday_peak = kernel.params.starting_balance
intraday_dd = _ZERO
intraday_peak = max(intraday_peak, high)
self._intraday_peak = intraday_peak
self._intraday_dd = max(intraday_dd, intraday_peak - low)
headroom = low - floor
if self._min_headroom is None or headroom < self._min_headroom:
self._min_headroom = headroom
self._min_headroom_ts = ts_ns
# Snapshot cadence: a closing trade landed this frame, or the run just
# died — never per bar, so the recording stays sparse. Stamped BEFORE
# the frame advances so "last snapshot at/before the cursor" reflects
# a trade on the very bar it settled.
if len(self._closes) > self._closes_at_snapshot or breach_new:
self._append_snapshot()
self._frame += 1
|
Seal the recording against the frozen result: apply the terminal
equity mark exactly as compute_summary does, recompute exposure in
full, and append the terminal snapshot.
Source code in src/topstep_backtest/replay.py
| def finalize(self, result: BacktestResult) -> None:
"""Seal the recording against the frozen result: apply the terminal
equity mark exactly as ``compute_summary`` does, recompute exposure in
full, and append the terminal snapshot."""
# A breach can land AFTER the last commit: the session-close check
# fails a closed balance at/below the floor with no further bar to
# commit, so the transition is caught here and stamped on the epilogue.
breach = result.breach
if breach is not None and not self._breach_recorded:
self._breach_recorded = True
self._session_events.append(
SessionEvent(
seq=self.next_seq(),
frame=self._frame,
kind="mll_breach",
ts_ns=breach.ts_ns,
detail=f"equity {breach.equity} <= floor {breach.limit}",
)
)
self._exposure_last = exposure_fraction(self._bar_stamps, result.round_trips)
final = self._build_stats(
verdict=result.verdict,
balance=result.ending_balance,
floor=result.floor,
total_profit=result.total_profit,
best_day=result.best_day,
days_traded=result.days_traded,
day_records=result.day_records,
trips=result.round_trips,
terminal_mark=result.ending_balance,
)
self._snapshots.append(StatsSnapshot(frame=self._frame, stats=final))
self._replay = Replay(
frame_count=self._frame,
strategy_contract_id=self._strategy_contract,
bars_gated=tuple(self._gated),
indicator_series=tuple(
IndicatorSeries(name=name, values=tuple(values))
for name, values in self._series.items()
),
cross_series=tuple(
CrossSeries(name=cross.name, fires=tuple(cross.fires)) for cross in self._crosses
),
intents=tuple(self._intents),
order_events=tuple(self._order_events),
fill_events=tuple(self._fill_events),
position_events=tuple(self._position_events),
notes=tuple(self._notes),
session_events=tuple(self._session_events),
positions=tuple(self._positions),
working=tuple(self._working),
balances=tuple(self._balances),
snapshots=tuple(self._snapshots),
final_stats=final,
)
# The sink dies with the run: a reused strategy instance must not keep
# writing into a sealed recording.
strategy = self._strategy
if strategy is not None:
strategy._note_sink = None # pyright: ignore[reportPrivateUsage]
|