execution.sim_broker

The simulated broker: full order lifecycle, FIFO lots, OCO brackets, and flat-to-flat RoundTrip records.

sim_broker

SimBroker: deterministic execution simulation behind the Broker protocol.

Implements the SAME structural protocol AsyncTopstepClient satisfies, so a strategy wired to a SimBroker runs unchanged live. The broker owns order lifecycle, OCO/bracket linkage, trailing-stop recomputation, position netting, P&L (exact Decimal), and wires every equity change through the Combine rule kernel — including intrabar breach detection along the deterministic price path, and forced liquidation with its own (worse) slippage plus Topstep's $10/contract automatic-liquidation fee.

Accounting: positions are FIFO lots with an exact Decimal COST BASIS. Every stored lot price is on the tick grid; unrealized P&L is division-free (core.money.position_unrealized), so scale-ins can never produce an off-grid average or rounding dust. Per-half-turn profit_and_loss uses FIFO lot attribution (documented assumption pending gateway calibration — docs/DESIGN.md §15).

Key correctness properties

• No look-ahead: orders participate in a bar only if accepted at/before its open (the fill model's accepted_ts firewall); trailing stops likewise only ratchet from bars the order actually lived through; bracket children created on an intrabar fill first participate the NEXT bar.
• Deterministic intrabar resolution: fills and rule-breach liquidations are ordered along ONE shared pessimistic price path by TRIGGER level (never by slippage-adjusted fill prices), and equity is re-checked AT each fill point after it applies (slippage + fees can themselves breach).
• Every rejection is the SDK's APIError with the gateway's error code.

Documented Tier-0 divergences from live (see docs/DESIGN.md §13.8): market orders — including positions.close/partial_close — fill at the NEXT bar's open (not instantly); stored bars are not live tape-built bars; wait_for_fill raises UnsupportedInBacktestError; STOP_LIMIT and JOIN_BID/JOIN_ASK are rejected (they need quote data, Tier-1+).

UserEvent module-attribute

UserEvent = OrderModel | HalfTradeModel | PositionModel

Events queued for strategy dispatch (mirrors the SDK user hub payloads).

SimBrokerConfig

SimBrokerConfig(*, forced_liq_slippage_ticks: int = 2, liquidation_fee_per_contract: Decimal = Decimal('10'), max_trail_ticks: int = 1000, history_depth: int = 20000)

Tunables that are broker-level (not fill-model-level).

Source code in src/topstep_backtest/execution/sim_broker.py

def __init__(
    self,
    *,
    forced_liq_slippage_ticks: int = 2,
    liquidation_fee_per_contract: Decimal = Decimal("10"),
    max_trail_ticks: int = 1000,
    history_depth: int = 20_000,
) -> None:
    self.forced_liq_slippage_ticks = forced_liq_slippage_ticks
    self.liquidation_fee_per_contract = liquidation_fee_per_contract
    self.max_trail_ticks = max_trail_ticks
    self.history_depth = history_depth

forced_liq_slippage_ticks instance-attribute

forced_liq_slippage_ticks = forced_liq_slippage_ticks

liquidation_fee_per_contract instance-attribute

liquidation_fee_per_contract = liquidation_fee_per_contract

max_trail_ticks instance-attribute

max_trail_ticks = max_trail_ticks

history_depth instance-attribute

history_depth = history_depth

RoundTrip

Bases: Struct

One flat-to-flat excursion in a single contract — a "trade" in the colloquial sense, as opposed to the half-turns the gateway reports.

Boundaries are unambiguous and need no new convention: a round trip opens when a contract goes from flat to positioned and closes when it returns to flat. A FLIP (long straight to short in one fill) closes one round trip and opens the next at that instant; the flip's single half-turn and its costs are attributed to the round trip being CLOSED, so no fill is counted twice.

This is a REPORTING grouping over the broker's own FIFO half-turns. It does not re-derive P&L: gross_pnl is the sum of the profit_and_loss figures the broker already computed, so the open question of how the live gateway pairs fills (docs/topstep-rules.md §9) cannot change these numbers without changing the half-turns first.

contract_id instance-attribute

contract_id: str

direction instance-attribute

direction: int

+1 if the excursion was net long, -1 if net short.

opened_ts_ns instance-attribute

opened_ts_ns: int

closed_ts_ns instance-attribute

closed_ts_ns: int

max_qty instance-attribute

max_qty: int

Largest position size held during the excursion (scale-ins included).

half_turns instance-attribute

half_turns: int

gross_pnl instance-attribute

gross_pnl: Decimal

Sum of the closing half-turns' realized P&L. GROSS — no fees.

costs instance-attribute

costs: Decimal

Fees + commissions charged on every half-turn in the excursion.

net_pnl instance-attribute

net_pnl: Decimal

gross_pnl - costs. This is what the excursion actually earned.

initial_risk instance-attribute

initial_risk: Decimal | None

Dollars at risk when the position was opened, from the bracket stop distance at entry (stop_loss_ticks x tick value x size, summed over every opening fill).

None when ANY opening fill carried no bracket stop — a signal-exit strategy has no defined risk, and guessing one would manufacture an R-multiple out of nothing. Captured AT ENTRY: moving or trailing the stop afterwards does not change it, which is the conventional meaning of R.

r_multiple instance-attribute

r_multiple: Decimal | None

net_pnl / initial_risk — the excursion's return in units of what it actually risked. NET basis deliberately: R answers "what did I make against what I put up", and the fees were genuinely paid. None whenever initial_risk is.

SimBroker

SimBroker(*, account_id: int, instruments: Mapping[str, InstrumentSpec], fill_model: FillModel, fee_model: FeeModel, kernel: CombineKernel, clock: Clock, ids: IdGenerator | None = None, config: SimBrokerConfig | None = None, session: SessionTimes = TOPSTEP_SESSION)

Deterministic simulated venue satisfying the Broker protocol.

Source code in src/topstep_backtest/execution/sim_broker.py

def __init__(
    self,
    *,
    account_id: int,
    instruments: Mapping[str, InstrumentSpec],
    fill_model: FillModel,
    fee_model: FeeModel,
    kernel: CombineKernel,
    clock: Clock,
    ids: IdGenerator | None = None,
    config: SimBrokerConfig | None = None,
    session: SessionTimes = TOPSTEP_SESSION,
) -> None:
    self._account_id = account_id
    self._instruments = dict(instruments)
    self._fill_model = fill_model
    self._fee_model = fee_model
    self._kernel = kernel
    self._clock = clock
    self._ids = ids or IdGenerator()
    self._config = config or SimBrokerConfig()
    self._session = session

    self._balance: Decimal = kernel.params.starting_balance
    self._orders: dict[int, WorkingOrder] = {}
    self._brackets: dict[int, _PendingBrackets] = {}
    self._oco: dict[int, int] = {}  # order id -> OCO sibling id (both directions)
    self._positions: dict[str, _Position] = {}
    self._last_bar: dict[str, Bar] = {}
    self._history: dict[str, list[Bar]] = {}
    # (ts_init, equity_high, equity_low, mll_floor) per bar processed.
    self._bar_equity: list[tuple[int, Decimal, Decimal, Decimal]] = []
    self._round_trips: list[RoundTrip] = []
    self._rt: dict[str, _RoundTripAcc] = {}  # in-flight, keyed by contract
    self._events: list[UserEvent] = []
    self._trades: list[HalfTradeModel] = []
    # Rejected placements, by gateway error_code. A strategy that silently
    # swallows APIError (the SymbolStrategy sugar routes it to on_reject,
    # whose default is a no-op) otherwise produces a clean-looking report
    # in which nothing was ever executed. Surfaced on BacktestResult.
    self._rejections: dict[int, int] = {}
    self._used_tags: set[str] = set()
    self._halted = False

orders property

orders: SimOrderApi

positions property

positions: SimPositionApi

history property

history: SimHistoryApi

rejections property

rejections: dict[int, int]

Count of rejected placements, keyed by gateway error_code.

round_trips property

round_trips: tuple[RoundTrip, ...]

Completed flat-to-flat excursions, in close order.

Only FINALIZED trips appear: a position still open has no close price and no realized P&L, so reporting it would invent both. The engine flattens at the session roll, so a completed run leaves none in flight.

bar_equity property

bar_equity: tuple[tuple[int, Decimal, Decimal, Decimal], ...]

Per-bar (ts_init, equity_high, equity_low, mll_floor).

Equity is realized + unrealized (the same figure the rule kernel breach-checks), sampled at the four points of the modelled intrabar path. This is what lets analytics reconstruct intraday-trailing drawdown and distance-to-floor over time; the engine copies it onto BacktestResult.bar_equity.

last_bar_equity property

last_bar_equity: tuple[int, Decimal, Decimal, Decimal] | None

The most recent bar_equity sample, or None before any bar.

O(1), unlike bar_equity (which copies the whole capture): per-bar consumers — the replay recorder reads this every frame — must not pay a full-run copy per bar.

account_id property

account_id: int

instruments property

instruments: dict[str, InstrumentSpec]

balance property

balance: Decimal

kernel property

kernel: CombineKernel

trades property

trades: tuple[HalfTradeModel, ...]

dead property

dead: bool

drain_events

drain_events() -> list[UserEvent]

Source code in src/topstep_backtest/execution/sim_broker.py

def drain_events(self) -> list[UserEvent]:
    events, self._events = self._events, []
    return events

equity

equity() -> Decimal

Realized balance + open P&L marked at each contract's last close.

Source code in src/topstep_backtest/execution/sim_broker.py

def equity(self) -> Decimal:
    """Realized balance + open P&L marked at each contract's last close."""
    total = self._balance
    for cid, pos in self._positions.items():
        last = self._last_bar.get(cid)
        if last is not None:
            total += self._unrealized(pos, last.close)
    return total

on_bar

on_bar(bar: Bar) -> None

Phase-1 matching: trailing ratchet, path walk with interleaved

fill/breach resolution, then the end-of-bar equity check.

Source code in src/topstep_backtest/execution/sim_broker.py

def on_bar(self, bar: Bar) -> None:
    """Phase-1 matching: trailing ratchet, path walk with interleaved

    fill/breach resolution, then the end-of-bar equity check.
    """
    cid = bar.bar_type.contract_id
    spec = self._instruments.get(cid)
    if spec is None:
        raise KeyError(f"bar for unknown contract {cid!r}: register it in `instruments`")

    prev = self._last_bar.get(cid)
    self._ratchet_trailing(cid, prev, spec)

    # Equity extremes are sampled with the bar-START position and again
    # with the bar-END position, and the wider pair is kept. A position
    # opened or closed mid-bar makes this an approximation (the sampler
    # cannot know which extreme the position was actually on for), and it
    # errs WIDE — consistent with the path model's adverse bias.
    pre_high, pre_low = self._sample_bar_equity(bar, cid)
    if self._kernel.verdict is not Verdict.FAILED:
        self._walk_bar(bar, spec, prev)
    post_high, post_low = self._sample_bar_equity(bar, cid)
    self._bar_equity.append(
        (
            bar.ts_init,
            max(pre_high, post_high),
            min(pre_low, post_low),
            self._kernel.floor,
        )
    )

    self._last_bar[cid] = bar
    bucket = self._history.setdefault(cid, [])
    bucket.append(bar)
    if len(bucket) > self._config.history_depth:
        del bucket[: len(bucket) - self._config.history_depth]

flatten_all

flatten_all(ts_ns: int, *, reason: str) -> None

Topstep auto-flatten enforcement (16:10 ET / day roll): market-dump

every position with forced slippage AND the automatic-liquidation fee, then cancel every working order. Strategy-initiated exits should happen earlier via positions.close (which does NOT pay this fee).

Source code in src/topstep_backtest/execution/sim_broker.py

def flatten_all(self, ts_ns: int, *, reason: str) -> None:
    """Topstep auto-flatten enforcement (16:10 ET / day roll): market-dump

    every position with forced slippage AND the automatic-liquidation fee,
    then cancel every working order. Strategy-initiated exits should
    happen earlier via ``positions.close`` (which does NOT pay this fee).
    """
    for cid in list(self._positions):
        last = self._last_bar.get(cid)
        if last is None:  # pragma: no cover - position implies a seen bar
            continue
        self._close_position_at(
            cid,
            last.close,
            ts_ns,
            note=reason,
            slippage_ticks=self._config.forced_liq_slippage_ticks,
            liquidation=True,
        )
    self._cancel_all_working(ts_ns)

session_close

session_close(ts_ns: int) -> None

EOD hook: report the closed balance to the rule kernel (MLL ratchet).

Source code in src/topstep_backtest/execution/sim_broker.py

def session_close(self, ts_ns: int) -> None:
    """EOD hook: report the closed balance to the rule kernel (MLL ratchet)."""
    self._kernel.on_session_close(ts_ns, self._balance)

SimOrderApi

SimOrderApi(broker: SimBroker)

Source code in src/topstep_backtest/execution/sim_broker.py

def __init__(self, broker: SimBroker) -> None:
    self._b = broker

place 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/execution/sim_broker.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 self._b._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,
    )

buy 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/execution/sim_broker.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.place(
        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,
    )

sell 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/execution/sim_broker.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.place(
        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,
    )

modify 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/execution/sim_broker.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:
    self._b._modify(
        account_id,
        order_id,
        size=size,
        limit_price=limit_price,
        stop_price=stop_price,
        trail_price=trail_price,
    )

cancel async

cancel(account_id: int, order_id: int) -> None

Source code in src/topstep_backtest/execution/sim_broker.py

async def cancel(self, account_id: int, order_id: int) -> None:
    self._b._cancel(account_id, order_id)

cancel_all async

cancel_all(account_id: int) -> list[int]

Source code in src/topstep_backtest/execution/sim_broker.py

async def cancel_all(self, account_id: int) -> list[int]:
    if account_id != self._b._account_id:
        reject_cancel(1, f"unknown account {account_id}")
    return self._b._cancel_all_working(self._b._clock.now_ns())

search_open async

search_open(account_id: int) -> list[OrderModel]

Source code in src/topstep_backtest/execution/sim_broker.py

async def search_open(self, account_id: int) -> list[OrderModel]:
    if account_id != self._b._account_id:
        raise APIError("AccountNotFound", error_code=1)
    return [
        self._b._order_model(o)
        for o in self._b._orders.values()
        if o.status is OrderStatus.OPEN
    ]

get async

get(account_id: int, order_id: int) -> OrderModel | None

Source code in src/topstep_backtest/execution/sim_broker.py

async def get(self, account_id: int, order_id: int) -> OrderModel | None:
    if account_id != self._b._account_id:
        return None  # the gateway reports OrderNotFound for foreign accounts
    order = self._b._orders.get(order_id)
    return None if order is None else self._b._order_model(order)

wait_for_fill 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/execution/sim_broker.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:
    order = self._b._orders.get(order_id) if account_id == self._b._account_id else None
    if order is not None and order.status in _TERMINAL:
        return self._b._order_model(order)
    raise UnsupportedInBacktestError(
        "wait_for_fill cannot busy-poll under a deterministic TestClock; "
        "handle fills in Strategy.on_order/on_fill callbacks instead (the "
        "parity-safe idiom in both sim and live)."
    )

SimPositionApi

SimPositionApi(broker: SimBroker)

Source code in src/topstep_backtest/execution/sim_broker.py

def __init__(self, broker: SimBroker) -> None:
    self._b = broker

search_open async

search_open(account_id: int) -> list[PositionModel]

Source code in src/topstep_backtest/execution/sim_broker.py

async def search_open(self, account_id: int) -> list[PositionModel]:
    if account_id != self._b._account_id:
        raise APIError("AccountNotFound", error_code=1)
    return [
        self._b._position_model(p, self._b._instruments[cid])
        for cid, p in self._b._positions.items()
    ]

close async

close(account_id: int, contract_id: str) -> None

Source code in src/topstep_backtest/execution/sim_broker.py

async def close(self, account_id: int, contract_id: str) -> None:
    if account_id != self._b._account_id:
        raise APIError("AccountNotFound", error_code=1)
    pos = self._b._positions.get(contract_id)
    if pos is None:
        raise APIError("PositionNotFound", error_code=2)
    self._b._submit_reduce_market(contract_id, pos.qty)

partial_close async

partial_close(account_id: int, contract_id: str, size: int) -> None

Source code in src/topstep_backtest/execution/sim_broker.py

async def partial_close(self, account_id: int, contract_id: str, size: int) -> None:
    if account_id != self._b._account_id:
        raise APIError("AccountNotFound", error_code=1)
    pos = self._b._positions.get(contract_id)
    if pos is None:
        raise APIError("PositionNotFound", error_code=2)
    if size <= 0 or size > pos.qty:
        raise APIError("InvalidCloseSize", error_code=5)
    self._b._submit_reduce_market(contract_id, size)

close_all async

close_all(account_id: int) -> list[str]

Source code in src/topstep_backtest/execution/sim_broker.py

async def close_all(self, account_id: int) -> list[str]:
    if account_id != self._b._account_id:
        raise APIError("AccountNotFound", error_code=1)
    closed: list[str] = []
    for cid, pos in list(self._b._positions.items()):
        self._b._submit_reduce_market(cid, pos.qty)
        closed.append(cid)
    return closed

SimHistoryApi

SimHistoryApi(broker: SimBroker)

Source code in src/topstep_backtest/execution/sim_broker.py

def __init__(self, broker: SimBroker) -> None:
    self._b = broker

retrieve_bars async

retrieve_bars(contract_id: str, *, unit: AggregateBarUnit | int, unit_number: int, start_time: datetime | str, end_time: datetime | str, limit: int = 1000, live: bool = False, include_partial_bar: bool = False) -> list[AggregateBarModel]

Serve ONLY already-seen bars (zero look-ahead), newest-first like the

gateway (bars stamped at open time, matching the SDK model). The requested unit/unit_number must match the feed's native bar spec — Tier-0 does no resampling and refuses to silently serve wrong aggregation. start_time/end_time filter when passed as datetimes (ISO strings accepted for parity but treated as unbounded).

Source code in src/topstep_backtest/execution/sim_broker.py

async def retrieve_bars(
    self,
    contract_id: str,
    *,
    unit: AggregateBarUnit | int,
    unit_number: int,
    start_time: datetime | str,
    end_time: datetime | str,
    limit: int = 1000,
    live: bool = False,
    include_partial_bar: bool = False,
) -> list[AggregateBarModel]:
    """Serve ONLY already-seen bars (zero look-ahead), newest-first like the

    gateway (bars stamped at open time, matching the SDK model). The
    requested ``unit``/``unit_number`` must match the feed's native bar
    spec — Tier-0 does no resampling and refuses to silently serve wrong
    aggregation. ``start_time``/``end_time`` filter when passed as
    datetimes (ISO strings accepted for parity but treated as unbounded).
    """
    if limit > 20_000:
        raise ValueError(f"limit {limit} exceeds the gateway maximum of 20000")
    bars = self._b._history.get(contract_id, [])
    if bars:
        native = bars[-1].bar_type
        if (int(unit), unit_number) != (int(native.unit), native.unit_number):
            raise ValueError(
                f"Tier-0 history serves only the feed's native bar spec "
                f"({native.unit.name} x{native.unit_number}); requested "
                f"{AggregateBarUnit(int(unit)).name} x{unit_number}. "
                "Resampling arrives with the multi-timeframe data layer."
            )
    start_ns = dt_to_ns(start_time) if isinstance(start_time, datetime) else None
    end_ns = dt_to_ns(end_time) if isinstance(end_time, datetime) else None
    out: list[AggregateBarModel] = []
    for bar in reversed(bars):
        if len(out) >= limit:
            break
        if start_ns is not None and bar.ts_event < start_ns:
            break  # history is time-ascending; everything earlier is out of range
        if end_ns is not None and bar.ts_event > end_ns:
            continue
        out.append(
            AggregateBarModel(
                t=ns_to_dt(bar.ts_event),
                o=bar.open,
                h=bar.high,
                l=bar.low,
                c=bar.close,
                v=bar.volume,
            )
        )
    return out