L10 — Più parametri, adattamenti perfetti e inutili

import matplotlib.pyplot as plt
import numpy as np

from cvbook import seed_for
from cvbook.dati import carica
from cvbook.metriche import rendimenti

r = rendimenti(carica("btcusdt").sort("data")["chiusura"].to_numpy())
meta = len(r) // 2
dentro_campione, fuori_campione = r[:meta], r[meta:]
print(f"{len(r)} giorni: {len(dentro_campione)} per costruire, "
      f"{len(fuori_campione)} mai visti")

CONDIZIONI_MAX = 25   # <- quante aggiunte al massimo
DISPONIBILI = 400     # <- quante condizioni ci sono nel cassetto
                      # PROVA / TRY: 2000 (vedi esercizio 1)

# In QUESTO esperimento i costi si tengono a zero, e va detto perche'.
# Aumentando le condizioni il segnale cambia stato piu' spesso, quindi opera di
# piu': con i costi dentro, la curva mescolerebbe due effetti diversi — i gradi
# di liberta' e la frequenza. Qui vogliamo isolare il primo. Il secondo lo
# guardiamo a parte, nella cella 2, che e' altrettanto istruttiva.
# In THIS experiment costs are held at zero, and here's why: more conditions
# means the signal flips more often, so it trades more — with costs in, the
# curve would blend two different effects (degrees of freedom and frequency).
# Here we want to isolate the first. We look at the second separately in
# cell 2, which is just as instructive.
COSTO = 0.0
# NON TOCCARE / DO NOT CHANGE: deve restare zero qui — è la cella 2 che
# rimette i costi per isolare il secondo effetto, non questa.
# It must stay zero here — it's cell 2 that puts costs back in to isolate
# the second effect, not this one.


def risultato(rend: np.ndarray, posizione: np.ndarray, costo: float = COSTO) -> float:
    movimenti = np.abs(np.diff(np.concatenate([[0.0], posizione])))
    return float(np.prod(1 + posizione * rend - movimenti * costo))


def ricerca_per_aggiunte(rumore: np.ndarray, passi: int) -> tuple[list[float], list[float]]:
    """Costruzione per aggiunte successive, come si fa davvero.

    A ogni passo si prova ad aggiungere ciascuna delle condizioni disponibili e
    si tiene quella che migliora di piu' il risultato **sui dati che si stanno
    guardando**. Ci si ferma quando nessuna aggiunta migliora — cioe' quando una
    persona reale smetterebbe.

    La regola sta dentro al mercato quando la somma delle condizioni scelte e'
    positiva: cosi' l'esposizione resta attorno alla meta' del tempo e l'unica
    cosa che cambia e' il numero di gradi di liberta'.
    """
    scelte: set[int] = set()
    somma = np.zeros(rumore.shape[1])
    corrente = 0.0
    curva_dentro, curva_fuori = [], []

    for _ in range(passi):
        migliore_indice, migliore_valore = None, -np.inf
        for k in range(len(rumore)):
            if k in scelte:
                continue
            posizione = ((somma + rumore[k]) > 0).astype(float)
            valore = risultato(dentro_campione, posizione[:meta])
            if valore > migliore_valore:
                migliore_indice, migliore_valore = k, valore

        if migliore_valore <= corrente:
            break  # nessuna aggiunta migliora: ci si ferma

        scelte.add(migliore_indice)
        somma = somma + rumore[migliore_indice]
        corrente = migliore_valore
        posizione = (somma > 0).astype(float)
        curva_dentro.append(migliore_valore)
        curva_fuori.append(risultato(fuori_campione, posizione[meta:]))

    return curva_dentro, curva_fuori


rng = np.random.default_rng(seed_for("lab-dimensionalita"))
# Le condizioni sono rumore puro allineato ai giorni: nessuna informazione dentro.
rumore = rng.normal(size=(DISPONIBILI, len(r)))
dentro, fuori = ricerca_per_aggiunte(rumore, CONDIZIONI_MAX)
CONDIZIONI = list(range(1, len(dentro) + 1))

for n, d, f in zip(CONDIZIONI, dentro, fuori):
    print(f"{n:2d} condizioni →  dentro campione {d:10.2f}x   fuori campione {f:8.2f}x")
print(f"\nla ricerca si e' fermata dopo {len(dentro)} aggiunte: nessuna condizione "
      f"rimasta migliorava piu' il risultato che si stava guardando.")

with avvio.figura("schermo"):
    fig, ax = plt.subplots()
    ax.plot(CONDIZIONI, dentro, marker="o", linewidth=2,
            label="sulla parte usata per costruire")
    ax.plot(CONDIZIONI, fuori, marker="s", linewidth=2, linestyle="--",
            label="sulla parte mai vista")
    ax.axhline(1.0, linestyle=":", linewidth=1)
    ax.set_yscale("log")
    ax.set_xlabel("Condizioni aggiunte alla regola (tutte generate a caso)")
    ax.set_ylabel("Capitale finale (volte, scala log)")
    ax.legend()
    plt.show()

rng2 = np.random.default_rng(seed_for("lab-dimensionalita-costi"))
print(f"{'condizioni':>11s} {'operazioni':>11s} {'senza costi':>13s} {'con 0,12%':>12s}")
for n in CONDIZIONI:
    indici = rng2.integers(0, DISPONIBILI, size=n)
    posizione = (rumore[indici].sum(axis=0) > 0).astype(float)
    movimenti = float(np.abs(np.diff(np.concatenate([[0.0], posizione]))).sum())
    senza = risultato(r, posizione, costo=0.0)
    con = risultato(r, posizione, costo=0.0012)
    print(f"{n:11d} {movimenti:11.0f} {senza:12.2f}x {con:11.2f}x")

print("\nDue effetti diversi che spesso vengono confusi: i gradi di liberta' "
      "gonfiano il risultato apparente, la frequenza lo erode. Nella pratica "
      "agiscono insieme, ed e' per questo che vanno misurati separatamente.")

scelte = {
    "quale asset": 3,
    "quale intervallo (giorn./orario/settim.)": 3,
    "da quando parte la serie": 4,
    "quale prezzo (chiusura, apertura, medio)": 3,
    "media semplice o pesata": 2,
    "cosa fare quando si e' fuori": 2,
    "quale costo per operazione": 3,
    "quando si esegue": 3,
    "ogni quanto si controlla il segnale": 3,
}

totale = 1
for nome, quante in scelte.items():
    totale *= quante
    print(f"  {nome:>42s}: {quante} valori plausibili")
print(f"\ncombinazioni nascoste dietro «un solo parametro»: {totale:,}")
print("Non le hai provate tutte. Ne hai provata UNA, per abitudine, e non l'hai "
      "contata. Se il risultato fosse venuto brutto ne avresti cambiata qualcuna.")

prezzi = carica("btcusdt").sort("data")["chiusura"].to_numpy()


def sopra_media(p: np.ndarray, finestra: int, ritardo: int = 1) -> np.ndarray:
    cumulata = np.concatenate([[0.0], np.cumsum(p)])
    media = np.full(len(p), np.nan)
    media[finestra - 1:] = (cumulata[finestra:] - cumulata[:-finestra]) / finestra
    segnale = np.nan_to_num(np.where(p > media, 1.0, 0.0))
    posizione = np.zeros(len(p))
    posizione[ritardo:] = segnale[:-ritardo]
    return posizione


varianti = []
for finestra in (20, 50, 100, 200):
    for partenza in (0, 200, 400, 600):  # PROVA / TRY: aggiungi altre date (esercizio 3)
        p = prezzi[partenza:]
        rend = rendimenti(p)
        varianti.append(risultato(rend, sopra_media(p, finestra)[1:], costo=0.0012))

varianti = np.array(varianti)
print(f"{len(varianti)} varianti della STESSA idea (4 finestre × 4 date d'inizio)\n")
print(f"  peggiore  {varianti.min():8.2f}x")
print(f"  mediana   {np.median(varianti):8.2f}x")
print(f"  migliore  {varianti.max():8.2f}x")
print(f"  rapporto migliore/peggiore: {varianti.max() / varianti.min():.1f} volte")
print("\nQuesto intervallo e' l'informazione onesta. Il numero singolo che di "
      "solito si pubblica e' un punto scelto dentro di esso.")