# 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.