Backtest sin humo: ¿de golpe o poco a poco?

# -*- coding: utf-8 -*-
"""
Backtest honesto de "DCA vs invertir todo de golpe" sobre el S&P 500.

La pregunta: te cae una cantidad de dinero (herencia, bonus, venta de un piso).
¿La metes toda hoy, o la repartes en cuotas mensuales para "promediar"?

Reglas objetivas, sin letra pequeña:

  - Datos: S&P 500 (^GSPC) diario desde 1990, precios ajustados (yfinance).
  - Se prueban TODOS los puntos de entrada posibles (primer día hábil de cada
    mes), no unos cuantos elegidos a dedo.
  - De golpe: se invierte el 100% el día 1.
  - DCA: N cuotas iguales, la primera el día 1 y el resto el primer día hábil
    de cada mes siguiente. La comparación se cierra cuando el DCA termina de
    invertir, que es cuando ambos están al 100% y la diferencia queda fijada.
  - La liquidez pendiente NO está muerta: renta al tipo de las letras del
    Tesoro a 13 semanas (^IRX) de cada día. Si no hay datos, 0% (y se avisa).
  - SIN comisiones, SIN impuestos, SIN deslizamiento.

Alimenta el segundo "Backtest sin humo" de Aprende (articulos.py).

    python backtest_dca_vs_golpe.py

Guarda cache/gspc_daily.csv y cache/irx_daily.csv para repetirlo sin red.
"""

import os

import pandas as pd

AQUI = os.path.dirname(os.path.abspath(__file__))
CACHE_SPX = os.path.join(AQUI, "cache", "gspc_daily.csv")
CACHE_IRX = os.path.join(AQUI, "cache", "irx_daily.csv")

DESDE = "1990-01-01"
PLAZOS = [6, 12, 24]      # meses de reparto del DCA que se comparan
PLAZO_BASE = 12           # el que se detalla en el informe

# Ventanas de 12 meses famosas, para enseñar los extremos
EPISODIOS = [
    ("Antes del pinchazo puntocom", "2000-03-01"),
    ("Antes de la caida de 2008", "2007-10-01"),
    ("Justo en el suelo de 2009", "2009-03-01"),
    ("Antes del Covid", "2020-01-02"),
    ("Antes de la inflacion 2022", "2022-01-03"),
]


def _fix_ssl() -> None:
    """Avast rompe el SSL de yfinance: usar el CA bundle propio si existe.

    CURL_CA_BUNDLE hace falta porque yfinance ya descarga con curl_cffi, que
    ignora REQUESTS_CA_BUNDLE.
    """
    ca = os.path.expanduser("~/.trading_ca/cacert_trading.pem")
    if os.path.exists(ca):
        for var in ("REQUESTS_CA_BUNDLE", "SSL_CERT_FILE", "CURL_CA_BUNDLE"):
            os.environ.setdefault(var, ca)


def _sesion(verificar):
    """Sesion curl_cffi (patron del Crisis Screener contra el bloqueo de Yahoo)."""
    from curl_cffi import requests as cffi
    return cffi.Session(impersonate="chrome120", verify=verificar)


def _descargar(ticker: str, destino: str) -> pd.Series:
    _fix_ssl()
    import yfinance as yf
    ca = os.path.expanduser("~/.trading_ca/cacert_trading.pem")
    # El CA propio se queda obsoleto cada vez que Avast rota su certificado;
    # si falla se repite sin verificar, como hace el resto del sistema.
    df = None
    for verificar in ([ca] if os.path.exists(ca) else []) + [False]:
        try:
            df = yf.download(ticker, start=DESDE, auto_adjust=True,
                             progress=False, threads=False,
                             session=_sesion(verificar))
        except Exception:
            continue
        if df is not None and not df.empty:
            break
    if df is None or df.empty:
        raise RuntimeError(f"descarga vacia de {ticker}")
    close = df["Close"]
    if isinstance(close, pd.DataFrame):   # yfinance nuevo: columnas multinivel
        close = close.iloc[:, 0]
    close = close.rename("close").dropna()
    if close.empty:
        raise RuntimeError(f"descarga vacia de {ticker}")
    os.makedirs(os.path.dirname(destino), exist_ok=True)
    close.to_frame().to_csv(destino)
    return close


def cargar(ticker: str, destino: str) -> pd.Series:
    if os.path.exists(destino):
        s = pd.read_csv(destino, index_col=0, parse_dates=True)["close"]
        if not s.empty:
            return s
        os.remove(destino)          # cache corrupta de un intento fallido
    return _descargar(ticker, destino)


def factor_liquidez(precios: pd.Index) -> pd.Series:
    """Factor de capitalizacion diario de la liquidez segun ^IRX (letras 13s).

    ^IRX cotiza el tipo anualizado en %. Se convierte a rendimiento diario
    simple (base 252) y se rellena hacia delante en los dias sin dato.
    """
    try:
        irx = cargar("^IRX", CACHE_IRX)
    except Exception as exc:                                  # sin red y sin cache
        print(f"  [aviso] sin datos de ^IRX ({exc}); liquidez al 0%")
        return pd.Series(1.0, index=precios)
    tipo = irx.reindex(precios).ffill().fillna(0.0) / 100.0
    tipo = tipo.clip(lower=0.0)
    return 1.0 + tipo / 252.0


def primeros_dias_de_mes(indice: pd.DatetimeIndex) -> list:
    """Primer dia habil de cada mes presente en el indice."""
    s = pd.Series(indice, index=indice)
    return list(s.groupby([indice.year, indice.month]).first())


def simular(precios: pd.Series, cap_dia: pd.Series, fechas_compra: list,
            fin) -> tuple:
    """Devuelve (valor_final_dca, valor_final_golpe, maxdd_dca, maxdd_golpe).

    Capital inicial = 1. El DCA compra 1/N en cada fecha de `fechas_compra`;
    la liquidez pendiente capitaliza a diario con `cap_dia`.
    """
    n = len(fechas_compra)
    tramo = precios.loc[fechas_compra[0]:fin]
    cuota = 1.0 / n

    # --- de golpe -------------------------------------------------------
    part_golpe = 1.0 / precios.loc[fechas_compra[0]]
    valor_golpe = part_golpe * tramo

    # --- DCA: recorrido dia a dia ---------------------------------------
    pendientes = set(fechas_compra)
    caja, participaciones = 1.0, 0.0
    valores = []
    for fecha, precio in tramo.items():
        caja *= cap_dia.loc[fecha]          # la liquidez renta cada dia
        if fecha in pendientes:
            invertir = min(cuota, caja) if fecha != fechas_compra[-1] else caja
            participaciones += invertir / precio
            caja -= invertir
        valores.append(participaciones * precio + caja)
    valor_dca = pd.Series(valores, index=tramo.index)

    dd = lambda s: (s / s.cummax() - 1).min() * 100
    return (valor_dca.iloc[-1], valor_golpe.iloc[-1],
            dd(valor_dca), dd(valor_golpe))


def main() -> None:
    precios = cargar("^GSPC", CACHE_SPX)
    cap_dia = factor_liquidez(precios.index)
    meses = primeros_dias_de_mes(precios.index)

    print(f"S&P 500 · {precios.index[0].date()} -> {precios.index[-1].date()}"
          f"  (sin costes, liquidez a letras 13 semanas)\n")

    resultados = {}
    for n in PLAZOS:
        filas = []
        for i in range(len(meses) - n):
            compras = meses[i:i + n]
            fin = meses[i + n]
            v_dca, v_gol, dd_dca, dd_gol = simular(precios, cap_dia, compras, fin)
            filas.append({
                "inicio": compras[0], "dca": v_dca, "golpe": v_gol,
                "dd_dca": dd_dca, "dd_golpe": dd_gol,
            })
        r = pd.DataFrame(filas).set_index("inicio")
        r["dif"] = (r["golpe"] - r["dca"]) * 100      # puntos a favor del golpe
        resultados[n] = r

    print(f"{'Reparto DCA':>14}{'Gana de golpe':>15}{'Ventaja media':>15}"
          f"{'Peor caso golpe':>17}{'Peor caso DCA':>15}")
    for n, r in resultados.items():
        gana = (r["golpe"] > r["dca"]).mean() * 100
        print(f"{str(n) + ' meses':>14}{gana:>14.0f}%{r['dif'].mean():>+14.1f}%"
              f"{(r['golpe'].min() - 1) * 100:>+16.1f}%"
              f"{(r['dca'].min() - 1) * 100:>+14.1f}%")

    r = resultados[PLAZO_BASE]
    print(f"\nDetalle del reparto en {PLAZO_BASE} meses "
          f"({len(r)} puntos de entrada distintos):")
    print(f"  Rentabilidad media   de golpe {(r['golpe'].mean() - 1) * 100:+.1f}%"
          f"   ·  DCA {(r['dca'].mean() - 1) * 100:+.1f}%")
    print(f"  Peor caida media     de golpe {r['dd_golpe'].mean():.1f}%"
          f"   ·  DCA {r['dd_dca'].mean():.1f}%")
    print(f"  Peor caida extrema   de golpe {r['dd_golpe'].min():.1f}%"
          f"   ·  DCA {r['dd_dca'].min():.1f}%")
    peor = r["dif"].idxmin()
    print(f"  El DCA gano por mas: entrando en {peor.date()}"
          f"  ({-r.loc[peor, 'dif']:+.1f} puntos a su favor)")

    print(f"\nQue habria pasado entrando justo en cada momento "
          f"(reparto {PLAZO_BASE} meses):")
    for nombre, fecha in EPISODIOS:
        f = pd.Timestamp(fecha)
        idx = r.index[r.index >= f]
        if idx.empty:
            continue
        fila = r.loc[idx[0]]
        print(f"  {nombre:30} {idx[0].date()}"
              f"   de golpe {(fila['golpe'] - 1) * 100:>+7.1f}%"
              f"   DCA {(fila['dca'] - 1) * 100:>+7.1f}%")


if __name__ == "__main__":
    main()