PythagorasGoal/src/live/livefeed.py

"""FEED LIVE EFFIMERO per il segnale SKH01 (book a 230m) — NON tocca i dati certificati su disco.

SKH01 decide su griglia 230m: per eseguirlo fedelmente il segnale serve fresco all'ultima barra
chiusa. Il rebuild certificato (rebuild_history.py) gira 1×/giorno e fa un rebuild COMPLETO (pesante):
girarlo ogni ora sarebbe sbagliato e violerebbe la regola "aggiornare lo storico SOLO con
rebuild_history + certificare". Quindi qui NON scriviamo su disco: carichiamo il 5m CERTIFICATO e gli
appendiamo IN MEMORIA una coda recente presa da Deribit PUBBLICO (ccxt, tokenless, STESSO simbolo
inverse del feed certificato -> prezzi entro ~3 bps). I dati certificati restano la verità su disco;
questa estensione vive solo nel processo live e per il calcolo del segnale.

Robusto ai fallimenti: qualunque errore di rete/fetch -> ritorna il feed certificato invariato (il
runner degrada a "fermo all'ultimo dato certificato", mai opera a cieco). Solo SKH01 ne ha bisogno:
TP01 è giornaliero e gira bene sul feed certificato.
"""
from __future__ import annotations

import logging
import time

import pandas as pd

from src.data.downloader import load_data

# STESSO simbolo del feed certificato (vedi scripts/analysis/rebuild_history.DERIBIT_INSTR):
# inverse USD perp, storia lunga, entro ~3 bps dal lineare USDC su cui eseguiamo.
DERIBIT_SYMBOL = {"BTC": "BTC/USD:BTC", "ETH": "ETH/USD:ETH"}
SCHEMA = ["timestamp", "open", "high", "low", "close", "volume"]

_LOG = logging.getLogger(__name__)

# CAUSA dell'ultimo fallback. Il fallback resta SILENZIOSO come comportamento (mai operare a cieco
# > mai operare, scelta del 26/07): quello che NON deve restare silenzioso e' il PERCHE'.
#
# Perche' esiste (2026-07-29): dopo il riavvio della VPS delle 04:11 UTC il feed e' ricaduto sul
# certificato in 6 giri orari su 8 fra le 05:00 e le 12:00 -> fino a 685 min di eta', cioe' la
# latenza d'uscita di SKH01 passata da ~1h a ~11h. L'allerta del 26/07 ha segnalato ogni volta
# (ha funzionato), ma la CAUSA non era recuperabile dai log: l'eccezione di pagina viene ingoiata
# con un `break` e `fresh_5m` ritorna il certificato senza lasciare traccia. L'unica riga sulla
# causa era CABLATA nell'allerta ("fetch pubblico KO") = una presunzione stampata come misura.
# ⚠️ La causa di QUEL giorno resta IGNOTA e tale deve restare scritta: le prove sono solo
# circostanziali (i giri falliti durano quanto quelli riusciti, 17-47s -> errore immediato, non
# timeout; nello stesso giorno il percorso del CONTO, rete diversa ma stesso venue a valle, dava
# ReadTimeout/404/502). Ipotesi non distinguibili a posteriori: rate limit per-IP del venue,
# contesa di rete/CPU sulla VPS al minuto tondo, degrado post-riavvio.
# Questa variabile serve a che la PROSSIMA occorrenza sia un dato e non un'ipotesi.
_LAST_ERROR: str | None = None


def last_fetch_error() -> str | None:
    """Perche' l'ultima `fresh_5m` non ha attaccato una coda fresca COMPLETA.

    `None` = coda attaccata senza errori. Non-None = fallback al certificato **oppure** coda
    troncata a meta' paginazione (la paginazione va in avanti: se cade a pagina N mancano le
    barre piu' RECENTI, quindi in entrambi i casi il feed e' piu' vecchio di quanto sembri).
    Chi decide guarda `feed_age_minutes`; questa dice il perche'."""
    return _LAST_ERROR


def _note_error(msg: str) -> None:
    """Registra E logga la ragione del fallback. Non solleva: il chiamante prosegue sul certificato."""
    global _LAST_ERROR
    _LAST_ERROR = msg
    _LOG.warning("fresh_5m: coda fresca non attaccata, fallback al feed certificato — %s", msg)


def _fetch_recent_5m(symbol: str, lookback_days: int) -> pd.DataFrame:
    """Coda recente di 5m da Deribit pubblico (ccxt). Paginazione in avanti. Solo letture pubbliche."""
    import ccxt
    ex = ccxt.deribit({"enableRateLimit": True})
    tf_ms = 5 * 60 * 1000
    since = int((time.time() - lookback_days * 86400) * 1000)
    rows: dict[int, list] = {}
    guard = 0
    while guard < 200:
        guard += 1
        try:
            r = ex.fetch_ohlcv(symbol, "5m", since=since, limit=1000)
        except Exception as e:
            # E' QUI che il guasto diventa invisibile: l'errore di pagina viene ingoiato e la
            # funzione ritorna comunque cio' che ha raccolto finora. Se a fallire e' la PRIMA
            # pagina il risultato e' un DataFrame VUOTO, che a valle e' indistinguibile da
            # "Deribit non ha barre" — ed e' il ramo che il 29/07 ha cancellato la diagnosi.
            _note_error(f"{type(e).__name__}: {e} (pagina {guard}, {symbol})")
            break
        r = [x for x in r if int(x[0]) >= since]
        if not r:
            break
        for x in r:
            t = int(x[0])
            rows[t] = [t, float(x[1]), float(x[2]), float(x[3]), float(x[4]), float(x[5] or 0)]
        nxt = int(r[-1][0]) + tf_ms
        if nxt <= since:
            break
        since = nxt
    if not rows:
        return pd.DataFrame(columns=SCHEMA)
    return pd.DataFrame(rows.values(), columns=SCHEMA).sort_values("timestamp").reset_index(drop=True)


def merge_tail(base: pd.DataFrame, tail: pd.DataFrame) -> pd.DataFrame:
    """Fonde la coda fresca sul feed certificato: concat, dedup per timestamp (la coda VINCE sui
    duplicati, ma le barre certificate storiche restano), riordina. Mantiene lo schema di load_data
    (inclusa 'datetime' se presente). PURA, testabile senza rete."""
    if tail is None or tail.empty:
        return base
    cols = [c for c in SCHEMA if c in base.columns]
    t = tail[[c for c in SCHEMA if c in tail.columns]].copy()
    merged = pd.concat([base[cols], t], ignore_index=True)
    merged = merged.drop_duplicates("timestamp", keep="last").sort_values("timestamp").reset_index(drop=True)
    # ricostruisci 'datetime' coerente (build_frames non la usa, ma load_data la espone)
    merged["datetime"] = pd.to_datetime(merged["timestamp"], unit="ms", utc=True)
    return merged


def feed_age_minutes(df5: pd.DataFrame, now_ms: int | None = None) -> float | None:
    """Eta' in MINUTI dell'ultima barra 5m del feed passato. None se non interpretabile.

    ⚠️ E' la metrica che dice se `fresh_5m` ha davvero attaccato la coda fresca o se e' ricaduta
    in SILENZIO sul certificato: il fallback non solleva e non logga, quindi senza questa misura
    un fetch pubblico rotto e' invisibile. Conta perche' la latenza d'uscita di SKH01 e' dove vive
    la qualita' del path live (misurata il 2026-07-26): col feed fresco l'uscita e' rilevata entro
    ~1 ora dal tocco del livello, sul solo certificato si allunga fino a ~1 GIORNO (il rebuild
    certificato gira 1x/giorno). PURA, testabile senza rete."""
    if df5 is None or len(df5) == 0 or "timestamp" not in df5:
        return None
    try:
        last = int(df5["timestamp"].iloc[-1])
    except (ValueError, TypeError):
        return None
    now = int(pd.Timestamp.now(tz="UTC").timestamp() * 1000) if now_ms is None else int(now_ms)
    return max(0.0, (now - (last + 5 * 60_000)) / 60_000.0)


def fresh_5m(asset: str, lookback_days: int = 12) -> pd.DataFrame:
    """Feed 5m certificato + coda recente effimera (in-memory). Fallback al certificato su errore.

    NB: il fallback e' SILENZIOSO per scelta (mai operare a cieco > mai operare). Chi esegue deve
    misurare la freschezza di cio' che riceve con `feed_age_minutes` — vedi book.book_report, che
    la espone come `skh_feed_age_min`, e book_execute, che allerta. La RAGIONE del fallback resta
    disponibile in `last_fetch_error()` fino alla chiamata successiva."""
    global _LAST_ERROR
    _LAST_ERROR = None
    base = load_data(asset, "5m")
    sym = DERIBIT_SYMBOL.get(asset)
    if sym is None:
        _note_error(f"nessun simbolo Deribit per l'asset {asset}")
        return base
    try:
        tail = _fetch_recent_5m(sym, lookback_days)
    except Exception as e:
        _note_error(f"{type(e).__name__}: {e} ({sym})")
        return base
    if tail is None or len(tail) == 0:
        # Coda vuota senza eccezione propagata: o _fetch_recent_5m ha gia' registrato l'errore
        # di pagina, o Deribit ha davvero risposto senza barre. Sono due guasti diversi e vanno
        # detti diversi, senno' si ricade nel "non vedo = va tutto bene" gia' pagato due volte.
        if _LAST_ERROR is None:
            _note_error(f"nessuna barra restituita da {sym} (risposta vuota, nessuna eccezione)")
        return base
    return merge_tail(base, tail)