feat(cli): subcommand 'audit data' — qualità dati market + chain

Usa il group 'audit' esistente (sibling di 'audit verify' per la hash
chain). Opzioni: --db, --since DAYS, --asset ETH|BTC, --json.
Output stdout Rich di default, dump JSON con --json. Esce con codice 2
su sqlite3.OperationalError (DB malformato/schema mancante).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/cerbero_bite/cli.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import asyncio
 import os
+import sqlite3
 import sys
 from collections.abc import Callable
 from datetime import UTC, datetime, timedelta
@@ -23,6 +24,12 @@ from rich.console import Console
 from rich.table import Table
 
 from cerbero_bite import __version__
+from cerbero_bite.analysis.data_audit import (
+    ChainAuditReport,
+    MarketAuditReport,
+    audit_market_snapshots,
+    audit_option_chain,
+)
 from cerbero_bite.clients import HttpToolClient, McpError
 from cerbero_bite.clients.deribit import DeribitClient
 from cerbero_bite.clients.hyperliquid import HyperliquidClient
@@ -898,6 +905,171 @@ def audit_verify(audit_path: Path) -> None:
         console.print(f"[green]ok[/green] {count} entries verified")
 
 
+def _market_to_dict(r: MarketAuditReport) -> dict[str, Any]:
+    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: ChainAuditReport) -> dict[str, Any]:
+    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),
+    }
+
+
+def _render_market_report(asset: str, r: MarketAuditReport) -> None:
+    console.print(
+        f"\n[bold cyan]=== {asset} — market_snapshots "
+        f"({r.since.date()} → {r.until.date()}) ===[/bold cyan]"
+    )
+    console.print(
+        f"  ticks:           {r.actual_ticks}  expected: {r.expected_ticks}   "
+        f"coverage: {r.coverage_pct}%"
+    )
+    console.print(f"  gaps > 20min:    {len(r.gaps)}")
+    console.print(
+        f"  fetch_ok=0:      {r.fetch_ok_zero_count} rows "
+        f"(max streak: {r.max_fetch_ok_zero_streak})"
+    )
+    bad_nulls = {k: v for k, v in r.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%")
+
+
+def _render_chain_report(asset: str, r: ChainAuditReport) -> None:
+    console.print(
+        f"\n[bold cyan]=== {asset} — option_chain_snapshots "
+        f"({r.since.date()} → {r.until.date()}) ===[/bold cyan]"
+    )
+    console.print(
+        f"  snapshots:       {r.actual_snapshots}  expected: {r.expected_snapshots}   "
+        f"coverage: {r.coverage_pct}%"
+    )
+    console.print(
+        f"  quotes/snap:     median {r.quotes_per_snap_median}  "
+        f"p10 {r.quotes_per_snap_p10}  p90 {r.quotes_per_snap_p90}"
+    )
+    console.print(f"  bid > ask:       {r.bid_gt_ask_count}")
+    console.print(
+        f"  IV null:         {r.iv_null_count} quotes ({r.iv_null_pct}%)"
+    )
+
+
+@audit.command(name="data")
+@click.option(
+    "--db",
+    "db_path",
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    default=_DEFAULT_DB_PATH,
+    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_data(
+    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
+
+    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
+        }
+    except sqlite3.OperationalError as exc:
+        console.print(f"[red]DB error[/red]: {exc}")
+        sys.exit(2)
+    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
+
+    for a in assets:
+        _render_market_report(a, market_reports[a])
+        _render_chain_report(a, chain_reports[a])
+
+
 @main.group()
 def state() -> None:
     """State inspection utilities."""

tests/unit/test_cli_audit_data.py

@@ -0,0 +1,84 @@
+"""Smoke tests for the ``cerbero-bite audit data`` CLI subcommand."""
+
+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 as cli_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):
+            for asset, spot in (("ETH", "3000"), ("BTC", "80000")):
+                conn.execute(
+                    "INSERT INTO market_snapshots(timestamp, asset, spot, "
+                    "dvol, fetch_ok) VALUES (?,?,?,?,1)",
+                    (now.isoformat(), asset, spot, "55"),
+                )
+            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_data_human_output(tmp_path: Path) -> None:
+    db = tmp_path / "state.sqlite"
+    _seed_minimal_db(db)
+    result = CliRunner().invoke(
+        cli_main, ["audit", "data", "--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_data_json_output(tmp_path: Path) -> None:
+    db = tmp_path / "state.sqlite"
+    _seed_minimal_db(db)
+    result = CliRunner().invoke(
+        cli_main,
+        ["audit", "data", "--db", str(db), "--since", "1", "--json"],
+    )
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert "since" in payload
+    assert "until" in payload
+    assert set(payload["assets"]) == {"ETH", "BTC"}
+    assert "market" in payload["assets"]["ETH"]
+    assert "chain" in payload["assets"]["ETH"]
+    assert payload["assets"]["ETH"]["market"]["asset"] == "ETH"
+
+
+def test_audit_data_single_asset_filter(tmp_path: Path) -> None:
+    db = tmp_path / "state.sqlite"
+    _seed_minimal_db(db)
+    result = CliRunner().invoke(
+        cli_main,
+        ["audit", "data", "--db", str(db), "--asset", "BTC", "--json"],
+    )
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert set(payload["assets"]) == {"BTC"}
+
+
+def test_audit_data_missing_db_exits_2(tmp_path: Path) -> None:
+    # click.Path(exists=True) returns exit_code=2 (UsageError) for missing files.
+    result = CliRunner().invoke(
+        cli_main,
+        ["audit", "data", "--db", str(tmp_path / "absent.sqlite")],
+    )
+    assert result.exit_code == 2