"""Data quality audit over market_snapshots + option_chain_snapshots. Pure functions: each takes a ``sqlite3.Connection`` and a UTC time window, returns a frozen dataclass. No side effects, no MCP, no writes. The CLI layer (``cli.audit``) is responsible for I/O and formatting. Thresholds are module-level constants by design: the audit and the runtime live in different contexts and must not share operational parameters. To tune a threshold, edit this file. """ from __future__ import annotations import math import sqlite3 import statistics from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from decimal import Decimal __all__ = [ "ChainAuditReport", "GapRecord", "MarketAuditReport", "audit_market_snapshots", "audit_option_chain", ] # Tick cadence + gap tolerance. Cron is */15; +5 min tolerance covers # late-arriving MCP responses. _TICK_INTERVAL_MIN: int = 15 _GAP_THRESHOLD_MIN: int = 20 # fetch_ok=0 streak threshold: 1-2 are transient MCP failures, 3+ is a # pattern worth flagging. _FETCH_OK_STREAK_THRESHOLD: int = 3 # A numeric column with >10% NULL in the window is too unreliable for # backtesting that metric. _NULL_RATE_FLAG: Decimal = Decimal("0.10") # Columns to NULL-audit on market_snapshots. fetch_ok / fetch_errors_json # are excluded (they are status fields, not metrics). _MARKET_NUMERIC_COLUMNS: tuple[str, ...] = ( "spot", "dvol", "realized_vol_30d", "iv_minus_rv", "funding_perp_annualized", "funding_cross_annualized", "dealer_net_gamma", "gamma_flip_level", "oi_delta_pct_4h", "macro_days_to_event", ) @dataclass(frozen=True) class GapRecord: """One gap between consecutive market_snapshots ticks.""" prev_timestamp: datetime next_timestamp: datetime gap_minutes: int @dataclass(frozen=True) class MarketAuditReport: asset: str since: datetime until: datetime expected_ticks: int actual_ticks: int coverage_pct: Decimal gaps: tuple[GapRecord, ...] = field(default_factory=tuple) fetch_ok_zero_count: int = 0 max_fetch_ok_zero_streak: int = 0 null_rate_by_column: dict[str, Decimal] = field(default_factory=dict) @dataclass(frozen=True) class ChainAuditReport: asset: str since: datetime until: datetime expected_snapshots: int actual_snapshots: int coverage_pct: Decimal quotes_per_snap_median: int = 0 quotes_per_snap_p10: int = 0 quotes_per_snap_p90: int = 0 bid_gt_ask_count: int = 0 iv_null_count: int = 0 iv_null_pct: Decimal = Decimal("0") depth_zero_pct: Decimal = Decimal("0") def _expected_ticks(since: datetime, until: datetime) -> int: """Number of `*/15` ticks in ``[since, until)`` aligned to wall clock. A tick is any UTC instant where ``minute % 15 == 0``. The first tick at or after ``since`` is computed by rounding ``since`` up; every subsequent tick is +15 minutes. The window is half-open on the right. """ if until <= since: return 0 # Round `since` up to the next */15 boundary. minute = since.minute remainder = minute % _TICK_INTERVAL_MIN if remainder == 0 and since.second == 0 and since.microsecond == 0: first_tick = since else: bump = _TICK_INTERVAL_MIN - remainder first_tick = (since + timedelta(minutes=bump)).replace( second=0, microsecond=0 ) if first_tick >= until: return 0 # Count ticks in [first_tick, until): the largest k with # first_tick + k*15min < until is ceil(span/15) - 1, so the # count is ceil(span_minutes / 15). floor() under-counts at # aligned multiples and would mis-count non-aligned spans. span_seconds = (until - first_tick).total_seconds() return math.ceil(span_seconds / (_TICK_INTERVAL_MIN * 60)) def _max_zero_streak(flags: list[int]) -> int: """Longest run of consecutive zeros.""" longest = 0 current = 0 for v in flags: if v == 0: current += 1 longest = max(longest, current) else: current = 0 return longest def _detect_gaps(timestamps: list[datetime]) -> tuple[GapRecord, ...]: """Return gaps where consecutive timestamps differ by > threshold.""" out: list[GapRecord] = [] for prev, nxt in zip(timestamps, timestamps[1:], strict=False): delta_min = int((nxt - prev).total_seconds() // 60) if delta_min > _GAP_THRESHOLD_MIN: out.append( GapRecord( prev_timestamp=prev, next_timestamp=nxt, gap_minutes=delta_min, ) ) return tuple(out) def _fetch_market_rows( conn: sqlite3.Connection, *, asset: str, since: datetime, until: datetime, ) -> list[sqlite3.Row]: cols = ", ".join(("timestamp", "fetch_ok", *_MARKET_NUMERIC_COLUMNS)) rows = conn.execute( f"SELECT {cols} FROM market_snapshots " "WHERE asset = ? AND timestamp >= ? AND timestamp < ? " "ORDER BY timestamp ASC", (asset, since.isoformat(), until.isoformat()), ).fetchall() return list(rows) def _compute_null_rate( rows: list[sqlite3.Row], columns: tuple[str, ...] ) -> dict[str, Decimal]: if not rows: return {c: Decimal("0") for c in columns} total = Decimal(len(rows)) out: dict[str, Decimal] = {} for c in columns: nulls = sum(1 for r in rows if r[c] is None) out[c] = (Decimal(nulls) / total).quantize(Decimal("0.0001")) return out def audit_market_snapshots( conn: sqlite3.Connection, *, asset: str, since: datetime, until: datetime, ) -> MarketAuditReport: """Compute the market_snapshots audit report for an asset in [since, until).""" rows = _fetch_market_rows(conn, asset=asset, since=since, until=until) timestamps = [datetime.fromisoformat(r["timestamp"]) for r in rows] expected = _expected_ticks(since, until) actual = len(rows) coverage = ( (Decimal(actual) / Decimal(expected) * Decimal("100")).quantize( Decimal("0.01") ) if expected > 0 else Decimal("0") ) gaps = _detect_gaps(timestamps) fetch_ok_flags = [int(r["fetch_ok"]) for r in rows] fetch_ok_zero_count = sum(1 for v in fetch_ok_flags if v == 0) max_streak = _max_zero_streak(fetch_ok_flags) null_rates = _compute_null_rate(rows, _MARKET_NUMERIC_COLUMNS) return MarketAuditReport( asset=asset, since=since, until=until, expected_ticks=expected, actual_ticks=actual, coverage_pct=coverage, gaps=gaps, fetch_ok_zero_count=fetch_ok_zero_count, max_fetch_ok_zero_streak=max_streak, null_rate_by_column=null_rates, )