Revisión automática de backtests y migraciones de código.

import ast
import textwrap

import numpy as np
import pandas as pd

ajustados = pd.read_parquet("universo2_ajustados.parquet")
P = ajustados.pivot(index="fecha", columns="ticker",
                    values="cierre_ajustado")
ultima = P.apply(pd.Series.last_valid_index)
print(P.shape, "· sesiones", P.index[0].date(), "a",
      P.index[-1].date())
print("dejan de cotizar antes del final:",
      int((ultima < P.index[-1]).sum()))

BASE = """
def backtest(P):
    R = P.pct_change(fill_method=None)
{cuerpo}
    w = w.div(w.abs().sum(axis=1), axis=0)
    return (w.shift(1) * R).sum(axis=1)
"""
LARGO_CORTO = "    w = s.sub(s.mean(axis=1), axis=0)"
# versión correcta del cuerpo, línea correcta y línea con el error
BANCO = {
    "B1": ("""
    s = P.pct_change(20, fill_method=None)
""" + LARGO_CORTO, "(20, fill_method=None)",
           "(20, fill_method=None).shift(-1)"),
    "B2": ("""
    m = P.pct_change(120, fill_method=None)
    z = (m - m.expanding(250).mean()) / m.expanding(250).std()
    w = (z > 1).astype(float).where(P.notna())""",
           "(m - m.expanding(250).mean()) / m.expanding(250).std()",
           "(m - m.mean()) / m.std()"),
    "B3": ("""
    s = P.resample("ME").last().pct_change(12, fill_method=None)
    s = s.reindex(P.index).ffill()
    w = s.rank(axis=1, pct=True).ge(0.8).astype(float)
    w = w.where(P.notna())""", ".ffill()", ".bfill()"),
    "B4": ("""
    s = P.pct_change(5, fill_method=None)
""" + LARGO_CORTO, "(w.shift(1) * R)", "(w * R)"),
    "B5": ("""
    vivas = P.columns
    w = P[vivas].notna().astype(float)""",
           "vivas = P.columns",
           "vivas = P.columns[P.iloc[-1].notna()]"),
    "B6": ("""
    s = P.resample("ME").last().pct_change(fill_method=None)
    s = s.reindex(P.index, method="ffill")
""" + LARGO_CORTO,
           """s = P.resample("ME").last().pct_change(fill_method=None)
    s = s.reindex(P.index, method="ffill")""",
           """g = P.groupby(P.index.to_period("M"))
    s = g.transform("last") / g.transform("first") - 1"""),
}

def codigos(nombre):
    """(código con el error, código corregido) de un backtest."""
    cuerpo, buena, mala = BANCO[nombre]
    bien = BASE.format(cuerpo=cuerpo)
    return bien.replace(buena, mala), bien

def compilar(codigo):
    espacio = {"np": np, "pd": pd}
    exec(textwrap.dedent(codigo), espacio)
    return espacio["backtest"]

print(codigos("B3")[0])

def sharpe(r):
    r = r.loc["2007":]                  # sin el primer año
    return float(r.mean() / r.std() * np.sqrt(252))

coste = {}
for nombre in BANCO:
    mal, bien = (compilar(c)(P) for c in codigos(nombre))
    coste[nombre] = (sharpe(mal), sharpe(bien))
    print(f"{nombre}  Sharpe con error {coste[nombre][0]:6.2f}"
          f"   corregido {coste[nombre][1]:6.2f}")

VENTANAS = {"rolling", "expanding", "ewm", "groupby", "resample"}
TODA_MUESTRA = {"mean", "std", "min", "max", "quantile", "median"}

def metodo(nodo):
    """Nombre del método si el nodo es una llamada x.metodo(...)."""
    if (isinstance(nodo, ast.Call)
            and isinstance(nodo.func, ast.Attribute)):
        return nodo.func.attr
    return None

def negativo(nodo):
    return (isinstance(nodo, ast.UnaryOp)
            and isinstance(nodo.op, ast.USub))

def revision_estatica(codigo):
    avisos = []
    def aviso(nodo, texto):
        avisos.append(f"l.{nodo.lineno - 1} {texto}")
    for n in ast.walk(ast.parse(textwrap.dedent(codigo))):
        m = metodo(n)
        kw = {k.arg: k.value for k in getattr(n, "keywords", [])}
        if m == "shift" and (n.args and negativo(n.args[0])
                             or negativo(kw.get("periods"))):
            aviso(n, "desplazamiento negativo")
        if m == "rolling" and getattr(kw.get("center"),
                                      "value", False):
            aviso(n, "ventana centrada")
        if m in ("bfill", "backfill") or getattr(
                kw.get("method"), "value", "") in ("bfill",
                                                   "backfill"):
            aviso(n, "relleno hacia atrás")
        if m in TODA_MUESTRA and metodo(n.func.value) not in VENTANAS:
            eje = getattr(kw.get("axis"), "value", 0)
            if eje not in (1, "columns"):
                aviso(n, f".{m}() con toda la muestra")
        if (isinstance(n, ast.BinOp) and isinstance(n.op, ast.Mult)
                and any(isinstance(x, ast.Name) and x.id == "R"
                        for x in (n.left, n.right))):
            otro = n.right if isinstance(n.left, ast.Name) else n.left
            if metodo(otro) != "shift":
                aviso(n, "pesos sin retrasar (misma barra)")
        if (isinstance(n, ast.Subscript)
                and isinstance(n.value, ast.Attribute)
                and n.value.attr == "iloc" and negativo(n.slice)):
            aviso(n, "universo con el último dato")
        if m == "dropna" and getattr(kw.get("axis"), "value",
                                     0) in (1, "columns"):
            aviso(n, "universo con el último dato")
    return avisos

estatica = {}
for nombre in BANCO:
    mal, bien = codigos(nombre)
    estatica[nombre] = (revision_estatica(mal),
                        revision_estatica(bien))
    print(f"{nombre} con error: {estatica[nombre][0] or '-'}")
    print(f"   corregido: {estatica[nombre][1] or '-'}")

informe = """
def resumen(curva):
    final = curva.iloc[-1]
    return {"capital final": final, "media": curva.mean()}
"""
revision_estatica(informe)

CORTES = ["2010-11-15", "2014-06-13", "2018-12-14", "2022-06-15"]

def prueba_truncado(f, P):
    """Sesiones hasta cada corte en las que la P&L cambia."""
    completo = f(P)
    distintas = 0
    for corte in CORTES:
        parcial = f(P.loc[:corte])
        a, b = completo.loc[:corte], parcial
        distintas += int((~np.isclose(a, b, equal_nan=True)).sum())
    return distintas

def placebo(P, semilla=652):
    """Rendimientos independientes de media cero con la volatilidad
    de cada acción y su mismo periodo de cotización."""
    rng = np.random.default_rng(semilla)
    vol = P.pct_change(fill_method=None).std().to_numpy()
    r = rng.standard_normal(P.shape) * vol
    falso = 50 * np.cumprod(1 + r, axis=0)
    return pd.DataFrame(falso, P.index, P.columns).where(P.notna())

P0 = placebo(P)
dinamica = {}
for nombre in BANCO:
    fila = []
    for codigo in codigos(nombre):
        f = compilar(codigo)
        fila.append((prueba_truncado(f, P), sharpe(f(P0))))
    dinamica[nombre] = fila
    (t_mal, s_mal), (t_bien, s_bien) = fila
    print(f"{nombre} con error: truncado {t_mal:5d} sesiones "
          f"distintas, placebo {s_mal:6.2f} | corregido: "
          f"{t_bien} y {s_bien:5.2f}")

filas = []
for nombre in BANCO:
    (t_mal, s_mal), (t_bien, s_bien) = dinamica[nombre]
    filas.append({
        "backtest": nombre,
        "estática": bool(estatica[nombre][0]),
        "truncado": t_mal > 0,
        "placebo": abs(s_mal) > 1,
        "falsas alarmas": int(bool(estatica[nombre][1]))
        + int(t_bien > 0) + int(abs(s_bien) > 1),
    })
matriz = pd.DataFrame(filas).set_index("backtest")
matriz["alguna"] = matriz[["estática", "truncado",
                           "placebo"]].any(axis=1)
matriz

import subprocess
import sys
from pathlib import Path

PY3 = sys.executable
PY2 = str(Path(sys.executable).parents[2]
          / "entornos" / "zipline" / ".venv" / "bin" / "python")

Path("casos.py").write_text('''
import warnings
import numpy as np
import pandas as pd

def caso(nombre, f):
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        try:
            r = f()
        except Exception as e:
            r = type(e).__name__
    avisos = sorted({x.category.__name__ for x in w})
    print(f"{nombre}|{r}|{' '.join(avisos)}")

idx = pd.date_range("2024-01-02", periods=6, freq="B")
s = pd.Series([1.0, np.nan, 3.0, np.nan, 5.0, 6.0], idx)
df = pd.DataFrame({"tk": ["AAA", "BBB", None, "AAA", "BBB", "AAA"],
                   "p": [10.0, 11.0, 12.0, 13.0, 14.0, 15.0]}, idx)

def encadenada():
    d = df.copy()
    d["p"][d["p"] > 13] = 0.0
    return d["p"].tolist()[-3:]

caso("fillna(method=)", lambda: s.fillna(method="ffill").iloc[1])
caso("applymap", lambda: df[["p"]].applymap(abs).iloc[0, 0])
caso("resample('M')", lambda: len(s.resample("M").last()))
caso("first('3D')", lambda: len(s.first("3D")))
caso("s[0] con fechas", lambda: s[0])
caso("asignación encadenada", encadenada)
caso("dtype == object", lambda: df["tk"].dtype == object)
caso("astype(str) de None",
     lambda: repr(df["tk"].astype(str).iloc[2]))
caso("fecha a int64", lambda: int(pd.to_datetime(
    ["2024-01-02"]).astype("int64")[0]))
print(f"versión|pandas {pd.__version__}|")
''')

def ejecutar(python, script):
    r = subprocess.run([python, "-I", script], capture_output=True,
                       text=True, check=True)
    return [linea.split("|") for linea in r.stdout.splitlines()]

def celda(valor, aviso):
    aviso = aviso.removesuffix("Warning")
    return f"{valor} ({aviso})" if aviso else valor

for a, b in zip(ejecutar(PY2, "casos.py"),
                ejecutar(PY3, "casos.py")):
    print(f"{a[0]:<22}{celda(a[1], a[2]):<35}{celda(b[1], b[2])}")

b5 = pd.read_parquet("indice2_5min.parquet")
sesion = (b5["ts"].dt.tz_convert("America/New_York")
          .dt.tz_localize(None).dt.normalize())
kapa = b5.groupby(sesion)["cierre"].last().rename("close")
kapa.index.name = "fecha"
kapa.to_csv("kapa_diario.csv")

ANTIGUO = '''import pandas as pd
precios = pd.read_csv("kapa_diario.csv", index_col=0,
                      parse_dates=True)["close"]
precios = precios.fillna(method="ffill")
mensual = precios.resample("M").last()
res = pd.DataFrame({"ret": mensual.pct_change()})
res["pos"] = (res["ret"] > 0).astype(float).shift(1)
res["pnl"] = res["pos"] * res["ret"]
cambios = res["pos"].diff().abs() > 0
res["pnl"][cambios] -= 0.001
print(f"primer cierre {precios[0]:.2f}, meses {len(res)}, "
      f"capital {(1 + res['pnl'].fillna(0)).prod():.6f}")
'''

ALIAS = {"M": "ME", "Q": "QE", "Y": "YE", "A": "YE", "H": "h",
         "T": "min"}

def escaner_migracion(codigo):
    avisos = []
    for n in ast.walk(ast.parse(codigo)):
        m = metodo(n)
        kw = {k.arg: k.value for k in getattr(n, "keywords", [])}
        if m == "fillna" and "method" in kw:
            avisos.append((n.lineno, "fillna(method=) retirado: "
                           "ffill() / bfill()"))
        if m in ("resample", "asfreq", "date_range") and n.args:
            f = getattr(n.args[0], "value", None)
            if f in ALIAS:
                avisos.append((n.lineno, f"alias '{f}' retirado: "
                               f"'{ALIAS[f]}'"))
        if m == "applymap":
            avisos.append((n.lineno, "applymap retirado: map"))
        if isinstance(n, ast.Assign):
            destinos = n.targets
        else:
            destinos = [getattr(n, "target", None)]
        if any(isinstance(d, ast.Subscript)
               and isinstance(d.value, ast.Subscript)
               for d in destinos):
            avisos.append((n.lineno, "asignación encadenada: "
                           "no asigna con copy-on-write; usar .loc"))
        if (isinstance(n, ast.Subscript)
                and isinstance(n.slice, ast.Constant)
                and type(n.slice.value) is int):
            avisos.append((n.lineno, "clave entera: en pandas 3 es "
                           "una etiqueta; usar .iloc"))
    return sorted(avisos)

for linea, texto in escaner_migracion(ANTIGUO):
    print(f"línea {linea}: {texto}")

import re

SOLO_ERRORES = (ANTIGUO
    .replace('precios.fillna(method="ffill")', "precios.ffill()")
    .replace('resample("M")', 'resample("ME")')
    .replace("precios[0]", "precios.iloc[0]"))
MIGRADO = SOLO_ERRORES.replace('res["pnl"][cambios] -= 0.001',
                               'res.loc[cambios, "pnl"] -= 0.001')

def correr(python, codigo, nombre="bt.py"):
    """Salida del script y categorías de los avisos o del error."""
    Path(nombre).write_text(codigo)
    r = subprocess.run([python, "-I", nombre], capture_output=True,
                       text=True)
    avisos = sorted(set(re.findall(r": (\w+(?:Warning|Error)): ",
                                   r.stderr)))
    if r.returncode:
        return r.stderr.strip().splitlines()[-1].split(":")[0], []
    return r.stdout.strip(), avisos

for etiqueta, python, codigo in [
        ("antiguo, pandas 2.3", PY2, ANTIGUO),
        ("antiguo, pandas 3.0", PY3, ANTIGUO),
        ("solo errores, pandas 3.0", PY3, SOLO_ERRORES),
        ("migrado, pandas 3.0", PY3, MIGRADO),
        ("migrado, pandas 2.3", PY2, MIGRADO)]:
    salida, avisos = correr(python, codigo)
    print(f"{etiqueta:<25} {salida}")
    if avisos:
        print(" " * 26 + "avisos: " + ", ".join(avisos))

import warnings

from pandas.errors import Pandas4Warning

# El motor llama a Timestamp.utcnow(), obsoleto en pandas 3:
# un aviso por ejecución, fuera de nuestro control (lección 2.4.5)
warnings.filterwarnings("ignore", message="Timestamp.utcnow",
                        category=Pandas4Warning)

from nautilus_trader.model.currencies import USD
from nautilus_trader.model.data import BarType
from nautilus_trader.model.identifiers import (
    InstrumentId, Symbol, Venue)
from nautilus_trader.model.instruments import Equity
from nautilus_trader.model.objects import Money, Price, Quantity
from nautilus_trader.persistence.wranglers import BarDataWrangler

diario = b5.groupby(sesion).agg(
    open=("apertura", "first"), high=("maximo", "max"),
    low=("minimo", "min"), close=("cierre", "last"),
    volume=("volumen", "sum"))

def en_utc(fechas, hora, minuto):
    return ((fechas + pd.Timedelta(hours=hora, minutes=minuto))
            .tz_localize("America/New_York").tz_convert("UTC"))

VENUE = Venue("XNYS")
kapa_n = Equity(
    instrument_id=InstrumentId(Symbol("KAPA"), VENUE),
    raw_symbol=Symbol("KAPA"), currency=USD, price_precision=2,
    price_increment=Price.from_str("0.01"),
    lot_size=Quantity.from_int(1), ts_event=0, ts_init=0)

dia = diario.set_axis(en_utc(diario.index, 16, 0))
BT_DIA = BarType.from_str("KAPA.XNYS-1-DAY-LAST-EXTERNAL")
barras_dia = BarDataWrangler(BT_DIA, kapa_n).process(dia)

apertura = pd.DataFrame(
    {c: diario["open"] for c in ("open", "high", "low", "close")})
apertura["volume"] = 1_000.0
apertura.index = en_utc(diario.index, 9, 31)
BT_AP = BarType.from_str("KAPA.XNYS-1-MINUTE-LAST-EXTERNAL")
barras_ap = BarDataWrangler(BT_AP, kapa_n).process(apertura)
print(len(barras_dia), "barras diarias;", len(barras_ap),
      "de apertura")

from nautilus_trader.backtest.engine import BacktestEngine
from nautilus_trader.config import (
    BacktestEngineConfig, LoggingConfig, StrategyConfig)
from nautilus_trader.indicators import SimpleMovingAverage
from nautilus_trader.model.enums import (
    AccountType, OmsType, OrderSide)
from nautilus_trader.trading.strategy import Strategy


class SmaCrossConfig(StrategyConfig, frozen=True):
    instrument_id: InstrumentId
    bar_type: BarType
    barra_apertura: BarType | None = None
    rapida: int = 10
    lenta: int = 30
    acciones: int = 100      # backtrader compraría 1 (SizerFix)


class SmaCross(Strategy):
    def __init__(self, config: SmaCrossConfig) -> None:
        super().__init__(config)
        self.rapida = SimpleMovingAverage(config.rapida)
        self.lenta = SimpleMovingAverage(config.lenta)
        self.previo = None          # signo de rápida - lenta
        self.pendiente = None

    def on_start(self) -> None:
        cfg = self.config
        self.register_indicator_for_bars(cfg.bar_type, self.rapida)
        self.register_indicator_for_bars(cfg.bar_type, self.lenta)
        self.subscribe_bars(cfg.bar_type)
        if cfg.barra_apertura is not None:
            self.subscribe_bars(cfg.barra_apertura)

    def cruce(self):
        """BUY, SELL o None, como bt.ind.CrossOver."""
        if not self.indicators_initialized():
            return None
        d = np.sign(self.rapida.value - self.lenta.value)
        previo, self.previo = self.previo, d
        largo = self.portfolio.is_net_long(self.config.instrument_id)
        if previo is None:
            return None
        if d > 0 and previo <= 0 and not largo:
            return OrderSide.BUY
        if d < 0 and previo >= 0 and largo:
            return OrderSide.SELL
        return None

    def on_bar(self, bar) -> None:
        if bar.bar_type == self.config.bar_type:
            lado = self.cruce()
            con_apertura = self.config.barra_apertura is not None
            if lado is not None and con_apertura:
                self.pendiente = lado       # a la apertura
            elif lado is not None:
                self.enviar(lado)           # al cierre
        elif self.pendiente is not None:
            self.enviar(self.pendiente)
            self.pendiente = None

    def enviar(self, lado) -> None:
        self.submit_order(self.order_factory.market(
            self.config.instrument_id, lado,
            Quantity.from_int(self.config.acciones)))


def ejecutar_nautilus(con_apertura):
    motor = BacktestEngine(BacktestEngineConfig(
        logging=LoggingConfig(log_level="ERROR")))
    motor.add_venue(venue=VENUE, oms_type=OmsType.NETTING,
                    account_type=AccountType.CASH,
                    base_currency=USD,
                    starting_balances=[Money(100_000, USD)])
    motor.add_instrument(kapa_n)
    motor.add_data(barras_dia)
    if con_apertura:
        motor.add_data(barras_ap)
    motor.add_strategy(SmaCross(SmaCrossConfig(
        instrument_id=kapa_n.id, bar_type=BT_DIA,
        barra_apertura=BT_AP if con_apertura else None)))
    motor.run()
    f = motor.trader.generate_order_fills_report()
    sesion_ny = (pd.to_datetime(f["ts_last"])
                 .dt.tz_convert("America/New_York")
                 .dt.tz_localize(None).dt.normalize())
    return pd.DataFrame({"fecha": sesion_ny.to_numpy(),
                         "lado": f["side"].astype(str).to_numpy(),
                         "precio": f["avg_px"].astype(float)
                         .to_numpy()})

literal = ejecutar_nautilus(False)
imitada = ejecutar_nautilus(True)
print("literal:", len(literal), "órdenes; imitada:", len(imitada))
imitada.head(3)

c, o = diario["close"], diario["open"]
x = np.sign(c.rolling(10).mean() - c.rolling(30).mean())
alza = (x > 0) & (x.shift(1) <= 0)
baja = (x < 0) & (x.shift(1) >= 0)
filas, dentro = [], False
for i in range(len(c)):
    if (alza.iloc[i] and not dentro) or (baja.iloc[i] and dentro):
        dentro = not dentro
        filas.append((i, "BUY" if dentro else "SELL"))

def referencia(al_cierre):
    out = []
    for i, lado in filas:
        j = i if al_cierre else i + 1
        if j < len(c):
            precio = c.iloc[j] if al_cierre else o.iloc[j]
            out.append((c.index[j], lado, round(float(precio), 2)))
    return pd.DataFrame(out, columns=["fecha", "lado", "precio"])

def coinciden(a, b):
    n = min(len(a), len(b))
    a, b = a.iloc[:n].reset_index(drop=True), b.iloc[:n]
    b = b.reset_index(drop=True)
    fecha = (a["fecha"] == b["fecha"]) & (a["lado"] == b["lado"])
    precio = fecha & np.isclose(a["precio"], b["precio"])
    return int(fecha.sum()), int(precio.sum()), len(a), len(b)

for nombre, nau, ref in [
        ("literal frente a cierre", literal, referencia(True)),
        ("literal frente a apertura sig.", literal,
         referencia(False)),
        ("imitada frente a apertura sig.", imitada,
         referencia(False))]:
    f, p, n1, n2 = coinciden(nau, ref)
    print(f"{nombre:<31} {n1} y {n2} órdenes; fecha y lado {f},"
          f" también precio {p}")