Compare commits
23 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1a46abd5cf | |||
| 1eeb53650e | |||
| 589d003e5c | |||
| ce475da173 | |||
| 647e3e565f | |||
| a5a39a6d27 | |||
| 839285d75b | |||
| 16d73c309a | |||
| 374822a931 | |||
| 923acd79f6 | |||
| 197c9db74d | |||
| 06ce2c0eb2 | |||
| bb2ca425a7 | |||
| d6af69f4cb | |||
| aeac8f2a95 | |||
| 75fe803296 | |||
| ea5c612446 | |||
| 9e2216d202 | |||
| 35ac92e938 | |||
| 07a8bbf5c8 | |||
| 0c6e462545 | |||
| 569df334dc | |||
| 19695e4730 |
+4
-1
@@ -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
|
||||
|
||||
@@ -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
@@ -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`
|
||||
|
||||
@@ -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`).
|
||||
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
"""
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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."""
|
||||
|
||||
@@ -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]:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)"
|
||||
|
||||
@@ -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"))
|
||||
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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="
|
||||
|
||||
@@ -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;
|
||||
@@ -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):
|
||||
|
||||
@@ -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
@@ -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
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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) == []
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user