PythagorasGoal/scripts/research/r0807_asset_compare.py

#!/usr/bin/env python
"""r0807_asset_compare.py — e se invece del book ci mettessi un ETF S&P 500 o MSCI World?

DOMANDA DELL'OPERATORE. Confronto onesto fra tre motori per lo STESSO piano di accumulo:

    BOOK   il book live Deribit (TP01 75% + SKH01 25%) alle ancore canoniche, de-luckato x0.89
    SP500  SPY, ADJUSTED_LAST (dividendi reinvestiti), 1996+ -> 30 anni
    WORLD  MSCI World: URTH se certificato, altrimenti proxy 70% SPY + 30% EFA

Il confronto e' interessante SOLO se sono comparabili quattro cose, e tre di queste il progetto
le ha gia' sbagliate almeno una volta:

 (1) LA GRIGLIA. Le azioni hanno ~252 barre/anno, il book 365. Una serie di borsa passata a un
     motore che annualizza a 365 esce con Sharpe x1.20 e CAGR x1.45 (lezione del 25/07 su GTAA01).
     Qui le azioni vanno su griglia di CALENDARIO con 0.0 nei giorni di chiusura — la convenzione
     gia' adottata per GTAA01: il capitale e' fermo, non riciclato.

 (2) LA FINESTRA. Il book ha 7.4 anni (2019-2026, due tori cripto); SPY ne ha 30 (2000, 2008,
     2020, 2022 dentro). Confrontare 7 anni scelti dalla storia recente contro 30 anni e' un
     confronto truccato a favore del book -> si riporta ANCHE la finestra comune.

 (3) IL FISCO, che NON e' lo stesso e non e' un dettaglio:
       book  -> realizza in continuo, imposta ANNUA sulle plusvalenze (33% assunto) + 0.2% sul valore
       ETF   -> UCITS ad accumulazione: nessuna distribuzione, imposta DIFFERITA alla vendita (26%)
                + 0.2% annuo sul valore
     Il differimento e' un vantaggio strutturale dell'ETF che il book non ha, e va nel modello.

 (4) IL BERSAGLIO. $272.061 e' il capitale-rendita calcolato con la rendita perpetua DEL BOOK e
     l'aliquota del 33%. Un ETF ha un'altra distribuzione e un'altra aliquota -> ha un ALTRO
     bersaglio, che qui si ricalcola per ciascun motore invece di riusare il numero del book.

⚠️ Le assunzioni fiscali sono dichiarate e NON sono un parere fiscale.

    uv run python scripts/research/r0807_asset_compare.py [--dep 500] [--lump 5000] [--anni 15]
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts" / "research"))

import eqlib as EQ                       # noqa: E402
import r0725_capcurve as CC              # noqa: E402
import r0726_venue_risk as VR            # noqa: E402
import r0726_capwall_refresh as WR       # noqa: E402
import r0727_lumpsum_split as LS         # noqa: E402
import r0807_growth_yearly as G          # noqa: E402

TAX_CRYPTO = 0.33        # L. 207/2024 art. 1 c.23-29 (assunzione gia' cablata nel progetto)
TAX_ETF = 0.26           # redditi diversi di natura finanziaria
PATRIM = 0.002           # 0.2% annuo sul valore (IVAFE / imposta di bollo)
COMM_EUR = 2.0           # commissione per acquisto mensile su un broker retail europeo
N_PATHS = 4000
SEED = 807


# ------------------------------------------------------------------ serie
def _calendar_daily(close: pd.Series) -> pd.Series:
    """Ritorni giornalieri su griglia di CALENDARIO: 0.0 quando la borsa e' chiusa.

    E' la convenzione di GTAA01 (capitale fermo nel weekend, non riciclato) ed e' cio' che rende
    la serie confrontabile con quella del book, che ha 365 barre l'anno.
    """
    r = close.astype(float).pct_change().dropna()
    r.index = pd.to_datetime(r.index).tz_convert(None).normalize()
    r = r[~r.index.duplicated(keep="last")]
    idx = pd.date_range(r.index[0], r.index[-1], freq="D")
    return r.reindex(idx).fillna(0.0)


def carica_serie() -> dict:
    out = {}
    b = VR.venue_series()["Deribit"]
    b.index = pd.to_datetime(b.index).tz_localize(None).normalize()
    out["BOOK"] = b.astype(float)
    out["SP500"] = _calendar_daily(EQ.load_eq("SPY")["close"])

    # MSCI World: il vero ETF se e' su disco, altrimenti il proxy dichiarato
    world_path = ROOT / "data" / "raw" / "eq_urth_1d.parquet"
    if world_path.exists():
        out["WORLD"] = _calendar_daily(EQ.load_eq("URTH")["close"])
        out["_world_fonte"] = "URTH (iShares MSCI World, reale)"
    else:
        spy, efa = _calendar_daily(EQ.load_eq("SPY")["close"]), _calendar_daily(EQ.load_eq("EFA")["close"])
        j = pd.concat({"SPY": spy, "EFA": efa}, axis=1).dropna()
        out["WORLD"] = (0.70 * j["SPY"] + 0.30 * j["EFA"])
        out["_world_fonte"] = "PROXY 70% SPY + 30% EFA (ribil. giornaliero) — non e' l'indice"
    return out


def metriche(r: pd.Series) -> dict:
    eq = (1 + r).cumprod()
    dd = float((eq / eq.cummax() - 1).min())
    mu, sd = float(r.mean()) * 365, float(r.std()) * np.sqrt(365)
    return dict(n=len(r), da=str(r.index[0].date()), a=str(r.index[-1].date()),
                anni=len(r) / 365.0, drift=mu, vol=sd, sharpe=mu / sd if sd > 0 else 0.0, maxdd=dd)


# ------------------------------------------------------------------ accumulo
def accumula(r: np.ndarray, lump_eur: float, dep_eur: float, anni: int,
             differita: bool, aliquota: float, comm_eur: float) -> list[dict]:
    """Traiettoria annuale con due regimi fiscali.

    * `differita=False` (book): imposta ANNUA sulle plusvalenze realizzate + patrimoniale.
      E' esattamente `r0807_growth_yearly.traiettoria`, riusata senza riscriverla.
    * `differita=True` (ETF ad accumulazione): nessuna imposta annua sul reddito, solo la
      patrimoniale; l'imposta sul guadagno si paga alla VENDITA -> la curva riportata e' il
      valore di LIQUIDAZIONE, `cap - aliquota * max(0, cap - versato)`, che e' l'unica cosa
      confrontabile con un book gia' tassato ogni anno.

    La commissione d'acquisto si modella come deposito netto piu' piccolo, ma il `versato`
    riportato resta quello USCITO DI TASCA: cosi' la commissione pesa sul guadagno, dov'e' giusto.
    """
    dep_netto = max(0.0, dep_eur - comm_eur) if dep_eur > 0 else 0.0
    righe = G.traiettoria(r, lump_eur, dep_netto, anni,
                          0.0 if differita else aliquota, PATRIM,
                          n_paths=N_PATHS, seed=SEED)
    out = []
    for x in righe:
        n_dep = round((x["versato"] - LS.START - lump_eur * CC.EURUSD) / (dep_netto * CC.EURUSD)) \
            if dep_netto > 0 else 0
        versato_reale = LS.START + lump_eur * CC.EURUSD + n_dep * dep_eur * CC.EURUSD
        med, p25, p75 = x["capitale_med"], x["capitale_p25"], x["capitale_p75"]
        if differita:
            netto = lambda c: c - aliquota * max(0.0, c - versato_reale)   # noqa: E731
            med, p25, p75 = netto(med), netto(p25), netto(p75)
        out.append(dict(anno=x["anno"], versato=versato_reale, med=med, p25=p25, p75=p75,
                        guadagno=med - versato_reale, phit_book=x["p_traguardo"]))
    return out


def bersaglio(r: np.ndarray, aliquota: float) -> tuple[float, float]:
    """(rendita perpetua, capitale necessario) per €50/giorno NETTI con l'aliquota di quel motore."""
    lordo = CC.TARGET_EUR_DAY * 365 * CC.EURUSD / (1 - aliquota)
    _, perp, cap = WR.perp_and_wall(r, 1.0, lordo)
    return perp, cap


# ------------------------------------------------------------------ main
def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--lump", type=float, default=5000.0)
    ap.add_argument("--dep", type=float, default=500.0)
    ap.add_argument("--anni", type=int, default=15)
    ap.add_argument("--json", type=str, default="")
    a = ap.parse_args()

    S = carica_serie()
    motori = ["BOOK", "SP500", "WORLD"]

    print("=" * 104)
    print("  r0807 — BOOK vs S&P 500 vs MSCI WORLD, stesso piano di accumulo")
    print("=" * 104)
    print(f"  piano: ${LS.START:,.0f} + €{a.lump:,.0f} + €{a.dep:,.0f}/mese · {a.anni} anni · "
          f"{N_PATHS:,} path, blocchi 20g")
    print(f"  MSCI World = {S['_world_fonte']}")

    print("\n  (1) LE SERIE — griglia di calendario, 0.0 a borsa chiusa")
    print(f"  {'motore':8}{'barre':>8}{'da':>13}{'a':>13}{'anni':>7}{'drift':>9}{'vol':>8}"
          f"{'Sharpe':>8}{'maxDD':>8}")
    for m in motori:
        x = metriche(S[m])
        print(f"  {m:8}{x['n']:>8}{x['da']:>13}{x['a']:>13}{x['anni']:>7.1f}"
              f"{x['drift']:>8.1%}{x['vol']:>8.1%}{x['sharpe']:>8.2f}{x['maxdd']:>8.1%}")

    # finestra comune: il confronto onesto sul REGIME
    lo = max(S[m].index[0] for m in motori)
    hi = min(S[m].index[-1] for m in motori)
    print(f"\n  (2) STESSA FINESTRA {lo.date()} → {hi.date()} — toglie il vantaggio di aver scelto "
          f"gli anni")
    print(f"  {'motore':8}{'anni':>7}{'drift':>9}{'vol':>8}{'Sharpe':>8}{'maxDD':>8}")
    comune = {}
    for m in motori:
        comune[m] = S[m].loc[lo:hi]
        x = metriche(comune[m])
        print(f"  {m:8}{x['anni']:>7.1f}{x['drift']:>8.1%}{x['vol']:>8.1%}{x['sharpe']:>8.2f}"
              f"{x['maxdd']:>8.1%}")

    # bersagli propri
    print("\n  (3) IL BERSAGLIO E' DIVERSO PER OGNI MOTORE (rendita perpetua + aliquota propria)")
    print(f"  {'motore':8}{'aliquota':>10}{'rendita perpetua':>19}{'capitale per €50/g netti':>27}")
    berg = {}
    for m in motori:
        al = TAX_CRYPTO if m == "BOOK" else TAX_ETF
        perp, cap = bersaglio(S[m].values.astype(float), al)
        berg[m] = dict(perp=perp, cap=cap, aliquota=al)
        cs = f"${cap:,.0f}" if np.isfinite(cap) else "non sostenibile"
        print(f"  {m:8}{al:>9.0%}{perp:>18.2%}{cs:>27}")

    # bersaglio anche sulla finestra comune — altrimenti si confronta un accumulo di 7 anni
    # con un bersaglio stimato su 30. ⚠️ Questa colonna e' la MENO affidabile delle due: un
    # prelievo perpetuo e' una domanda a 20 anni e 7 anni di storia non la risolvono.
    print("\n  (3-bis) LO STESSO BERSAGLIO, ma stimato sulla sola finestra comune (poco "
          "affidabile: 7 anni)")
    print(f"  {'motore':8}{'rendita perpetua':>19}{'capitale':>14}{'vs finestra piena':>20}")
    berg_c = {}
    for m in motori:
        al = berg[m]["aliquota"]
        perp, cap = bersaglio(comune[m].values.astype(float), al)
        berg_c[m] = dict(perp=perp, cap=cap, aliquota=al)
        rel = cap / berg[m]["cap"] - 1 if np.isfinite(cap) and np.isfinite(berg[m]["cap"]) else float("nan")
        cs = f"${cap:,.0f}" if np.isfinite(cap) else "non sostenibile"
        print(f"  {m:8}{perp:>18.2%}{cs:>14}{rel:>19.0%}")

    # accumulo, ciascuno col proprio regime
    print(f"\n  (4) ACCUMULO a {a.anni} anni — ciascuno col PROPRIO fisco e i propri costi")
    print(f"  {'motore':8}{'versato':>13}{'capitale med':>15}{'p25':>13}{'p75':>13}"
          f"{'guadagno':>13}{'% del cap.':>12}")
    traj = {}
    for m in motori:
        t = accumula(S[m].values.astype(float), a.lump, a.dep, a.anni,
                     differita=(m != "BOOK"),
                     aliquota=TAX_CRYPTO if m == "BOOK" else TAX_ETF,
                     comm_eur=0.0 if m == "BOOK" else COMM_EUR)
        traj[m] = t
        L = t[-1]
        print(f"  {m:8}{L['versato']:>12,.0f}${L['med']:>14,.0f}${L['p25']:>12,.0f}$"
              f"{L['p75']:>12,.0f}${L['guadagno']:>12,.0f}${L['guadagno']/L['med']:>11.0%}")

    # (4-bis) l'accumulo sulla STESSA finestra: e' il confronto che toglie il regime
    print(f"\n  (4-bis) LO STESSO ACCUMULO, ma ricampionando SOLO {lo.date()}→{hi.date()} "
          f"per tutti e tre")
    print(f"  {'motore':8}{'capitale med':>15}{'p25':>13}{'p75':>13}{'guadagno':>13}"
          f"{'vs finestra piena':>20}")
    traj_c = {}
    for m in motori:
        t = accumula(comune[m].values.astype(float), a.lump, a.dep, a.anni,
                     differita=(m != "BOOK"),
                     aliquota=TAX_CRYPTO if m == "BOOK" else TAX_ETF,
                     comm_eur=0.0 if m == "BOOK" else COMM_EUR)
        traj_c[m] = t
        L, F = t[-1], traj[m][-1]
        print(f"  {m:8}{L['med']:>14,.0f}${L['p25']:>12,.0f}${L['p75']:>12,.0f}$"
              f"{L['guadagno']:>12,.0f}${L['med']/F['med']-1:>19.0%}")

    print("\n  (5) QUOTA DEL PROPRIO BERSAGLIO RAGGIUNTA (ognuno il suo)")
    for m in motori:
        L, c = traj[m][-1], berg[m]["cap"]
        if np.isfinite(c) and c > 0:
            print(f"  {m:8} capitale ${L['med']:>10,.0f}  ·  bersaglio ${c:>10,.0f}"
                  f"  ·  {L['med'] / c:>6.0%}")
        else:
            print(f"  {m:8} capitale ${L['med']:>10,.0f}  ·  bersaglio non sostenibile")

    # sensibilita' del proxy World: la quota USA dell'indice e' cambiata molto nel tempo
    if "PROXY" in S["_world_fonte"]:
        spy, efa = _calendar_daily(EQ.load_eq("SPY")["close"]), _calendar_daily(EQ.load_eq("EFA")["close"])
        j = pd.concat({"SPY": spy, "EFA": efa}, axis=1).dropna()
        print("\n  (6) SENSIBILITA' DEL PROXY — la quota USA di MSCI World e' passata da ~50% "
              "(2008) a ~72% (oggi)")
        print(f"  {'peso USA':>10}{'drift':>9}{'vol':>8}{'Sharpe':>8}{'maxDD':>8}")
        for w in (0.5, 0.6, 0.7, 0.8):
            x = metriche(w * j["SPY"] + (1 - w) * j["EFA"])
            print(f"  {w:>9.0%}{x['drift']:>8.1%}{x['vol']:>8.1%}{x['sharpe']:>8.2f}{x['maxdd']:>8.1%}")

    if a.json:
        Path(a.json).write_text(json.dumps(dict(
            piano=dict(lump=a.lump, dep=a.dep, anni=a.anni, start=LS.START, eurusd=CC.EURUSD,
                       n_paths=N_PATHS, comm_eur=COMM_EUR, tax_crypto=TAX_CRYPTO,
                       tax_etf=TAX_ETF, patrim=PATRIM, world=S["_world_fonte"]),
            metriche={m: metriche(S[m]) for m in motori},
            metriche_comune={m: metriche(comune[m]) for m in motori},
            finestra_comune=dict(da=str(lo.date()), a=str(hi.date())),
            bersaglio={m: berg[m] for m in motori},
            bersaglio_comune={m: berg_c[m] for m in motori},
            traiettoria=traj, traiettoria_comune=traj_c), indent=1))
        print(f"\n  scritto {a.json}")


if __name__ == "__main__":
    main()