23 Commits

Author SHA1 Message Date
Adriano Dal Pastro 1a46abd5cf 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>
2026-06-09 07:45:05 +00:00
Adriano Dal Pastro 1eeb53650e feat(research): collettore full-chain opzioni con book_depth e colonna source
Aggiunge un secondo collettore opzioni indipendente dal live, pensato per
trasformare il dataset da "skew/premi medi" a backtest opzioni vero
(per-trade e standing put).

- option_chain_research_cycle.py: cattura tutte le scadenze <= expiry_max_days
  (1g..3mesi) ed entrambe le ali (OI>=100, filtro moneyness opzionale),
  popolando book_depth_top3 via orderbook_depth_top3 (concorrenza limitata)
  cosi' lo slippage reale e' modellabile. Best-effort come il collettore live.
- migrazione 0007: colonna `source` su option_chain_snapshots ('live' default
  per le righe storiche, 'research' per il nuovo collettore) + indice
  (asset, source, timestamp). user_version 6 -> 7.
- ResearchCollectorConfig (schema): blocco `research_collector`, enabled=false
  di default (costo API non trascurabile); cron orario, expiry_max_days=95.
- orchestrator: job `option_chain_research` schedulato solo se data-analysis
  attiva E research_collector.enabled.
- repository: insert+mapper estesi con `source`; list_option_chain_snapshots
  accetta un filtro `source` per le query di backtest.

Verificato in isolamento (immagine + interprete reali): import, migrazione,
round-trip repo. Suite: 527 passed, zero regressioni vs baseline pristina.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-09 07:44:47 +00:00
Adriano Dal Pastro 589d003e5c chore(compose): disabilita watchtower auto-update
Niente aggiornamenti automatici dell'immagine per un bot di trading:
i deploy devono essere deliberati (build + up manuali), non innescati
da un push del registry.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 09:23:18 +00:00
Adriano Dal Pastro ce475da173 docs(safety): disarm via coda manual_actions, gap storici audit, trigger _safe
Aggiornamenti post-incidente 31/05 (kill switch armato 6 giorni da un
singolo McpTimeoutError transitorio) e post-ripristino 05/06:

- 07: tabella trigger — documentato orchestrator._safe (qualsiasi
  eccezione di tick → arm CRITICAL al primo errore) e la tolleranza
  al lag benigno dell'anchor (647e3e5)
- 07: sezione Disarm — percorso raccomandato via coda manual_actions
  a engine acceso; CLI solo a engine fermo (race CLI<->engine = gap
  permanente nella hash chain)
- 07: nuova sezione "Limiti noti" — vincolo single-writer e i 7 gap
  storici benigni del log di produzione (01/05 e 29/05); audit verify
  full-chain fallisce alla riga 11 by design
- 06: nota in Flusso 5 — la tolleranza "3 fallimenti" vale solo per i
  probe health, _safe arma al primo errore su ogni altro tick

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 09:23:07 +00:00
Adriano Dal Pastro 647e3e565f fix(safety): il boot-check dell'audit tollera il lag benigno dell'anchor
L'anchor dell'audit (system_state.last_audit_hash) e' persistito best-effort:
sotto contesa di lock SQLite il mirror puo' restare indietro rispetto al log,
che invece cresce in avanti integro. Il check di boot armava il kill-switch
su qualsiasi disuguaglianza anchor!=coda, scambiando questo lag per
manomissione.

- audit_log: nuova tail_continues_from() — True solo se l'anchor e' un
  antenato valido della coda (presente in catena + chain forward integra a EOF)
- orchestrator._verify_audit_anchor: se lag benigno -> re-sync anchor + warning;
  solo anchor assente o chain rotta (troncamento/tamper) -> critical -> arma

Verificato live: anchor in ritardo -> re-sync senza armare; anchor bogus -> arma.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 19:55:19 +00:00
Adriano Dal Pastro a5a39a6d27 feat(entry): finestra evento macro configurabile, ridotta a 1g (config v1.6.0)
Il gate macro riusava structure.dte_target (18g) per decidere entro quanti
giorni un evento high-severity blocca l'entrata, accoppiando la protezione
di rischio-evento alla scelta delle scadenze opzioni. Disaccoppiato con un
parametro dedicato.

- schema: nuovo entry.exclude_macro_within_days (default 18 = comportamento storico)
- entry_validator: il gate macro usa il nuovo campo; rimosso structure_cfg inutilizzato
- entry_cycle: la fetch del calendario macro guarda avanti exclude_macro_within_days giorni
- strategy.yaml: exclude_macro_within_days=1, config_version 1.5.0->1.6.0, hash rigenerato

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 19:55:19 +00:00
Adriano Dal Pastro 839285d75b fix(mcp): allinea il client all'interfaccia unificata MCP V2
Cerbero MCP V2 ha spostato get_instruments/get_historical/get_indicators
sul router unificato /mcp (rimossi dagli endpoint per-exchange) e ha
rinominato get_technical_indicators in get_indicators. Il client chiamava
ancora questi tool su /mcp-deribit -> 404 -> kill-switch armato, raccolta
option-chain ferma e bias direzionale rotto.

- mcp_endpoints: nuovo endpoint `unified` (CERBERO_BITE_MCP_UNIFIED_URL,
  default http://cerbero-mcp:9000/mcp)
- deribit: i 3 tool dati passano dal client unificato con exchange="deribit",
  interval minuscolo (1D->1d) e parsing del nuovo shape (symbol + native.*);
  adx_14 usa get_indicators (indicators.adx.adx)
- dependencies/gui: iniettano il client unificato in DeribitClient
- docker-compose: default in-cluster per l'endpoint /mcp

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 19:55:04 +00:00
Adriano Dal Pastro 16d73c309a fix(analysis): distingui ask<=0 (missing ask) da bid>ask (cross reale)
Nuovo campo ChainAuditReport.ask_zero_count. La logica del loop
parsifica prima l'ask: se ask<=0 è "missing ask side" (su Deribit
significa nessun ordine in vendita), conta in ask_zero_count e non
applica il check bid>ask. Solo con ask>0 si confronta con bid.

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

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

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

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-16 19:46:54 +00:00
Adriano Dal Pastro 197c9db74d feat(cli): subcommand 'audit data' — qualità dati market + chain
Usa il group 'audit' esistente (sibling di 'audit verify' per la hash
chain). Opzioni: --db, --since DAYS, --asset ETH|BTC, --json.
Output stdout Rich di default, dump JSON con --json. Esce con codice 2
su sqlite3.OperationalError (DB malformato/schema mancante).

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

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

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-16 19:22:31 +00:00
root d6af69f4cb feat(analysis): audit_market_snapshots — coverage, gap, fetch_ok, NULL rate
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-13 09:33:28 +00:00
root aeac8f2a95 feat(analysis): _max_zero_streak su flag fetch_ok 2026-05-13 09:31:38 +00:00
root 75fe803296 style(analysis): consolidate test imports at top of file (PEP 8) 2026-05-13 09:30:41 +00:00
root ea5c612446 feat(analysis): _detect_gaps su timestamp consecutivi (> 20 min)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-13 09:29:48 +00:00
root 9e2216d202 fix(analysis): _expected_ticks usa ceiling division (no off-by-one)
Il piano originale aveva `floor(span/15) + 1` che over-conta a span allineati
(span=60min → 5 invece di 4). Il primo fix dell'implementer (`floor(span/15)`)
under-conta a span non-allineati (span=16min → 1 invece di 2). Solo
`ceil(span/15)` è corretto in entrambi i casi. Aggiunti 2 test che
coprono gli scenari non-allineato e boundary-esatto per impedire
regressioni. Plan doc allineato.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 09:28:24 +00:00
root 35ac92e938 feat(analysis): _expected_ticks per finestre */15 allineate
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 22:32:12 +00:00
root 07a8bbf5c8 feat(analysis): skeleton modulo data_audit (dataclass + soglie) 2026-05-12 22:01:44 +00:00
root 0c6e462545 docs(plan): data quality audit implementation plan (10 task TDD)
Piano dettagliato task-by-task per `cerbero-bite audit`:
analysis/data_audit.py (helper puri + dataclass), CLI subcommand,
test unit + smoke test, end-to-end su DB produzione. Ogni task ha
i suoi step TDD con codice completo, comandi e commit.

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

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

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 13:38:34 +00:00
33 changed files with 3304 additions and 102 deletions
+4 -1
View File
@@ -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
+1 -1
View File
@@ -242,7 +242,7 @@ async def run_monitor_cycle(ctx: RuntimeContext, *, now):
dvol = await deribit.latest_dvol(currency="ETH", now=now)
return_4h = await _fetch_return_4h(ctx, now=now) # usa dvol_history o
# fallback get_historical
repo.record_dvol_snapshot(DvolSnapshot(timestamp=now, dvol=dvol, eth_spot=spot))
repo.record_dvol_snapshot(DvolSnapshot(timestamp=now, asset="ETH", dvol=dvol, spot=spot))
for record in repo.list_positions(status="open"):
snapshot = await _build_position_snapshot(...)
+14 -5
View File
@@ -137,16 +137,25 @@ CREATE INDEX idx_decisions_proposal ON decisions(proposal_id);
### `dvol_history`
Snapshot DVOL + ETH spot ad ogni evaluation. Utile per il calcolo di
`return_4h` durante il monitor (vedi `runtime/monitor_cycle.py
_fetch_return_4h`) e per analisi post-mortem.
Snapshot DVOL + spot per asset ad ogni evaluation. Utile per il
calcolo di `return_4h` durante il monitor (vedi
`runtime/monitor_cycle.py _fetch_return_4h`, che filtra `asset='ETH'`)
e per analisi post-mortem comparate fra ETH e BTC. Lo schema è
multi-asset dal migration `0006_dvol_history_multi_asset.sql`; le
righe storiche pre-migration sono state backfillate con
`asset='ETH'`.
```sql
CREATE TABLE dvol_history (
timestamp TEXT PRIMARY KEY,
timestamp TEXT NOT NULL,
asset TEXT NOT NULL, -- "ETH", "BTC"
dvol NUMERIC NOT NULL,
eth_spot NUMERIC NOT NULL
spot NUMERIC NOT NULL,
PRIMARY KEY (timestamp, asset)
);
CREATE INDEX idx_dvol_history_asset_ts
ON dvol_history(asset, timestamp DESC);
```
### `manual_actions`
+8
View File
@@ -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`).
+64 -3
View File
@@ -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
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,201 @@
# Data Quality Audit — Design Spec
**Date:** 2026-05-12
**Status:** Approved (design phase)
**Author:** session-driven (operator + agent)
## Motivation
Prima di costruire il backtest non-stilizzato su `option_chain_snapshots`
(prossimo macro-step del progetto), serve confermare che i dati
raccolti negli ultimi 11 giorni (ETH) e 8 giorni (BTC) siano usabili:
copertura temporale piena, niente buchi sistemici, niente quote
malformate (bid > ask, IV mancante, depth book a zero). Lo stesso
audit dev'essere ri-eseguibile periodicamente come check operativo.
`market_snapshots` rientra nello scope per simmetria (entrambe le
tabelle alimentano la decisione di entrata e il monitoring), mentre
`dvol_history` è escluso: appena migrato a multi-asset (commit
`19695e4`), serie troppo corta per BTC (29 righe al momento del
design) e copertura ETH già implicita in `market_snapshots`.
## Scope
**In scope:**
- `market_snapshots`: continuità temporale, fetch_ok streaks, NULL rate
per colonna numerica, parità ETH/BTC.
- `option_chain_snapshots`: snapshot mancanti, distribuzione quote per
snap, bid/ask sanity, IV null rate, book depth.
- CLI subcommand `cerbero-bite audit data`, output stdout + opzionale `--json`.
**Out of scope:**
- `dvol_history`, `decisions`, `positions`, `instructions`,
`manual_actions` (non rilevanti per il backtest non-stilizzato).
- Audit di consistenza cross-tabella (es: per ogni snapshot chain esiste
uno snapshot market) — interessante ma rinviato.
- Persistenza dei risultati audit nello stesso DB.
## Architecture
```
src/cerbero_bite/analysis/
__init__.py
data_audit.py # logica pura, no I/O lato MCP
src/cerbero_bite/cli.py # nuovo subcommand `audit data`
tests/unit/
test_data_audit.py # DB temporaneo + seed deterministico
```
`data_audit.py` espone funzioni pure che prendono una `sqlite3.Connection`
e una finestra temporale, ritornano `dataclass` di risultati. Il CLI
apre la connection in read-only, chiama le funzioni, formatta l'output.
Funzioni principali:
```python
@dataclass(frozen=True)
class MarketAuditReport:
asset: str
expected_ticks: int
actual_ticks: int
coverage_pct: Decimal
gaps_over_threshold: list[GapRecord]
fetch_ok_zero_count: int
max_fetch_ok_zero_streak: int
null_rate_by_column: dict[str, Decimal]
@dataclass(frozen=True)
class ChainAuditReport:
asset: str
expected_snapshots: int
actual_snapshots: int
coverage_pct: Decimal
quotes_per_snap_median: int
quotes_per_snap_p10: int
quotes_per_snap_p90: int
bid_gt_ask_count: int
ask_zero_count: int
iv_null_count: int
iv_null_pct: Decimal
def audit_market_snapshots(conn, *, asset, since, now) -> MarketAuditReport: ...
def audit_option_chain(conn, *, asset, since, now) -> ChainAuditReport: ...
```
## Checks & Thresholds
| Tabella | Check | Soglia "bad" | Rationale |
|---|---|---|---|
| market_snapshots | gap tra tick consecutivi | > 20 min | cron è `*/15`; +5 min tolleranza |
| market_snapshots | streak `fetch_ok=0` | ≥ 3 consecutivi | 1-2 = transient MCP, 3+ = pattern |
| market_snapshots | NULL rate per colonna | > 10% nella finestra | una metrica con >10% NULL non è affidabile per backtest |
| option_chain_snapshots | snap mancanti | qualsiasi (count visibile) | cron `*/15`, ogni miss è significativo |
| option_chain_snapshots | quote/snap < 50% mediana 24h | qualsiasi | rilevatore di chain truncate (mismatch with width filter) |
| option_chain_snapshots | bid > ask (con ask > 0) | qualsiasi | dato corrotto, da indagare |
| option_chain_snapshots | ask ≤ 0 | qualsiasi | missing ask side: best ask vuoto al momento della query (≠ cross BBO) |
| option_chain_snapshots | IV null/non-parseable | conteggio + % | IV è chiave per BS skew calibration |
> **Nota:** il check `book_depth_top3 = 0` originariamente previsto è
> stato rimosso. La colonna è NULL by design sugli snapshot (il collector
> `option_chain_snapshot_cycle` non chiama l'orderbook per ogni strike
> per non saturare l'API Deribit); viene popolata solo dal path
> `entry_cycle` per gli strike candidati al picker. Il check di liquidità
> ha senso solo lì, non in questo audit.
Le soglie sono costanti modulo (non config YAML) per ridurre il blast
radius dei cambi: il backtest e l'audit girano in contesti diversi,
non condividono parametri operativi.
## CLI
```
cerbero-bite audit data [--db PATH] [--since DAYS] [--json] [--asset ETH|BTC]
```
> **Deviazione dal design originale**: il group `audit` era già usato per
> `audit verify` (hash chain dell'audit log). Il data-quality audit è
> stato inserito come sibling `audit data` per evitare collisione di
> namespace, anziché come comando top-level `cerbero-bite audit`.
- `--db PATH` (default `data/state.sqlite`): path al DB SQLite.
- `--since DAYS` (default `7`): finestra di analisi, retro dal `now()` corrente.
- `--json` (default off): stampa solo dump JSON serializzabile, niente tabelle umane.
- `--asset` (default `tutti`): filtra ad un singolo asset.
Exit code:
- `0`: audit completato (a prescindere dai problemi trovati).
- `2`: errori di connessione/DB (DB inesistente, schema mancante).
Niente exit code per "found issues": l'audit è informativo, decide
l'umano. Far diventare l'audit un gate CI è out of scope.
## Output
**Stdout (default):**
```
=== ETH — market_snapshots (last 7d, 2026-05-05 → 2026-05-12) ===
ticks: 672 expected: 672 coverage: 100.0%
gaps > 20min: 0
fetch_ok=0: 4 rows (max streak: 1)
null rate: dealer_net_gamma 2.1% oi_delta_pct_4h 0.3%
=== ETH — option_chain_snapshots (last 7d) ===
snapshots: 672 expected: 672 coverage: 100.0%
quotes/snap: median 55 p10 50 p90 60
bid > ask: 0
ask <= 0: 37 (missing ask side)
IV null: 12 quotes (0.03%)
=== BTC — ...
```
**JSON (`--json`):**
```json
{
"since": "2026-05-05T20:46:00+00:00",
"until": "2026-05-12T20:46:00+00:00",
"assets": {
"ETH": {
"market": {"expected_ticks": 672, "actual_ticks": 672, ...},
"chain": {"expected_snapshots": 672, ...}
},
"BTC": {...}
}
}
```
## Testing
`tests/unit/test_data_audit.py`. Per ogni funzione:
- DB temporaneo (`tmp_path`), schema migrato via `run_migrations`.
- Seed deterministico: insert manuali per riprodurre lo scenario.
- Test cases:
- market: copertura piena → 100%; un gap iniettato → conteggio gap=1;
streak `fetch_ok=0` lunga 3 → flagged.
- chain: snap mancante → expected actual = 1; quote dimezzate in
un tick → quotes/snap p10 cala; `bid=10 ask=5` → bid>ask=1.
- 0 dipendenze nuove (sqlite + pytest standard).
## Performance
Tabelle attuali: ~57k quote chain. Le query usano gli index
`idx_option_chain_asset_ts` e `(asset, timestamp)` di
`market_snapshots`. L'audit deve girare in < 2s su 7gg.
## Anti-goals (esplicito)
- Nessun salvataggio dei risultati nello stato del DB.
- Nessun trigger automatico (no cron job, no APScheduler).
- Nessun alert/notifica: stdout + JSON sono lo strumento, l'operatore
decide cosa farne.
- Nessun ML / detection di anomalie sofisticate. Soglie costanti.
## Open Questions
Nessuna al momento della scrittura. Eventuali punti emergeranno durante
l'implementazione e andranno annotati qui.
+5
View File
@@ -0,0 +1,5 @@
"""Analysis utilities — pure functions over the state DB.
Modules here read SQLite, never write. They are ergonomic to call
from CLI commands, notebooks, or one-off scripts.
"""
+336
View File
@@ -0,0 +1,336 @@
"""Data quality audit over market_snapshots + option_chain_snapshots.
Pure functions: each takes a ``sqlite3.Connection`` and a UTC time
window, returns a frozen dataclass. No side effects, no MCP, no
writes. The CLI layer (``cli.audit``) is responsible for I/O and
formatting.
Thresholds are module-level constants by design: the audit and the
runtime live in different contexts and must not share operational
parameters. To tune a threshold, edit this file.
"""
from __future__ import annotations
import math
import sqlite3
import statistics
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from decimal import Decimal
__all__ = [
"ChainAuditReport",
"GapRecord",
"MarketAuditReport",
"audit_market_snapshots",
"audit_option_chain",
]
# Tick cadence + gap tolerance. Cron is */15; +5 min tolerance covers
# late-arriving MCP responses.
_TICK_INTERVAL_MIN: int = 15
_GAP_THRESHOLD_MIN: int = 20
# fetch_ok=0 streak threshold: 1-2 are transient MCP failures, 3+ is a
# pattern worth flagging.
_FETCH_OK_STREAK_THRESHOLD: int = 3
# A numeric column with >10% NULL in the window is too unreliable for
# backtesting that metric.
_NULL_RATE_FLAG: Decimal = Decimal("0.10")
# Columns to NULL-audit on market_snapshots. fetch_ok / fetch_errors_json
# are excluded (they are status fields, not metrics).
_MARKET_NUMERIC_COLUMNS: tuple[str, ...] = (
"spot",
"dvol",
"realized_vol_30d",
"iv_minus_rv",
"funding_perp_annualized",
"funding_cross_annualized",
"dealer_net_gamma",
"gamma_flip_level",
"oi_delta_pct_4h",
"macro_days_to_event",
)
@dataclass(frozen=True)
class GapRecord:
"""One gap between consecutive market_snapshots ticks."""
prev_timestamp: datetime
next_timestamp: datetime
gap_minutes: int
@dataclass(frozen=True)
class MarketAuditReport:
asset: str
since: datetime
until: datetime
expected_ticks: int
actual_ticks: int
coverage_pct: Decimal
gaps: tuple[GapRecord, ...] = field(default_factory=tuple)
fetch_ok_zero_count: int = 0
max_fetch_ok_zero_streak: int = 0
null_rate_by_column: dict[str, Decimal] = field(default_factory=dict)
@dataclass(frozen=True)
class ChainAuditReport:
asset: str
since: datetime
until: datetime
expected_snapshots: int
actual_snapshots: int
coverage_pct: Decimal
quotes_per_snap_median: int = 0
quotes_per_snap_p10: int = 0
quotes_per_snap_p90: int = 0
bid_gt_ask_count: int = 0
ask_zero_count: int = 0
iv_null_count: int = 0
iv_null_pct: Decimal = Decimal("0")
def _expected_ticks(since: datetime, until: datetime) -> int:
"""Number of `*/15` ticks in ``[since, until)`` aligned to wall clock.
A tick is any UTC instant where ``minute % 15 == 0``. The first
tick at or after ``since`` is computed by rounding ``since`` up;
every subsequent tick is +15 minutes. The window is half-open on
the right.
"""
if until <= since:
return 0
# Round `since` up to the next */15 boundary.
minute = since.minute
remainder = minute % _TICK_INTERVAL_MIN
if remainder == 0 and since.second == 0 and since.microsecond == 0:
first_tick = since
else:
bump = _TICK_INTERVAL_MIN - remainder
first_tick = (since + timedelta(minutes=bump)).replace(
second=0, microsecond=0
)
if first_tick >= until:
return 0
# Count ticks in [first_tick, until): the largest k with
# first_tick + k*15min < until is ceil(span/15) - 1, so the
# count is ceil(span_minutes / 15). floor() under-counts at
# aligned multiples and would mis-count non-aligned spans.
span_seconds = (until - first_tick).total_seconds()
return math.ceil(span_seconds / (_TICK_INTERVAL_MIN * 60))
def _max_zero_streak(flags: list[int]) -> int:
"""Longest run of consecutive zeros."""
longest = 0
current = 0
for v in flags:
if v == 0:
current += 1
longest = max(longest, current)
else:
current = 0
return longest
def _detect_gaps(timestamps: list[datetime]) -> tuple[GapRecord, ...]:
"""Return gaps where consecutive timestamps differ by > threshold."""
out: list[GapRecord] = []
for prev, nxt in zip(timestamps, timestamps[1:], strict=False):
delta_min = int((nxt - prev).total_seconds() // 60)
if delta_min > _GAP_THRESHOLD_MIN:
out.append(
GapRecord(
prev_timestamp=prev,
next_timestamp=nxt,
gap_minutes=delta_min,
)
)
return tuple(out)
def _fetch_market_rows(
conn: sqlite3.Connection,
*,
asset: str,
since: datetime,
until: datetime,
) -> list[sqlite3.Row]:
cols = ", ".join(("timestamp", "fetch_ok", *_MARKET_NUMERIC_COLUMNS))
rows = conn.execute(
f"SELECT {cols} FROM market_snapshots "
"WHERE asset = ? AND timestamp >= ? AND timestamp < ? "
"ORDER BY timestamp ASC",
(asset, since.isoformat(), until.isoformat()),
).fetchall()
return list(rows)
def _compute_null_rate(
rows: list[sqlite3.Row], columns: tuple[str, ...]
) -> dict[str, Decimal]:
if not rows:
return {c: Decimal("0") for c in columns}
total = Decimal(len(rows))
out: dict[str, Decimal] = {}
for c in columns:
nulls = sum(1 for r in rows if r[c] is None)
out[c] = (Decimal(nulls) / total).quantize(Decimal("0.0001"))
return out
def audit_market_snapshots(
conn: sqlite3.Connection,
*,
asset: str,
since: datetime,
until: datetime,
) -> MarketAuditReport:
"""Compute the market_snapshots audit report for an asset in [since, until)."""
rows = _fetch_market_rows(conn, asset=asset, since=since, until=until)
timestamps = [datetime.fromisoformat(r["timestamp"]) for r in rows]
expected = _expected_ticks(since, until)
actual = len(rows)
coverage = (
(Decimal(actual) / Decimal(expected) * Decimal("100")).quantize(
Decimal("0.01")
)
if expected > 0
else Decimal("0")
)
gaps = _detect_gaps(timestamps)
fetch_ok_flags = [int(r["fetch_ok"]) for r in rows]
fetch_ok_zero_count = sum(1 for v in fetch_ok_flags if v == 0)
max_streak = _max_zero_streak(fetch_ok_flags)
null_rates = _compute_null_rate(rows, _MARKET_NUMERIC_COLUMNS)
return MarketAuditReport(
asset=asset,
since=since,
until=until,
expected_ticks=expected,
actual_ticks=actual,
coverage_pct=coverage,
gaps=gaps,
fetch_ok_zero_count=fetch_ok_zero_count,
max_fetch_ok_zero_streak=max_streak,
null_rate_by_column=null_rates,
)
def _pct(values: list[int], q: int) -> int:
"""Integer percentile via `statistics.quantiles` (inclusive).
Returns 0 on empty input; the single value on len==1. ``q`` is in 1..100.
"""
if not values:
return 0
if len(values) == 1:
return values[0]
sorted_vals = sorted(values)
if q >= 100:
return sorted_vals[-1]
cuts = statistics.quantiles(sorted_vals, n=100, method="inclusive")
return int(round(cuts[q - 1]))
def audit_option_chain(
conn: sqlite3.Connection,
*,
asset: str,
since: datetime,
until: datetime,
) -> ChainAuditReport:
"""Compute the option_chain_snapshots audit report in [since, until)."""
expected = _expected_ticks(since, until)
snap_counts = conn.execute(
"SELECT timestamp, COUNT(*) AS n FROM option_chain_snapshots "
"WHERE asset = ? AND timestamp >= ? AND timestamp < ? "
"GROUP BY timestamp ORDER BY timestamp",
(asset, since.isoformat(), until.isoformat()),
).fetchall()
actual = len(snap_counts)
coverage = (
(Decimal(actual) / Decimal(expected) * Decimal("100")).quantize(
Decimal("0.01")
)
if expected > 0
else Decimal("0")
)
quotes_per_snap = [r["n"] for r in snap_counts]
median_q = _pct(quotes_per_snap, 50)
p10_q = _pct(quotes_per_snap, 10)
p90_q = _pct(quotes_per_snap, 90)
quote_rows = conn.execute(
"SELECT bid, ask, iv FROM option_chain_snapshots "
"WHERE asset = ? AND timestamp >= ? AND timestamp < ?",
(asset, since.isoformat(), until.isoformat()),
).fetchall()
total_quotes = len(quote_rows)
if total_quotes == 0:
return ChainAuditReport(
asset=asset,
since=since,
until=until,
expected_snapshots=expected,
actual_snapshots=actual,
coverage_pct=coverage,
quotes_per_snap_median=median_q,
quotes_per_snap_p10=p10_q,
quotes_per_snap_p90=p90_q,
)
bid_gt_ask = 0
ask_zero = 0
iv_null = 0
for r in quote_rows:
bid_s, ask_s, iv_s = r["bid"], r["ask"], r["iv"]
ask_d: Decimal | None = None
if ask_s is not None:
try:
ask_d = Decimal(ask_s)
except (ValueError, ArithmeticError):
ask_d = None
if ask_d is not None and ask_d <= 0:
# Su Deribit ask=0 significa "nessun ordine in vendita",
# non bid>ask reale: contiamolo a parte.
ask_zero += 1
elif ask_d is not None and bid_s is not None:
try:
if Decimal(bid_s) > ask_d:
bid_gt_ask += 1
except (ValueError, ArithmeticError):
pass
if iv_s is None:
iv_null += 1
else:
try:
Decimal(iv_s)
except (ValueError, ArithmeticError):
iv_null += 1
iv_null_pct = (
Decimal(iv_null) / Decimal(total_quotes) * Decimal("100")
).quantize(Decimal("0.01"))
return ChainAuditReport(
asset=asset,
since=since,
until=until,
expected_snapshots=expected,
actual_snapshots=actual,
coverage_pct=coverage,
quotes_per_snap_median=median_q,
quotes_per_snap_p10=p10_q,
quotes_per_snap_p90=p90_q,
bid_gt_ask_count=bid_gt_ask,
ask_zero_count=ask_zero,
iv_null_count=iv_null,
iv_null_pct=iv_null_pct,
)
+174
View File
@@ -11,6 +11,7 @@ from __future__ import annotations
import asyncio
import os
import sqlite3
import sys
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
@@ -23,6 +24,12 @@ from rich.console import Console
from rich.table import Table
from cerbero_bite import __version__
from cerbero_bite.analysis.data_audit import (
ChainAuditReport,
MarketAuditReport,
audit_market_snapshots,
audit_option_chain,
)
from cerbero_bite.clients import HttpToolClient, McpError
from cerbero_bite.clients.deribit import DeribitClient
from cerbero_bite.clients.hyperliquid import HyperliquidClient
@@ -898,6 +905,173 @@ def audit_verify(audit_path: Path) -> None:
console.print(f"[green]ok[/green] {count} entries verified")
def _market_to_dict(r: MarketAuditReport) -> dict[str, Any]:
return {
"asset": r.asset,
"since": r.since.isoformat(),
"until": r.until.isoformat(),
"expected_ticks": r.expected_ticks,
"actual_ticks": r.actual_ticks,
"coverage_pct": str(r.coverage_pct),
"gaps": [
{
"prev": g.prev_timestamp.isoformat(),
"next": g.next_timestamp.isoformat(),
"gap_minutes": g.gap_minutes,
}
for g in r.gaps
],
"fetch_ok_zero_count": r.fetch_ok_zero_count,
"max_fetch_ok_zero_streak": r.max_fetch_ok_zero_streak,
"null_rate_by_column": {
k: str(v) for k, v in r.null_rate_by_column.items()
},
}
def _chain_to_dict(r: ChainAuditReport) -> dict[str, Any]:
return {
"asset": r.asset,
"since": r.since.isoformat(),
"until": r.until.isoformat(),
"expected_snapshots": r.expected_snapshots,
"actual_snapshots": r.actual_snapshots,
"coverage_pct": str(r.coverage_pct),
"quotes_per_snap_median": r.quotes_per_snap_median,
"quotes_per_snap_p10": r.quotes_per_snap_p10,
"quotes_per_snap_p90": r.quotes_per_snap_p90,
"bid_gt_ask_count": r.bid_gt_ask_count,
"ask_zero_count": r.ask_zero_count,
"iv_null_count": r.iv_null_count,
"iv_null_pct": str(r.iv_null_pct),
}
def _render_market_report(asset: str, r: MarketAuditReport) -> None:
console.print(
f"\n[bold cyan]=== {asset} — market_snapshots "
f"({r.since.date()}{r.until.date()}) ===[/bold cyan]"
)
console.print(
f" ticks: {r.actual_ticks} expected: {r.expected_ticks} "
f"coverage: {r.coverage_pct}%"
)
console.print(f" gaps > 20min: {len(r.gaps)}")
console.print(
f" fetch_ok=0: {r.fetch_ok_zero_count} rows "
f"(max streak: {r.max_fetch_ok_zero_streak})"
)
bad_nulls = {k: v for k, v in r.null_rate_by_column.items() if v > Decimal("0")}
if bad_nulls:
parts = " ".join(
f"{k} {(v * 100).quantize(Decimal('0.1'))}%"
for k, v in bad_nulls.items()
)
console.print(f" null rate: {parts}")
else:
console.print(" null rate: all columns 0%")
def _render_chain_report(asset: str, r: ChainAuditReport) -> None:
console.print(
f"\n[bold cyan]=== {asset} — option_chain_snapshots "
f"({r.since.date()}{r.until.date()}) ===[/bold cyan]"
)
console.print(
f" snapshots: {r.actual_snapshots} expected: {r.expected_snapshots} "
f"coverage: {r.coverage_pct}%"
)
console.print(
f" quotes/snap: median {r.quotes_per_snap_median} "
f"p10 {r.quotes_per_snap_p10} p90 {r.quotes_per_snap_p90}"
)
console.print(f" bid > ask: {r.bid_gt_ask_count}")
console.print(f" ask <= 0: {r.ask_zero_count} (missing ask side)")
console.print(
f" IV null: {r.iv_null_count} quotes ({r.iv_null_pct}%)"
)
@audit.command(name="data")
@click.option(
"--db",
"db_path",
type=click.Path(exists=True, dir_okay=False, path_type=Path),
default=_DEFAULT_DB_PATH,
show_default=True,
help="Path al DB SQLite di stato.",
)
@click.option(
"--since",
"since_days",
type=int,
default=7,
show_default=True,
help="Finestra di analisi (giorni indietro da ora).",
)
@click.option(
"--asset",
"asset",
type=click.Choice(["ETH", "BTC"], case_sensitive=False),
default=None,
help="Limita l'audit a un singolo asset (default: entrambi).",
)
@click.option(
"--json",
"as_json",
is_flag=True,
default=False,
help="Stampa solo dump JSON, niente tabelle umane.",
)
def audit_data(
db_path: Path,
since_days: int,
asset: str | None,
as_json: bool,
) -> None:
"""Audit qualità dati: market_snapshots + option_chain_snapshots."""
import json as _json # noqa: PLC0415
until = datetime.now(UTC)
since = until - timedelta(days=since_days)
assets = [asset.upper()] if asset else ["ETH", "BTC"]
conn = connect_state(db_path)
try:
market_reports = {
a: audit_market_snapshots(conn, asset=a, since=since, until=until)
for a in assets
}
chain_reports = {
a: audit_option_chain(conn, asset=a, since=since, until=until)
for a in assets
}
except sqlite3.OperationalError as exc:
console.print(f"[red]DB error[/red]: {exc}")
sys.exit(2)
finally:
conn.close()
if as_json:
payload = {
"since": since.isoformat(),
"until": until.isoformat(),
"assets": {
a: {
"market": _market_to_dict(market_reports[a]),
"chain": _chain_to_dict(chain_reports[a]),
}
for a in assets
},
}
click.echo(_json.dumps(payload, indent=2, default=str))
return
for a in assets:
_render_market_report(a, market_reports[a])
_render_chain_report(a, chain_reports[a])
@main.group()
def state() -> None:
"""State inspection utilities."""
+59 -20
View File
@@ -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(
"get_technical_indicators",
# MCP V2: indicators are computed by the unified ``get_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
# indicator keyed by name, or a list of points. Be tolerant.
adx_payload = raw.get("adx") or raw.get("ADX") or raw.get("indicators", {})
if isinstance(adx_payload, list) and adx_payload:
tail = adx_payload[-1]
value = tail.get("value") if isinstance(tail, dict) else tail
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")
indicators = raw.get("indicators")
if not isinstance(indicators, dict):
return None
adx_block = indicators.get("adx")
if isinstance(adx_block, dict):
value = adx_block.get("adx")
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]:
+13
View File
@@ -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)
+34
View File
@@ -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)
+1 -2
View File
@@ -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)"
+1 -1
View File
@@ -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"))
+1 -1
View File
@@ -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"))
+1 -1
View File
@@ -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,
)
@@ -181,19 +181,18 @@ async def collect_market_snapshot(
try:
with transaction(conn):
ctx.repository.record_market_snapshot(conn, record)
# Mirror ETH spot+DVOL into dvol_history so monitor_cycle's
# return_4h lookup has local samples even in data-only mode.
if (
record.asset == "ETH"
and record.spot is not None
and record.dvol is not None
):
# Mirror spot+DVOL into dvol_history (per asset) so
# monitor_cycle's return_4h lookup has local samples even
# in data-only mode. dvol_history enforces NOT NULL on
# dvol/spot so skip if either is missing.
if record.spot is not None and record.dvol is not None:
ctx.repository.record_dvol_snapshot(
conn,
DvolSnapshot(
timestamp=record.timestamp,
asset=record.asset,
dvol=record.dvol,
eth_spot=record.spot,
spot=record.spot,
),
)
finally:
+3 -3
View File
@@ -173,8 +173,8 @@ async def _fetch_return_4h(ctx: RuntimeContext, *, now: datetime) -> Decimal:
conn = connect_state(ctx.db_path)
try:
row = conn.execute(
"SELECT timestamp, eth_spot FROM dvol_history "
"WHERE timestamp <= ? AND timestamp >= ? "
"SELECT timestamp, spot FROM dvol_history "
"WHERE asset = 'ETH' AND timestamp <= ? AND timestamp >= ? "
"ORDER BY timestamp DESC LIMIT 1",
(cutoff.isoformat(), floor.isoformat()),
).fetchone()
@@ -239,7 +239,7 @@ async def run_monitor_cycle(
with transaction(conn):
ctx.repository.record_dvol_snapshot(
conn,
DvolSnapshot(timestamp=when, dvol=dvol, eth_spot=spot),
DvolSnapshot(timestamp=when, asset="ETH", dvol=dvol, spot=spot),
)
positions = ctx.repository.list_positions(conn, status="open")
finally:
@@ -0,0 +1,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
+61 -11
View File
@@ -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_MONITOR = "0 2,14 * * *"
_CRON_ENTRY = "0 */2 * * *" # crypto 24/7: valutazione ogni 2h; i gate decidono se entrare
_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:
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",
if actual_tail == state.last_audit_hash:
return
# The anchor is persisted best-effort (see build_runtime): under
# SQLite write contention the mirror can fall behind the log while
# the log itself keeps growing forward, intact. Treat that benign
# lag — anchor is a valid ancestor of the current tail — as a
# re-sync, not tampering. Only a missing anchor or a broken
# post-anchor chain (truncation/tampering) arms the kill switch.
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="
+50
View File
@@ -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)
@@ -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;
@@ -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;
+5 -1
View File
@@ -116,8 +116,9 @@ class DvolSnapshot(BaseModel):
model_config = ConfigDict(extra="forbid")
timestamp: datetime
asset: str # "ETH", "BTC"
dvol: Decimal
eth_spot: Decimal
spot: Decimal
class MarketSnapshotRecord(BaseModel):
@@ -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):
+16 -5
View File
@@ -339,12 +339,13 @@ class Repository:
self, conn: sqlite3.Connection, snapshot: DvolSnapshot
) -> None:
conn.execute(
"INSERT OR REPLACE INTO dvol_history(timestamp, dvol, eth_spot) "
"VALUES (?,?,?)",
"INSERT OR REPLACE INTO dvol_history(timestamp, asset, dvol, spot) "
"VALUES (?,?,?,?)",
(
_enc_dt(snapshot.timestamp),
snapshot.asset,
_enc_dec(snapshot.dvol),
_enc_dec(snapshot.eth_spot),
_enc_dec(snapshot.spot),
),
)
@@ -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) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
"open_interest, volume_24h, book_depth_top3, source) "
"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"
),
)
+33 -12
View File
@@ -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_hash: "7fa9b0be5b56517293421bc19838b700da595725360fe018a1be13b802dea859"
last_review: "2026-04-26"
config_version: "1.5.0-aggressiva"
config_hash: "a5e23c289315d738289f79e6b8c0e05e880e07c6ef878b013fc9849918e8b37a"
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_min: "0.10"
delta_max: "0.15"
distance_otm_pct_min: "0.15"
delta_target: "0.18"
delta_min: "0.12"
delta_max: "0.22"
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: "70", delta_target: "0.12", delta_min: "0.10", delta_max: "0.15"}
- {dvol_under: "90", delta_target: "0.10", delta_min: "0.08", delta_max: "0.12"}
- {dvol_under: "50", delta_target: "0.22", delta_min: "0.18", delta_max: "0.25"}
- {dvol_under: "70", delta_target: "0.18", delta_min: "0.12", delta_max: "0.22"}
- {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
+49 -15
View File
@@ -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_hash: "22182814216190331e0b69b3bc99493e6d69cc813f7ed937394986eecc1f5d11"
last_review: "2026-04-26"
config_version: "1.7.0"
config_hash: "1171380de6d3334be1f6eed04797cebe15e5b8ec2124e130b582c2e2097bde37"
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_bear_threshold_annualized: "-0.20"
funding_bull_threshold_annualized: "0.10"
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_min: "0.10"
delta_max: "0.15"
distance_otm_pct_min: "0.15"
delta_target: "0.18"
delta_min: "0.12"
delta_max: "0.22"
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:
+2 -1
View File
@@ -87,7 +87,8 @@ def _seed_dvol_history(ctx, *, when: datetime, spot: Decimal, dvol: Decimal):
try:
with transaction(conn):
ctx.repository.record_dvol_snapshot(
conn, DvolSnapshot(timestamp=when, dvol=dvol, eth_spot=spot)
conn,
DvolSnapshot(timestamp=when, asset="ETH", dvol=dvol, spot=spot),
)
finally:
conn.close()
+84
View File
@@ -0,0 +1,84 @@
"""Smoke tests for the ``cerbero-bite audit data`` CLI subcommand."""
from __future__ import annotations
import json
from datetime import UTC, datetime
from pathlib import Path
from click.testing import CliRunner
from cerbero_bite.cli import main as cli_main
from cerbero_bite.state import connect, run_migrations, transaction
def _seed_minimal_db(db_path: Path) -> None:
conn = connect(db_path)
try:
run_migrations(conn)
now = datetime.now(UTC).replace(microsecond=0)
with transaction(conn):
for asset, spot in (("ETH", "3000"), ("BTC", "80000")):
conn.execute(
"INSERT INTO market_snapshots(timestamp, asset, spot, "
"dvol, fetch_ok) VALUES (?,?,?,?,1)",
(now.isoformat(), asset, spot, "55"),
)
conn.execute(
"INSERT INTO option_chain_snapshots(timestamp, asset, "
"instrument_name, strike, expiry, option_type, bid, ask, iv) "
"VALUES (?, 'ETH', 'ETH-X', '3000', ?, 'C', '0.01', '0.02', '0.7')",
(now.isoformat(), now.isoformat()),
)
finally:
conn.close()
def test_audit_data_human_output(tmp_path: Path) -> None:
db = tmp_path / "state.sqlite"
_seed_minimal_db(db)
result = CliRunner().invoke(
cli_main, ["audit", "data", "--db", str(db), "--since", "1"]
)
assert result.exit_code == 0, result.output
assert "ETH — market_snapshots" in result.output
assert "BTC — market_snapshots" in result.output
assert "ETH — option_chain_snapshots" in result.output
def test_audit_data_json_output(tmp_path: Path) -> None:
db = tmp_path / "state.sqlite"
_seed_minimal_db(db)
result = CliRunner().invoke(
cli_main,
["audit", "data", "--db", str(db), "--since", "1", "--json"],
)
assert result.exit_code == 0, result.output
payload = json.loads(result.output)
assert "since" in payload
assert "until" in payload
assert set(payload["assets"]) == {"ETH", "BTC"}
assert "market" in payload["assets"]["ETH"]
assert "chain" in payload["assets"]["ETH"]
assert payload["assets"]["ETH"]["market"]["asset"] == "ETH"
def test_audit_data_single_asset_filter(tmp_path: Path) -> None:
db = tmp_path / "state.sqlite"
_seed_minimal_db(db)
result = CliRunner().invoke(
cli_main,
["audit", "data", "--db", str(db), "--asset", "BTC", "--json"],
)
assert result.exit_code == 0, result.output
payload = json.loads(result.output)
assert set(payload["assets"]) == {"BTC"}
def test_audit_data_missing_db_exits_2(tmp_path: Path) -> None:
# click.Path(exists=True) returns exit_code=2 (UsageError) for missing files.
result = CliRunner().invoke(
cli_main,
["audit", "data", "--db", str(tmp_path / "absent.sqlite")],
)
assert result.exit_code == 2
+448
View File
@@ -0,0 +1,448 @@
"""Unit tests for analysis.data_audit."""
from __future__ import annotations
import sqlite3
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from cerbero_bite.analysis.data_audit import (
ChainAuditReport,
MarketAuditReport,
audit_market_snapshots,
audit_option_chain,
)
from cerbero_bite.analysis.data_audit import ( # noqa: PLC2701
_detect_gaps,
_expected_ticks,
_max_zero_streak,
_pct,
)
from cerbero_bite.state import connect, run_migrations, transaction
def test_expected_ticks_basic() -> None:
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
# 12:00, 12:15, 12:30, 12:45 → 4 ticks before 13:00
assert _expected_ticks(since, until) == 4
def test_expected_ticks_inclusive_left_exclusive_right() -> None:
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 15, tzinfo=UTC)
# Only 12:00
assert _expected_ticks(since, until) == 1
def test_expected_ticks_unaligned_since_rounds_up() -> None:
since = datetime(2026, 5, 12, 12, 7, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 30, tzinfo=UTC)
# First aligned tick after 12:07 is 12:15, then 12:30 is excluded
assert _expected_ticks(since, until) == 1
def test_expected_ticks_non_multiple_span_uses_ceiling() -> None:
# Span = 16 min (12:00 → 12:16): ticks 12:00 AND 12:15 are < 12:16.
# floor(16/15) = 1 would under-count; the correct answer is 2.
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 16, tzinfo=UTC)
assert _expected_ticks(since, until) == 2
def test_expected_ticks_exact_quarter_boundary_is_excluded() -> None:
# Until landing exactly on a */15 tick: that tick is NOT counted
# because the window is half-open on the right.
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 45, tzinfo=UTC)
# Ticks at 12:00, 12:15, 12:30 → 3 (12:45 excluded)
assert _expected_ticks(since, until) == 3
def test_detect_gaps_returns_empty_when_no_gap() -> None:
ts = [
datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
datetime(2026, 5, 12, 12, 15, tzinfo=UTC),
datetime(2026, 5, 12, 12, 30, tzinfo=UTC),
]
assert _detect_gaps(ts) == ()
def test_detect_gaps_flags_above_threshold() -> None:
ts = [
datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
datetime(2026, 5, 12, 12, 45, tzinfo=UTC), # 45-min gap
datetime(2026, 5, 12, 13, 0, tzinfo=UTC),
]
gaps = _detect_gaps(ts)
assert len(gaps) == 1
assert gaps[0].gap_minutes == 45
assert gaps[0].prev_timestamp == ts[0]
assert gaps[0].next_timestamp == ts[1]
def test_detect_gaps_ignores_threshold_boundary() -> None:
# 20-min gap is exactly the threshold → NOT flagged (strict >)
ts = [
datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
datetime(2026, 5, 12, 12, 20, tzinfo=UTC),
]
assert _detect_gaps(ts) == ()
def test_max_zero_streak_empty() -> None:
assert _max_zero_streak([]) == 0
def test_max_zero_streak_no_zeros() -> None:
assert _max_zero_streak([1, 1, 1, 1]) == 0
def test_max_zero_streak_single_zero_block() -> None:
assert _max_zero_streak([1, 0, 0, 0, 1, 0, 1]) == 3
def test_max_zero_streak_all_zeros() -> None:
assert _max_zero_streak([0, 0, 0]) == 3
def _make_conn(tmp_path: Path) -> sqlite3.Connection:
conn = connect(tmp_path / "state.sqlite")
run_migrations(conn)
return conn
def _seed_market(
conn: sqlite3.Connection,
*,
asset: str,
ts: datetime,
spot: Decimal | None = Decimal("3000"),
dvol: Decimal | None = Decimal("55"),
dealer_net_gamma: Decimal | None = Decimal("-50000000"),
fetch_ok: int = 1,
) -> None:
conn.execute(
"INSERT INTO market_snapshots(timestamp, asset, spot, dvol, "
"dealer_net_gamma, fetch_ok) VALUES (?,?,?,?,?,?)",
(
ts.isoformat(),
asset,
str(spot) if spot is not None else None,
str(dvol) if dvol is not None else None,
str(dealer_net_gamma) if dealer_net_gamma is not None else None,
fetch_ok,
),
)
def test_audit_market_full_coverage(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
with transaction(conn):
for minute in (0, 15, 30, 45):
_seed_market(
conn,
asset="ETH",
ts=since.replace(minute=minute),
)
report = audit_market_snapshots(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.asset == "ETH"
assert report.expected_ticks == 4
assert report.actual_ticks == 4
assert report.coverage_pct == Decimal("100")
assert report.gaps == ()
assert report.fetch_ok_zero_count == 0
assert report.max_fetch_ok_zero_streak == 0
def test_audit_market_detects_gap_and_streak(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 30, tzinfo=UTC)
try:
with transaction(conn):
# 12:00 OK, 12:15 OK, gap (12:30, 12:45 missing), 13:00 fail,
# 13:15 fail, 13:30 outside window.
_seed_market(conn, asset="ETH", ts=since.replace(minute=0))
_seed_market(conn, asset="ETH", ts=since.replace(minute=15))
_seed_market(
conn, asset="ETH", ts=since.replace(hour=13, minute=0),
fetch_ok=0,
)
_seed_market(
conn, asset="ETH", ts=since.replace(hour=13, minute=15),
fetch_ok=0,
)
report = audit_market_snapshots(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.expected_ticks == 6
assert report.actual_ticks == 4
# 12:15 → 13:00 is a 45-min gap → flagged
assert len(report.gaps) == 1
assert report.gaps[0].gap_minutes == 45
assert report.fetch_ok_zero_count == 2
assert report.max_fetch_ok_zero_streak == 2
def test_audit_market_null_rate_per_column(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
with transaction(conn):
# 4 ticks, 1 with dealer_net_gamma=NULL → 25% null
for minute in (0, 15, 30, 45):
_seed_market(
conn,
asset="ETH",
ts=since.replace(minute=minute),
dealer_net_gamma=None if minute == 30 else Decimal("-50000000"),
)
report = audit_market_snapshots(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.null_rate_by_column["dealer_net_gamma"] == Decimal("0.25")
assert report.null_rate_by_column["spot"] == Decimal("0")
def test_pct_empty_returns_zero() -> None:
assert _pct([], 50) == 0
def test_pct_single_value_returns_that_value() -> None:
assert _pct([42], 50) == 42
assert _pct([42], 10) == 42
assert _pct([42], 90) == 42
def test_pct_median_odd_count() -> None:
assert _pct([10, 20, 30, 40, 50], 50) == 30
def test_pct_p10_p90_on_100_values() -> None:
# inclusive method on [1..100]:
# p10 interpolates to 10.9 → 11; p90 interpolates to 90.1 → 90.
values = list(range(1, 101))
assert _pct(values, 10) == 11
assert _pct(values, 90) == 90
def _seed_chain_row(
conn: sqlite3.Connection,
*,
asset: str,
ts: datetime,
instrument: str,
strike: str = "3000",
expiry: datetime | None = None,
option_type: str = "C",
bid: str | None = "0.01",
ask: str | None = "0.02",
iv: str | None = "0.7",
) -> None:
if expiry is None:
expiry = datetime(2026, 6, 12, 8, 0, tzinfo=UTC)
conn.execute(
"INSERT INTO option_chain_snapshots(timestamp, asset, instrument_name, "
"strike, expiry, option_type, bid, ask, iv) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(
ts.isoformat(),
asset,
instrument,
strike,
expiry.isoformat(),
option_type,
bid,
ask,
iv,
),
)
def test_audit_chain_full_coverage(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
with transaction(conn):
for minute in (0, 15, 30, 45):
ts = since.replace(minute=minute)
for i in range(10):
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument=f"ETH-EXP-{i}-C",
strike=str(3000 + i * 50),
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert isinstance(report, ChainAuditReport)
assert report.asset == "ETH"
assert report.expected_snapshots == 4
assert report.actual_snapshots == 4
assert report.coverage_pct == Decimal("100")
assert report.quotes_per_snap_median == 10
assert report.quotes_per_snap_p10 == 10
assert report.quotes_per_snap_p90 == 10
assert report.bid_gt_ask_count == 0
assert report.ask_zero_count == 0
assert report.iv_null_count == 0
assert report.iv_null_pct == Decimal("0")
def test_audit_chain_detects_bid_gt_ask_and_iv_null(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 30, tzinfo=UTC)
try:
with transaction(conn):
ts = since.replace(minute=0)
# Crossed BBO: bid > ask
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-A",
bid="0.10",
ask="0.05",
)
# Missing IV
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-B",
strike="3050",
iv=None,
)
# Normal row (control)
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-C",
strike="3100",
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.bid_gt_ask_count == 1
assert report.ask_zero_count == 0
assert report.iv_null_count == 1
# 1 of 3 quotes → 33.33%
assert report.iv_null_pct == Decimal("33.33")
def test_audit_chain_ask_zero_counted_separately(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 30, tzinfo=UTC)
try:
with transaction(conn):
ts = since.replace(minute=0)
# ask=0 → conta in ask_zero, NON in bid_gt_ask
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-A",
bid="0.05",
ask="0",
)
# ask negativo (impossibile economicamente) → ask_zero
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-B",
strike="3050",
bid="0.05",
ask="-0.01",
)
# Crossed BBO reale (ask>0) → bid_gt_ask
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-C",
strike="3100",
bid="0.10",
ask="0.05",
)
# Normal control
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-D",
strike="3150",
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.ask_zero_count == 2
assert report.bid_gt_ask_count == 1
def test_audit_chain_missing_snapshots(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
with transaction(conn):
# Only 2 of the expected 4 ticks have data
for minute in (0, 15):
_seed_chain_row(
conn,
asset="ETH",
ts=since.replace(minute=minute),
instrument=f"ETH-{minute}",
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.expected_snapshots == 4
assert report.actual_snapshots == 2
assert report.coverage_pct == Decimal("50")
def test_audit_chain_empty_window_returns_zero_coverage(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.expected_snapshots == 4
assert report.actual_snapshots == 0
assert report.coverage_pct == Decimal("0")
assert report.quotes_per_snap_median == 0
assert report.bid_gt_ask_count == 0
assert report.iv_null_count == 0
+15 -9
View File
@@ -38,7 +38,9 @@ def _ctx(tmp_path: Path) -> MagicMock:
# Default: every feed succeeds with sane mock values.
ctx.deribit = MagicMock()
ctx.deribit.spot_perp_price = AsyncMock(return_value=Decimal("3000"))
ctx.deribit.spot_perp_price = AsyncMock(
side_effect=lambda asset: Decimal("65000") if asset == "BTC" else Decimal("3000")
)
ctx.deribit.latest_dvol = AsyncMock(return_value=Decimal("55"))
ctx.deribit.realized_vol = AsyncMock(
return_value={
@@ -181,31 +183,35 @@ def _read_dvol_history(ctx: MagicMock) -> list[dict]:
@pytest.mark.asyncio
async def test_eth_snapshot_mirrors_into_dvol_history(tmp_path: Path) -> None:
async def test_snapshot_mirrors_each_asset_into_dvol_history(tmp_path: Path) -> None:
ctx = _ctx(tmp_path)
await collect_market_snapshot(ctx, assets=("ETH", "BTC"), now=_now())
rows = _read_dvol_history(ctx)
assert len(rows) == 1
assert Decimal(str(rows[0]["dvol"])) == Decimal("55")
assert Decimal(str(rows[0]["eth_spot"])) == Decimal("3000")
by_asset = {r["asset"]: r for r in rows}
assert set(by_asset) == {"ETH", "BTC"}
assert Decimal(str(by_asset["ETH"]["spot"])) == Decimal("3000")
assert Decimal(str(by_asset["BTC"]["spot"])) == Decimal("65000")
@pytest.mark.asyncio
async def test_btc_only_snapshot_does_not_touch_dvol_history(
async def test_btc_only_snapshot_mirrors_into_dvol_history(
tmp_path: Path,
) -> None:
ctx = _ctx(tmp_path)
await collect_market_snapshot(ctx, assets=("BTC",), now=_now())
assert _read_dvol_history(ctx) == []
rows = _read_dvol_history(ctx)
assert len(rows) == 1
assert rows[0]["asset"] == "BTC"
assert Decimal(str(rows[0]["spot"])) == Decimal("65000")
@pytest.mark.asyncio
async def test_eth_snapshot_skips_dvol_history_when_dvol_missing(
async def test_snapshot_skips_dvol_history_when_dvol_missing(
tmp_path: Path,
) -> None:
ctx = _ctx(tmp_path)
ctx.deribit.latest_dvol = AsyncMock(side_effect=RuntimeError("no dvol"))
await collect_market_snapshot(ctx, assets=("ETH",), now=_now())
# market_snapshots row still persisted, but dvol_history must stay empty
# because its schema enforces NOT NULL on dvol/eth_spot.
# because its schema enforces NOT NULL on dvol/spot.
assert _read_dvol_history(ctx) == []
+33 -2
View File
@@ -277,15 +277,46 @@ def test_record_dvol_snapshot_replaces_on_duplicate_timestamp(
ts = datetime(2026, 4, 27, 14, 0, tzinfo=UTC)
with transaction(conn):
repo.record_dvol_snapshot(
conn, DvolSnapshot(timestamp=ts, dvol=Decimal("50"), eth_spot=Decimal("3000"))
conn,
DvolSnapshot(
timestamp=ts, asset="ETH", dvol=Decimal("50"), spot=Decimal("3000")
),
)
repo.record_dvol_snapshot(
conn, DvolSnapshot(timestamp=ts, dvol=Decimal("55"), eth_spot=Decimal("3050"))
conn,
DvolSnapshot(
timestamp=ts, asset="ETH", dvol=Decimal("55"), spot=Decimal("3050")
),
)
rows = conn.execute("SELECT COUNT(*) FROM dvol_history").fetchone()
assert rows[0] == 1
def test_record_dvol_snapshot_keeps_assets_distinct_on_same_timestamp(
conn: sqlite3.Connection, repo: Repository
) -> None:
ts = datetime(2026, 4, 27, 14, 0, tzinfo=UTC)
with transaction(conn):
repo.record_dvol_snapshot(
conn,
DvolSnapshot(
timestamp=ts, asset="ETH", dvol=Decimal("50"), spot=Decimal("3000")
),
)
repo.record_dvol_snapshot(
conn,
DvolSnapshot(
timestamp=ts, asset="BTC", dvol=Decimal("45"), spot=Decimal("65000")
),
)
rows = conn.execute(
"SELECT asset, dvol, spot FROM dvol_history ORDER BY asset"
).fetchall()
assert [r["asset"] for r in rows] == ["BTC", "ETH"]
assert Decimal(str(rows[0]["spot"])) == Decimal("65000")
assert Decimal(str(rows[1]["spot"])) == Decimal("3000")
def test_manual_action_enqueue_consume_cycle(
conn: sqlite3.Connection, repo: Repository
) -> None: