data.synthetic

Deterministic seeded bars, for tests and for examples that must run without shipping a data file.

synthetic

Deterministic synthetic OHLCV bars for examples and tests.

A seeded random.Random walk in INTEGER TICKS — prices are reconstructed via from_ticks so every value is exactly on the instrument's grid and no float ever touches a price. Bars are stamped in ET (one per unit x unit_number step), weekends are skipped, and the same seed always reproduces the identical bar tuple.

hours selects the span stamped for each trading day:

• "rth" (default) — regular trading hours, starting 09:30 ET on the trading day itself. Covers the NY session only.
• "globex" — the full 23-hour electronic session, starting 18:00 ET on the PREVIOUS calendar day (Sunday 18:00 for a Monday) and running to 17:00 ET. This is the mode that produces Asia and London bars, and therefore the only one against which session-scoped behaviour means anything.

Exchange holidays are NOT skipped — this package ships no holiday calendar (see core/time.py). A synthetic tape spanning a market holiday will emit a session that would not exist in real data.

days counts WEEKDAYS emitted: the generator scans forward from start_day and skips Sat/Sun until days sessions exist. The weekday test is applied to the TRADING day, so a Monday tape correctly opens on the preceding Sunday evening under "globex".

bars_per_day is capped so the last one-minute bar closes at 17:00 ET — 450 from the 09:30 RTH open, 1380 from the 18:00 Globex open — and no synthetic bar can ever sit inside the 17:00-18:00 ET maintenance halt.

Hours module-attribute

Hours = Literal['rth', 'globex']

synthetic_bars

synthetic_bars(*, contract_id: str, spec: InstrumentSpec, start_day: date, days: int, seed: int, start_price: Decimal, bars_per_day: int | None = None, unit: AggregateBarUnit = MINUTE, unit_number: int = 1, drift_ticks_per_day: int = 0, vol_ticks: int = 8, hours: Hours = 'rth') -> tuple[Bar, ...]

Generate days trading sessions of consistent, on-grid OHLCV bars.

start_price must lie on spec.tick_size (raises OffGridError otherwise) and be comfortably above zero for the chosen vol_ticks. drift_ticks_per_day is distributed across the day's bars in exact integer ticks; vol_ticks bounds the per-bar random move.

hours="rth" (default) stamps from 09:30 ET on the trading day and caps bars_per_day at 450; hours="globex" stamps the full electronic session from 18:00 ET on the PREVIOUS calendar day and caps it at 1380. Either way the cap is the point at which a one-minute bar would close past 17:00 ET and land inside the maintenance halt. bars_per_day defaults to a full session for the chosen mode (390 RTH, 1380 Globex).

Only "globex" emits Asia and London bars — an "rth" tape is NY-only and cannot exercise anything session-scoped.

Source code in src/topstep_backtest/data/synthetic.py

def synthetic_bars(
    *,
    contract_id: str,
    spec: InstrumentSpec,
    start_day: date,
    days: int,
    seed: int,
    start_price: Decimal,
    bars_per_day: int | None = None,
    unit: AggregateBarUnit = AggregateBarUnit.MINUTE,
    unit_number: int = 1,
    drift_ticks_per_day: int = 0,
    vol_ticks: int = 8,
    hours: Hours = "rth",
) -> tuple[Bar, ...]:
    """Generate ``days`` trading sessions of consistent, on-grid OHLCV bars.

    ``start_price`` must lie on ``spec.tick_size`` (raises ``OffGridError``
    otherwise) and be comfortably above zero for the chosen ``vol_ticks``.
    ``drift_ticks_per_day`` is distributed across the day's bars in exact
    integer ticks; ``vol_ticks`` bounds the per-bar random move.

    ``hours="rth"`` (default) stamps from 09:30 ET on the trading day and caps
    ``bars_per_day`` at 450; ``hours="globex"`` stamps the full electronic
    session from 18:00 ET on the PREVIOUS calendar day and caps it at 1380.
    Either way the cap is the point at which a one-minute bar would close past
    17:00 ET and land inside the maintenance halt. ``bars_per_day`` defaults to
    a full session for the chosen mode (390 RTH, 1380 Globex).

    Only ``"globex"`` emits Asia and London bars — an ``"rth"`` tape is NY-only
    and cannot exercise anything session-scoped.
    """
    if days < 0:
        raise ValueError(f"days must be >= 0, got {days}")
    globex = hours == "globex"
    if bars_per_day is None:
        bars_per_day = _DEFAULT_BARS_PER_DAY_GLOBEX if globex else _DEFAULT_BARS_PER_DAY
    max_bars = _MAX_BARS_PER_DAY_GLOBEX if globex else _MAX_BARS_PER_DAY
    open_label = "18:00" if globex else "09:30"
    if bars_per_day < 1:
        raise ValueError(f"bars_per_day must be >= 1, got {bars_per_day}")
    if bars_per_day > max_bars:
        raise ValueError(
            f"bars_per_day must be <= {max_bars}, got {bars_per_day}: bars "
            f"are stamped from the {open_label} ET session open and {open_label} + "
            f"{max_bars} minutes = 17:00 ET, the start of the 17:00-18:00 ET "
            "maintenance halt — extra bars would land inside the halt and fail "
            "validation"
        )
    if vol_ticks < 0:
        raise ValueError(f"vol_ticks must be >= 0, got {vol_ticks}")

    step = step_ns(unit, unit_number)
    tick = spec.tick_size
    start_ticks = to_ticks(start_price, tick)  # raises OffGridError off-grid

    wick_max = max(1, vol_ticks // 2)
    # Keep the walk high enough that low = min(o, c) - wick stays positive.
    floor_ticks = vol_ticks + wick_max + 1
    if start_ticks < floor_ticks:
        raise ValueError(
            f"start_price {start_price} ({start_ticks} ticks) is too low for "
            f"vol_ticks={vol_ticks}; need at least {floor_ticks} ticks"
        )

    rng = Random(seed)
    bar_type = BarType(contract_id=contract_id, unit=unit, unit_number=unit_number)
    # Exact integer split of the daily drift across bars (works for negatives).
    base_drift, extra_drift_bars = divmod(drift_ticks_per_day, bars_per_day)

    bars: list[Bar] = []
    current = start_ticks
    day = start_day
    emitted = 0
    while emitted < days:
        if day.weekday() >= 5:
            day += timedelta(days=1)
            continue
        # Globex reuses the canonical 18:00-previous-day open from core/time.py
        # rather than restating it, so the generator cannot drift from the
        # boundary the engine and the kernel actually enforce.
        session_open_ns = TOPSTEP_SESSION.open_ns(day) if globex else et_time_of(day, _RTH_OPEN)
        for i in range(bars_per_day):
            open_ticks = current
            delta = base_drift + (1 if i < extra_drift_bars else 0)
            delta += rng.randint(-vol_ticks, vol_ticks)
            close_ticks = max(open_ticks + delta, floor_ticks)
            high_ticks = max(open_ticks, close_ticks) + rng.randint(0, wick_max)
            low_ticks = min(open_ticks, close_ticks) - rng.randint(0, wick_max)
            ts_event = session_open_ns + i * step
            bars.append(
                Bar(
                    bar_type=bar_type,
                    ts_event=ts_event,
                    ts_init=ts_event + step,
                    open=from_ticks(open_ticks, tick),
                    high=from_ticks(high_ticks, tick),
                    low=from_ticks(low_ticks, tick),
                    close=from_ticks(close_ticks, tick),
                    volume=rng.randint(_MIN_VOLUME, _MAX_VOLUME),
                )
            )
            current = close_ticks
        emitted += 1
        day += timedelta(days=1)
    return tuple(bars)