diff --git a/docs/superpowers/plans/2026-05-12-data-quality-audit.md b/docs/superpowers/plans/2026-05-12-data-quality-audit.md new file mode 100644 index 0000000..3b6a3dd --- /dev/null +++ b/docs/superpowers/plans/2026-05-12-data-quality-audit.md @@ -0,0 +1,1335 @@ +# Data Quality Audit Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build `cerbero-bite audit` CLI subcommand che verifica copertura temporale, gap, fetch_ok streaks, NULL rate per `market_snapshots` e gap/quote stats/bid-ask sanity/IV null/depth zero per `option_chain_snapshots` per ETH e BTC. + +**Architecture:** Modulo `analysis/data_audit.py` con funzioni pure (sqlite3.Connection + finestra temporale → dataclass frozen). Niente I/O verso MCP. CLI subcommand legge il DB in read-only, chiama le funzioni, formatta stdout (rich Table) o JSON. Test unitari con DB temporaneo e seed deterministico. + +**Tech Stack:** Python 3.13, sqlite3 stdlib, click (CLI), rich (output), pytest. Zero dipendenze nuove. + +**Spec:** `docs/superpowers/specs/2026-05-12-data-quality-audit-design.md` + +--- + +## File Structure + +**Create:** +- `src/cerbero_bite/analysis/__init__.py` — package marker +- `src/cerbero_bite/analysis/data_audit.py` — funzioni pure + dataclass + soglie costanti +- `tests/unit/test_data_audit.py` — test del modulo + +**Modify:** +- `src/cerbero_bite/cli.py` — aggiunge il subcommand `audit` + +**No DB schema changes.** L'audit legge esclusivamente; non scrive mai. + +--- + +### Task 1: Setup package + dataclass + soglie costanti + +**Files:** +- Create: `src/cerbero_bite/analysis/__init__.py` +- Create: `src/cerbero_bite/analysis/data_audit.py` +- Test: `tests/unit/test_data_audit.py` (creato nel Task 2) + +- [ ] **Step 1: Create empty package marker** + +Create `src/cerbero_bite/analysis/__init__.py`: + +```python +"""Analysis utilities — pure functions over the state DB. + +Modules here read SQLite, never write. They are ergonomic to call +from CLI commands, notebooks, or one-off scripts. +""" +``` + +- [ ] **Step 2: Create data_audit module with constants + dataclasses** + +Create `src/cerbero_bite/analysis/data_audit.py`: + +```python +"""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 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") +``` + +- [ ] **Step 3: Verify module imports** + +Run: `.venv/bin/python -c "from cerbero_bite.analysis.data_audit import MarketAuditReport, ChainAuditReport; print('ok')"` +Expected: `ok` + +- [ ] **Step 4: Commit** + +```bash +git add src/cerbero_bite/analysis/__init__.py src/cerbero_bite/analysis/data_audit.py +git commit -m "feat(analysis): skeleton modulo data_audit (dataclass + soglie)" +``` + +--- + +### Task 2: Helper `_expected_ticks` + first failing test + +**Files:** +- Create: `tests/unit/test_data_audit.py` +- Modify: `src/cerbero_bite/analysis/data_audit.py` (aggiunge `_expected_ticks`) + +- [ ] **Step 1: Write the failing test** + +Create `tests/unit/test_data_audit.py`: + +```python +"""Unit tests for analysis.data_audit.""" + +from __future__ import annotations + +import sqlite3 +from datetime import UTC, datetime +from decimal import Decimal +from pathlib import Path + +import pytest + +from cerbero_bite.analysis.data_audit import ( + ChainAuditReport, + MarketAuditReport, + audit_market_snapshots, + audit_option_chain, +) +from cerbero_bite.analysis.data_audit import _expected_ticks # noqa: PLC2701 +from cerbero_bite.state import connect, run_migrations, transaction + + +def test_expected_ticks_basic() -> None: + since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC) + until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC) + # 12:00, 12:15, 12:30, 12:45 → 4 ticks before 13:00 + assert _expected_ticks(since, until) == 4 + + +def test_expected_ticks_inclusive_left_exclusive_right() -> None: + since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC) + until = datetime(2026, 5, 12, 12, 15, tzinfo=UTC) + # Only 12:00 + assert _expected_ticks(since, until) == 1 + + +def test_expected_ticks_unaligned_since_rounds_up() -> None: + since = datetime(2026, 5, 12, 12, 7, tzinfo=UTC) + until = datetime(2026, 5, 12, 12, 30, tzinfo=UTC) + # First aligned tick after 12:07 is 12:15, then 12:30 is excluded + assert _expected_ticks(since, until) == 1 +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py::test_expected_ticks_basic -v` +Expected: FAIL with `ImportError` or `AttributeError` on `_expected_ticks`. + +- [ ] **Step 3: Implement `_expected_ticks`** + +Append to `src/cerbero_bite/analysis/data_audit.py`: + +```python +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 + span = until - first_tick + return int(span.total_seconds() // (_TICK_INTERVAL_MIN * 60)) + 1 +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -v` +Expected: 3 passed. + +- [ ] **Step 5: Commit** + +```bash +git add src/cerbero_bite/analysis/data_audit.py tests/unit/test_data_audit.py +git commit -m "feat(analysis): _expected_ticks per finestre */15 allineate" +``` + +--- + +### Task 3: Helper `_detect_gaps` + +**Files:** +- Modify: `src/cerbero_bite/analysis/data_audit.py` +- Modify: `tests/unit/test_data_audit.py` + +- [ ] **Step 1: Write the failing test** + +Append to `tests/unit/test_data_audit.py`: + +```python +from cerbero_bite.analysis.data_audit import _detect_gaps # noqa: PLC2701 + + +def test_detect_gaps_returns_empty_when_no_gap() -> None: + ts = [ + datetime(2026, 5, 12, 12, 0, tzinfo=UTC), + datetime(2026, 5, 12, 12, 15, tzinfo=UTC), + datetime(2026, 5, 12, 12, 30, tzinfo=UTC), + ] + assert _detect_gaps(ts) == () + + +def test_detect_gaps_flags_above_threshold() -> None: + ts = [ + datetime(2026, 5, 12, 12, 0, tzinfo=UTC), + datetime(2026, 5, 12, 12, 45, tzinfo=UTC), # 45-min gap + datetime(2026, 5, 12, 13, 0, tzinfo=UTC), + ] + gaps = _detect_gaps(ts) + assert len(gaps) == 1 + assert gaps[0].gap_minutes == 45 + assert gaps[0].prev_timestamp == ts[0] + assert gaps[0].next_timestamp == ts[1] + + +def test_detect_gaps_ignores_threshold_boundary() -> None: + # 20-min gap is exactly the threshold → NOT flagged (strict >) + ts = [ + datetime(2026, 5, 12, 12, 0, tzinfo=UTC), + datetime(2026, 5, 12, 12, 20, tzinfo=UTC), + ] + assert _detect_gaps(ts) == () +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py::test_detect_gaps_flags_above_threshold -v` +Expected: FAIL on import of `_detect_gaps`. + +- [ ] **Step 3: Implement `_detect_gaps`** + +Append to `src/cerbero_bite/analysis/data_audit.py`: + +```python +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) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -v` +Expected: all green. + +- [ ] **Step 5: Commit** + +```bash +git add src/cerbero_bite/analysis/data_audit.py tests/unit/test_data_audit.py +git commit -m "feat(analysis): _detect_gaps su timestamp consecutivi (> 20 min)" +``` + +--- + +### Task 4: Helper `_max_zero_streak` + +**Files:** +- Modify: `src/cerbero_bite/analysis/data_audit.py` +- Modify: `tests/unit/test_data_audit.py` + +- [ ] **Step 1: Write the failing test** + +Append to `tests/unit/test_data_audit.py`: + +```python +from cerbero_bite.analysis.data_audit import _max_zero_streak # noqa: PLC2701 + + +def test_max_zero_streak_empty() -> None: + assert _max_zero_streak([]) == 0 + + +def test_max_zero_streak_no_zeros() -> None: + assert _max_zero_streak([1, 1, 1, 1]) == 0 + + +def test_max_zero_streak_single_zero_block() -> None: + assert _max_zero_streak([1, 0, 0, 0, 1, 0, 1]) == 3 + + +def test_max_zero_streak_all_zeros() -> None: + assert _max_zero_streak([0, 0, 0]) == 3 +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -k max_zero_streak -v` +Expected: FAIL on import. + +- [ ] **Step 3: Implement** + +Append to `data_audit.py`: + +```python +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 +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -v` +Expected: all green. + +- [ ] **Step 5: Commit** + +```bash +git add src/cerbero_bite/analysis/data_audit.py tests/unit/test_data_audit.py +git commit -m "feat(analysis): _max_zero_streak su flag fetch_ok" +``` + +--- + +### Task 5: `audit_market_snapshots` composer + +**Files:** +- Modify: `src/cerbero_bite/analysis/data_audit.py` +- Modify: `tests/unit/test_data_audit.py` + +- [ ] **Step 1: Write the failing test (full integration on tmp DB)** + +Append to `tests/unit/test_data_audit.py`: + +```python +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") +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -k audit_market -v` +Expected: FAIL on `audit_market_snapshots` not implemented or returning placeholder. + +- [ ] **Step 3: Implement `audit_market_snapshots`** + +Append to `src/cerbero_bite/analysis/data_audit.py`: + +```python +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, + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -v` +Expected: all green (now ~10 tests). + +- [ ] **Step 5: Commit** + +```bash +git add src/cerbero_bite/analysis/data_audit.py tests/unit/test_data_audit.py +git commit -m "feat(analysis): audit_market_snapshots — coverage, gap, fetch_ok, NULL rate" +``` + +--- + +### Task 6: Helper percentiles `_pct` + +**Files:** +- Modify: `src/cerbero_bite/analysis/data_audit.py` +- Modify: `tests/unit/test_data_audit.py` + +- [ ] **Step 1: Write the failing test** + +Append to `tests/unit/test_data_audit.py`: + +```python +from cerbero_bite.analysis.data_audit import _pct # noqa: PLC2701 + + +def test_pct_empty_returns_zero() -> None: + assert _pct([], 50) == 0 + + +def test_pct_median_odd_count() -> None: + assert _pct([10, 20, 30, 40, 50], 50) == 30 + + +def test_pct_p10_p90() -> None: + values = list(range(1, 101)) # 1..100 + assert _pct(values, 10) == 10 + assert _pct(values, 90) == 90 +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -k pct -v` +Expected: FAIL on import. + +- [ ] **Step 3: Implement** + +Append to `data_audit.py`: + +```python +def _pct(values: list[int], q: int) -> int: + """Integer percentile (linear interpolation, then rounded). + + Returns 0 on empty input. ``q`` is in 0..100. + """ + if not values: + return 0 + sorted_vals = sorted(values) + return int(round(statistics.quantiles( + sorted_vals, n=100, method="inclusive" + )[q - 1])) if q < 100 else sorted_vals[-1] +``` + +Note: `statistics.quantiles(n=100)` returns 99 cut-points (q=1..99). For +q=50 / median, the 50th cut-point is at index 49. + +Wait — re-read the docs: `quantiles(data, *, n=4, method='exclusive')` +returns ``n - 1`` cut points. With ``n=100`` → 99 cut points +representing q=1..99. So index ``q - 1`` for q in 1..99. For q=100 we +return the max. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -v` +Expected: all green. + +- [ ] **Step 5: Commit** + +```bash +git add src/cerbero_bite/analysis/data_audit.py tests/unit/test_data_audit.py +git commit -m "feat(analysis): _pct helper per percentili interi" +``` + +--- + +### Task 7: `audit_option_chain` composer + +**Files:** +- Modify: `src/cerbero_bite/analysis/data_audit.py` +- Modify: `tests/unit/test_data_audit.py` + +- [ ] **Step 1: Write the failing test** + +Append to `tests/unit/test_data_audit.py`: + +```python +def _seed_chain_row( + conn: sqlite3.Connection, + *, + asset: str, + ts: datetime, + instrument: str, + strike: str = "3000", + expiry: datetime | None = None, + option_type: str = "C", + bid: str | None = "0.01", + ask: str | None = "0.02", + iv: str | None = "0.7", + depth: int | None = 5, +) -> None: + if expiry is None: + expiry = datetime(2026, 6, 12, 8, 0, tzinfo=UTC) + conn.execute( + "INSERT INTO option_chain_snapshots(timestamp, asset, instrument_name, " + "strike, expiry, option_type, bid, ask, iv, book_depth_top3) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", + ( + ts.isoformat(), + asset, + instrument, + strike, + expiry.isoformat(), + option_type, + bid, + ask, + iv, + depth, + ), + ) + + +def test_audit_chain_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): + ts = since.replace(minute=minute) + for i in range(10): + _seed_chain_row( + conn, + asset="ETH", + ts=ts, + instrument=f"ETH-EXP-3000-{i}-C", + strike=str(3000 + i * 50), + ) + report = audit_option_chain( + conn, asset="ETH", since=since, until=until + ) + finally: + conn.close() + assert report.expected_snapshots == 4 + assert report.actual_snapshots == 4 + assert report.coverage_pct == Decimal("100") + assert report.quotes_per_snap_median == 10 + assert report.bid_gt_ask_count == 0 + assert report.iv_null_count == 0 + + +def test_audit_chain_detects_bid_gt_ask_and_iv_null(tmp_path: Path) -> None: + conn = _make_conn(tmp_path) + since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC) + until = datetime(2026, 5, 12, 12, 30, tzinfo=UTC) + try: + with transaction(conn): + ts = since.replace(minute=0) + _seed_chain_row( + conn, + asset="ETH", + ts=ts, + instrument="ETH-A", + bid="0.10", + ask="0.05", # inverted + ) + _seed_chain_row( + conn, + asset="ETH", + ts=ts, + instrument="ETH-B", + iv=None, # missing IV + strike="3050", + ) + _seed_chain_row( + conn, + asset="ETH", + ts=ts, + instrument="ETH-C", + depth=0, # zero depth + strike="3100", + ) + report = audit_option_chain( + conn, asset="ETH", since=since, until=until + ) + finally: + conn.close() + assert report.bid_gt_ask_count == 1 + assert report.iv_null_count == 1 + assert report.iv_null_pct == Decimal("33.33") + assert report.depth_zero_pct == Decimal("33.33") + + +def test_audit_chain_missing_snapshot(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): + # Only 2 of the expected 4 ticks have data + for minute in (0, 15): + _seed_chain_row( + conn, + asset="ETH", + ts=since.replace(minute=minute), + instrument=f"ETH-{minute}", + ) + report = audit_option_chain( + conn, asset="ETH", since=since, until=until + ) + finally: + conn.close() + assert report.expected_snapshots == 4 + assert report.actual_snapshots == 2 + assert report.coverage_pct == Decimal("50") +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -k audit_chain -v` +Expected: FAIL. + +- [ ] **Step 3: Implement `audit_option_chain`** + +Append to `data_audit.py`: + +```python +def audit_option_chain( + conn: sqlite3.Connection, + *, + asset: str, + since: datetime, + until: datetime, +) -> ChainAuditReport: + """Audit dell'option chain per un asset in ``[since, until)``.""" + expected = _expected_ticks(since, until) + # Coverage by distinct snapshot timestamp. + snap_counts: list[tuple[str, int]] = list( + conn.execute( + "SELECT timestamp, COUNT(*) AS n FROM option_chain_snapshots " + "WHERE asset = ? AND timestamp >= ? AND timestamp < ? " + "GROUP BY timestamp ORDER BY timestamp", + (asset, since.isoformat(), until.isoformat()), + ).fetchall() + ) + actual = len(snap_counts) + coverage = ( + (Decimal(actual) / Decimal(expected) * Decimal("100")).quantize( + Decimal("0.01") + ) + if expected > 0 + else Decimal("0") + ) + quotes_per_snap = [n for _, n in snap_counts] + median_q = _pct(quotes_per_snap, 50) + p10_q = _pct(quotes_per_snap, 10) + p90_q = _pct(quotes_per_snap, 90) + + quote_rows = conn.execute( + "SELECT bid, ask, iv, book_depth_top3 FROM option_chain_snapshots " + "WHERE asset = ? AND timestamp >= ? AND timestamp < ?", + (asset, since.isoformat(), until.isoformat()), + ).fetchall() + total_quotes = len(quote_rows) + if total_quotes == 0: + return ChainAuditReport( + asset=asset, + since=since, + until=until, + expected_snapshots=expected, + actual_snapshots=actual, + coverage_pct=coverage, + ) + + bid_gt_ask = 0 + iv_null = 0 + depth_zero = 0 + for r in quote_rows: + bid_s, ask_s, iv_s, depth = r["bid"], r["ask"], r["iv"], r["book_depth_top3"] + if bid_s is not None and ask_s is not None: + try: + if Decimal(bid_s) > Decimal(ask_s): + bid_gt_ask += 1 + except (ValueError, ArithmeticError): + pass + if iv_s is None: + iv_null += 1 + else: + try: + Decimal(iv_s) + except (ValueError, ArithmeticError): + iv_null += 1 + if depth == 0: + depth_zero += 1 + + pct = lambda c: ( + Decimal(c) / Decimal(total_quotes) * Decimal("100") + ).quantize(Decimal("0.01")) + + return ChainAuditReport( + asset=asset, + since=since, + until=until, + expected_snapshots=expected, + actual_snapshots=actual, + coverage_pct=coverage, + quotes_per_snap_median=median_q, + quotes_per_snap_p10=p10_q, + quotes_per_snap_p90=p90_q, + bid_gt_ask_count=bid_gt_ask, + iv_null_count=iv_null, + iv_null_pct=pct(iv_null), + depth_zero_pct=pct(depth_zero), + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/unit/test_data_audit.py -v` +Expected: all green (~13 tests). + +- [ ] **Step 5: Commit** + +```bash +git add src/cerbero_bite/analysis/data_audit.py tests/unit/test_data_audit.py +git commit -m "feat(analysis): audit_option_chain — coverage, quote stats, bid>ask, IV null, depth zero" +``` + +--- + +### Task 8: CLI subcommand `audit` + +**Files:** +- Modify: `src/cerbero_bite/cli.py` +- Test: subprocess smoke test in Task 9 + +- [ ] **Step 1: Read where to insert the new command** + +Run: `.venv/bin/python -c "from cerbero_bite.cli import main; print('ok')"` +Expected: `ok`. This confirms the module imports. + +- [ ] **Step 2: Add the `audit` command at the bottom of cli.py (before the `if __name__ ...` block)** + +Find the section after the existing `backtest` command (look for `def backtest(`). Insert immediately AFTER its closing — before the `if __name__ == "__main__"` block at end of file. + +Add this code: + +```python +@main.command("audit") +@click.option( + "--db", + "db_path", + type=click.Path(exists=True, dir_okay=False, path_type=Path), + default=Path("data/state.sqlite"), + show_default=True, + help="Path al DB SQLite di stato.", +) +@click.option( + "--since", + "since_days", + type=int, + default=7, + show_default=True, + help="Finestra di analisi (giorni indietro da ora).", +) +@click.option( + "--asset", + "asset", + type=click.Choice(["ETH", "BTC"], case_sensitive=False), + default=None, + help="Limita l'audit a un singolo asset (default: entrambi).", +) +@click.option( + "--json", + "as_json", + is_flag=True, + default=False, + help="Stampa solo dump JSON, niente tabelle umane.", +) +def audit_command( + db_path: Path, + since_days: int, + asset: str | None, + as_json: bool, +) -> None: + """Audit qualità dati: market_snapshots + option_chain_snapshots.""" + import json as _json # noqa: PLC0415 + from cerbero_bite.analysis.data_audit import ( # noqa: PLC0415 + audit_market_snapshots, + audit_option_chain, + ) + + until = datetime.now(UTC) + since = until - timedelta(days=since_days) + assets = [asset.upper()] if asset else ["ETH", "BTC"] + + conn = connect_state(db_path) + try: + market_reports = { + a: audit_market_snapshots(conn, asset=a, since=since, until=until) + for a in assets + } + chain_reports = { + a: audit_option_chain(conn, asset=a, since=since, until=until) + for a in assets + } + finally: + conn.close() + + if as_json: + payload = { + "since": since.isoformat(), + "until": until.isoformat(), + "assets": { + a: { + "market": _market_to_dict(market_reports[a]), + "chain": _chain_to_dict(chain_reports[a]), + } + for a in assets + }, + } + click.echo(_json.dumps(payload, indent=2, default=str)) + return + + console = Console() + for a in assets: + m = market_reports[a] + console.print( + f"\n[bold cyan]=== {a} — market_snapshots " + f"({m.since.date()} → {m.until.date()}) ===[/bold cyan]" + ) + console.print( + f" ticks: {m.actual_ticks} expected: {m.expected_ticks} " + f"coverage: {m.coverage_pct}%" + ) + console.print(f" gaps > 20min: {len(m.gaps)}") + console.print( + f" fetch_ok=0: {m.fetch_ok_zero_count} rows " + f"(max streak: {m.max_fetch_ok_zero_streak})" + ) + bad_nulls = { + k: v + for k, v in m.null_rate_by_column.items() + if v > Decimal("0") + } + if bad_nulls: + parts = " ".join( + f"{k} {(v * 100).quantize(Decimal('0.1'))}%" + for k, v in bad_nulls.items() + ) + console.print(f" null rate: {parts}") + else: + console.print(" null rate: all columns 0%") + + c = chain_reports[a] + console.print( + f"\n[bold cyan]=== {a} — option_chain_snapshots " + f"({c.since.date()} → {c.until.date()}) ===[/bold cyan]" + ) + console.print( + f" snapshots: {c.actual_snapshots} expected: {c.expected_snapshots} " + f"coverage: {c.coverage_pct}%" + ) + console.print( + f" quotes/snap: median {c.quotes_per_snap_median} " + f"p10 {c.quotes_per_snap_p10} p90 {c.quotes_per_snap_p90}" + ) + console.print(f" bid > ask: {c.bid_gt_ask_count}") + console.print( + f" IV null: {c.iv_null_count} quotes ({c.iv_null_pct}%)" + ) + console.print(f" depth_top3 = 0: {c.depth_zero_pct}% of quotes") + + +def _market_to_dict(r) -> dict: # type: ignore[no-untyped-def] + return { + "asset": r.asset, + "since": r.since.isoformat(), + "until": r.until.isoformat(), + "expected_ticks": r.expected_ticks, + "actual_ticks": r.actual_ticks, + "coverage_pct": str(r.coverage_pct), + "gaps": [ + { + "prev": g.prev_timestamp.isoformat(), + "next": g.next_timestamp.isoformat(), + "gap_minutes": g.gap_minutes, + } + for g in r.gaps + ], + "fetch_ok_zero_count": r.fetch_ok_zero_count, + "max_fetch_ok_zero_streak": r.max_fetch_ok_zero_streak, + "null_rate_by_column": {k: str(v) for k, v in r.null_rate_by_column.items()}, + } + + +def _chain_to_dict(r) -> dict: # type: ignore[no-untyped-def] + return { + "asset": r.asset, + "since": r.since.isoformat(), + "until": r.until.isoformat(), + "expected_snapshots": r.expected_snapshots, + "actual_snapshots": r.actual_snapshots, + "coverage_pct": str(r.coverage_pct), + "quotes_per_snap_median": r.quotes_per_snap_median, + "quotes_per_snap_p10": r.quotes_per_snap_p10, + "quotes_per_snap_p90": r.quotes_per_snap_p90, + "bid_gt_ask_count": r.bid_gt_ask_count, + "iv_null_count": r.iv_null_count, + "iv_null_pct": str(r.iv_null_pct), + "depth_zero_pct": str(r.depth_zero_pct), + } +``` + +- [ ] **Step 3: Verify CLI registration** + +Run: `.venv/bin/python -m cerbero_bite.cli audit --help 2>&1 | head -20` +Expected: shows the audit command help with `--db`, `--since`, `--asset`, `--json` options. + +- [ ] **Step 4: Run the existing test suite to ensure no regression** + +Run: `.venv/bin/python -m pytest --tb=short -q 2>&1 | tail -5` +Expected: all tests pass (no new tests yet, just confirming nothing broke). + +- [ ] **Step 5: Commit** + +```bash +git add src/cerbero_bite/cli.py +git commit -m "feat(cli): subcommand audit — qualita dati market + chain" +``` + +--- + +### Task 9: CLI smoke test + +**Files:** +- Create: `tests/unit/test_cli_audit.py` + +- [ ] **Step 1: Write a smoke test that invokes the click command via CliRunner** + +Create `tests/unit/test_cli_audit.py`: + +```python +"""Smoke test for the `cerbero-bite audit` CLI command.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path + +from click.testing import CliRunner + +from cerbero_bite.cli import main +from cerbero_bite.state import connect, run_migrations, transaction + + +def _seed_minimal_db(db_path: Path) -> None: + conn = connect(db_path) + try: + run_migrations(conn) + now = datetime.now(UTC).replace(microsecond=0) + with transaction(conn): + conn.execute( + "INSERT INTO market_snapshots(timestamp, asset, spot, dvol, fetch_ok) " + "VALUES (?, 'ETH', '3000', '55', 1)", + (now.isoformat(),), + ) + conn.execute( + "INSERT INTO market_snapshots(timestamp, asset, spot, dvol, fetch_ok) " + "VALUES (?, 'BTC', '80000', '40', 1)", + (now.isoformat(),), + ) + conn.execute( + "INSERT INTO option_chain_snapshots(timestamp, asset, instrument_name, " + "strike, expiry, option_type, bid, ask, iv) " + "VALUES (?, 'ETH', 'ETH-X', '3000', ?, 'C', '0.01', '0.02', '0.7')", + (now.isoformat(), now.isoformat()), + ) + finally: + conn.close() + + +def test_audit_human_output(tmp_path: Path) -> None: + db = tmp_path / "state.sqlite" + _seed_minimal_db(db) + runner = CliRunner() + result = runner.invoke(main, ["audit", "--db", str(db), "--since", "1"]) + assert result.exit_code == 0, result.output + assert "ETH — market_snapshots" in result.output + assert "BTC — market_snapshots" in result.output + assert "ETH — option_chain_snapshots" in result.output + + +def test_audit_json_output(tmp_path: Path) -> None: + db = tmp_path / "state.sqlite" + _seed_minimal_db(db) + runner = CliRunner() + result = runner.invoke( + main, ["audit", "--db", str(db), "--since", "1", "--json"] + ) + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert "assets" in payload + assert set(payload["assets"]) == {"ETH", "BTC"} + assert "market" in payload["assets"]["ETH"] + assert "chain" in payload["assets"]["ETH"] + + +def test_audit_single_asset(tmp_path: Path) -> None: + db = tmp_path / "state.sqlite" + _seed_minimal_db(db) + runner = CliRunner() + result = runner.invoke( + main, ["audit", "--db", str(db), "--asset", "BTC", "--json"] + ) + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert set(payload["assets"]) == {"BTC"} +``` + +- [ ] **Step 2: Run the smoke tests** + +Run: `.venv/bin/python -m pytest tests/unit/test_cli_audit.py -v` +Expected: 3 passed. + +- [ ] **Step 3: Run the full test suite to confirm no regressions** + +Run: `.venv/bin/python -m pytest --tb=short -q 2>&1 | tail -5` +Expected: all tests pass. + +- [ ] **Step 4: Commit** + +```bash +git add tests/unit/test_cli_audit.py +git commit -m "test(cli): smoke test per cerbero-bite audit (human + json + single asset)" +``` + +--- + +### Task 10: Run audit against production DB + +**Files:** none (read-only end-to-end check) + +- [ ] **Step 1: Copy production DB out and run audit on it** + +```bash +docker cp cerbero-bite-cerbero-bite-1:/app/data/state.sqlite /tmp/state.audit.sqlite +.venv/bin/python -m cerbero_bite.cli audit --db /tmp/state.audit.sqlite --since 7 +``` + +Expected: structured output for ETH + BTC with coverage near 100% for `market_snapshots` (window started 2026-03-26 so 7d ≪ history) and `option_chain_snapshots` (started 2026-05-04 for BTC). + +- [ ] **Step 2: Run audit with --json and pipe to jq for spot-check** + +```bash +.venv/bin/python -m cerbero_bite.cli audit --db /tmp/state.audit.sqlite --since 7 --json | jq '.assets.ETH.chain.coverage_pct' +``` + +Expected: a Decimal string like `"99.40"` or `"100.00"`. + +- [ ] **Step 3: Cleanup** + +```bash +rm /tmp/state.audit.sqlite +``` + +- [ ] **Step 4: Document the result (optional commit if interesting findings)** + +If the audit surfaces any issue worth recording (a gap, a high NULL rate), +append the finding as a note to the spec at +`docs/superpowers/specs/2026-05-12-data-quality-audit-design.md` under +an "Initial findings" section, and commit. Otherwise skip this step. + +--- + +## Self-Review + +**Spec coverage:** + +| Spec section | Covered by task | +|---|---| +| `audit_market_snapshots` API + dataclass | Tasks 1, 2, 3, 4, 5 | +| `audit_option_chain` API + dataclass | Tasks 1, 6, 7 | +| Checks: gap > 20 min | Task 3 | +| Checks: fetch_ok streak ≥ 3 | Task 4 | +| Checks: NULL rate per column | Task 5 | +| Checks: snap missing | Task 7 | +| Checks: quotes/snap stats | Tasks 6, 7 | +| Checks: bid > ask, IV null, depth zero | Task 7 | +| CLI subcommand `--since --json --asset` | Task 8 | +| Stdout format (rich) | Task 8 | +| JSON output format | Task 8 | +| Tests via tmp DB + seed | Tasks 2–7 | +| Smoke test CLI | Task 9 | +| Anti-goals (no DB writes, no cron, no alerts) | Enforced by architecture | + +No gaps. + +**Placeholder scan:** no TBD/TODO/"implement later"/"add appropriate handling" — all code is concrete and complete. + +**Type consistency:** dataclass field names defined in Task 1 are used identically in Task 5 (`expected_ticks`, `actual_ticks`, `coverage_pct`, `gaps`, `fetch_ok_zero_count`, `max_fetch_ok_zero_streak`, `null_rate_by_column`) and Task 7 (`expected_snapshots`, etc.). CLI in Task 8 reads the same field names. ✓ + +Helper function names: `_expected_ticks` (Task 2), `_detect_gaps` (Task 3), `_max_zero_streak` (Task 4), `_pct` (Task 6), `_fetch_market_rows` and `_compute_null_rate` (Task 5), `_market_to_dict` / `_chain_to_dict` (Task 8) — all distinct, all used in the order defined. + +Plan is internally consistent.