feat(research): collettore full-chain opzioni con book_depth e colonna source
Aggiunge un secondo collettore opzioni indipendente dal live, pensato per
trasformare il dataset da "skew/premi medi" a backtest opzioni vero
(per-trade e standing put).
- option_chain_research_cycle.py: cattura tutte le scadenze <= expiry_max_days
(1g..3mesi) ed entrambe le ali (OI>=100, filtro moneyness opzionale),
popolando book_depth_top3 via orderbook_depth_top3 (concorrenza limitata)
cosi' lo slippage reale e' modellabile. Best-effort come il collettore live.
- migrazione 0007: colonna `source` su option_chain_snapshots ('live' default
per le righe storiche, 'research' per il nuovo collettore) + indice
(asset, source, timestamp). user_version 6 -> 7.
- ResearchCollectorConfig (schema): blocco `research_collector`, enabled=false
di default (costo API non trascurabile); cron orario, expiry_max_days=95.
- orchestrator: job `option_chain_research` schedulato solo se data-analysis
attiva E research_collector.enabled.
- repository: insert+mapper estesi con `source`; list_option_chain_snapshots
accetta un filtro `source` per le query di backtest.
Verificato in isolamento (immagine + interprete reali): import, migrazione,
round-trip repo. Suite: 527 passed, zero regressioni vs baseline pristina.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -359,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
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -383,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)
|
||||
|
||||
@@ -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,6 +37,9 @@ 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
|
||||
@@ -320,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),
|
||||
@@ -354,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="
|
||||
|
||||
@@ -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;
|
||||
@@ -178,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):
|
||||
|
||||
@@ -547,6 +547,7 @@ class Repository:
|
||||
q.open_interest,
|
||||
q.volume_24h,
|
||||
q.book_depth_top3,
|
||||
q.source,
|
||||
)
|
||||
for q in quotes
|
||||
]
|
||||
@@ -554,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)
|
||||
@@ -569,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))
|
||||
@@ -925,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"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user