Adriano/Cerbero-Bite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
839285d75bf1d20f3aabde75cc962b63dbf01416
Cerbero-Bite/docs/superpowers/plans/2026-05-12-data-quality-audit.md
T
root 9e2216d202 fix(analysis): _expected_ticks usa ceiling division (no off-by-one) 
Il piano originale aveva `floor(span/15) + 1` che over-conta a span allineati
(span=60min → 5 invece di 4). Il primo fix dell'implementer (`floor(span/15)`)
under-conta a span non-allineati (span=16min → 1 invece di 2). Solo
`ceil(span/15)` è corretto in entrambi i casi. Aggiunti 2 test che
coprono gli scenari non-allineato e boundary-esatto per impedire
regressioni. Plan doc allineato.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 09:28:24 +00:00

1340 lines
40 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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
# 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**
```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.
Reference in New Issue View Git Blame Copy Permalink