05-async-events

⚡ ASYNC EVENT ENGINE — MOTOR EVENT-DRIVEN v3.0

PRINCIPIO CENTRAL

Los motores (Fase B/C) nunca esperan activamente datos. Se suscriben al Event Bus y reaccionan cuando llegan snapshots. Toda espera activa (while True: check_for_data()) es un anti-patrón.

ARQUITECTURA DEL BUS

MarketDataHub (Phase A output)
        │ publish(MarketSnapshot)
        ▼
┌─────────────────────────┐
│  Standard Queue          │  asyncio.Queue(maxsize=10_000)
│  Fase B/C consumers      │  Drop-Oldest si se llena
└──────────┬──────────────┘
           │
    ┌──────┴──────┐
    ▼             ▼
 Phase B        Phase C
    │             │
    └──────┬──────┘
           │ publish(OptionContract)
           ▼
┌─────────────────────────┐
│  Priority Queue          │  asyncio.Queue(maxsize=1_000)
│  Phase D EXCLUSIVO       │  CRITICAL log si se llena
└──────────┬──────────────┘
           ▼
      Phase D Monitor → ExecutionSignal → Frontend

WORKER PATTERN — Aislamiento de fallos

class MicrostructureEngine:
    """Phase B: consume snapshots, calcula VPIN/OFI. ZERO imports de red."""

    async def run(self) -> None:
        """Loop principal. Falla individual → log + continúa. Bus nunca para."""
        async for snapshot in self._bus.consume():
            try:
                await self._process_snapshot(snapshot)
            except Exception:
                logger.error(
                    "Phase B: fallo en snapshot — continuando [PD-6]",
                    extra={"ticker": snapshot.ticker},
                    exc_info=True,
                )
                # ← El bus continúa. Un fallo no mata el worker.

    async def _process_snapshot(self, snapshot: MarketSnapshot) -> None:
        """CPU-bound → ProcessPoolExecutor para no bloquear event loop."""
        loop = asyncio.get_running_loop()
        enriched = await loop.run_in_executor(
            self._executor,
            calculate_vpin_ofi_sync,  # función pura, sin async
            snapshot,
        )