feat(gui): pannello informativo Gate IV-RV adattivo in Calibrazione

Mostra status (warmup/attivo), soglia P25 rolling corrente, IV-RV
ultimo tick, floor assoluto, decisione hypothetical e sezione
Vol-of-Vol guard. Read-only: i percentili statici esistenti
restano per analisi.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-05-08 23:14:34 +00:00
parent 080acf829d
commit 64f4d4e09e
@@ -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()