L13 — Lookahead, test di invarianza e i cinque controlli sui dati

import matplotlib.pyplot as plt
import numpy as np

from cvbook.dati import carica
from cvbook.regole import esegui

df = carica("btcusdt").sort("data")
prezzi = df["chiusura"].to_numpy()
FINESTRA = 20  # PROVA / TRY: qualunque valore (vedi esercizio 1) — il rapporto resta enorme
COSTO = 0.0012  # PROVA / TRY: 0,0006 · 0,0012 · 0,0025


def media_mobile(p: np.ndarray, finestra: int) -> np.ndarray:
    cumulata = np.concatenate([[0.0], np.cumsum(p)])
    m = np.full(len(p), np.nan)
    m[finestra - 1:] = (cumulata[finestra:] - cumulata[:-finestra]) / finestra
    return m

media = media_mobile(prezzi, FINESTRA)
segnale = np.nan_to_num(np.where(prezzi > media, 1.0, 0.0))

# Versione CAUSALE: la posizione di oggi usa il segnale di ieri.
causale = np.zeros(len(prezzi))
causale[1:] = segnale[:-1]

# Versione CON LOOKAHEAD: la posizione di oggi usa il segnale di oggi, cioe'
# un'informazione che al momento di decidere non esisteva ancora.
# NON TOCCARE / DO NOT CHANGE: è sbagliata apposta, per il confronto — non è
# un bug da sistemare.
# It's wrong on purpose, for the comparison — not a bug to fix.
con_lookahead = segnale.copy()

a = esegui(prezzi, causale, costo=COSTO)
b = esegui(prezzi, con_lookahead, costo=COSTO)

with avvio.figura("schermo"):
    fig, ax = plt.subplots()
    ax.semilogy(b["curva"], linewidth=2, label=f"con lookahead — {b['finale']:,.0f}x")
    ax.semilogy(a["curva"], linewidth=2, linestyle="--",
                label=f"causale — {a['finale']:,.1f}x")
    ax.set_ylabel("Capitale (scala log)")
    ax.set_xlabel("Giorni")
    ax.legend()
    plt.show()

print(f"versione causale:      {a['finale']:12,.2f}x")
print(f"versione con lookahead:{b['finale']:12,.2f}x")
print(f"rapporto:              {b['finale'] / a['finale']:12,.0f} volte")
print("\nUna riga. Il risultato non e' un po' piu' ottimista: e' impossibile, "
      "ottenuto da una macchina che sapeva in anticipo come sarebbe finita la "
      "giornata.")

def test_invarianza(funzione, p: np.ndarray, tagli=(400, 1200, 2400)) -> bool:
    """True se `funzione(p)` non cambia il passato quando arrivano dati nuovi."""
    completa = funzione(p)
    tutto_bene = True
    for taglio in tagli:
        parziale = funzione(p[:taglio])
        uguali = np.allclose(parziale, completa[:taglio], equal_nan=True)
        if not uguali:
            primo = int(np.argmax(~np.isclose(parziale, completa[:taglio],
                                              equal_nan=True)))
            print(f"  troncando a {taglio}: DIVERSO, prima differenza al giorno {primo}")
            tutto_bene = False
        else:
            print(f"  troncando a {taglio}: identico")
    return tutto_bene


def regola_causale(p: np.ndarray) -> np.ndarray:
    s = np.nan_to_num(np.where(p > media_mobile(p, FINESTRA), 1.0, 0.0))
    pos = np.zeros(len(p))
    pos[1:] = s[:-1]
    return pos


def regola_con_lookahead(p: np.ndarray) -> np.ndarray:
    return np.nan_to_num(np.where(p > media_mobile(p, FINESTRA), 1.0, 0.0))


def regola_normalizzata_male(p: np.ndarray) -> np.ndarray:
    """Errore diffusissimo: normalizzare usando media e deviazione di TUTTO."""
    z = (p - p.mean()) / p.std()
    return (z > 0).astype(float)


print("regola causale:")
print("  esito:", "PASSA" if test_invarianza(regola_causale, prezzi) else "FALLISCE")
print("\nregola con lookahead:")
print("  esito:", "PASSA" if test_invarianza(regola_con_lookahead, prezzi) else "FALLISCE")
print("\nregola normalizzata sull'intero periodo:")
print("  esito:", "PASSA" if test_invarianza(regola_normalizzata_male, prezzi) else "FALLISCE")

def controlla(nome: str) -> None:
    d = carica(nome).sort("data")
    date = d["data"].to_list()
    chiusura = d["chiusura"].to_numpy()
    r = chiusura[1:] / chiusura[:-1] - 1.0

    print(f"\n=== {nome} ===")

    # 1. righe contro calendario
    attesi = (date[-1] - date[0]).days + 1
    print(f"1. righe: {len(date)} su {attesi} giorni di calendario "
          f"({len(date) / attesi:.1%} di copertura)")

    # 2. i venti movimenti piu' grandi
    estremi = np.argsort(np.abs(r))[-5:][::-1]
    print("2. i cinque movimenti piu' grandi:")
    for i in estremi:
        print(f"     {date[i + 1]}  {r[i]:+7.1%}")

    # 3. giorni a variazione esattamente zero
    zeri = int(np.sum(r == 0.0))
    print(f"3. giorni a variazione esattamente zero: {zeri} "
          f"({'sospetti: probabile riempimento' if zeri > 3 else 'ok'})")

    # 4. coerenza fra massimo, minimo, apertura e chiusura
    m, mi = d["massimo"].to_numpy(), d["minimo"].to_numpy()
    ap, ch = d["apertura"].to_numpy(), d["chiusura"].to_numpy()
    incoerenti = int(np.sum((m < ch) | (m < ap) | (mi > ch) | (mi > ap) | (m < mi)))
    print(f"4. barre incoerenti (max sotto la chiusura, ecc.): {incoerenti}")

    # 5. volume all'inizio e alla fine
    v = d["volume"].to_numpy()
    print(f"5. volume mediano dei primi 30 giorni: {np.median(v[:30]):,.0f}")
    print(f"   volume mediano degli ultimi 30:     {np.median(v[-30:]):,.0f}")
    if np.median(v[:30]) < np.median(v[-30:]) / 10:
        print("   → l'inizio della serie e' molto sottile: quei prezzi esistono, "
              "ma non erano ottenibili in quantita'.")


controlla("btcusdt")
controlla("lunausdt")