data.validator

What a bad bar stream looks like, and which defects are fatal versus merely worth knowing about.

validator

Bar-stream sanity validation: catch bad data BEFORE it reaches the engine.

Severity encoding (documented contract): severity is derived from the issue code — codes in INFO_CODES (currently only "session_gap") are informational; every other code is an ERROR. ValidationReport.ok is True iff no ERROR-severity issue is present.

Codes emitted

• ohlc_inconsistent ERROR high < max(open, close) or low > min(open, close)
• stamp_inverted ERROR ts_event >= ts_init — the bar's open does not precede its close (the classic off-by-one look-ahead stamp bug)
• negative_volume ERROR volume < 0
• duplicate_ts ERROR same instrument, repeated ts_init
• non_monotonic ERROR ts_init decreases along the sequence
• off_grid ERROR a price off the instrument's tick grid
• in_maintenance_halt ERROR ts_event inside the 17:00-18:00 ET halt
• weekend_bar ERROR bar's trading day falls on Sat/Sun
• session_gap INFO > 3x step missing between consecutive bars of the same instrument within one trading day

Ordering checks (duplicate_ts, non_monotonic) anchor ts_ns to the bar's ts_init; event-time checks (halt/weekend) anchor to ts_event.

There is deliberately no exchange-holiday check: this package ships no holiday calendar (see the note in core/time.py), so a bar landing on a market holiday is indistinguishable from any other weekday bar here.

INFO_CODES module-attribute

INFO_CODES: frozenset[str] = frozenset({'session_gap'})

ValidationIssue

Bases: Struct

One finding; severity is derived from code (see module docstring).

code instance-attribute

code: str

message instance-attribute

message: str

ts_ns class-attribute instance-attribute

ts_ns: int | None = None

ValidationReport

Bases: Struct

All findings for a bar sequence; ok means no ERROR-severity issues.

issues instance-attribute

issues: tuple[ValidationIssue, ...]

ok property

ok: bool

validate_bars

validate_bars(bars: Sequence[Bar], spec: InstrumentSpec) -> ValidationReport

Run every check over bars and return the full report (never raises).

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

def validate_bars(bars: Sequence[Bar], spec: InstrumentSpec) -> ValidationReport:
    """Run every check over ``bars`` and return the full report (never raises)."""
    issues: list[ValidationIssue] = []
    last_ts_init: int | None = None
    last_per_instrument: dict[str, Bar] = {}
    step_cache: dict[BarType, int | None] = {}

    for index, bar in enumerate(bars):
        # --- stamp sanity: open must strictly precede close ----------------
        if bar.ts_event >= bar.ts_init:
            issues.append(
                ValidationIssue(
                    code="stamp_inverted",
                    message=(
                        f"bar {index}: ts_event {bar.ts_event} >= ts_init "
                        f"{bar.ts_init} — the bar's open does not precede its "
                        "close (stamp-inverted bar; acting on it hands the "
                        "strategy one bar of the future)"
                    ),
                    ts_ns=bar.ts_init,
                )
            )

        # --- volume sanity -------------------------------------------------
        if bar.volume < 0:
            issues.append(
                ValidationIssue(
                    code="negative_volume",
                    message=f"bar {index}: volume {bar.volume} is negative",
                    ts_ns=bar.ts_init,
                )
            )

        # --- OHLC internal consistency -----------------------------------
        if bar.high < max(bar.open, bar.close) or bar.low > min(bar.open, bar.close):
            issues.append(
                ValidationIssue(
                    code="ohlc_inconsistent",
                    message=(
                        f"bar {index}: OHLC inconsistent (o={bar.open} h={bar.high} "
                        f"l={bar.low} c={bar.close})"
                    ),
                    ts_ns=bar.ts_init,
                )
            )

        # --- tick grid ----------------------------------------------------
        for field, price in (
            ("open", bar.open),
            ("high", bar.high),
            ("low", bar.low),
            ("close", bar.close),
        ):
            if not is_on_grid(price, spec.tick_size):
                issues.append(
                    ValidationIssue(
                        code="off_grid",
                        message=(
                            f"bar {index}: {field}={price} is not on the {spec.tick_size} tick grid"
                        ),
                        ts_ns=bar.ts_init,
                    )
                )

        # --- ordering (global) and per-instrument duplicates/gaps ---------
        if last_ts_init is not None and bar.ts_init < last_ts_init:
            issues.append(
                ValidationIssue(
                    code="non_monotonic",
                    message=(f"bar {index}: ts_init {bar.ts_init} < previous {last_ts_init}"),
                    ts_ns=bar.ts_init,
                )
            )
        last_ts_init = bar.ts_init

        contract_id = bar.bar_type.contract_id
        prev = last_per_instrument.get(contract_id)
        if prev is not None:
            if bar.ts_init == prev.ts_init:
                issues.append(
                    ValidationIssue(
                        code="duplicate_ts",
                        message=(
                            f"bar {index}: duplicate ts_init {bar.ts_init} for "
                            f"instrument {contract_id}"
                        ),
                        ts_ns=bar.ts_init,
                    )
                )
            else:
                step = _bar_step_ns(bar.bar_type, step_cache)
                same_day = trading_day_of(prev.ts_init) == trading_day_of(bar.ts_init)
                if step is not None and same_day and bar.ts_init - prev.ts_init > 3 * step:
                    missing = (bar.ts_init - prev.ts_init) // step - 1
                    issues.append(
                        ValidationIssue(
                            code="session_gap",
                            message=(
                                f"bar {index}: ~{missing} bars missing before ts_init "
                                f"{bar.ts_init} for {contract_id} (intraday gap > 3x step)"
                            ),
                            ts_ns=bar.ts_init,
                        )
                    )
        last_per_instrument[contract_id] = bar

        # --- calendar / session placement (keyed off ts_event) ------------
        event_et = ns_to_et(bar.ts_event)
        if _HALT_START <= event_et.time() < _HALT_END:
            issues.append(
                ValidationIssue(
                    code="in_maintenance_halt",
                    message=(
                        f"bar {index}: ts_event {event_et:%Y-%m-%d %H:%M:%S %Z} is "
                        "inside the 17:00-18:00 ET maintenance halt"
                    ),
                    ts_ns=bar.ts_event,
                )
            )
        session_day = trading_day_of(bar.ts_event)
        if session_day.weekday() >= 5:
            issues.append(
                ValidationIssue(
                    code="weekend_bar",
                    message=(
                        f"bar {index}: ts_event {event_et:%Y-%m-%d %H:%M:%S %Z} falls "
                        f"in the weekend session gap (trading day {session_day})"
                    ),
                    ts_ns=bar.ts_event,
                )
            )

    return ValidationReport(issues=tuple(issues))