feat(config): profilo B credit-spread + blocco research_collector (v1.7.0)

Due modifiche di config, raggruppate perche' condividono i file YAML e il
config_hash (rigenerato sullo stato finale di ciascun profilo).

1) Profilo B credit-spread. L'analisi su 3.689 snapshot (1mag-8giu) mostra che
   con delta short 0.10-0.15 il miglior credit/width fisicamente ottenibile e'
   ~6%, quindi credit_to_width_ratio_min=0.30 era irraggiungibile: 0 spread
   fattibili. La sola leva efficace e' vendere piu' vicino.
   - short_strike: delta 0.12-0.22 (target 0.18), distance_otm_pct_min 0.10
   - spread_width.max_pct_of_spot 0.05 -> 0.06 (la griglia strike ETH a 100pt
     sotto i $1500 e' ~6% a spot <$2k: con cap 5% nessuna gamba long in banda)
   - credit_to_width_ratio_min 0.30 -> 0.08 (allineato al realizzabile)
   - aggressiva: ritarata anche la delta_by_dvol step-function
   Atteso: ~1,5 trade/mese (tetto dato da posizione-singola + tenuta 18g).
   Nota rischio: delta ~0.18-0.22 alza la prob. di assegnazione da ~12% a ~18-22%.

2) Blocco research_collector in entrambi i profili (enabled=false, opt-in):
   wiring del collettore full-chain introdotto nel commit precedente.

config_version 1.6.0 -> 1.7.0 (conservativo), 1.4.0 -> 1.5.0-aggressiva;
config_hash rigenerato e verificato stabile su entrambi.

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

docker-compose.yml

@@ -42,6 +42,9 @@ x-bite-env: &bite-env
   CERBERO_BITE_MCP_HYPERLIQUID_URL: ${CERBERO_BITE_MCP_HYPERLIQUID_URL:-http://cerbero-mcp:9000/mcp-hyperliquid}
   CERBERO_BITE_MCP_MACRO_URL: ${CERBERO_BITE_MCP_MACRO_URL:-http://cerbero-mcp:9000/mcp-macro}
   CERBERO_BITE_MCP_SENTIMENT_URL: ${CERBERO_BITE_MCP_SENTIMENT_URL:-http://cerbero-mcp:9000/mcp-sentiment}
+  # MCP V2 unified interface (/mcp): get_instruments, get_historical,
+  # get_indicators — removed from the per-exchange routers.
+  CERBERO_BITE_MCP_UNIFIED_URL: ${CERBERO_BITE_MCP_UNIFIED_URL:-http://cerbero-mcp:9000/mcp}
 
 services:
   cerbero-bite:
@@ -129,4 +132,4 @@ services:
       - traefik.http.routers.cerbero-bite.entrypoints=websecure
       - traefik.http.routers.cerbero-bite.tls.certresolver=mytlschallenge
       - traefik.http.services.cerbero-bite.loadbalancer.server.port=8765
-      - com.centurylinklabs.watchtower.enable=true
+      - com.centurylinklabs.watchtower.enable=false

docs/02-architecture.md

@@ -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(...)

docs/05-data-model.md

@@ -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
+Snapshot DVOL + spot per asset ad ogni evaluation. Utile per il
-`return_4h` durante il monitor (vedi `runtime/monitor_cycle.py
+calcolo di `return_4h` durante il monitor (vedi
-_fetch_return_4h`) e per analisi post-mortem.
+`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`

docs/06-operational-flow.md

@@ -156,6 +156,14 @@ Trigger: ogni 5 minuti.
 Il dead-man (`scripts/dead_man.sh`) sorveglia che `HEALTH_OK` venga
 scritto: silenzio > 15 min → kill switch via SQLite e alert.
 
+> **Nota — la tolleranza "3 fallimenti" vale solo per i probe del
+> health check.** Tutti i tick schedulati (entry, monitor, snapshot,
+> …) girano dentro `orchestrator._safe`, che escala **qualsiasi**
+> eccezione non gestita a `alert_manager.critical` → kill switch
+> armato al **primo** errore, anche transitorio. È il percorso che ha
+> fermato il bot il 31/05/2026 (singolo `McpTimeoutError` nel tick
+> entry). Vedi la tabella trigger in `07-risk-controls.md`.
+
 ## Flusso 5b — Manual actions consumer
 
 Trigger: cron `*/1 * * * *` (job APScheduler `manual_actions`).

docs/07-risk-controls.md

@@ -36,26 +36,51 @@ infrastrutturali o decisioni umane fuori posto.
 | MCP `cerbero-deribit` non risponde per 3 health check consecutivi | Sì | `runtime/health_check.py` | Severity HIGH |
 | MCP `cerbero-macro` / `cerbero-hyperliquid` / `cerbero-sentiment` non risponde per 3 health check consecutivi | Sì | `runtime/health_check.py` | Severity HIGH |
 | `mcp-deribit.environment_info.environment` ≠ `strategy.execution.environment` | Sì | `runtime/orchestrator.boot` + health check | Severity CRITICAL al boot, HIGH a runtime |
-| Mismatch tra il tail del file `data/audit.log` e `system_state.last_audit_hash` (truncation o tampering) | Sì | `runtime/orchestrator._verify_audit_anchor` | Severity CRITICAL al boot |
+| Mismatch tra il tail del file `data/audit.log` e `system_state.last_audit_hash` (truncation o tampering) | Sì | `runtime/orchestrator._verify_audit_anchor` | Severity CRITICAL al boot. Dal fix `647e3e5` il **lag benigno** dell'anchor (anchor indietro rispetto al tail, ma antenato genuino — vedi `safety/audit_log.tail_continues_from`) è tollerato e ri-sincronizzato senza armare |
+| Eccezione non gestita in **qualsiasi** tick schedulato (`entry`, `monitor`, `health`, snapshot, backup, …) | Sì | `orchestrator._safe` → `alert_manager.critical` | Severity CRITICAL. Attenzione: anche un errore **transitorio una tantum** arma in modo permanente. Incidente 31/05/2026: un singolo `McpTimeoutError` su `sentiment.get_cross_exchange_funding` nel tick entry → bot fermo 6 giorni nonostante il servizio si fosse ripreso in pochi secondi. Possibile hardening futuro: tolleranza N-consecutivi o auto-pause temporanea per errori MCP transitori |
 | Stato SQLite incoerente con il broker (recovery non risolutivo) | Sì | `runtime/recovery.py` | Severity CRITICAL al boot |
 | `place_combo_order` di chiusura respinto dal broker | Sì | `runtime/monitor_cycle.py` | Severity CRITICAL; la posizione torna in `open` per ritentare |
 | `place_combo_order` di apertura respinto dal broker | Sì | `runtime/entry_cycle.py` | Severity HIGH; la posizione viene marcata `cancelled` |
-| Hash chain audit non verifica (`audit verify` fallisce) | Manuale per ora; CLI `audit verify` segnala l'anomalia con exit 2 | `cli.py audit verify` + `safety/audit_log.verify_chain` | Severity CRITICAL quando integrata nel boot |
+| Hash chain audit non verifica (`audit verify` fallisce) | Manuale per ora; CLI `audit verify` segnala l'anomalia con exit 2 | `cli.py audit verify` + `safety/audit_log.verify_chain` | ⚠️ Il log di produzione ha 7 gap storici benigni (vedi «Limiti noti» sotto): `audit verify` full-chain fallisce alla riga 11 by design. Non integrare il full-chain verify nel boot senza prima ruotare il log con un nuovo genesis |
 | Comando manuale via `cerbero-bite kill-switch arm` | Sì | `cli.py kill_switch_arm` | Severity HIGH (operator-initiated) |
 
 ### Disarm
 
+Due percorsi, a seconda che l'engine sia in esecuzione o fermo.
+
+**Engine in esecuzione → coda `manual_actions` (raccomandato).**
+Si accoda una riga `kind="disarm_kill"` (dalla GUI, o a mano via
+SQLite); il job `manual_actions` (cron `*/1`) la consuma e chiama
+`KillSwitch.disarm` **in-process** entro un minuto:
+
+```sql
+INSERT INTO manual_actions (kind, payload_json, created_at)
+VALUES ('disarm_kill', '{"reason": "<motivo>"}',
+        strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
+```
+
+**Engine fermo → CLI.**
+
 ```bash
 cerbero-bite kill-switch disarm --reason "<motivo>" \
     --db data/state.sqlite \
     --audit data/audit.log
 ```
 
+⚠️ **Non usare il CLI con l'engine in esecuzione**: il CLI appende
+all'audit log da un processo separato e va in race con gli append
+dell'engine — il risultato è un fork/gap permanente nella hash chain
+(`prev_hash` che non aggancia la riga precedente). I 7 gap storici
+del log (01/05 e 29/05/2026, vedi «Limiti noti» sotto) sono stati
+causati esattamente da questo pattern. Il percorso a coda non ha il
+problema perché l'unico writer resta l'engine.
+
 L'operazione è transazionale: SQLite `system_state.kill_switch = 0` +
 una linea `KILL_SWITCH_DISARMED` nella audit chain con il motivo. Il
 disarm non riavvia automaticamente lo scheduler; è il prossimo tick
 naturale (entry giornaliero o monitor 12h) a far ripartire la
-decisione.
+decisione. Il disarm **persiste attraverso i restart** del container,
+così come l'arm (verificato 29/05 e 05/06/2026).
 
 ## Cap di rischio (oltre alle regole di strategia)
 
@@ -156,6 +181,42 @@ il tail del file: in caso di mismatch (truncation, sostituzione, file
 mancante) viene armato il kill switch CRITICAL prima che qualsiasi
 ciclo trading parta.
 
+Dal fix `647e3e5` (29/05/2026) il check distingue il **lag benigno**
+dal tampering: se l'anchor persistito è indietro rispetto al tail ma è
+un *antenato genuino* (la chain dall'anchor al tail verifica —
+`safety/audit_log.tail_continues_from`), il boot ri-sincronizza
+l'anchor e prosegue senza armare. L'anchor è infatti persistito
+best-effort e può restare indietro sotto write contention SQLite. Un
+anchor assente dal file o una chain post-anchor rotta restano
+tampering e armano CRITICAL (verificato con test negativo il 29/05).
+
+### Limiti noti: gap da scritture concorrenti
+
+La chain assume un **single writer** (l'engine). Un processo separato
+che appende mentre l'engine gira (CLI `kill-switch arm/disarm`,
+script di resync) legge il tail, calcola `prev_hash` e scrive — ma se
+l'engine appende nel frattempo, una riga viene persa o la chain si
+biforca: la riga successiva ha un `prev_hash` che non aggancia nulla.
+
+Il file `data/audit.log` di produzione contiene **7 gap storici di
+questo tipo, tutti benigni e attribuiti** (verifica completa del
+05/06/2026):
+
+| Righe | Data | Causa |
+|---|---|---|
+| 11, 16 | 01/05/2026, primo boot | write contention durante il caos env mismatch testnet/mainnet |
+| 8130, 8262, 8287 | 29/05/2026 | restart/resync manuali durante il debug anchor (8287 ha persino timestamp fuori ordine) |
+| 8301, 8323 | 29/05/2026 | disarm via CLI con engine in esecuzione (race CLI ↔ engine) |
+
+Conseguenza operativa: `cerbero-bite audit verify` (full-chain dal
+genesis) fallisce **per sempre** alla riga 11 — è atteso, non è
+tampering. Il controllo operativo in vigore è quello dell'anchor al
+boot (tail-continuity), che resta pienamente efficace per truncation
+e tampering del tail. Eventuali gap **nuovi** (oltre ai 7 elencati)
+vanno invece investigati. Mitigazione: usare la coda `manual_actions`
+per arm/disarm a engine acceso (mai il CLI), così l'unico writer
+resta l'engine.
+
 ## Dry-run mode
 
 Il comando `cerbero-bite dry-run --cycle entry|monitor|health` esegue

docs/superpowers/specs/2026-05-12-data-quality-audit-design.md

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

src/cerbero_bite/analysis/__init__.py

@@ -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.
+"""

src/cerbero_bite/analysis/data_audit.py

@@ -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,
+    )

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

src/cerbero_bite/clients/deribit.py

@@ -120,12 +120,29 @@ def _to_decimal(value: Any) -> Decimal | None:
 class DeribitClient:
     SERVICE = "deribit"
 
-    def __init__(self, http: HttpToolClient) -> None:
+    def __init__(
+        self, http: HttpToolClient, unified: HttpToolClient | None = None
+    ) -> None:
         if http.service != self.SERVICE:
             raise ValueError(
                 f"DeribitClient requires service '{self.SERVICE}', got '{http.service}'"
             )
         self._http = http
+        # Cerbero MCP V2 moved the data tools (get_instruments,
+        # get_historical, get_indicators) to the unified ``/mcp`` router;
+        # they are no longer on ``/mcp-deribit``. This second client points
+        # at ``/mcp`` and is only needed by those three methods — diagnostic
+        # paths (e.g. ``environment_info`` for ping) leave it ``None``.
+        self._unified = unified
+
+    def _unified_client(self, tool: str) -> HttpToolClient:
+        if self._unified is None:
+            raise McpDataAnomalyError(
+                f"unified MCP client not configured; '{tool}' lives on /mcp",
+                service=self.SERVICE,
+                tool=tool,
+            )
+        return self._unified
 
     # ------------------------------------------------------------------
     # Environment / health
@@ -206,7 +223,16 @@ class DeribitClient:
         limit: int = 500,
     ) -> list[InstrumentMeta]:
         """Return option instruments matching the filters as typed metadata."""
-        body: dict[str, Any] = {"currency": currency, "kind": "option", "limit": limit}
+        # MCP V2: get_instruments is unified-only (/mcp). The venue is
+        # selected via ``exchange`` and each row is normalized — the native
+        # symbol is ``symbol`` and Deribit-specific fields live under
+        # ``native``.
+        body: dict[str, Any] = {
+            "exchange": self.SERVICE,
+            "currency": currency,
+            "kind": "option",
+            "limit": limit,
+        }
         if expiry_from is not None:
             body["expiry_from"] = expiry_from.date().isoformat()
         if expiry_to is not None:
@@ -214,28 +240,31 @@ class DeribitClient:
         if min_open_interest is not None:
             body["min_open_interest"] = min_open_interest
 
-        raw = await self._http.call("get_instruments", body)
+        raw = await self._unified_client("get_instruments").call(
+            "get_instruments", body
+        )
         instruments = raw.get("instruments") or []
         out: list[InstrumentMeta] = []
         for entry in instruments:
             if not isinstance(entry, dict):
                 continue
-            name = entry.get("name")
+            name = entry.get("symbol")
             if not isinstance(name, str):
                 continue
             try:
                 strike, expiry, option_type = _parse_instrument(name)
             except McpDataAnomalyError:
                 continue
+            native = entry.get("native") if isinstance(entry.get("native"), dict) else {}
             out.append(
                 InstrumentMeta(
                     name=name,
                     strike=strike,
                     expiry=expiry,
                     option_type=option_type,
-                    open_interest=_to_decimal(entry.get("open_interest")),
+                    open_interest=_to_decimal(native.get("open_interest")),
                     tick_size=_to_decimal(entry.get("tick_size")),
-                    min_trade_amount=_to_decimal(entry.get("min_trade_amount")),
+                    min_trade_amount=_to_decimal(native.get("min_trade_amount")),
                 )
             )
         return out
@@ -291,13 +320,17 @@ class DeribitClient:
         in :func:`compute_bias`. Returns ``None`` when the chain has no
         data in the window.
         """
-        raw = await self._http.call(
+        # MCP V2: get_historical is unified-only (/mcp); it takes
+        # ``exchange`` + lowercase ``interval`` (e.g. "1d", "1h") instead of
+        # the old per-exchange ``resolution``.
+        raw = await self._unified_client("get_historical").call(
             "get_historical",
             {
+                "exchange": self.SERVICE,
                 "instrument": instrument,
                 "start_date": start.date().isoformat(),
                 "end_date": end.date().isoformat(),
-                "resolution": resolution,
+                "interval": resolution.lower(),
             },
         )
         candles = (raw or {}).get("candles") or []
@@ -422,28 +455,34 @@ class DeribitClient:
         resolution: str = "1h",
     ) -> Decimal | None:
         """Return the most recent ADX(14) value, or ``None`` when missing."""
-        raw = await self._http.call(
+        # MCP V2: indicators are computed by the unified ``get_indicators``
-            "get_technical_indicators",
+        # tool (/mcp), which replaced the per-exchange
+        # ``get_technical_indicators``. Body takes ``exchange`` + lowercase
+        # ``interval``; the response nests each indicator under
+        # ``indicators`` (ADX as ``{"adx": <float>, "+di":…, "-di":…}``).
+        raw = await self._unified_client("get_indicators").call(
+            "get_indicators",
             {
+                "exchange": self.SERVICE,
                 "instrument": instrument,
                 "indicators": ["adx"],
                 "start_date": start.date().isoformat(),
                 "end_date": end.date().isoformat(),
-                "resolution": resolution,
+                "interval": resolution.lower(),
             },
         )
         if not isinstance(raw, dict):
             return None
-        # The MCP server returns either a top-level dict with the
+        indicators = raw.get("indicators")
-        # indicator keyed by name, or a list of points. Be tolerant.
+        if not isinstance(indicators, dict):
-        adx_payload = raw.get("adx") or raw.get("ADX") or raw.get("indicators", {})
+            return None
-        if isinstance(adx_payload, list) and adx_payload:
+        adx_block = indicators.get("adx")
-            tail = adx_payload[-1]
+        if isinstance(adx_block, dict):
-            value = tail.get("value") if isinstance(tail, dict) else tail
+            value = adx_block.get("adx")
-            return None if value is None else Decimal(str(value))
-        if isinstance(adx_payload, dict):
-            value = adx_payload.get("latest") or adx_payload.get("value")
             return None if value is None else Decimal(str(value))
+        # Tolerate a flattened scalar shape just in case.
+        if adx_block is not None and not isinstance(adx_block, list):
+            return Decimal(str(adx_block))
         return None
 
     async def get_account_summary(self, currency: str = "USDC") -> dict[str, Any]:

src/cerbero_bite/config/mcp_endpoints.py

@@ -62,6 +62,14 @@ DEFAULT_ENDPOINTS: dict[str, str] = {
 }
 
 
+# Cerbero MCP V2 unified interface (``/mcp``). The data tools
+# ``get_instruments`` / ``get_historical`` / ``get_indicators`` live here
+# ONLY — they were removed from the per-exchange routers in MCP V2. The
+# caller passes ``exchange="deribit"`` in the body to target one venue.
+_UNIFIED_ENV = "CERBERO_BITE_MCP_UNIFIED_URL"
+_DEFAULT_UNIFIED_URL = "http://cerbero-mcp:9000/mcp"
+
+
 @dataclass(frozen=True)
 class McpEndpoints:
     """Resolved per-service URLs."""
@@ -70,6 +78,7 @@ class McpEndpoints:
     hyperliquid: str
     macro: str
     sentiment: str
+    unified: str = _DEFAULT_UNIFIED_URL
 
     def for_service(self, name: str) -> str:
         try:
@@ -85,6 +94,10 @@ def load_endpoints(env: dict[str, str] | None = None) -> McpEndpoints:
     for name, (host, port, env_var) in MCP_SERVICES.items():
         override = e.get(env_var)
         resolved[name] = override.rstrip("/") if override else _default_url(host, port)
+    unified_override = e.get(_UNIFIED_ENV)
+    resolved["unified"] = (
+        unified_override.rstrip("/") if unified_override else _DEFAULT_UNIFIED_URL
+    )
     return McpEndpoints(**resolved)
 
 

src/cerbero_bite/config/schema.py

@@ -59,6 +59,11 @@ class EntryConfig(BaseModel):
     no_position_concurrent: bool = True
     exclude_macro_severity: list[str] = Field(default_factory=lambda: ["high"])
     exclude_macro_countries: list[str] = Field(default_factory=lambda: ["US", "EU"])
+    # Finestra (giorni) entro cui un evento macro high-severity blocca
+    # l'entry. Disaccoppiata da `structure.dte_target`: il filtro macro
+    # è una protezione di rischio evento, indipendente dalla scadenza
+    # scelta per le opzioni. Default 18 = comportamento storico.
+    exclude_macro_within_days: int = 18
 
     # directional bias (§3.1)
     trend_window_days: int = 30
@@ -354,6 +359,31 @@ class McpConfig(_LooseSection): ...
 class TelegramConfig(_LooseSection): ...
 
 
+class ResearchCollectorConfig(BaseModel):
+    """Collettore *research* full-chain (§13-bis).
+
+    Indipendente dal collettore operativo: cattura ogni ciclo tutte le
+    scadenze liquide entro ``expiry_max_days`` ed entrambe le ali,
+    popolando ``book_depth_top3`` (1 call orderbook/strumento). Trasforma
+    il dataset da "skew/premi medi" a backtest opzioni per-trade e
+    standing. Disabilitato di default: ha un costo API non trascurabile.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    enabled: bool = False
+    cron: str = "0 * * * *"  # orario, indipendente dal */15 del live
+    expiry_max_days: int = 95  # 1g..3mesi
+    # None = nessun filtro moneyness (catena completa, entrambe le ali).
+    # Valorizzato (es. 0.30) = tiene solo gli strike entro ±band dallo spot.
+    moneyness_band_pct: Decimal | None = None
+    open_interest_min: int = 100
+    fetch_book_depth: bool = True
+    # Concorrenza max delle call orderbook depth (bound sul rate-limit).
+    book_depth_concurrency: int = 8
+    assets: list[str] = Field(default_factory=lambda: ["ETH", "BTC"])
+
+
 # ---------------------------------------------------------------------------
 # Root
 # ---------------------------------------------------------------------------
@@ -378,6 +408,10 @@ class StrategyConfig(BaseModel):
     kelly_recalibration: KellyConfig = Field(default_factory=KellyConfig)
     auto_pause: AutoPauseConfig = Field(default_factory=AutoPauseConfig)
 
+    research_collector: ResearchCollectorConfig = Field(
+        default_factory=ResearchCollectorConfig
+    )
+
     execution: ExecutionConfig = Field(default_factory=ExecutionConfig)
     monitoring: MonitoringConfig = Field(default_factory=MonitoringConfig)
     storage: StorageConfig = Field(default_factory=StorageConfig)

src/cerbero_bite/core/entry_validator.py

@@ -101,7 +101,6 @@ def validate_entry(ctx: EntryContext, cfg: StrategyConfig) -> EntryDecision:
     """
     reasons: list[str] = []
     entry_cfg = cfg.entry
-    structure_cfg = cfg.structure
 
     if ctx.has_open_position:
         reasons.append("open position already exists")
@@ -118,7 +117,7 @@ def validate_entry(ctx: EntryContext, cfg: StrategyConfig) -> EntryDecision:
 
     if (
         ctx.next_macro_event_in_days is not None
-        and ctx.next_macro_event_in_days <= structure_cfg.dte_target
+        and ctx.next_macro_event_in_days <= entry_cfg.exclude_macro_within_days
     ):
         reasons.append(
             f"macro event within DTE window ({ctx.next_macro_event_in_days} days)"

src/cerbero_bite/gui/live_data.py

@@ -191,7 +191,7 @@ async def _fetch_balances_async(*, timeout_s: float = 8.0) -> BalancesSnapshot:
                 client=http_client,
             )
 
-        deribit = DeribitClient(_client("deribit"))
+        deribit = DeribitClient(_client("deribit"), _client("unified"))
         hl = HyperliquidClient(_client("hyperliquid"))
         macro = MacroClient(_client("macro"))
 

src/cerbero_bite/runtime/dependencies.py

@@ -158,7 +158,7 @@ def build_runtime(
         telegram=telegram, audit_log=audit_log, kill_switch=kill_switch
     )
 
-    deribit = DeribitClient(_client("deribit"))
+    deribit = DeribitClient(_client("deribit"), _client("unified"))
     macro = MacroClient(_client("macro"))
     sentiment = SentimentClient(_client("sentiment"))
     hyperliquid = HyperliquidClient(_client("hyperliquid"))

src/cerbero_bite/runtime/entry_cycle.py

@@ -145,7 +145,7 @@ async def _gather_snapshot(
     )
     macro_t: asyncio.Task[int | None] = asyncio.create_task(
         macro.next_high_severity_within(
-            days=cfg.structure.dte_target,
+            days=cfg.entry.exclude_macro_within_days,
             countries=list(cfg.entry.exclude_macro_countries),
             now=now,
         )

src/cerbero_bite/runtime/market_snapshot_cycle.py

@@ -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
+                    # Mirror spot+DVOL into dvol_history (per asset) so
-                    # return_4h lookup has local samples even in data-only mode.
+                    # monitor_cycle's return_4h lookup has local samples even
-                    if (
+                    # in data-only mode. dvol_history enforces NOT NULL on
-                        record.asset == "ETH"
+                    # dvol/spot so skip if either is missing.
-                        and record.spot is not None
+                    if record.spot is not None and record.dvol 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:

src/cerbero_bite/runtime/monitor_cycle.py

@@ -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 "
+            "SELECT timestamp, spot FROM dvol_history "
-            "WHERE timestamp <= ? AND timestamp >= ? "
+            "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:

src/cerbero_bite/runtime/option_chain_research_cycle.py

@@ -0,0 +1,194 @@
+"""Full-chain *research* option collector (§13-bis).
+
+Diverso dal collettore operativo (`option_chain_snapshot_cycle`):
+
+* finestra scadenze ``[now, now + expiry_max_days]`` — cattura TUTTE le
+  scadenze liquide (1g/1sett/2sett/1mese/3mesi), non solo quella nella
+  finestra DTE della strategia;
+* entrambe le ali (nessun filtro moneyness di default), oppure entro
+  ``±moneyness_band_pct`` se configurato;
+* popola ``book_depth_top3`` chiamando l'orderbook per ogni strumento
+  tenuto (1 call/strumento, concorrenza limitata da
+  ``book_depth_concurrency``) — così lo slippage reale è modellabile;
+* scrive con ``source='research'`` per non confondersi con le righe
+  'live'.
+
+Questo trasforma il dataset da "skew/premi medi" a backtest opzioni
+vero, per-trade e standing. Best-effort come l'altro collettore: un
+batch o un orderbook che falliscono non invalidano il resto.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from typing import TYPE_CHECKING, Any
+
+from cerbero_bite.state import connect, transaction
+from cerbero_bite.state.models import OptionChainQuoteRecord
+
+from cerbero_bite.runtime.option_chain_snapshot_cycle import (
+    DEFAULT_BATCH_SIZE,
+    _fetch_tickers_in_batches,
+    _to_decimal_or_none,
+)
+
+if TYPE_CHECKING:
+    from cerbero_bite.runtime.dependencies import RuntimeContext
+
+__all__ = ["collect_option_chain_research"]
+
+_log = logging.getLogger("cerbero_bite.runtime.option_chain_research")
+
+
+def _underlying_price(ticker: dict[str, Any]) -> Decimal | None:
+    """Spot/index dell'underlying dal ticker, per il filtro moneyness."""
+    for key in ("underlying_price", "index_price", "estimated_delivery_price"):
+        val = _to_decimal_or_none(ticker.get(key))
+        if val is not None and val > 0:
+            return val
+    return None
+
+
+async def _depth_for(
+    ctx: RuntimeContext, name: str, sem: asyncio.Semaphore
+) -> int | None:
+    """Best-effort top-3 book depth per ``name`` (None se fallisce)."""
+    async with sem:
+        try:
+            return await ctx.deribit.orderbook_depth_top3(name)
+        except Exception as exc:
+            _log.debug("orderbook_depth_top3 failed for %s: %s", name, exc)
+            return None
+
+
+async def collect_option_chain_research(
+    ctx: RuntimeContext,
+    *,
+    asset: str = "ETH",
+    now: datetime | None = None,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> int:
+    """Collect + persist un singolo snapshot full-chain ``research`` per
+    ``asset``. Ritorna il numero di quote persistiti (0 su fallimento
+    best-effort o se il collettore è disabilitato)."""
+    rc = getattr(ctx.cfg, "research_collector", None)
+    if rc is None or not rc.enabled:
+        return 0
+
+    when = (now or datetime.now(UTC)).astimezone(UTC)
+
+    expiry_from = when
+    expiry_to = when + timedelta(days=rc.expiry_max_days)
+
+    try:
+        chain = await ctx.deribit.options_chain(
+            currency=asset.upper(),
+            expiry_from=expiry_from,
+            expiry_to=expiry_to,
+            min_open_interest=int(rc.open_interest_min),
+        )
+    except Exception:
+        _log.exception("research option chain fetch failed")
+        return 0
+
+    if not chain:
+        _log.info("research option chain empty for %s in window", asset)
+        return 0
+
+    names = [meta.name for meta in chain]
+    tickers = await _fetch_tickers_in_batches(ctx, names, batch_size=batch_size)
+
+    band = rc.moneyness_band_pct  # Decimal | None
+
+    # 1) costruisci i quote, applicando l'eventuale filtro moneyness.
+    kept: list[tuple[OptionChainQuoteRecord, dict[str, Any] | None]] = []
+    for meta in chain:
+        ticker = tickers.get(meta.name)
+
+        if band is not None and ticker is not None:
+            spot = _underlying_price(ticker)
+            if spot is not None:
+                moneyness = abs(meta.strike - spot) / spot
+                if moneyness > band:
+                    continue  # fuori dall'ala richiesta
+
+        if ticker is None:
+            rec = OptionChainQuoteRecord(
+                timestamp=when,
+                asset=asset.upper(),
+                instrument_name=meta.name,
+                strike=meta.strike,
+                expiry=meta.expiry,
+                option_type=meta.option_type,
+                open_interest=int(meta.open_interest)
+                if meta.open_interest is not None
+                else None,
+                source="research",
+            )
+            kept.append((rec, None))
+            continue
+
+        greeks = ticker.get("greeks") or {}
+        rec = OptionChainQuoteRecord(
+            timestamp=when,
+            asset=asset.upper(),
+            instrument_name=meta.name,
+            strike=meta.strike,
+            expiry=meta.expiry,
+            option_type=meta.option_type,
+            bid=_to_decimal_or_none(ticker.get("bid")),
+            ask=_to_decimal_or_none(ticker.get("ask")),
+            mid=_to_decimal_or_none(ticker.get("mark_price")),
+            iv=_to_decimal_or_none(ticker.get("mark_iv")),
+            delta=_to_decimal_or_none(greeks.get("delta")),
+            gamma=_to_decimal_or_none(greeks.get("gamma")),
+            theta=_to_decimal_or_none(greeks.get("theta")),
+            vega=_to_decimal_or_none(greeks.get("vega")),
+            open_interest=int(meta.open_interest)
+            if meta.open_interest is not None
+            else None,
+            volume_24h=(
+                int(ticker["volume_24h"])
+                if ticker.get("volume_24h") is not None
+                else None
+            ),
+            source="research",
+        )
+        kept.append((rec, ticker))
+
+    # 2) popola book_depth_top3 (concorrenza limitata) sugli strumenti tenuti.
+    if rc.fetch_book_depth and kept:
+        sem = asyncio.Semaphore(max(1, int(rc.book_depth_concurrency)))
+        depths = await asyncio.gather(
+            *(_depth_for(ctx, rec.instrument_name, sem) for rec, _ in kept)
+        )
+        kept = [
+            (rec.model_copy(update={"book_depth_top3": depth}), tk)
+            for (rec, tk), depth in zip(kept, depths, strict=True)
+        ]
+
+    quotes = [rec for rec, _ in kept]
+
+    persisted = 0
+    try:
+        conn = connect(ctx.db_path)
+        try:
+            with transaction(conn):
+                persisted = ctx.repository.record_option_chain_snapshot(conn, quotes)
+        finally:
+            conn.close()
+    except Exception:
+        _log.exception("persist research option chain snapshot failed")
+        return 0
+
+    _log.info(
+        "option_chain_research persisted %d quote(s) for %s (%d expiries window<=%dd)",
+        persisted,
+        asset.upper(),
+        len({q.expiry for q in quotes}),
+        rc.expiry_max_days,
+    )
+    return persisted

src/cerbero_bite/runtime/orchestrator.py

@@ -37,10 +37,15 @@ from cerbero_bite.runtime.market_snapshot_cycle import (
 from cerbero_bite.runtime.option_chain_snapshot_cycle import (
     collect_option_chain_snapshot,
 )
+from cerbero_bite.runtime.option_chain_research_cycle import (
+    collect_option_chain_research,
+)
 from cerbero_bite.runtime.monitor_cycle import MonitorCycleResult, run_monitor_cycle
 from cerbero_bite.runtime.recovery import recover_state
 from cerbero_bite.runtime.scheduler import JobSpec, build_scheduler
+from cerbero_bite.safety.audit_log import tail_continues_from
 from cerbero_bite.state import connect as connect_state
+from cerbero_bite.state import transaction
 
 __all__ = ["Orchestrator"]
 
@@ -50,8 +55,8 @@ _log = logging.getLogger("cerbero_bite.runtime.orchestrator")
 Environment = Literal["testnet", "mainnet"]
 
 # Default cron schedule (matches docs/06-operational-flow.md table).
-_CRON_ENTRY = "0 14 * * *"  # crypto 24/7: candidatura giornaliera; i gate decidono se entrare
+_CRON_ENTRY = "0 */2 * * *"  # crypto 24/7: valutazione ogni 2h; i gate decidono se entrare
-_CRON_MONITOR = "0 2,14 * * *"
+_CRON_MONITOR = "0 * * * *"  # stop/take-profit check ogni ora
 _CRON_HEALTH = "*/5 * * * *"
 _CRON_BACKUP = "0 * * * *"
 _CRON_MANUAL_ACTIONS = "*/1 * * * *"
@@ -158,16 +163,39 @@ class Orchestrator:
         if state is None or state.last_audit_hash is None:
             return  # first boot, nothing to compare against
         actual_tail = self._ctx.audit_log.last_hash
-        if actual_tail != state.last_audit_hash:
+        if actual_tail == state.last_audit_hash:
-            await self._ctx.alert_manager.critical(
+            return
-                source="orchestrator.boot",
+        # The anchor is persisted best-effort (see build_runtime): under
-                message=(
+        # SQLite write contention the mirror can fall behind the log while
-                    f"audit log anchor mismatch: anchor="
+        # the log itself keeps growing forward, intact. Treat that benign
-                    f"{state.last_audit_hash[:12]}…, file tail="
+        # lag — anchor is a valid ancestor of the current tail — as a
-                    f"{actual_tail[:12]}… — possible tampering or truncation"
+        # re-sync, not tampering. Only a missing anchor or a broken
-                ),
+        # post-anchor chain (truncation/tampering) arms the kill switch.
-                component="safety.audit_log",
+        if tail_continues_from(self._ctx.audit_log.path, state.last_audit_hash):
+            conn = connect_state(self._ctx.db_path)
+            try:
+                with transaction(conn):
+                    self._ctx.repository.set_last_audit_hash(
+                        conn, hex_hash=actual_tail
+                    )
+            finally:
+                conn.close()
+            _log.warning(
+                "audit anchor lagged behind tail; re-synced "
+                "(anchor=%s…, tail=%s…)",
+                state.last_audit_hash[:12],
+                actual_tail[:12],
             )
+            return
+        await self._ctx.alert_manager.critical(
+            source="orchestrator.boot",
+            message=(
+                f"audit log anchor mismatch: anchor="
+                f"{state.last_audit_hash[:12]}…, file tail="
+                f"{actual_tail[:12]}… — possible tampering or truncation"
+            ),
+            component="safety.audit_log",
+        )
 
     async def run_entry(
         self, *, now: datetime | None = None
@@ -295,6 +323,13 @@ class Orchestrator:
 
             await _safe("option_chain_snapshot", _do)
 
+        async def _option_chain_research() -> None:
+            async def _do() -> None:
+                for asset in self._ctx.cfg.research_collector.assets:
+                    await collect_option_chain_research(self._ctx, asset=asset)
+
+            await _safe("option_chain_research", _do)
+
         jobs: list[JobSpec] = [
             JobSpec(name="health", cron=health_cron, coro_factory=_health),
             JobSpec(name="backup", cron=backup_cron, coro_factory=_backup),
@@ -329,6 +364,21 @@ class Orchestrator:
                     coro_factory=_option_chain_snapshot,
                 )
             )
+            rc = self._ctx.cfg.research_collector
+            if rc.enabled:
+                jobs.append(
+                    JobSpec(
+                        name="option_chain_research",
+                        cron=rc.cron,
+                        coro_factory=_option_chain_research,
+                    )
+                )
+                _log.info(
+                    "research collector ENABLED (cron=%s, window<=%dd, depth=%s)",
+                    rc.cron,
+                    rc.expiry_max_days,
+                    rc.fetch_book_depth,
+                )
         else:
             _log.warning(
                 "data analysis disabled (CERBERO_BITE_ENABLE_DATA_ANALYSIS="

src/cerbero_bite/safety/audit_log.py

@@ -28,6 +28,7 @@ __all__ = [
     "AuditChainError",
     "AuditEntry",
     "AuditLog",
+    "tail_continues_from",
     "verify_chain",
 ]
 
@@ -157,6 +158,55 @@ def verify_chain(path: str | Path) -> int:
     return count
 
 
+def tail_continues_from(path: str | Path, anchor_hash: str) -> bool:
+    """Return ``True`` when *anchor_hash* is a genuine ancestor of the tail.
+
+    That is: *anchor_hash* is the ``hash`` of some line in the log AND every
+    line after it forms a valid forward chain to EOF. This distinguishes a
+    **benign anchor lag** — the best-effort anchor persisted in
+    :func:`cerbero_bite.runtime.dependencies.build_runtime` fell behind the
+    file under SQLite write contention, yet the log grew forward
+    legitimately — from real **truncation/tampering** (anchor absent, or the
+    post-anchor chain broken).
+
+    Returns ``False`` if the file is missing/empty, the anchor is not found,
+    or the chain from the anchor to the tail does not verify. The single-
+    writer invariant of :class:`AuditLog` still holds; this only makes the
+    boot-time anchor check tolerant of the durability gap it documents.
+    """
+    p = Path(path)
+    if not p.exists() or p.stat().st_size == 0:
+        return False
+    seen = False
+    expected_prev = ""
+    with p.open("r", encoding="utf-8") as fh:
+        for line in fh:
+            if not line.strip():
+                continue
+            try:
+                entry = _parse_line(line)
+            except AuditChainError:
+                if seen:
+                    return False
+                continue
+            if seen:
+                if entry.prev_hash != expected_prev:
+                    return False
+                recomputed = _compute_hash(
+                    entry.timestamp.isoformat(),
+                    entry.event,
+                    _canonical_payload(entry.payload),
+                    entry.prev_hash,
+                )
+                if recomputed != entry.hash:
+                    return False
+                expected_prev = entry.hash
+            elif entry.hash == anchor_hash:
+                seen = True
+                expected_prev = entry.hash
+    return seen
+
+
 def iter_entries(path: str | Path) -> Iterator[AuditEntry]:
     """Yield each :class:`AuditEntry` from *path* without verifying."""
     p = Path(path)

src/cerbero_bite/state/migrations/0006_dvol_history_multi_asset.sql

@@ -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;

src/cerbero_bite/state/migrations/0007_option_chain_source.sql

@@ -0,0 +1,18 @@
+-- 0007_option_chain_source.sql — distingue le righe del collettore
+--
+-- Due collettori scrivono ora su option_chain_snapshots:
+--   * 'live'     — collettore operativo, finestra DTE della strategia
+--                  (cfg.structure.dte_min..dte_max), 1 scadenza, no depth.
+--   * 'research' — collettore full-chain (tutte le scadenze <=95g,
+--                  entrambe le ali, book_depth_top3 popolato) per il
+--                  backtest opzioni vero (per-trade e standing put).
+--
+-- Le righe storiche pre-migrazione sono tutte 'live' (DEFAULT). Il
+-- backtest per-trade/standing filtra su source='research'.
+
+ALTER TABLE option_chain_snapshots ADD COLUMN source TEXT NOT NULL DEFAULT 'live';
+
+CREATE INDEX idx_option_chain_source
+    ON option_chain_snapshots(asset, source, timestamp DESC);
+
+PRAGMA user_version = 7;

src/cerbero_bite/state/models.py

@@ -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):
@@ -177,6 +178,9 @@ class OptionChainQuoteRecord(BaseModel):
     open_interest: int | None = None
     volume_24h: int | None = None
     book_depth_top3: int | None = None
+    # 'live' = collettore operativo (finestra DTE strategia, no depth);
+    # 'research' = collettore full-chain con book_depth popolato.
+    source: str = "live"
 
 
 class ManualAction(BaseModel):

src/cerbero_bite/state/repository.py

@@ -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) "
+            "INSERT OR REPLACE INTO dvol_history(timestamp, asset, dvol, spot) "
-            "VALUES (?,?,?)",
+            "VALUES (?,?,?,?)",
             (
                 _enc_dt(snapshot.timestamp),
+                snapshot.asset,
                 _enc_dec(snapshot.dvol),
-                _enc_dec(snapshot.eth_spot),
+                _enc_dec(snapshot.spot),
             ),
         )
 
@@ -546,6 +547,7 @@ class Repository:
                 q.open_interest,
                 q.volume_24h,
                 q.book_depth_top3,
+                q.source,
             )
             for q in quotes
         ]
@@ -553,8 +555,8 @@ class Repository:
             "INSERT OR REPLACE INTO option_chain_snapshots("
             "timestamp, asset, instrument_name, strike, expiry, option_type, "
             "bid, ask, mid, iv, delta, gamma, theta, vega, "
-            "open_interest, volume_24h, book_depth_top3) "
+            "open_interest, volume_24h, book_depth_top3, source) "
-            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
             rows,
         )
         return len(rows)
@@ -568,10 +570,14 @@ class Repository:
         end: datetime | None = None,
         expiry_from: datetime | None = None,
         expiry_to: datetime | None = None,
+        source: str | None = None,
         limit: int = 50000,
     ) -> list[OptionChainQuoteRecord]:
         clauses: list[str] = ["asset = ?"]
         params: list[Any] = [asset]
+        if source is not None:
+            clauses.append("source = ?")
+            params.append(source)
         if start is not None:
             clauses.append("timestamp >= ?")
             params.append(_enc_dt(start))
@@ -924,6 +930,11 @@ def _row_to_option_chain_quote(row: sqlite3.Row) -> OptionChainQuoteRecord:
             if row["book_depth_top3"] is not None
             else None
         ),
+        source=(
+            row["source"]
+            if "source" in row.keys() and row["source"] is not None
+            else "live"
+        ),
     )
 
 

strategy.aggressiva.yaml

@@ -28,9 +28,9 @@
 # 2× via "ETH + BTC" indicato in `📚 Strategia` è una **stima ex-ante**
 # di cosa otterresti DOPO quel lavoro di codice.
 
-config_version: "1.4.0-aggressiva"
+config_version: "1.5.0-aggressiva"
-config_hash: "7fa9b0be5b56517293421bc19838b700da595725360fe018a1be13b802dea859"
+config_hash: "a5e23c289315d738289f79e6b8c0e05e880e07c6ef878b013fc9849918e8b37a"
-last_review: "2026-04-26"
+last_review: "2026-06-09"
 last_reviewer: "Adriano"
 
 asset:
@@ -83,31 +83,52 @@ entry:
   vol_of_vol_lookback_hours: 24
 
 
+# §13-bis: collettore research full-chain (vedi strategy.yaml). Cattura
+# tutte le scadenze ed entrambe le ali con book_depth_top3 popolato per
+# il backtest opzioni per-trade/standing. Disabilitato di default.
+research_collector:
+  enabled: false
+  cron: "0 * * * *"
+  expiry_max_days: 95
+  moneyness_band_pct: null
+  open_interest_min: 100
+  fetch_book_depth: true
+  book_depth_concurrency: 8
+  assets: ["ETH", "BTC"]
+
+
 structure:
   dte_target: 18
   dte_min: 14
   dte_max: 21
 
+  # PROFILO B (tune 2026-06-09): stessa correzione del profilo
+  # conservativo. La delta step-function aggressiva (max 0.17 a DVOL<50)
+  # NON bastava: il credit/width satura a ~6% e 0.30 resta irraggiungibile
+  # (0 spread fattibili su 3.689 snapshot). Si alza il delta band.
   short_strike:
-    delta_target: "0.12"
+    delta_target: "0.18"
-    delta_min: "0.10"
+    delta_min: "0.12"
-    delta_max: "0.15"
+    delta_max: "0.22"
-    distance_otm_pct_min: "0.15"
+    distance_otm_pct_min: "0.10"
     distance_otm_pct_max: "0.25"
 
     # §3.2 (A): step-function delta-target per regime DVOL.
     # DVOL bassa (≤50) → più premio; alta (>70) → più safety.
     delta_by_dvol:
-      - {dvol_under: "50", delta_target: "0.15", delta_min: "0.13", delta_max: "0.17"}
+      - {dvol_under: "50", delta_target: "0.22", delta_min: "0.18", delta_max: "0.25"}
-      - {dvol_under: "70", delta_target: "0.12", delta_min: "0.10", delta_max: "0.15"}
+      - {dvol_under: "70", delta_target: "0.18", delta_min: "0.12", delta_max: "0.22"}
-      - {dvol_under: "90", delta_target: "0.10", delta_min: "0.08", delta_max: "0.12"}
+      - {dvol_under: "90", delta_target: "0.15", delta_min: "0.10", delta_max: "0.18"}
 
   spread_width:
     target_pct_of_spot: "0.04"
     min_pct_of_spot: "0.03"
-    max_pct_of_spot: "0.05"
+    # 0.06: griglia strike ETH a 100pt (~6% a spot <$2k); cap 0.05
+    # escludeva ogni gamba long.
+    max_pct_of_spot: "0.06"
 
-  credit_to_width_ratio_min: "0.30"
+  # Era 0.30 — fisicamente irraggiungibile. 0.08 allineato al realizzabile.
+  credit_to_width_ratio_min: "0.08"
 
 liquidity:
   open_interest_min: 100

strategy.yaml

@@ -6,9 +6,9 @@
 # config hash), and lands as a separate commit with the motivation in
 # the commit message.
 
-config_version: "1.4.0"
+config_version: "1.7.0"
-config_hash: "22182814216190331e0b69b3bc99493e6d69cc813f7ed937394986eecc1f5d11"
+config_hash: "1171380de6d3334be1f6eed04797cebe15e5b8ec2124e130b582c2e2097bde37"
-last_review: "2026-04-26"
+last_review: "2026-06-09"
 last_reviewer: "Adriano"
 
 asset:
@@ -16,7 +16,7 @@ asset:
   exchange: "deribit"
 
 entry:
-  cron: "0 14 * * *"
+  cron: "0 */2 * * *"
   skip_holidays_country: "IT"
 
   capital_min_usd: "720"
@@ -27,12 +27,15 @@ entry:
   no_position_concurrent: true
   exclude_macro_severity: ["high"]
   exclude_macro_countries: ["US", "EU"]
+  # Finestra evento macro ridotta a 1 giorno: blocca l'entry solo se un
+  # evento high-severity cade entro 24h, invece dei 18gg (dte_target).
+  exclude_macro_within_days: 1
 
   trend_window_days: 30
   trend_bull_threshold_pct: "0.05"
   trend_bear_threshold_pct: "-0.05"
-  funding_bull_threshold_annualized: "0.20"
+  funding_bull_threshold_annualized: "0.10"
-  funding_bear_threshold_annualized: "-0.20"
+  funding_bear_threshold_annualized: "-0.10"
   iron_condor_dvol_min: "55"
   iron_condor_adx_max: "20"
   iron_condor_trend_neutral_band_pct: "0.05"
@@ -43,11 +46,33 @@ entry:
   # per vendere credit spread. Soglia conservativa, da rifinire dopo
   # paper trading.
   dealer_gamma_min: "0"
-  dealer_gamma_filter_enabled: true
+  dealer_gamma_filter_enabled: false
   liquidation_filter_enabled: true
   # IV richness gate (§2.9). Disabilitato di default.
   iv_minus_rv_min: "0"
-  iv_minus_rv_filter_enabled: false
+  iv_minus_rv_filter_enabled: true
+
+  # IV richness gate adattivo — soglia P25 rolling su 60 giorni
+  iv_minus_rv_adaptive_enabled: true
+  iv_minus_rv_percentile: "0.25"
+  iv_minus_rv_window_target_days: 60
+  iv_minus_rv_window_min_days: 30
+
+
+# §13-bis: collettore research full-chain. INDIPENDENTE dal collettore
+# live (che resta sulla finestra DTE della strategia, 1 scadenza, no
+# depth). Cattura tutte le scadenze <=expiry_max_days ed entrambe le
+# ali, popolando book_depth_top3 → backtest opzioni per-trade e
+# standing put + slippage reale. Disabilitato di default (costo API).
+research_collector:
+  enabled: false
+  cron: "0 * * * *"            # orario
+  expiry_max_days: 95          # 1g..3mesi
+  moneyness_band_pct: null     # null = catena completa (entrambe le ali)
+  open_interest_min: 100
+  fetch_book_depth: true
+  book_depth_concurrency: 8
+  assets: ["ETH", "BTC"]
 
 
 structure:
@@ -55,19 +80,28 @@ structure:
   dte_min: 14
   dte_max: 21
 
+  # PROFILO B (tune 2026-06-09): vendere più vicino sblocca credito
+  # realizzabile. Analisi su 3.689 snapshot (1mag-8giu): a delta
+  # 0.10-0.15 il miglior credit/width ottenibile è ~6%, quindi 0.30 era
+  # fisicamente irraggiungibile (0 spread fattibili). Alzando il delta a
+  # ~0.18-0.22 il ratio sale a ~8-10% e l'eleggibilità a ~11%.
   short_strike:
-    delta_target: "0.12"
+    delta_target: "0.18"
-    delta_min: "0.10"
+    delta_min: "0.12"
-    delta_max: "0.15"
+    delta_max: "0.22"
-    distance_otm_pct_min: "0.15"
+    distance_otm_pct_min: "0.10"
     distance_otm_pct_max: "0.25"
 
   spread_width:
     target_pct_of_spot: "0.04"
     min_pct_of_spot: "0.03"
-    max_pct_of_spot: "0.05"
+    # 0.06: la griglia strike ETH sotto i $1500 è a 100pt (~6% a spot
+    # <$2k). Con cap 0.05 nessuna gamba long rientrava in banda.
+    max_pct_of_spot: "0.06"
 
-  credit_to_width_ratio_min: "0.30"
+  # Era 0.30 — irraggiungibile con delta <=0.30 in questo mercato.
+  # 0.08 = allineato al credit/width fisicamente realizzabile a delta ~0.18.
+  credit_to_width_ratio_min: "0.08"
 
 liquidity:
   open_interest_min: 100
@@ -107,7 +141,7 @@ exit:
   # atomica. Pipeline runtime non ancora attiva (hook futuro).
   profit_take_partial_levels: []
 
-  monitor_cron: "0 2,14 * * *"
+  monitor_cron: "0 * * * *"
   user_confirmation_timeout_min: 30
 
   escalate_on_timeout:

tests/integration/test_monitor_cycle.py

@@ -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()

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

tests/unit/test_data_audit.py

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

tests/unit/test_market_snapshot_cycle.py

@@ -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
+    by_asset = {r["asset"]: r for r in rows}
-    assert Decimal(str(rows[0]["dvol"])) == Decimal("55")
+    assert set(by_asset) == {"ETH", "BTC"}
-    assert Decimal(str(rows[0]["eth_spot"])) == Decimal("3000")
+    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) == []

tests/unit/test_state_repository.py

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