Arquitectura de un sistema en vivo: de los datos a la reconciliación.

import hashlib
import heapq
import itertools
from collections import Counter, defaultdict
from dataclasses import dataclass

import exchange_calendars as xcals
import numpy as np
import pandas as pd

@dataclass(frozen=True)
class Evento:
    t: pd.Timestamp       # hora lógica (UTC)
    tipo: str
    datos: dict

class Bus:
    """Cola de eventos por hora; el reloj es el del último evento."""

    def __init__(self):
        self.cola, self.subs = [], defaultdict(list)
        self.n = itertools.count()   # desempate: orden de publicación
        self.ahora = None
        self.registro = []

    def suscribir(self, tipo, fn):
        self.subs[tipo].append(fn)

    def publicar(self, t, tipo, **datos):
        if self.ahora is not None and t < self.ahora:
            raise ValueError(f"{tipo} con hora pasada: {t}")
        heapq.heappush(self.cola,
                       (t, next(self.n), Evento(t, tipo, datos)))

    def correr(self):
        while self.cola:
            t, _, ev = heapq.heappop(self.cola)
            self.ahora = t
            self.registro.append(ev)
            for fn in self.subs[ev.tipo]:
                fn(ev)

bus = Bus()
bus.suscribir("eco", lambda ev: print(ev.t.time(), ev.datos))
t0 = pd.Timestamp("2025-09-02 19:45", tz="UTC")
bus.publicar(t0 + pd.Timedelta(seconds=2), "eco", n=2)
bus.publicar(t0, "eco", n=1)
bus.correr()

MAPA = {"ORUS": "RV_EEUU", "ORSC": "RV_EEUU_P", "OREU": "RV_EUR",
        "OREM": "RV_EMG", "ORTN": "RF_T710", "ORTL": "RF_T20",
        "ORIG": "RF_CIG", "ORAU": "MP_ORO", "ORCM": "MP_AMP",
        "ORRE": "INM_EEUU"}
ETF = list(MAPA)
u = pd.read_parquet("universo4_precios.parquet").set_index("fecha")
ext = pd.read_csv("vivo4_extracto_posiciones.csv",
                  parse_dates=["fecha"])
prim = ext[ext["fecha"] < "2025-11-17"].groupby("instrumento").first()
escala = pd.Series({t: prim.loc[t, "precio_cierre"]
                    / u.loc[prim.loc[t, "fecha"], a]
                    for t, a in MAPA.items()})
precios = u[list(MAPA.values())].set_axis(ETF, axis=1) * escala

CAL = xcals.get_calendar("XNYS")
SESIONES = CAL.sessions_in_range("2025-09-02", "2025-12-31")
print(len(SESIONES), "sesiones;", "cierre del 2 de septiembre:",
      CAL.session_close(SESIONES[0]), "| del 31 de diciembre:",
      CAL.session_close(SESIONES[-1]))
precios.loc["2025-09-02"].round(2).to_dict()

class Datos:
    """Publica lo que se sabe en cada momento, con su hora."""

    def __init__(self, bus, precios):
        self.bus, self.p = bus, precios

    def programar(self, sesiones):
        for d in sesiones:
            cierre = CAL.session_close(d)
            hist = self.p.loc[:d].iloc[-65:-1]   # hasta la víspera
            self.bus.publicar(cierre - pd.Timedelta(minutes=15),
                              "precios", fecha=d, hist=hist)
            self.bus.publicar(cierre, "cierre", fecha=d,
                              px=self.p.loc[d])
            self.bus.publicar(cierre + pd.Timedelta(minutes=30),
                              "fin_dia", fecha=d)

def pesos_momentum(h):
    mom = h.iloc[-1] / h.iloc[-21] - 1
    vol = h.pct_change().iloc[-63:].std() * np.sqrt(252)
    w = np.minimum(0.025 / vol, 0.25).where(mom > 0, 0.0)
    return w / w.sum() if w.sum() > 1 else w

class Senal:
    def __init__(self, bus):
        self.bus = bus
        bus.suscribir("precios", self.calcular)

    def calcular(self, ev):
        h = ev.datos["hist"]
        self.bus.publicar(ev.t, "objetivo", fecha=ev.datos["fecha"],
                          pesos=pesos_momentum(h), ref=h.iloc[-1])

ordenes = pd.read_csv("vivo4_ordenes.csv")
pct = 100 * ordenes["cantidad"] * ordenes["precio_medio_envio"] / 1e6
print(f"Mayor orden: {pct.max():.1f} % del capital; "
      f"mediana: {pct.median():.1f} %")
for lim in (10, 20, 30):
    print(f"Límite {lim} %: bloquearía {(pct > lim).sum()} "
          "órdenes legítimas")
error_dedo = 10 * pct.max()
print(f"Error de dedo (x10): {error_dedo:.0f} %, bloqueada: "
      f"{error_dedo > 30}")

class Riesgo:
    """Límites de cartera y controles previos a cada orden."""

    def __init__(self, bus, peso_max=0.25, valor_max=0.30,
                 ordenes_dia=25):
        self.bus, self.peso_max = bus, peso_max
        self.valor_max, self.ordenes_dia = valor_max, ordenes_dia
        self.interruptor = False      # lección 4.2.5
        self.hoy = Counter()
        bus.suscribir("objetivo", self.revisar)

    def revisar(self, ev):
        if self.interruptor:
            self.bus.publicar(ev.t, "alerta", texto="interruptor")
            return
        w = ev.datos["pesos"].clip(upper=self.peso_max)
        w = w / w.sum() if w.sum() > 1 else w
        self.bus.publicar(ev.t, "objetivo_ok",
                          **{**ev.datos, "pesos": w})

    def comprobar(self, orden, patrimonio):
        """None si pasa; si no, el motivo del bloqueo."""
        if orden["cantidad"] * orden["ref"] > (self.valor_max
                                               * patrimonio):
            return "valor máximo"
        self.hoy[orden["fecha"]] += 1
        if self.hoy[orden["fecha"]] > self.ordenes_dia:
            return "órdenes por día"
        return None

MS = pd.Timedelta(milliseconds=1)

def comision(q, px):
    return round(min(max(0.0035 * q, 0.35), 0.01 * q * px), 2)

class Cartera:
    def __init__(self, bus, efectivo):
        self.efectivo, self.pos = efectivo, pd.Series(0, index=ETF)
        self.vistas = set()
        bus.suscribir("informe", self.anotar)

    def anotar(self, ev):
        d = ev.datos
        if d["evento"] not in ("parcial", "ejecutada") \
                or d["id_ejecucion"] in self.vistas:
            return
        self.vistas.add(d["id_ejecucion"])
        s = 1 if d["lado"] == "compra" else -1
        self.pos[d["inst"]] += s * d["cantidad"]
        self.efectivo -= s * d["cantidad"] * d["precio"] + d["comision"]

    def patrimonio(self, px):
        return self.efectivo + float((self.pos * px).sum())

class OMS:
    def __init__(self, bus, riesgo, cartera, umbral=0.004):
        self.bus, self.riesgo, self.cartera = bus, riesgo, cartera
        self.umbral, self.ordenes = umbral, {}
        bus.suscribir("objetivo_ok", self.rebalancear)
        bus.suscribir("informe", self.actualizar)

    def rebalancear(self, ev):
        ref, w = ev.datos["ref"], ev.datos["pesos"]
        pat = self.cartera.patrimonio(ref)
        delta = np.floor(w * pat / ref) - self.cartera.pos
        for inst, q in delta.sort_values().items():  # ventas antes
            if abs(q) * ref[inst] < self.umbral * pat:
                continue
            o = dict(id=f"C{len(self.ordenes) + 1:05d}",
                     fecha=ev.datos["fecha"], inst=inst,
                     lado="compra" if q > 0 else "venta",
                     cantidad=int(abs(q)), ref=float(ref[inst]))
            self.ordenes[o["id"]] = {**o, "estado": "nueva",
                                     "recibida": 0}
            motivo = self.riesgo.comprobar(o, pat)
            if motivo:
                self.ordenes[o["id"]]["estado"] = "bloqueada"
                self.bus.publicar(ev.t, "alerta", texto=motivo)
                continue
            self.ordenes[o["id"]]["estado"] = "enviada"
            self.bus.publicar(ev.t + 5 * 60_000 * MS, "orden", **o)

    def actualizar(self, ev):
        o = self.ordenes[ev.datos["id"]]
        o["estado"] = ev.datos["evento"]
        o["recibida"] += ev.datos.get("cantidad", 0)
        o["acumulada"] = ev.datos.get("acumulada", 0)

class BrokerSimulado:
    LAT = 42 * MS

    def __init__(self, bus, efectivo, media_horquilla_pb=1.0):
        self.bus, self.efectivo = bus, efectivo
        self.pos = pd.Series(0, index=ETF)
        self.hs, self.cola, self.seq = media_horquilla_pb / 1e4, [], 0
        self.perder = set()          # fallos de transporte a propósito
        bus.suscribir("orden", self.recibir)
        bus.suscribir("cierre", self.subasta)

    def informar(self, t, **d):
        self.seq += 1
        if d.get("id_ejecucion") not in self.perder:
            self.bus.publicar(t + self.LAT, "informe", seq=self.seq,
                              **d)

    def recibir(self, ev):
        o, t = ev.datos, ev.t + self.LAT
        ventas = sum(x["cantidad"] * x["ref"] for x in self.cola
                     if x["lado"] == "venta")
        compras = sum(x["cantidad"] * x["ref"] for x in self.cola
                      if x["lado"] == "compra")
        coste = o["cantidad"] * o["ref"] * (o["lado"] == "compra")
        if coste > 0 and compras + coste > self.efectivo + ventas:
            self.informar(t, id=o["id"], evento="rechazada")
            return
        self.cola.append(o)
        self.informar(t, id=o["id"], evento="aceptada")

    def subasta(self, ev):
        for o in self.cola:
            s = 1 if o["lado"] == "compra" else -1
            px = round(ev.datos["px"][o["inst"]] * (1 + s * self.hs), 2)
            n = 1 if o["cantidad"] < 300 else (
                2 if o["cantidad"] < 1500 else 3)
            partes = [o["cantidad"] // n] * n
            partes[-1] += o["cantidad"] - sum(partes)
            acum = 0
            for k, q in enumerate(partes, 1):
                acum += q
                com = comision(q, px)
                self.pos[o["inst"]] += s * q
                self.efectivo -= s * q * px + com
                fin = acum == o["cantidad"]
                self.informar(ev.t + 300 * k * MS, id=o["id"],
                              evento="ejecutada" if fin else "parcial",
                              inst=o["inst"], lado=o["lado"],
                              cantidad=q, acumulada=acum, precio=px,
                              comision=com,
                              id_ejecucion=f"{o['id']}.{k:02d}")
        self.cola = []

class Reconciliador:
    def __init__(self, bus, cartera, broker):
        self.bus, self.c, self.b = bus, cartera, broker
        self.diario = []
        bus.suscribir("fin_dia", self.cuadrar)

    def cuadrar(self, ev):
        dif = (self.b.pos - self.c.pos)
        dif = dif[dif != 0].to_dict()
        de = round(self.b.efectivo - self.c.efectivo, 2)
        self.diario.append(dict(fecha=ev.datos["fecha"].date(),
                                posiciones=dif, efectivo=de))
        if dif or de:
            self.bus.publicar(ev.t, "alerta", texto=f"descuadre {dif}")

def montar(perder=(), capital=1_000_000.0):
    bus = Bus()
    riesgo = Riesgo(bus)
    cartera = Cartera(bus, capital)
    broker = BrokerSimulado(bus, capital)
    broker.perder = set(perder)
    oms = OMS(bus, riesgo, cartera)
    rec = Reconciliador(bus, cartera, broker)
    Senal(bus)
    alertas = []
    bus.suscribir("alerta", lambda ev: alertas.append(
        (ev.t.date(), ev.datos["texto"])))
    Datos(bus, precios).programar(SESIONES)
    bus.correr()
    return dict(bus=bus, oms=oms, cartera=cartera, broker=broker,
                rec=rec, alertas=alertas)

s = montar()
print(Counter(ev.tipo for ev in s["bus"].registro))
print(Counter(o["estado"] for o in s["oms"].ordenes.values()))
cierre = precios.loc[SESIONES[-1]]
print(f"Patrimonio final: {s['cartera'].patrimonio(cierre):,.2f} USD")
print("Días con descuadre:",
      sum(bool(r["posiciones"] or r["efectivo"])
          for r in s["rec"].diario), "| alertas:", len(s["alertas"]))

obj = {ev.datos["fecha"]: ev.datos["pesos"]
       for ev in s["bus"].registro if ev.tipo == "objetivo"}
sen = pd.read_csv("vivo4_senales.csv")
sen["fecha"] = pd.to_datetime(sen["ts"].str[:10])
sen["nuestro"] = [obj[f][i] for f, i in
                  zip(sen["fecha"], sen["instrumento"])]
dif = (sen["nuestro"].round(4) - sen["peso_objetivo"]).abs()
print(f"{len(sen)} señales; diferencia máxima {dif.max():.4f}; "
      f"distintas a 4 decimales: {(dif > 1e-9).sum()}")

perdida = next(ev.datos for ev in s["bus"].registro
               if ev.tipo == "informe"
               and ev.datos["evento"] == "parcial"
               and ev.datos["inst"] == "ORSC")
print(perdida["id_ejecucion"], perdida["lado"], perdida["cantidad"],
      perdida["precio"], perdida["comision"])
f = montar(perder={perdida["id_ejecucion"]})
o_f = f["oms"].ordenes[perdida["id"]]
print("En el OMS:", o_f["estado"], "| pedida", o_f["cantidad"],
      "| sumada", o_f["recibida"], "| acumulada", o_f["acumulada"])
for r in [r for r in f["rec"].diario
          if r["posiciones"] or r["efectivo"]][:3]:
    print(r)

def ordenes_de(sim, fecha, inst):
    return [(o["lado"], o["cantidad"], o["estado"])
            for o in sim["oms"].ordenes.values()
            if o["fecha"] == pd.Timestamp(fecha)
            and o["inst"] == inst]

for dia in ("2025-09-03", "2025-09-04", "2025-09-05"):
    print(dia, "sin fallo:", ordenes_de(s, dia, "ORSC"),
          "| con fallo:", ordenes_de(f, dia, "ORSC"))
print("ORSC al final, broker y cartera interna:",
      f["broker"].pos["ORSC"], f["cartera"].pos["ORSC"],
      "| sistema sano:", s["broker"].pos["ORSC"])
fantasma = (f["broker"].pos["ORSC"] - f["cartera"].pos["ORSC"]) \
    * precios.loc[SESIONES[-1], "ORSC"]
print(f"Valor de la posición fantasma: {fantasma:,.0f} USD; días "
      f"con descuadre: {sum(bool(r['posiciones'])
                            for r in f['rec'].diario)}")

def huella(sim):
    h = hashlib.sha256()
    for ev in sim["bus"].registro:
        datos = {k: v for k, v in ev.datos.items()
                 if k not in ("hist", "px", "pesos", "ref")}
        h.update(f"{ev.t.isoformat()}|{ev.tipo}|{datos}".encode())
    return h.hexdigest()[:16]

print(huella(s), huella(montar()), huella(f))

import asyncio
from datetime import datetime, timezone

async def adaptador(bus, flujo_broker, flujo_datos):
    """Publica en el bus cada mensaje con su hora de llegada."""
    async def bombear(flujo, tipo):
        async for msg in flujo:
            ahora = pd.Timestamp(datetime.now(timezone.utc))
            bus.publicar(ahora, tipo, **msg)
            bus.correr()          # procesa lo que haya en la cola
    await asyncio.gather(bombear(flujo_broker, "informe"),
                         bombear(flujo_datos, "precios"))

ev = pd.read_csv("vivo4_eventos_broker.csv")
o = ordenes.set_index("id_orden_cliente").loc["R00071"]
t0 = pd.Timestamp(o["ts_creacion"])
ms = lambda ts: round((pd.Timestamp(ts) - t0).total_seconds() * 1e3)
print(o["id_senal"], o["instrumento"], o["lado"], o["cantidad"],
      o["tipo"], "| envío:", ms(o["ts_envio"]), "ms")
r = ev[ev["id_orden_cliente"] == "R00071"]
pd.DataFrame({"seq": r["seq"], "evento": r["evento"],
              "broker_ms": r["ts_broker"].map(ms),
              "recibido_ms": r["ts_recibido"].map(ms),
              "cantidad": r["cantidad_ultima"],
              "precio": r["precio_ultimo"]}).set_index("seq")