From d6af69f4cb27c92dd2eb6fcbb86758db8d1191d7 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 13 May 2026 09:33:28 +0000 Subject: [PATCH] =?UTF-8?q?feat(analysis):=20audit=5Fmarket=5Fsnapshots=20?= =?UTF-8?q?=E2=80=94=20coverage,=20gap,=20fetch=5Fok,=20NULL=20rate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- src/cerbero_bite/analysis/data_audit.py | 68 ++++++++++++++ tests/unit/test_data_audit.py | 119 ++++++++++++++++++++++++ 2 files changed, 187 insertions(+) diff --git a/src/cerbero_bite/analysis/data_audit.py b/src/cerbero_bite/analysis/data_audit.py index 561cea6..e8fb951 100644 --- a/src/cerbero_bite/analysis/data_audit.py +++ b/src/cerbero_bite/analysis/data_audit.py @@ -153,3 +153,71 @@ def _detect_gaps(timestamps: list[datetime]) -> tuple[GapRecord, ...]: ) ) 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, + ) diff --git a/tests/unit/test_data_audit.py b/tests/unit/test_data_audit.py index a114f06..ec3072e 100644 --- a/tests/unit/test_data_audit.py +++ b/tests/unit/test_data_audit.py @@ -2,13 +2,21 @@ from __future__ import annotations +import sqlite3 from datetime import UTC, datetime +from decimal import Decimal +from pathlib import Path +from cerbero_bite.analysis.data_audit import ( + MarketAuditReport, + audit_market_snapshots, +) from cerbero_bite.analysis.data_audit import ( # noqa: PLC2701 _detect_gaps, _expected_ticks, _max_zero_streak, ) +from cerbero_bite.state import connect, run_migrations, transaction def test_expected_ticks_basic() -> None: @@ -94,3 +102,114 @@ def test_max_zero_streak_single_zero_block() -> None: def test_max_zero_streak_all_zeros() -> None: assert _max_zero_streak([0, 0, 0]) == 3 + + +def _make_conn(tmp_path: Path) -> sqlite3.Connection: + conn = connect(tmp_path / "state.sqlite") + run_migrations(conn) + return conn + + +def _seed_market( + conn: sqlite3.Connection, + *, + asset: str, + ts: datetime, + spot: Decimal | None = Decimal("3000"), + dvol: Decimal | None = Decimal("55"), + dealer_net_gamma: Decimal | None = Decimal("-50000000"), + fetch_ok: int = 1, +) -> None: + conn.execute( + "INSERT INTO market_snapshots(timestamp, asset, spot, dvol, " + "dealer_net_gamma, fetch_ok) VALUES (?,?,?,?,?,?)", + ( + ts.isoformat(), + asset, + str(spot) if spot is not None else None, + str(dvol) if dvol is not None else None, + str(dealer_net_gamma) if dealer_net_gamma is not None else None, + fetch_ok, + ), + ) + + +def test_audit_market_full_coverage(tmp_path: Path) -> None: + conn = _make_conn(tmp_path) + since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC) + until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC) + try: + with transaction(conn): + for minute in (0, 15, 30, 45): + _seed_market( + conn, + asset="ETH", + ts=since.replace(minute=minute), + ) + report = audit_market_snapshots( + conn, asset="ETH", since=since, until=until + ) + finally: + conn.close() + assert report.asset == "ETH" + assert report.expected_ticks == 4 + assert report.actual_ticks == 4 + assert report.coverage_pct == Decimal("100") + assert report.gaps == () + assert report.fetch_ok_zero_count == 0 + assert report.max_fetch_ok_zero_streak == 0 + + +def test_audit_market_detects_gap_and_streak(tmp_path: Path) -> None: + conn = _make_conn(tmp_path) + since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC) + until = datetime(2026, 5, 12, 13, 30, tzinfo=UTC) + try: + with transaction(conn): + # 12:00 OK, 12:15 OK, gap (12:30, 12:45 missing), 13:00 fail, + # 13:15 fail, 13:30 outside window. + _seed_market(conn, asset="ETH", ts=since.replace(minute=0)) + _seed_market(conn, asset="ETH", ts=since.replace(minute=15)) + _seed_market( + conn, asset="ETH", ts=since.replace(hour=13, minute=0), + fetch_ok=0, + ) + _seed_market( + conn, asset="ETH", ts=since.replace(hour=13, minute=15), + fetch_ok=0, + ) + report = audit_market_snapshots( + conn, asset="ETH", since=since, until=until + ) + finally: + conn.close() + assert report.expected_ticks == 6 + assert report.actual_ticks == 4 + # 12:15 → 13:00 is a 45-min gap → flagged + assert len(report.gaps) == 1 + assert report.gaps[0].gap_minutes == 45 + assert report.fetch_ok_zero_count == 2 + assert report.max_fetch_ok_zero_streak == 2 + + +def test_audit_market_null_rate_per_column(tmp_path: Path) -> None: + conn = _make_conn(tmp_path) + since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC) + until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC) + try: + with transaction(conn): + # 4 ticks, 1 with dealer_net_gamma=NULL → 25% null + for minute in (0, 15, 30, 45): + _seed_market( + conn, + asset="ETH", + ts=since.replace(minute=minute), + dealer_net_gamma=None if minute == 30 else Decimal("-50000000"), + ) + report = audit_market_snapshots( + conn, asset="ETH", since=since, until=until + ) + finally: + conn.close() + assert report.null_rate_by_column["dealer_net_gamma"] == Decimal("0.25") + assert report.null_rate_by_column["spot"] == Decimal("0")