diff --git a/src/cerbero_bite/gui/pages/6_πŸ“_Calibrazione.py b/src/cerbero_bite/gui/pages/6_πŸ“_Calibrazione.py index ea97136..02a5ee5 100644 --- a/src/cerbero_bite/gui/pages/6_πŸ“_Calibrazione.py +++ b/src/cerbero_bite/gui/pages/6_πŸ“_Calibrazione.py @@ -16,6 +16,7 @@ from __future__ import annotations import os from dataclasses import dataclass from datetime import UTC, datetime, timedelta +from decimal import Decimal from pathlib import Path import pandas as pd @@ -23,6 +24,7 @@ import plotly.graph_objects as go import streamlit as st from cerbero_bite.config.loader import load_strategy +from cerbero_bite.core.adaptive_threshold import compute_adaptive_threshold from cerbero_bite.gui.data_layer import ( DEFAULT_DB_PATH, humanize_dt, @@ -242,6 +244,131 @@ def _render_metric(spec: MetricSpec, records: list[MarketSnapshotRecord]) -> Non _percentiles_strip(s) +def _render_adaptive_gate_panel( + strategy: object | None, + records: list[MarketSnapshotRecord], +) -> None: + """Pannello informativo sul gate IV-RV adattivo (read-only).""" + if strategy is None: + return + + try: + entry = strategy.entry # type: ignore[attr-defined] + except AttributeError: + return + + if not getattr(entry, "iv_minus_rv_filter_enabled", False): + st.subheader("🎯 Gate IV-RV adattivo") + st.info("Gate IV-RV disabilitato nel profilo corrente.") + st.divider() + return + + st.subheader("🎯 Gate IV-RV adattivo") + + # records DESC (newest first) β†’ history ASC con NULL/fetch_ok=0 esclusi + iv_rv_history: list[Decimal] = [] + for r in reversed(records): + if r.fetch_ok and r.iv_minus_rv is not None: + iv_rv_history.append(r.iv_minus_rv) + + n_ticks = len(iv_rv_history) + target = int(getattr(entry, "iv_minus_rv_window_target_days", 60)) + min_days = int(getattr(entry, "iv_minus_rv_window_min_days", 30)) + n_days = n_ticks // 96 + + if n_days < 1: + status = f"🟑 Warmup hard ({n_ticks}/96 tick)" + elif n_days < min_days: + status = f"🟑 Warmup ({n_days}/{min_days}g β€” finestra crescente)" + elif n_days < target: + status = f"🟒 Attivo (finestra {min_days}g, target {target}g)" + else: + status = f"🟒 Attivo (finestra stabile {target}g)" + st.markdown(f"**Status:** {status}") + + # Latest tick + iv_rv_now: Decimal | None = None + dvol_now: Decimal | None = None + latest_ts: datetime | None = None + for r in records: # records DESC + if r.fetch_ok: + iv_rv_now = r.iv_minus_rv + dvol_now = r.dvol + latest_ts = r.timestamp + break + + adaptive_on = bool(getattr(entry, "iv_minus_rv_adaptive_enabled", False)) + floor = Decimal(str(getattr(entry, "iv_minus_rv_min", "0"))) + + if adaptive_on: + percentile = Decimal( + str(getattr(entry, "iv_minus_rv_percentile", "0.25")) + ) + try: + threshold = compute_adaptive_threshold( + history=iv_rv_history, + percentile=percentile, + absolute_floor=floor, + min_days=min_days, + target_days=target, + ) + except ValueError as exc: + st.warning(f"Configurazione gate non valida: {exc}") + threshold = None + + c1, c2, c3 = st.columns(3) + pct_label = int(percentile * 100) + c1.metric( + f"Soglia P{pct_label} rolling", + f"{threshold:.2f}" if threshold is not None else "β€”", + help="Soglia adattiva = max(percentile, floor)", + ) + c2.metric( + "IV-RV ultimo tick", + f"{iv_rv_now:.2f}" if iv_rv_now is not None else "β€”", + ) + c3.metric("Floor assoluto", f"{floor:.2f}") + + if threshold is not None and iv_rv_now is not None: + verdict = "βœ… PASS" if iv_rv_now >= threshold else "❌ SKIP" + st.markdown(f"**Decisione hypothetical:** {verdict}") + else: + st.write(f"ModalitΓ  statica: floor = {floor} vol pts") + + if bool(getattr(entry, "vol_of_vol_guard_enabled", False)): + st.markdown("---") + st.markdown("**Vol-of-Vol guard**") + threshold_pt = Decimal( + str(getattr(entry, "vol_of_vol_threshold_pt", "5")) + ) + lookback_h = int(getattr(entry, "vol_of_vol_lookback_hours", 24)) + + # Find tick closest to latest_ts - lookback hours, tolerance 15 min + dvol_lookback: Decimal | None = None + if latest_ts is not None and dvol_now is not None: + target_ts = latest_ts - timedelta(hours=lookback_h) + best_delta = timedelta(minutes=15) + for r in records: + if not r.fetch_ok or r.dvol is None: + continue + d = abs(r.timestamp - target_ts) + if d <= best_delta: + best_delta = d + dvol_lookback = r.dvol + + if dvol_lookback is not None and dvol_now is not None: + delta = abs(dvol_now - dvol_lookback) + c1, c2 = st.columns(2) + c1.metric(f"|Ξ”DVOL {lookback_h}h|", f"{delta:.2f}") + c2.metric("Soglia VoV", f"{threshold_pt:.2f}") + verdict = "βœ… PASS" if delta < threshold_pt else "❌ SKIP" + st.markdown(f"**Verdict:** {verdict}") + else: + st.info(f"Lookback {lookback_h}h non disponibile (gap dati).") + + st.divider() + + def render() -> None: st.title("πŸ“ Calibrazione") st.caption( @@ -313,6 +440,8 @@ def render() -> None: specs = _metric_specs(strategy) + _render_adaptive_gate_panel(strategy, records) + for spec in specs: _render_metric(spec, records) st.divider()