Cerbero-Bite/docs/superpowers/plans/2026-05-12-data-quality-audit.md

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:

"""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:

"""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

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:

"""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:

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 count is
    # ceil(span_minutes / 15). floor() under-counts at non-multiple
    # spans; floor()+1 over-counts at aligned multiples. Only
    # ceil() works for both. Requires `import math` at the top.
    span_seconds = (until - first_tick).total_seconds()
    return math.ceil(span_seconds / (_TICK_INTERVAL_MIN * 60))

• 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

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:

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:

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

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:

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:

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

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:

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:

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

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:

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:

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

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:

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:

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

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:

@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

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:

"""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

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

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

.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

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.