16 Commits

Author SHA1 Message Date
Adriano Dal Pastro 16d73c309a fix(analysis): distingui ask<=0 (missing ask) da bid>ask (cross reale)
Nuovo campo ChainAuditReport.ask_zero_count. La logica del loop
parsifica prima l'ask: se ask<=0 è "missing ask side" (su Deribit
significa nessun ordine in vendita), conta in ask_zero_count e non
applica il check bid>ask. Solo con ask>0 si confronta con bid.

Sul DB prod gli "bid>ask" del primo audit (27 ETH + 10 BTC negli
ultimi 7d) erano tutti falsi positivi con ask=0 — concentrati sulla
weekly 2026-05-29, OI alti ma volume basso. Dopo il fix: bid>ask=0,
ask<=0=83 ETH + 13 BTC.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-16 19:59:16 +00:00
Adriano Dal Pastro 374822a931 docs(spec): rimuovi depth_zero_pct + nota su book_depth_top3 NULL-by-design
Allinea la spec all'implementazione: dataclass ChainAuditReport senza
depth_zero_pct, tabella checks senza la riga book_depth_top3=0, output
example senza la riga depth_top3. Aggiunta nota che spiega perché il
check è stato rimosso (colonna popolata solo dal path entry_cycle).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-16 19:48:23 +00:00
Adriano Dal Pastro 923acd79f6 docs(spec): rename CLI subcommand audit → audit data
Riflette la struttura effettiva del CLI: il group 'audit' contiene già
'verify' per la hash chain; il data-quality audit è entrato come
sibling 'data'. Documentata la deviazione in linea.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-16 19:46:54 +00:00
Adriano Dal Pastro 197c9db74d 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>
2026-05-16 19:37:33 +00:00
Adriano Dal Pastro 06ce2c0eb2 feat(analysis): audit_option_chain — coverage, quote stats, bid>ask, IV null
Implementa la funzione dichiarata in __all__ ma mancante. Helper _pct
usa statistics.quantiles(method="inclusive") con fallback per len<=1.
Niente check su book_depth_top3: per design è NULL sugli snapshot
(popolato solo da entry_cycle per gli strike candidati al picker).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-16 19:32:52 +00:00
Adriano Dal Pastro bb2ca425a7 refactor(analysis): rimuovi depth_zero_pct da ChainAuditReport
book_depth_top3 è popolato solo dal path entry_cycle (per gli strike
candidati al picker), mai dal collector option_chain_snapshot — il
controllo depth_zero su questi snapshot sarebbe strutturalmente 100%.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-16 19:22:31 +00:00
root d6af69f4cb feat(analysis): audit_market_snapshots — coverage, gap, fetch_ok, NULL rate
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-13 09:33:28 +00:00
root aeac8f2a95 feat(analysis): _max_zero_streak su flag fetch_ok 2026-05-13 09:31:38 +00:00
root 75fe803296 style(analysis): consolidate test imports at top of file (PEP 8) 2026-05-13 09:30:41 +00:00
root ea5c612446 feat(analysis): _detect_gaps su timestamp consecutivi (> 20 min)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-13 09:29:48 +00:00
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
root 35ac92e938 feat(analysis): _expected_ticks per finestre */15 allineate
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 22:32:12 +00:00
root 07a8bbf5c8 feat(analysis): skeleton modulo data_audit (dataclass + soglie) 2026-05-12 22:01:44 +00:00
root 0c6e462545 docs(plan): data quality audit implementation plan (10 task TDD)
Piano dettagliato task-by-task per `cerbero-bite audit`:
analysis/data_audit.py (helper puri + dataclass), CLI subcommand,
test unit + smoke test, end-to-end su DB produzione. Ogni task ha
i suoi step TDD con codice completo, comandi e commit.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 21:58:54 +00:00
root 569df334dc docs(spec): data quality audit design (chain + market_snapshots)
Spec del comando CLI `cerbero-bite audit`: copertura temporale, gap,
fetch_ok streaks, NULL rate per market_snapshots; snap mancanti,
quote/snap, bid>ask, IV null, depth zero per option_chain_snapshots.
Output stdout + opzionale --json. Pre-requisito al backtest
non-stilizzato.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 21:51:51 +00:00
root 19695e4730 feat(state): dvol_history multi-asset (ETH+BTC) + backfill ETH legacy rows
Migration 0006 promuove dvol_history da PK=(timestamp) mono-ETH a
PK=(timestamp, asset), rinomina eth_spot -> spot, e backfilla con
asset='ETH' le righe storiche. market_snapshot_cycle ora scrive sia
per ETH che per BTC; monitor_cycle resta ETH-only via WHERE asset='ETH'
nella lookup di return_4h.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 13:38:34 +00:00
17 changed files with 2698 additions and 33 deletions
+1 -1
View File
@@ -242,7 +242,7 @@ async def run_monitor_cycle(ctx: RuntimeContext, *, now):
dvol = await deribit.latest_dvol(currency="ETH", now=now)
return_4h = await _fetch_return_4h(ctx, now=now) # usa dvol_history o
# fallback get_historical
repo.record_dvol_snapshot(DvolSnapshot(timestamp=now, dvol=dvol, eth_spot=spot))
repo.record_dvol_snapshot(DvolSnapshot(timestamp=now, asset="ETH", dvol=dvol, spot=spot))
for record in repo.list_positions(status="open"):
snapshot = await _build_position_snapshot(...)
+14 -5
View File
@@ -137,16 +137,25 @@ CREATE INDEX idx_decisions_proposal ON decisions(proposal_id);
### `dvol_history`
Snapshot DVOL + ETH spot ad ogni evaluation. Utile per il calcolo di
`return_4h` durante il monitor (vedi `runtime/monitor_cycle.py
_fetch_return_4h`) e per analisi post-mortem.
Snapshot DVOL + spot per asset ad ogni evaluation. Utile per il
calcolo di `return_4h` durante il monitor (vedi
`runtime/monitor_cycle.py _fetch_return_4h`, che filtra `asset='ETH'`)
e per analisi post-mortem comparate fra ETH e BTC. Lo schema è
multi-asset dal migration `0006_dvol_history_multi_asset.sql`; le
righe storiche pre-migration sono state backfillate con
`asset='ETH'`.
```sql
CREATE TABLE dvol_history (
timestamp TEXT PRIMARY KEY,
timestamp TEXT NOT NULL,
asset TEXT NOT NULL, -- "ETH", "BTC"
dvol NUMERIC NOT NULL,
eth_spot NUMERIC NOT NULL
spot NUMERIC NOT NULL,
PRIMARY KEY (timestamp, asset)
);
CREATE INDEX idx_dvol_history_asset_ts
ON dvol_history(asset, timestamp DESC);
```
### `manual_actions`
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,201 @@
# Data Quality Audit — Design Spec
**Date:** 2026-05-12
**Status:** Approved (design phase)
**Author:** session-driven (operator + agent)
## Motivation
Prima di costruire il backtest non-stilizzato su `option_chain_snapshots`
(prossimo macro-step del progetto), serve confermare che i dati
raccolti negli ultimi 11 giorni (ETH) e 8 giorni (BTC) siano usabili:
copertura temporale piena, niente buchi sistemici, niente quote
malformate (bid > ask, IV mancante, depth book a zero). Lo stesso
audit dev'essere ri-eseguibile periodicamente come check operativo.
`market_snapshots` rientra nello scope per simmetria (entrambe le
tabelle alimentano la decisione di entrata e il monitoring), mentre
`dvol_history` è escluso: appena migrato a multi-asset (commit
`19695e4`), serie troppo corta per BTC (29 righe al momento del
design) e copertura ETH già implicita in `market_snapshots`.
## Scope
**In scope:**
- `market_snapshots`: continuità temporale, fetch_ok streaks, NULL rate
per colonna numerica, parità ETH/BTC.
- `option_chain_snapshots`: snapshot mancanti, distribuzione quote per
snap, bid/ask sanity, IV null rate, book depth.
- CLI subcommand `cerbero-bite audit data`, output stdout + opzionale `--json`.
**Out of scope:**
- `dvol_history`, `decisions`, `positions`, `instructions`,
`manual_actions` (non rilevanti per il backtest non-stilizzato).
- Audit di consistenza cross-tabella (es: per ogni snapshot chain esiste
uno snapshot market) — interessante ma rinviato.
- Persistenza dei risultati audit nello stesso DB.
## Architecture
```
src/cerbero_bite/analysis/
__init__.py
data_audit.py # logica pura, no I/O lato MCP
src/cerbero_bite/cli.py # nuovo subcommand `audit data`
tests/unit/
test_data_audit.py # DB temporaneo + seed deterministico
```
`data_audit.py` espone funzioni pure che prendono una `sqlite3.Connection`
e una finestra temporale, ritornano `dataclass` di risultati. Il CLI
apre la connection in read-only, chiama le funzioni, formatta l'output.
Funzioni principali:
```python
@dataclass(frozen=True)
class MarketAuditReport:
asset: str
expected_ticks: int
actual_ticks: int
coverage_pct: Decimal
gaps_over_threshold: list[GapRecord]
fetch_ok_zero_count: int
max_fetch_ok_zero_streak: int
null_rate_by_column: dict[str, Decimal]
@dataclass(frozen=True)
class ChainAuditReport:
asset: str
expected_snapshots: int
actual_snapshots: int
coverage_pct: Decimal
quotes_per_snap_median: int
quotes_per_snap_p10: int
quotes_per_snap_p90: int
bid_gt_ask_count: int
ask_zero_count: int
iv_null_count: int
iv_null_pct: Decimal
def audit_market_snapshots(conn, *, asset, since, now) -> MarketAuditReport: ...
def audit_option_chain(conn, *, asset, since, now) -> ChainAuditReport: ...
```
## Checks & Thresholds
| Tabella | Check | Soglia "bad" | Rationale |
|---|---|---|---|
| market_snapshots | gap tra tick consecutivi | > 20 min | cron è `*/15`; +5 min tolleranza |
| market_snapshots | streak `fetch_ok=0` | ≥ 3 consecutivi | 1-2 = transient MCP, 3+ = pattern |
| market_snapshots | NULL rate per colonna | > 10% nella finestra | una metrica con >10% NULL non è affidabile per backtest |
| option_chain_snapshots | snap mancanti | qualsiasi (count visibile) | cron `*/15`, ogni miss è significativo |
| option_chain_snapshots | quote/snap < 50% mediana 24h | qualsiasi | rilevatore di chain truncate (mismatch with width filter) |
| option_chain_snapshots | bid > ask (con ask > 0) | qualsiasi | dato corrotto, da indagare |
| option_chain_snapshots | ask ≤ 0 | qualsiasi | missing ask side: best ask vuoto al momento della query (≠ cross BBO) |
| option_chain_snapshots | IV null/non-parseable | conteggio + % | IV è chiave per BS skew calibration |
> **Nota:** il check `book_depth_top3 = 0` originariamente previsto è
> stato rimosso. La colonna è NULL by design sugli snapshot (il collector
> `option_chain_snapshot_cycle` non chiama l'orderbook per ogni strike
> per non saturare l'API Deribit); viene popolata solo dal path
> `entry_cycle` per gli strike candidati al picker. Il check di liquidità
> ha senso solo lì, non in questo audit.
Le soglie sono costanti modulo (non config YAML) per ridurre il blast
radius dei cambi: il backtest e l'audit girano in contesti diversi,
non condividono parametri operativi.
## CLI
```
cerbero-bite audit data [--db PATH] [--since DAYS] [--json] [--asset ETH|BTC]
```
> **Deviazione dal design originale**: il group `audit` era già usato per
> `audit verify` (hash chain dell'audit log). Il data-quality audit è
> stato inserito come sibling `audit data` per evitare collisione di
> namespace, anziché come comando top-level `cerbero-bite audit`.
- `--db PATH` (default `data/state.sqlite`): path al DB SQLite.
- `--since DAYS` (default `7`): finestra di analisi, retro dal `now()` corrente.
- `--json` (default off): stampa solo dump JSON serializzabile, niente tabelle umane.
- `--asset` (default `tutti`): filtra ad un singolo asset.
Exit code:
- `0`: audit completato (a prescindere dai problemi trovati).
- `2`: errori di connessione/DB (DB inesistente, schema mancante).
Niente exit code per "found issues": l'audit è informativo, decide
l'umano. Far diventare l'audit un gate CI è out of scope.
## Output
**Stdout (default):**
```
=== ETH — market_snapshots (last 7d, 2026-05-05 → 2026-05-12) ===
ticks: 672 expected: 672 coverage: 100.0%
gaps > 20min: 0
fetch_ok=0: 4 rows (max streak: 1)
null rate: dealer_net_gamma 2.1% oi_delta_pct_4h 0.3%
=== ETH — option_chain_snapshots (last 7d) ===
snapshots: 672 expected: 672 coverage: 100.0%
quotes/snap: median 55 p10 50 p90 60
bid > ask: 0
ask <= 0: 37 (missing ask side)
IV null: 12 quotes (0.03%)
=== BTC — ...
```
**JSON (`--json`):**
```json
{
"since": "2026-05-05T20:46:00+00:00",
"until": "2026-05-12T20:46:00+00:00",
"assets": {
"ETH": {
"market": {"expected_ticks": 672, "actual_ticks": 672, ...},
"chain": {"expected_snapshots": 672, ...}
},
"BTC": {...}
}
}
```
## Testing
`tests/unit/test_data_audit.py`. Per ogni funzione:
- DB temporaneo (`tmp_path`), schema migrato via `run_migrations`.
- Seed deterministico: insert manuali per riprodurre lo scenario.
- Test cases:
- market: copertura piena → 100%; un gap iniettato → conteggio gap=1;
streak `fetch_ok=0` lunga 3 → flagged.
- chain: snap mancante → expected actual = 1; quote dimezzate in
un tick → quotes/snap p10 cala; `bid=10 ask=5` → bid>ask=1.
- 0 dipendenze nuove (sqlite + pytest standard).
## Performance
Tabelle attuali: ~57k quote chain. Le query usano gli index
`idx_option_chain_asset_ts` e `(asset, timestamp)` di
`market_snapshots`. L'audit deve girare in < 2s su 7gg.
## Anti-goals (esplicito)
- Nessun salvataggio dei risultati nello stato del DB.
- Nessun trigger automatico (no cron job, no APScheduler).
- Nessun alert/notifica: stdout + JSON sono lo strumento, l'operatore
decide cosa farne.
- Nessun ML / detection di anomalie sofisticate. Soglie costanti.
## Open Questions
Nessuna al momento della scrittura. Eventuali punti emergeranno durante
l'implementazione e andranno annotati qui.
+5
View File
@@ -0,0 +1,5 @@
"""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.
"""
+336
View File
@@ -0,0 +1,336 @@
"""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 math
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
ask_zero_count: int = 0
iv_null_count: int = 0
iv_null_pct: Decimal = Decimal("0")
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 largest k with
# first_tick + k*15min < until is ceil(span/15) - 1, so the
# count is ceil(span_minutes / 15). floor() under-counts at
# aligned multiples and would mis-count non-aligned spans.
span_seconds = (until - first_tick).total_seconds()
return math.ceil(span_seconds / (_TICK_INTERVAL_MIN * 60))
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
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)
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,
)
def _pct(values: list[int], q: int) -> int:
"""Integer percentile via `statistics.quantiles` (inclusive).
Returns 0 on empty input; the single value on len==1. ``q`` is in 1..100.
"""
if not values:
return 0
if len(values) == 1:
return values[0]
sorted_vals = sorted(values)
if q >= 100:
return sorted_vals[-1]
cuts = statistics.quantiles(sorted_vals, n=100, method="inclusive")
return int(round(cuts[q - 1]))
def audit_option_chain(
conn: sqlite3.Connection,
*,
asset: str,
since: datetime,
until: datetime,
) -> ChainAuditReport:
"""Compute the option_chain_snapshots audit report in [since, until)."""
expected = _expected_ticks(since, until)
snap_counts = 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 = [r["n"] for r 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 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,
quotes_per_snap_median=median_q,
quotes_per_snap_p10=p10_q,
quotes_per_snap_p90=p90_q,
)
bid_gt_ask = 0
ask_zero = 0
iv_null = 0
for r in quote_rows:
bid_s, ask_s, iv_s = r["bid"], r["ask"], r["iv"]
ask_d: Decimal | None = None
if ask_s is not None:
try:
ask_d = Decimal(ask_s)
except (ValueError, ArithmeticError):
ask_d = None
if ask_d is not None and ask_d <= 0:
# Su Deribit ask=0 significa "nessun ordine in vendita",
# non bid>ask reale: contiamolo a parte.
ask_zero += 1
elif ask_d is not None and bid_s is not None:
try:
if Decimal(bid_s) > ask_d:
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
iv_null_pct = (
Decimal(iv_null) / 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,
ask_zero_count=ask_zero,
iv_null_count=iv_null,
iv_null_pct=iv_null_pct,
)
+174
View File
@@ -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,173 @@ 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,
"ask_zero_count": r.ask_zero_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" ask <= 0: {r.ask_zero_count} (missing ask side)")
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."""
@@ -181,19 +181,18 @@ async def collect_market_snapshot(
try:
with transaction(conn):
ctx.repository.record_market_snapshot(conn, record)
# Mirror ETH spot+DVOL into dvol_history so monitor_cycle's
# return_4h lookup has local samples even in data-only mode.
if (
record.asset == "ETH"
and record.spot is not None
and record.dvol is not None
):
# Mirror spot+DVOL into dvol_history (per asset) so
# monitor_cycle's return_4h lookup has local samples even
# in data-only mode. dvol_history enforces NOT NULL on
# dvol/spot so skip if either is missing.
if record.spot is not None and record.dvol is not None:
ctx.repository.record_dvol_snapshot(
conn,
DvolSnapshot(
timestamp=record.timestamp,
asset=record.asset,
dvol=record.dvol,
eth_spot=record.spot,
spot=record.spot,
),
)
finally:
+3 -3
View File
@@ -173,8 +173,8 @@ async def _fetch_return_4h(ctx: RuntimeContext, *, now: datetime) -> Decimal:
conn = connect_state(ctx.db_path)
try:
row = conn.execute(
"SELECT timestamp, eth_spot FROM dvol_history "
"WHERE timestamp <= ? AND timestamp >= ? "
"SELECT timestamp, spot FROM dvol_history "
"WHERE asset = 'ETH' AND timestamp <= ? AND timestamp >= ? "
"ORDER BY timestamp DESC LIMIT 1",
(cutoff.isoformat(), floor.isoformat()),
).fetchone()
@@ -239,7 +239,7 @@ async def run_monitor_cycle(
with transaction(conn):
ctx.repository.record_dvol_snapshot(
conn,
DvolSnapshot(timestamp=when, dvol=dvol, eth_spot=spot),
DvolSnapshot(timestamp=when, asset="ETH", dvol=dvol, spot=spot),
)
positions = ctx.repository.list_positions(conn, status="open")
finally:
@@ -0,0 +1,30 @@
-- 0006_dvol_history_multi_asset.sql — promote dvol_history to multi-asset
--
-- Original schema (0001_init.sql) treated dvol_history as ETH-only:
-- PRIMARY KEY (timestamp) and a column named eth_spot. With the
-- orchestrator now snapshotting BTC in addition to ETH (commit
-- e978a44), the table needs an asset dimension so we can store a
-- DVOL/spot sample per asset per tick.
--
-- Forward-only. The 1028 existing rows are all ETH (the only writer
-- was the ETH branch of market_snapshot_cycle) so we backfill
-- asset='ETH' before swapping the table in place.
CREATE TABLE dvol_history_v2 (
timestamp TEXT NOT NULL,
asset TEXT NOT NULL,
dvol NUMERIC NOT NULL,
spot NUMERIC NOT NULL,
PRIMARY KEY (timestamp, asset)
);
INSERT INTO dvol_history_v2(timestamp, asset, dvol, spot)
SELECT timestamp, 'ETH', dvol, eth_spot FROM dvol_history;
DROP TABLE dvol_history;
ALTER TABLE dvol_history_v2 RENAME TO dvol_history;
CREATE INDEX idx_dvol_history_asset_ts
ON dvol_history(asset, timestamp DESC);
PRAGMA user_version = 6;
+2 -1
View File
@@ -116,8 +116,9 @@ class DvolSnapshot(BaseModel):
model_config = ConfigDict(extra="forbid")
timestamp: datetime
asset: str # "ETH", "BTC"
dvol: Decimal
eth_spot: Decimal
spot: Decimal
class MarketSnapshotRecord(BaseModel):
+4 -3
View File
@@ -339,12 +339,13 @@ class Repository:
self, conn: sqlite3.Connection, snapshot: DvolSnapshot
) -> None:
conn.execute(
"INSERT OR REPLACE INTO dvol_history(timestamp, dvol, eth_spot) "
"VALUES (?,?,?)",
"INSERT OR REPLACE INTO dvol_history(timestamp, asset, dvol, spot) "
"VALUES (?,?,?,?)",
(
_enc_dt(snapshot.timestamp),
snapshot.asset,
_enc_dec(snapshot.dvol),
_enc_dec(snapshot.eth_spot),
_enc_dec(snapshot.spot),
),
)
+2 -1
View File
@@ -87,7 +87,8 @@ def _seed_dvol_history(ctx, *, when: datetime, spot: Decimal, dvol: Decimal):
try:
with transaction(conn):
ctx.repository.record_dvol_snapshot(
conn, DvolSnapshot(timestamp=when, dvol=dvol, eth_spot=spot)
conn,
DvolSnapshot(timestamp=when, asset="ETH", dvol=dvol, spot=spot),
)
finally:
conn.close()
+84
View File
@@ -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
+448
View File
@@ -0,0 +1,448 @@
"""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
from cerbero_bite.analysis.data_audit import (
ChainAuditReport,
MarketAuditReport,
audit_market_snapshots,
audit_option_chain,
)
from cerbero_bite.analysis.data_audit import ( # noqa: PLC2701
_detect_gaps,
_expected_ticks,
_max_zero_streak,
_pct,
)
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
def test_expected_ticks_non_multiple_span_uses_ceiling() -> None:
# Span = 16 min (12:00 → 12:16): ticks 12:00 AND 12:15 are < 12:16.
# floor(16/15) = 1 would under-count; the correct answer is 2.
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 16, tzinfo=UTC)
assert _expected_ticks(since, until) == 2
def test_expected_ticks_exact_quarter_boundary_is_excluded() -> None:
# Until landing exactly on a */15 tick: that tick is NOT counted
# because the window is half-open on the right.
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 45, tzinfo=UTC)
# Ticks at 12:00, 12:15, 12:30 → 3 (12:45 excluded)
assert _expected_ticks(since, until) == 3
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) == ()
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
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")
def test_pct_empty_returns_zero() -> None:
assert _pct([], 50) == 0
def test_pct_single_value_returns_that_value() -> None:
assert _pct([42], 50) == 42
assert _pct([42], 10) == 42
assert _pct([42], 90) == 42
def test_pct_median_odd_count() -> None:
assert _pct([10, 20, 30, 40, 50], 50) == 30
def test_pct_p10_p90_on_100_values() -> None:
# inclusive method on [1..100]:
# p10 interpolates to 10.9 → 11; p90 interpolates to 90.1 → 90.
values = list(range(1, 101))
assert _pct(values, 10) == 11
assert _pct(values, 90) == 90
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",
) -> 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) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(
ts.isoformat(),
asset,
instrument,
strike,
expiry.isoformat(),
option_type,
bid,
ask,
iv,
),
)
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-{i}-C",
strike=str(3000 + i * 50),
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert isinstance(report, ChainAuditReport)
assert report.asset == "ETH"
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.quotes_per_snap_p10 == 10
assert report.quotes_per_snap_p90 == 10
assert report.bid_gt_ask_count == 0
assert report.ask_zero_count == 0
assert report.iv_null_count == 0
assert report.iv_null_pct == Decimal("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)
# Crossed BBO: bid > ask
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-A",
bid="0.10",
ask="0.05",
)
# Missing IV
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-B",
strike="3050",
iv=None,
)
# Normal row (control)
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-C",
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.ask_zero_count == 0
assert report.iv_null_count == 1
# 1 of 3 quotes → 33.33%
assert report.iv_null_pct == Decimal("33.33")
def test_audit_chain_ask_zero_counted_separately(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)
# ask=0 → conta in ask_zero, NON in bid_gt_ask
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-A",
bid="0.05",
ask="0",
)
# ask negativo (impossibile economicamente) → ask_zero
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-B",
strike="3050",
bid="0.05",
ask="-0.01",
)
# Crossed BBO reale (ask>0) → bid_gt_ask
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-C",
strike="3100",
bid="0.10",
ask="0.05",
)
# Normal control
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-D",
strike="3150",
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.ask_zero_count == 2
assert report.bid_gt_ask_count == 1
def test_audit_chain_missing_snapshots(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")
def test_audit_chain_empty_window_returns_zero_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:
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.expected_snapshots == 4
assert report.actual_snapshots == 0
assert report.coverage_pct == Decimal("0")
assert report.quotes_per_snap_median == 0
assert report.bid_gt_ask_count == 0
assert report.iv_null_count == 0
+15 -9
View File
@@ -38,7 +38,9 @@ def _ctx(tmp_path: Path) -> MagicMock:
# Default: every feed succeeds with sane mock values.
ctx.deribit = MagicMock()
ctx.deribit.spot_perp_price = AsyncMock(return_value=Decimal("3000"))
ctx.deribit.spot_perp_price = AsyncMock(
side_effect=lambda asset: Decimal("65000") if asset == "BTC" else Decimal("3000")
)
ctx.deribit.latest_dvol = AsyncMock(return_value=Decimal("55"))
ctx.deribit.realized_vol = AsyncMock(
return_value={
@@ -181,31 +183,35 @@ def _read_dvol_history(ctx: MagicMock) -> list[dict]:
@pytest.mark.asyncio
async def test_eth_snapshot_mirrors_into_dvol_history(tmp_path: Path) -> None:
async def test_snapshot_mirrors_each_asset_into_dvol_history(tmp_path: Path) -> None:
ctx = _ctx(tmp_path)
await collect_market_snapshot(ctx, assets=("ETH", "BTC"), now=_now())
rows = _read_dvol_history(ctx)
assert len(rows) == 1
assert Decimal(str(rows[0]["dvol"])) == Decimal("55")
assert Decimal(str(rows[0]["eth_spot"])) == Decimal("3000")
by_asset = {r["asset"]: r for r in rows}
assert set(by_asset) == {"ETH", "BTC"}
assert Decimal(str(by_asset["ETH"]["spot"])) == Decimal("3000")
assert Decimal(str(by_asset["BTC"]["spot"])) == Decimal("65000")
@pytest.mark.asyncio
async def test_btc_only_snapshot_does_not_touch_dvol_history(
async def test_btc_only_snapshot_mirrors_into_dvol_history(
tmp_path: Path,
) -> None:
ctx = _ctx(tmp_path)
await collect_market_snapshot(ctx, assets=("BTC",), now=_now())
assert _read_dvol_history(ctx) == []
rows = _read_dvol_history(ctx)
assert len(rows) == 1
assert rows[0]["asset"] == "BTC"
assert Decimal(str(rows[0]["spot"])) == Decimal("65000")
@pytest.mark.asyncio
async def test_eth_snapshot_skips_dvol_history_when_dvol_missing(
async def test_snapshot_skips_dvol_history_when_dvol_missing(
tmp_path: Path,
) -> None:
ctx = _ctx(tmp_path)
ctx.deribit.latest_dvol = AsyncMock(side_effect=RuntimeError("no dvol"))
await collect_market_snapshot(ctx, assets=("ETH",), now=_now())
# market_snapshots row still persisted, but dvol_history must stay empty
# because its schema enforces NOT NULL on dvol/eth_spot.
# because its schema enforces NOT NULL on dvol/spot.
assert _read_dvol_history(ctx) == []
+33 -2
View File
@@ -277,15 +277,46 @@ def test_record_dvol_snapshot_replaces_on_duplicate_timestamp(
ts = datetime(2026, 4, 27, 14, 0, tzinfo=UTC)
with transaction(conn):
repo.record_dvol_snapshot(
conn, DvolSnapshot(timestamp=ts, dvol=Decimal("50"), eth_spot=Decimal("3000"))
conn,
DvolSnapshot(
timestamp=ts, asset="ETH", dvol=Decimal("50"), spot=Decimal("3000")
),
)
repo.record_dvol_snapshot(
conn, DvolSnapshot(timestamp=ts, dvol=Decimal("55"), eth_spot=Decimal("3050"))
conn,
DvolSnapshot(
timestamp=ts, asset="ETH", dvol=Decimal("55"), spot=Decimal("3050")
),
)
rows = conn.execute("SELECT COUNT(*) FROM dvol_history").fetchone()
assert rows[0] == 1
def test_record_dvol_snapshot_keeps_assets_distinct_on_same_timestamp(
conn: sqlite3.Connection, repo: Repository
) -> None:
ts = datetime(2026, 4, 27, 14, 0, tzinfo=UTC)
with transaction(conn):
repo.record_dvol_snapshot(
conn,
DvolSnapshot(
timestamp=ts, asset="ETH", dvol=Decimal("50"), spot=Decimal("3000")
),
)
repo.record_dvol_snapshot(
conn,
DvolSnapshot(
timestamp=ts, asset="BTC", dvol=Decimal("45"), spot=Decimal("65000")
),
)
rows = conn.execute(
"SELECT asset, dvol, spot FROM dvol_history ORDER BY asset"
).fetchall()
assert [r["asset"] for r in rows] == ["BTC", "ETH"]
assert Decimal(str(rows[0]["spot"])) == Decimal("65000")
assert Decimal(str(rows[1]["spot"])) == Decimal("3000")
def test_manual_action_enqueue_consume_cycle(
conn: sqlite3.Connection, repo: Repository
) -> None: