Capstone C1: un sistema de futuros multi-activo, de la previsión a la cuenta de papel.

import hashlib
import json

import numpy as np
import pandas as pd

curvas = pd.read_parquet("futuros3_curvas.parquet")
ctr = pd.read_csv("futuros3_contratos.csv",
                  parse_dates=["vencimiento", "fecha_roll"])
esp = pd.read_csv("futuros3_especificaciones.csv",
                  index_col="mercado")
MERCADOS = list(esp.index[esp["mercado_base"].isna()
                          & (esp["divisa"] == "USD")])


def serie(mercado):
    """Contrato que se tiene, su precio, el cambio diario del mismo
    contrato en puntos y el carry anual (lección 3.1.4)."""
    c = curvas[curvas["mercado"] == mercado]
    P = c.pivot(index="fecha", columns="contrato",
                values="liquidacion")
    cm = ctr[ctr["mercado"] == mercado].reset_index(drop=True)
    pos = np.searchsorted(cm["fecha_roll"].to_numpy(),
                          P.index.to_numpy(), side="right")
    tiene = cm["contrato"].to_numpy()[pos]
    sig = cm["contrato"].to_numpy()[np.minimum(pos + 1, len(cm) - 1)]
    i, j = P.columns.get_indexer(tiene), P.columns.get_indexer(sig)
    f, Pv = np.arange(len(P)), P.to_numpy()
    ayer = np.r_[i[0], i[:-1]]
    dif = np.r_[np.nan, Pv[f[1:], ayer[1:]] - Pv[f[1:] - 1, ayer[1:]]]
    v = cm.set_index("contrato")["vencimiento"]
    dt = (v.reindex(sig).to_numpy()
          - v.reindex(tiene).to_numpy()).astype("timedelta64[D]")
    F2 = np.where(j >= 0, Pv[f, np.maximum(j, 0)], np.nan)
    carry = np.log(Pv[f, i] / F2) / (dt.astype(float) / 365)
    return pd.DataFrame({"contrato": tiene, "precio": Pv[f, i],
                         "dif": dif, "carry": carry}, index=P.index)


S = {m: serie(m) for m in MERCADOS}
fechas = S["RVA"].index
tabla = lambda col: pd.DataFrame(
    {m: S[m][col] for m in MERCADOS}).reindex(fechas)
precio, dif = tabla("precio"), tabla("dif")
contrato, carry = tabla("contrato"), tabla("carry")
mult = esp.loc[MERCADOS, "multiplicador"]
huella = {a: hashlib.sha256(open(a, "rb").read()).hexdigest()[:12]
          for a in ("futuros3_curvas.parquet",
                    "futuros3_contratos.csv",
                    "futuros3_especificaciones.csv")}
print(len(MERCADOS), "mercados ·", len(fechas), "sesiones ·",
      fechas[0].date(), "a", fechas[-1].date())
print(" ".join(MERCADOS))
print(huella)

ESTIMA = slice("2011", "2013")      # escalares y FDM: solo estos años
ajustado = dif.fillna(0).cumsum()
sigma_p = dif.ewm(span=35, min_periods=20).std()     # puntos por día
vol_anual = (dif / precio.shift(1)).ewm(
    span=35, min_periods=20).std() * np.sqrt(252)


def escalada(bruta):
    k = 10 / bruta.loc[ESTIMA].abs().stack().mean()
    return (bruta * k).clip(-20, 20)


def ewmac(r):
    return escalada((ajustado.ewm(span=r, min_periods=r).mean()
                     - ajustado.ewm(span=4 * r, min_periods=4 * r)
                     .mean()) / sigma_p)


REGLAS = {f"ewmac{r}": ewmac(r) for r in (8, 16, 32, 64)}
REGLAS["carry"] = escalada(
    carry.ewm(span=30, min_periods=10).mean() / vol_anual)


def combinar(pesos):
    w = pd.Series(pesos)
    junt = pd.DataFrame({k: REGLAS[k].loc[ESTIMA].stack()
                         for k in w.index}).dropna()
    fdm = min(2.5, 1 / np.sqrt(w @ junt.corr() @ w))
    return (sum(REGLAS[k] * w[k] for k in w.index) * fdm
            ).clip(-20, 20), fdm


BASE = {"ewmac16": 0.25, "ewmac64": 0.25, "carry": 0.5}
prevision, fdm = combinar(BASE)
print(f"FDM {fdm:.3f}")
print(prevision.loc["2025-12-31"].round(1).sort_values()
      .iloc[[0, 1, -2, -1]].to_dict())

def armonico(d):
    """Día del máximo de la deriva anual de un cambio diario d:
    regresión sobre los incrementos de cos y sin del día del año."""
    a = 2 * np.pi * d.index.dayofyear.to_numpy() / 365.25
    b = 2 * np.pi * (d.index.dayofyear.to_numpy() - 1) / 365.25
    X = np.c_[np.cos(a) - np.cos(b), np.sin(a) - np.sin(b)]
    ok = d.notna().to_numpy()
    c, s = np.linalg.lstsq(X[ok], d.to_numpy()[ok], rcond=None)[0]
    return np.degrees(np.arctan2(s, c)) % 360 / 360 * 365.25


def calendario(indice, pico):
    """+1 de valle + 10 días a máximo - 10; -1 de máximo + 10 a
    valle - 10; 0 el resto (lección 3.2.6)."""
    doy = indice.dayofyear.to_numpy()
    valle = (pico - 182.6) % 365.25
    sube = ((doy - (valle + 10)) % 365.25) < 162.6
    baja = ((doy - (pico + 10)) % 365.25) < 162.6
    return np.where(sube, 1.0, np.where(baja, -1.0, 0.0))


r = dif / precio.shift(1)                 # rendimiento del contrato
d_log = np.log1p(r["GRT"]) - np.log1p(r["GRM"])
lado = pd.Series(0.0, index=fechas)
picos = {}
for a in range(2014, 2026):
    picos[a] = armonico(d_log[d_log.index.year < a])
    lado[fechas.year == a] = calendario(fechas[fechas.year == a],
                                        picos[a])
vol_sp = (r["GRT"] - r["GRM"]).ewm(span=60, min_periods=40
                                   ).std() * np.sqrt(252)
print({a: round(p) for a, p in picos.items()})

CAPITAL, TAU, W_SPREAD = 5_000_000, 0.20, 0.20
normal = (dif / sigma_p.shift(1)).loc[ESTIMA]
w_i = pd.Series(1 / len(MERCADOS), index=MERCADOS)
idm = min(2.5, 1 / np.sqrt(w_i @ normal.corr() @ w_i))
riesgo_ctr = sigma_p * np.sqrt(252) * mult          # USD al año
media = (1 - W_SPREAD) * idm * w_i * CAPITAL * TAU / riesgo_ctr
nominal = W_SPREAD * CAPITAL * TAU / vol_sp          # USD por pata
opt_sp = pd.DataFrame(0.0, index=fechas, columns=MERCADOS)
opt_sp["GRT"] = lado * nominal / (precio["GRT"] * mult["GRT"])
opt_sp["GRM"] = -lado * nominal / (precio["GRM"] * mult["GRM"])
opt_sp = opt_sp.fillna(0)
coste_lado = (0.5 * esp.loc[MERCADOS, "valor_tick"]
              + esp.loc[MERCADOS, "comision"])        # USD
roll = contrato.ne(contrato.shift(1))


def contratos(optimo, media, buffer=0.10):
    """Enteros; se opera solo fuera de óptimo ± buffer·media y hasta
    el borde. Decidido con datos de t - 1 (lección 3.1.4)."""
    P = optimo.shift(1).to_numpy()
    B = buffer * media.shift(1).to_numpy()
    N, actual = np.zeros_like(P), np.zeros(P.shape[1])
    for t in range(len(P)):
        nuevo = np.clip(actual, np.round(P[t] - B[t]),
                        np.round(P[t] + B[t]))
        actual = np.where(np.isnan(P[t] + B[t]), 0.0, nuevo)
        N[t] = actual
    return pd.DataFrame(N, index=optimo.index, columns=optimo.columns)


def backtest(N):
    pyg = (N.shift(1) * dif * mult).sum(axis=1)
    c = (N.diff().abs() * coste_lado
         + 2 * N.shift(1).abs() * roll * coste_lado).sum(axis=1)
    return (((pyg - c) / CAPITAL).loc["2014":],
            (c / CAPITAL).loc["2014":])


def sharpe(x):
    return x.mean() / x.std() * np.sqrt(252)


N = contratos(prevision / 10 * media + opt_sp, media)
neto, coste = backtest(N)
partes = {"tendencia y carry":
              backtest(contratos(prevision / 10 * media, media))[0],
          "spread trigo-maíz": backtest(contratos(opt_sp, media))[0],
          "sistema": neto}
print(pd.DataFrame({k: {"sharpe": sharpe(v),
                        "vol_pct": 100 * v.std() * np.sqrt(252),
                        "rent_pct": 100 * v.mean() * 252}
                    for k, v in partes.items()}).T.round(2))
print(f"IDM {idm:.2f} · coste {100 * coste.mean() * 252:.2f} % "
      f"al año · {int((N.diff().abs() > 0).loc['2014':].sum().sum())}"
      " cambios de posición")

import warnings
from decimal import Decimal

from pandas.errors import Pandas4Warning

warnings.filterwarnings("ignore", message="Timestamp.utcnow",
                        category=Pandas4Warning)   # lección 2.4.5

from nautilus_trader.backtest.engine import BacktestEngine
from nautilus_trader.config import (BacktestEngineConfig,
                                    LoggingConfig, StrategyConfig)
from nautilus_trader.model.currencies import USD
from nautilus_trader.model.data import Bar, BarType
from nautilus_trader.model.enums import (AccountType, AssetClass,
                                         OmsType, OrderSide)
from nautilus_trader.model.identifiers import (InstrumentId, Symbol,
                                               Venue)
from nautilus_trader.model.instruments import FuturesContract
from nautilus_trader.model.objects import Money, Price, Quantity
from nautilus_trader.trading.strategy import Strategy

VENUE = Venue("SIM")
ANO = fechas[fechas.year == 2025]
px = curvas.set_index(["fecha", "contrato"])["liquidacion"]
info = ctr.set_index("contrato")
ns = lambda x: pd.Timestamp(x, tz="UTC").value


def decimales(tick):
    s = np.format_float_positional(tick)
    return len(s.split(".")[1]) if "." in s else 0


def futuro(c, m):
    d = decimales(esp.at[m, "tick"])
    return FuturesContract(
        InstrumentId(Symbol(c), VENUE), Symbol(c),
        AssetClass.COMMODITY, USD, d, Price(esp.at[m, "tick"], d),
        Quantity.from_int(int(mult[m])), Quantity.from_int(1), m,
        ns(info.at[c, "fecha_alta"]), ns(info.at[c, "vencimiento"]),
        0, 0, margin_init=Decimal(0), margin_maint=Decimal(0))


inst, barras, objetivo = {}, [], {}
for t in ANO:
    cierre = ns(t) + 21 * 3_600 * 10**9      # 21:00 UTC
    for m in MERCADOS:
        hoy, ayer = contrato.at[t, m], contrato.shift(1).at[t, m]
        for c in sorted({hoy, ayer}):
            inst.setdefault(c, futuro(c, m))
            p = Price(px[(t, c)], inst[c].price_precision)
            barras.append(Bar(
                BarType.from_str(f"{c}.SIM-1-DAY-LAST-EXTERNAL"),
                p, p, p, p, Quantity.from_int(1_000), cierre, cierre))
            objetivo[(cierre, c)] = N.at[t, m] if c == hoy else 0.0


class Ejecutor(Strategy):
    """Lleva cada contrato a su objetivo. No decide nada."""

    def __init__(self, objetivo):
        super().__init__(StrategyConfig(strategy_id="EJECUTOR-001"))
        self.objetivo = objetivo

    def on_start(self):
        for i in inst.values():
            self.subscribe_bars(
                BarType.from_str(f"{i.id}-1-DAY-LAST-EXTERNAL"))

    def on_bar(self, bar):
        iid = bar.bar_type.instrument_id
        q = self.objetivo[(bar.ts_event, iid.symbol.value)]
        d = int(q - float(self.portfolio.net_position(iid)))
        if d:
            self.submit_order(self.order_factory.market(
                iid, OrderSide.BUY if d > 0 else OrderSide.SELL,
                Quantity.from_int(abs(d))))


motor = BacktestEngine(BacktestEngineConfig(
    logging=LoggingConfig(log_level="ERROR")))
motor.add_venue(venue=VENUE, oms_type=OmsType.NETTING,
                account_type=AccountType.MARGIN, base_currency=USD,
                starting_balances=[Money(CAPITAL, USD)])
for i in inst.values():
    motor.add_instrument(i)
motor.add_data(barras)
motor.add_strategy(Ejecutor(objetivo))
motor.run()

fl = motor.trader.generate_order_fills_report()
sym = fl["instrument_id"].str.split(".").str[0]
q = (fl["filled_qty"].astype(float)
     * np.where(fl["side"] == "BUY", 1, -1))
m_fl = mult.reindex(info.loc[sym, "mercado"]).to_numpy()
caja = -(q * fl["avg_px"].astype(float) * m_fl).sum()
fin = ANO[-1]
marca = (N.loc[fin] * precio.loc[fin] * mult).sum()
pyg_motor = caja + marca
pyg_vector = (N.shift(1) * dif * mult).loc[ANO[1:]].sum().sum()
print(len(inst), "contratos ·", len(barras), "barras ·",
      len(fl), "órdenes ejecutadas")
print(f"bruto 2025: motor {pyg_motor:,.2f} USD · "
      f"vectorizado {pyg_vector:,.2f} USD · "
      f"diferencia {pyg_motor - pyg_vector:,.2f}")

import itertools

from scipy.stats import norm

TENDENCIA = {"8+32": ("ewmac8", "ewmac32"),
             "16+64": ("ewmac16", "ewmac64"),
             "32+64": ("ewmac32", "ewmac64"),
             "las cuatro": ("ewmac8", "ewmac16", "ewmac32",
                            "ewmac64")}
registro, rend = [], {}
for (nt, reglas), wc, buf in itertools.product(
        TENDENCIA.items(), (0.3, 0.5, 0.7), (0.0, 0.10)):
    pesos = {k: (1 - wc) / len(reglas) for k in reglas}
    pesos["carry"] = wc
    p_i, _ = combinar(pesos)
    r_i = backtest(contratos(p_i / 10 * media + opt_sp, media,
                             buf))[0]
    cfg = {"tendencia": nt, "peso_carry": wc, "buffer": buf}
    pid = hashlib.sha256(json.dumps(cfg, sort_keys=True)
                         .encode()).hexdigest()[:8]
    rend[pid] = r_i
    registro.append({"id": pid, **cfg, "desde": "2014-01-02",
                     "sesiones": len(r_i),
                     "sharpe": round(sharpe(r_i), 4)})
with open("registro_pruebas.jsonl", "w") as f:
    for fila in registro:
        f.write(json.dumps(fila, ensure_ascii=False) + "\n")
reg = pd.DataFrame(registro).set_index("id")
rend = pd.DataFrame(rend)
print(len(reg), "pruebas; Sharpe de", reg["sharpe"].min(), "a",
      reg["sharpe"].max())

# walk-forward: cada año, la mejor con los años anteriores
oos, elegidas = [], []
for a in range(2017, 2026):
    antes = rend[rend.index.year < a]
    mejor = (antes.mean() / antes.std()).idxmax()
    elegidas.append(mejor)
    oos.append(rend.loc[rend.index.year == a, mejor])
oos = pd.concat(oos)
tramo = rend.loc["2017":]
print("elegidas:", " ".join(f"{a % 100:02d}:{reg.at[e, 'tendencia']}"
      f"/{reg.at[e, 'peso_carry']}" for a, e
      in zip(range(2017, 2026), elegidas)))
print(f"2017-2025 · walk-forward {sharpe(oos):.2f} · mejor a "
      f"posteriori {tramo.apply(sharpe).max():.2f} · media de las "
      f"24 {tramo.apply(sharpe).mean():.2f}")


def psr(sr, umbral, t, asim, curt):          # lección 2.5.4
    den = np.sqrt(1 - asim * sr + (curt - 1) / 4 * sr**2)
    return norm.cdf((sr - umbral) * np.sqrt(t - 1) / den)


def sr0(n, var_sr):
    g = np.euler_gamma
    return np.sqrt(var_sr) * ((1 - g) * norm.ppf(1 - 1 / n)
                              + g * norm.ppf(1 - 1 / (n * np.e)))


mejor = reg["sharpe"].idxmax()
x, sr_d = rend[mejor], rend.mean() / rend.std()   # diarios
for n in (24, 100, 1000):
    u = sr0(n, sr_d.var())
    print(f"N = {n:4d}: umbral {u * np.sqrt(252):.2f} anual · "
          f"DSR {psr(sr_d[mejor], u, len(x), x.skew(),
                     x.kurt() + 3):.3f}")
dsr = psr(sr_d[mejor], sr0(len(reg), sr_d.var()), len(x),
          x.skew(), x.kurt() + 3)

CRITERIOS = {"dsr_min": 0.95, "sharpe_wf_min": 0.5,
             "coste_max_frac": 0.25, "motor_max_usd": 1.0}
bruto = neto + coste
frac = coste.mean() / bruto.mean()
pasa = {"dsr": bool(dsr >= CRITERIOS["dsr_min"]),
        "walk_forward": bool(sharpe(oos) >= CRITERIOS["sharpe_wf_min"]),
        "coste": bool(frac <= CRITERIOS["coste_max_frac"]),
        "motor": bool(abs(pyg_motor - pyg_vector)
                      <= CRITERIOS["motor_max_usd"])}
informe = {
    "datos": huella, "mercados": len(MERCADOS),
    "configuracion": {**BASE, "w_spread": W_SPREAD, "tau": TAU,
                      "capital": CAPITAL, "buffer": 0.10},
    "periodo": ["2014-01-02", "2025-12-31"],
    "sharpe_neto": round(sharpe(neto), 3),
    "coste_sobre_bruto": round(float(frac), 3),
    "n_pruebas": len(reg), "dsr_mejor": round(float(dsr), 4),
    "sharpe_walk_forward": round(sharpe(oos), 3),
    "criterios": CRITERIOS, "pasa": pasa,
    "decision": "paper" if all(pasa.values()) else "parar"}
with open("informe_c1.json", "w") as f:
    json.dump(informe, f, ensure_ascii=False, indent=1)
print(json.dumps({k: informe[k] for k in
                  ("sharpe_neto", "coste_sobre_bruto", "n_pruebas",
                   "dsr_mejor", "sharpe_walk_forward", "decision")},
                 ensure_ascii=False))
print(hashlib.sha256(open("informe_c1.json", "rb").read())
      .hexdigest()[:12])

from datetime import timedelta

from nautilus_trader.adapters.interactive_brokers.common import IB
from nautilus_trader.adapters.interactive_brokers.config import (
    InteractiveBrokersDataClientConfig,
    InteractiveBrokersExecClientConfig,
    InteractiveBrokersInstrumentProviderConfig,
    SymbologyMethod,
)
from nautilus_trader.adapters.interactive_brokers.factories import (
    InteractiveBrokersLiveDataClientFactory,
    InteractiveBrokersLiveExecClientFactory,
)
from nautilus_trader.config import (
    LoggingConfig, RoutingConfig, TradingNodeConfig)
from nautilus_trader.live.node import TradingNode
from nautilus_trader.model.identifiers import InstrumentId

from c1.ejecucion import EjecutorEnVivo   # Ejecutor + temporizador
from c1.config import ajustes              # pydantic-settings, 7.1.1

CONTRATOS = ["ESZ6.CME", "ZNZ6.CBOT", "CLF7.NYMEX", "GCZ6.COMEX",
             "ZCZ6.CBOT", "ZWZ6.CBOT"]   # se regenera en cada roll
proveedor = InteractiveBrokersInstrumentProviderConfig(
    symbology_method=SymbologyMethod.IB_SIMPLIFIED,
    load_ids=frozenset(InstrumentId.from_str(c) for c in CONTRATOS),
)
config = TradingNodeConfig(
    trader_id="C1-PAPEL-001",
    logging=LoggingConfig(log_level="INFO", log_directory="logs"),
    data_clients={IB: InteractiveBrokersDataClientConfig(
        ibg_host="127.0.0.1", ibg_port=7497, ibg_client_id=1,
        instrument_provider=proveedor)},
    exec_clients={IB: InteractiveBrokersExecClientConfig(
        ibg_host="127.0.0.1", ibg_port=7497, ibg_client_id=6,
        account_id=ajustes.ib_cuenta_papel,      # «DU…», sin secretos
        instrument_provider=proveedor,
        routing=RoutingConfig(default=True))},
)
node = TradingNode(config=config)
node.trader.add_strategy(EjecutorEnVivo(
    fichero_objetivos="estado/objetivos.json",
    hora_envio=ajustes.hora_envio_utc,       # fija, en la sesión
    estado_permiso="redis://127.0.0.1:6379/0",   # lección 7.3.3
    intervalo=timedelta(days=1)))
node.add_data_client_factory(
    IB, InteractiveBrokersLiveDataClientFactory)
node.add_exec_client_factory(
    IB, InteractiveBrokersLiveExecClientFactory)
node.build()
try:
    node.run()
finally:
    node.dispose()