diff --git a/src/cerbero_bite/config/schema.py b/src/cerbero_bite/config/schema.py index 9ab541d..a20e39d 100644 --- a/src/cerbero_bite/config/schema.py +++ b/src/cerbero_bite/config/schema.py @@ -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) diff --git a/src/cerbero_bite/runtime/option_chain_research_cycle.py b/src/cerbero_bite/runtime/option_chain_research_cycle.py new file mode 100644 index 0000000..e90a2e1 --- /dev/null +++ b/src/cerbero_bite/runtime/option_chain_research_cycle.py @@ -0,0 +1,194 @@ +"""Full-chain *research* option collector (§13-bis). + +Diverso dal collettore operativo (`option_chain_snapshot_cycle`): + +* finestra scadenze ``[now, now + expiry_max_days]`` — cattura TUTTE le + scadenze liquide (1g/1sett/2sett/1mese/3mesi), non solo quella nella + finestra DTE della strategia; +* entrambe le ali (nessun filtro moneyness di default), oppure entro + ``±moneyness_band_pct`` se configurato; +* popola ``book_depth_top3`` chiamando l'orderbook per ogni strumento + tenuto (1 call/strumento, concorrenza limitata da + ``book_depth_concurrency``) — così lo slippage reale è modellabile; +* scrive con ``source='research'`` per non confondersi con le righe + 'live'. + +Questo trasforma il dataset da "skew/premi medi" a backtest opzioni +vero, per-trade e standing. Best-effort come l'altro collettore: un +batch o un orderbook che falliscono non invalidano il resto. +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from typing import TYPE_CHECKING, Any + +from cerbero_bite.state import connect, transaction +from cerbero_bite.state.models import OptionChainQuoteRecord + +from cerbero_bite.runtime.option_chain_snapshot_cycle import ( + DEFAULT_BATCH_SIZE, + _fetch_tickers_in_batches, + _to_decimal_or_none, +) + +if TYPE_CHECKING: + from cerbero_bite.runtime.dependencies import RuntimeContext + +__all__ = ["collect_option_chain_research"] + +_log = logging.getLogger("cerbero_bite.runtime.option_chain_research") + + +def _underlying_price(ticker: dict[str, Any]) -> Decimal | None: + """Spot/index dell'underlying dal ticker, per il filtro moneyness.""" + for key in ("underlying_price", "index_price", "estimated_delivery_price"): + val = _to_decimal_or_none(ticker.get(key)) + if val is not None and val > 0: + return val + return None + + +async def _depth_for( + ctx: RuntimeContext, name: str, sem: asyncio.Semaphore +) -> int | None: + """Best-effort top-3 book depth per ``name`` (None se fallisce).""" + async with sem: + try: + return await ctx.deribit.orderbook_depth_top3(name) + except Exception as exc: + _log.debug("orderbook_depth_top3 failed for %s: %s", name, exc) + return None + + +async def collect_option_chain_research( + ctx: RuntimeContext, + *, + asset: str = "ETH", + now: datetime | None = None, + batch_size: int = DEFAULT_BATCH_SIZE, +) -> int: + """Collect + persist un singolo snapshot full-chain ``research`` per + ``asset``. Ritorna il numero di quote persistiti (0 su fallimento + best-effort o se il collettore è disabilitato).""" + rc = getattr(ctx.cfg, "research_collector", None) + if rc is None or not rc.enabled: + return 0 + + when = (now or datetime.now(UTC)).astimezone(UTC) + + expiry_from = when + expiry_to = when + timedelta(days=rc.expiry_max_days) + + try: + chain = await ctx.deribit.options_chain( + currency=asset.upper(), + expiry_from=expiry_from, + expiry_to=expiry_to, + min_open_interest=int(rc.open_interest_min), + ) + except Exception: + _log.exception("research option chain fetch failed") + return 0 + + if not chain: + _log.info("research option chain empty for %s in window", asset) + return 0 + + names = [meta.name for meta in chain] + tickers = await _fetch_tickers_in_batches(ctx, names, batch_size=batch_size) + + band = rc.moneyness_band_pct # Decimal | None + + # 1) costruisci i quote, applicando l'eventuale filtro moneyness. + kept: list[tuple[OptionChainQuoteRecord, dict[str, Any] | None]] = [] + for meta in chain: + ticker = tickers.get(meta.name) + + if band is not None and ticker is not None: + spot = _underlying_price(ticker) + if spot is not None: + moneyness = abs(meta.strike - spot) / spot + if moneyness > band: + continue # fuori dall'ala richiesta + + if ticker is None: + rec = OptionChainQuoteRecord( + timestamp=when, + asset=asset.upper(), + instrument_name=meta.name, + strike=meta.strike, + expiry=meta.expiry, + option_type=meta.option_type, + open_interest=int(meta.open_interest) + if meta.open_interest is not None + else None, + source="research", + ) + kept.append((rec, None)) + continue + + greeks = ticker.get("greeks") or {} + rec = OptionChainQuoteRecord( + timestamp=when, + asset=asset.upper(), + instrument_name=meta.name, + strike=meta.strike, + expiry=meta.expiry, + option_type=meta.option_type, + bid=_to_decimal_or_none(ticker.get("bid")), + ask=_to_decimal_or_none(ticker.get("ask")), + mid=_to_decimal_or_none(ticker.get("mark_price")), + iv=_to_decimal_or_none(ticker.get("mark_iv")), + delta=_to_decimal_or_none(greeks.get("delta")), + gamma=_to_decimal_or_none(greeks.get("gamma")), + theta=_to_decimal_or_none(greeks.get("theta")), + vega=_to_decimal_or_none(greeks.get("vega")), + open_interest=int(meta.open_interest) + if meta.open_interest is not None + else None, + volume_24h=( + int(ticker["volume_24h"]) + if ticker.get("volume_24h") is not None + else None + ), + source="research", + ) + kept.append((rec, ticker)) + + # 2) popola book_depth_top3 (concorrenza limitata) sugli strumenti tenuti. + if rc.fetch_book_depth and kept: + sem = asyncio.Semaphore(max(1, int(rc.book_depth_concurrency))) + depths = await asyncio.gather( + *(_depth_for(ctx, rec.instrument_name, sem) for rec, _ in kept) + ) + kept = [ + (rec.model_copy(update={"book_depth_top3": depth}), tk) + for (rec, tk), depth in zip(kept, depths, strict=True) + ] + + quotes = [rec for rec, _ in kept] + + persisted = 0 + try: + conn = connect(ctx.db_path) + try: + with transaction(conn): + persisted = ctx.repository.record_option_chain_snapshot(conn, quotes) + finally: + conn.close() + except Exception: + _log.exception("persist research option chain snapshot failed") + return 0 + + _log.info( + "option_chain_research persisted %d quote(s) for %s (%d expiries window<=%dd)", + persisted, + asset.upper(), + len({q.expiry for q in quotes}), + rc.expiry_max_days, + ) + return persisted diff --git a/src/cerbero_bite/runtime/orchestrator.py b/src/cerbero_bite/runtime/orchestrator.py index 976f9bc..516cbd0 100644 --- a/src/cerbero_bite/runtime/orchestrator.py +++ b/src/cerbero_bite/runtime/orchestrator.py @@ -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=" diff --git a/src/cerbero_bite/state/migrations/0007_option_chain_source.sql b/src/cerbero_bite/state/migrations/0007_option_chain_source.sql new file mode 100644 index 0000000..def7de5 --- /dev/null +++ b/src/cerbero_bite/state/migrations/0007_option_chain_source.sql @@ -0,0 +1,18 @@ +-- 0007_option_chain_source.sql — distingue le righe del collettore +-- +-- Due collettori scrivono ora su option_chain_snapshots: +-- * 'live' — collettore operativo, finestra DTE della strategia +-- (cfg.structure.dte_min..dte_max), 1 scadenza, no depth. +-- * 'research' — collettore full-chain (tutte le scadenze <=95g, +-- entrambe le ali, book_depth_top3 popolato) per il +-- backtest opzioni vero (per-trade e standing put). +-- +-- Le righe storiche pre-migrazione sono tutte 'live' (DEFAULT). Il +-- backtest per-trade/standing filtra su source='research'. + +ALTER TABLE option_chain_snapshots ADD COLUMN source TEXT NOT NULL DEFAULT 'live'; + +CREATE INDEX idx_option_chain_source + ON option_chain_snapshots(asset, source, timestamp DESC); + +PRAGMA user_version = 7; diff --git a/src/cerbero_bite/state/models.py b/src/cerbero_bite/state/models.py index 336e092..0e39cb1 100644 --- a/src/cerbero_bite/state/models.py +++ b/src/cerbero_bite/state/models.py @@ -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): diff --git a/src/cerbero_bite/state/repository.py b/src/cerbero_bite/state/repository.py index b3c3373..504748d 100644 --- a/src/cerbero_bite/state/repository.py +++ b/src/cerbero_bite/state/repository.py @@ -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" + ), )