""" Agente Bridge Promoted de Leonex. LA PIEZA QUE FALTABA. Conecta las estrategias PROMOVIDAS (tabla asset_strategies: los SILVER/GOLD que activaste con --promote en el Strategy Lab) con el Executor de Alpaca paper. Hasta ahora habia dos carriles que no se hablaban: Strategy Tracker -> re-simula las promovidas sobre los precios (yfinance), NO toca Alpaca. Es paper-paper (simulacion pura). Alpaca Executor -> cogia senales SOLO del pipeline legacy (regime_adaptive_v1_lo + meta-labeling), mandaba a Alpaca paper. Resultado: aunque promovieras 5 estrategias, Alpaca nunca recibia una orden de ellas. Este bridge cierra el hueco. Recompila el MISMO trigger|filtro de cada estrategia promovida (la misma maquinaria del Strategy Lab, identico computo que usa el Tracker) sobre la ULTIMA barra disponible y, si la entrada esta viva HOY, emite una entrada de plan para el Executor. El Executor ya se encarga del resto que NO tocamos aqui: - Sizing: Half-Kelly + Vol Target + HRP (sizing.py). - Judge cualitativo + narrativa. - Registrar el Triple Barrier (open_trades) al abrir, que luego cierra el Close Monitor (TP 2sigma / SL 1sigma / timeout + salida por regimen). El bridge NUNCA envia ordenes. Solo construye entradas de plan. El envio sigue gobernado por el modo del Executor (read_only / dry_run / execute --confirm). Flujo completo ya cosido: Tracker (simula) -> Bridge (detecta entrada viva HOY) -> Executor (dimensiona + manda a Alpaca paper) -> Close Monitor (cierra). Uso directo (inspeccion, no envia nada): python agents/agente_bridge_promoted.py python agents/agente_bridge_promoted.py --json """ from __future__ import annotations import argparse import json import logging import os import sqlite3 import sys from datetime import datetime try: from datetime import UTC except ImportError: # Python 3.10 from datetime import timezone UTC = timezone.utc from pathlib import Path from typing import Optional import pandas as pd # Reutilizamos EXACTAMENTE la misma maquinaria de senales/salidas del Strategy # Lab — el bridge no reimplementa indicadores: recompila los mismos triggers y # filtros, asi una entrada "viva" aqui es identica a la que cuenta el Tracker. sys.path.insert(0, str(Path(__file__).resolve().parent)) from agente_strategy_lab import ( # noqa: E402 DB_PATH, DASHBOARD_DATA_DIR, build_triggers, build_filters, EXIT_CONFIGS, load_prices, _atr, _rsi, simulate_tb_trades, ) # Mergeamos las salidas swing (quick/balanced/wide/trend_ride) con las del # Scalping Lab (scalp_tight/scalp_quick/scalp_wide). Sin esto, las promovidas # que provienen del Horizon Champions SHORT (5m scalp_quick, etc.) fallan en # evaluate_promoted como "stale catalog" — y nunca disparan aunque la senal # este viva, porque el bridge no reconoce su exit_config. try: from agente_strategy_lab_scalping import SCALPING_EXIT_CONFIGS # noqa: E402 ALL_EXIT_CONFIGS = {**EXIT_CONFIGS, **SCALPING_EXIT_CONFIGS} except Exception: ALL_EXIT_CONFIGS = dict(EXIT_CONFIGS) # Estrategias custom (PineScript traducido). Tienen formato single-name (no # trigger|filter|exit). Solo cargamos el registry si esta disponible; si no, # evaluate_promoted devolvera "unknown custom strategy" para las single-name. try: import custom_strategies as _custom_mod # noqa: E402 # Tras importar custom_strategies.py se rellena CUSTOM_STRATEGIES (decorador # @register la popula al cargar el modulo). CUSTOM_STRATEGIES_BY_NAME = {s["name"]: s for s in _custom_mod.CUSTOM_STRATEGIES} from agente_custom_lab import simulate_pct_eod_trades # noqa: E402 _CUSTOM_AVAILABLE = True except Exception: # pragma: no cover CUSTOM_STRATEGIES_BY_NAME = {} simulate_pct_eod_trades = None # type: ignore _CUSTOM_AVAILABLE = False # Reporte persistido para que el dashboard tenga visibilidad del diagnostico # de cada promovida (vive hoy / por que no). Sin esto, el bridge corre pero # desde fuera solo se ven el plan resultante — si esta vacio no sabes por que. REPORT_PATH = DASHBOARD_DATA_DIR / "promoted_bridge_report.json" # Minimo de barras para poder recompilar indicadores con sentido (igual que el # Strategy Tracker). Con menos, no evaluamos la estrategia. MIN_BARS = 60 # Long-only: los triggers del Strategy Lab son todos eventos de entrada larga. SIDE_LONG = 1 # Cortacircuitos de calidad: una estrategia promovida cuyo DSR registrado caiga # por debajo de este suelo NO opera en el bridge (red de seguridad independiente # de la rotacion Top-20, que puede estar apagada). Las entries legacy sin DSR # (dsr<=0 / NULL) se conservan: no hay base para juzgarlas. Configurable por env. MIN_BRIDGE_DSR = float(os.environ.get("LEONEX_MIN_BRIDGE_DSR", "0.50")) def load_promoted(db_path: Path = DB_PATH) -> list[dict]: """Lee las estrategias promovidas de asset_strategies (ticker, strategy, tf). Mismo origen que usa el Strategy Tracker. Devuelve lista vacia si la tabla no existe todavia. Compat con la rotacion dinamica: si la columna `active` existe (creada por agente_top20_selector), filtra por active=1. Si no existe (BD pre-rotacion), devuelve todas — comportamiento original sin breaking change. """ with sqlite3.connect(db_path) as conn: try: cols = [r[1] for r in conn.execute("PRAGMA table_info(asset_strategies)")] if not cols: return [] has_tf = "timeframe" in cols has_dsr = "dsr" in cols has_active = "active" in cols sql = ( "SELECT ticker, strategy, " + ("timeframe" if has_tf else "'1d'") + ", " + ("dsr" if has_dsr else "0.0") + " FROM asset_strategies" ) # Filtro rotacion: solo activas (o active IS NULL para entries # legacy que existian antes de la migracion). if has_active: sql += " WHERE active = 1 OR active IS NULL" rows = conn.execute(sql).fetchall() except Exception: return [] out = [] skipped = 0 for r in rows: dsr = float(r[3]) if r[3] is not None else 0.0 # Cortacircuitos: descarta promociones con DSR positivo pero por debajo # del suelo. dsr<=0 = legacy/desconocido -> se conserva (no se juzga). if 0.0 < dsr < MIN_BRIDGE_DSR: skipped += 1 continue out.append({"ticker": r[0], "strategy": r[1], "timeframe": r[2] or "1d", "dsr": dsr}) if skipped: logging.getLogger("agente_bridge_promoted").info( "Bridge: %d promociones descartadas por DSR < %.2f (cortacircuitos)", skipped, MIN_BRIDGE_DSR) return out def _clean_ohlcv(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() for col in ("open", "high", "low", "close"): df[col] = pd.to_numeric(df[col], errors="coerce") df["volume"] = pd.to_numeric(df.get("volume", 0.0), errors="coerce").fillna(0.0) return df.dropna(subset=["open", "high", "low", "close"]) def _evaluate_custom_promoted(p: dict, strategy: str, out: dict, db_path: Path) -> dict: """Rama de evaluate_promoted para estrategias CUSTOM (single-name). Tienen formato distinto: una funcion entry_fn(df) → boolean Series y un exit_config en porcentaje con cierre fin-de-dia (replica el PineScript). No tienen filtro de regimen — su logica ya esta encapsulada en entry_fn. """ import numpy as np spec = CUSTOM_STRATEGIES_BY_NAME[strategy] ticker = p["ticker"] timeframe = p.get("timeframe", "1h") out["regime"] = "custom" df = load_prices(ticker, timeframe, db_path) if df.empty or len(df) < MIN_BARS: out["note"] = f"insufficient_data ({len(df)} bars)" return out df = _clean_ohlcv(df) if len(df) < MIN_BARS: out["note"] = "insufficient_data (post-clean)" return out dates = df.index out["last_bar"] = str(dates[-1])[:16] close = df["close"] rsi = _rsi(close, 14) mom20 = close / close.shift(20) - 1.0 out["rsi_14"] = round(float(rsi.iloc[-1]) if pd.notna(rsi.iloc[-1]) else 0.0, 2) out["momentum_20"] = round( float(mom20.iloc[-1]) if pd.notna(mom20.iloc[-1]) else 0.0, 4) # Ejecutamos entry_fn(df) → boolean Series. try: entry_series = spec["entry_fn"](df) except Exception as exc: out["note"] = f"entry_fn raised: {exc!r}" return out entry_arr = entry_series.fillna(False).astype(bool).to_numpy() # filtro siempre True (la logica ya esta en entry_fn) filt_arr = np.ones(len(df), dtype=bool) day_codes = pd.factorize(pd.to_datetime(df.index).normalize())[0] exit_cfg = spec.get("exit") or {} tp_pct = float(exit_cfg.get("tp_pct", 3.0)) sl_pct = float(exit_cfg.get("sl_pct", 1.5)) direction = spec.get("direction", "long") # Simulamos con el motor pct + EOD del Custom Lab para coherencia metrica trades = simulate_pct_eod_trades( close.to_numpy(dtype=float), df["high"].to_numpy(dtype=float), df["low"].to_numpy(dtype=float), entry_arr, filt_arr, tp_pct, sl_pct, day_codes, direction=direction, ) n_tr = len(trades) wins = sum(1 for t in trades if t.get("return", 0.0) > 0) prob_win = (wins / n_tr) if n_tr else 0.0 out["prob_win"] = round(prob_win, 4) out["backtest_n_trades"] = n_tr trig_on = bool(entry_arr[-1]) out["trigger_on"] = trig_on out["filter_on"] = True out["side"] = direction.upper() if trig_on: out["live"] = True out["note"] = (f"LIVE entry today: custom '{strategy}' ({direction}) " f"disparo en ultima barra de {timeframe}. " f"TP={tp_pct}% / SL={sl_pct}% / EOD. " f"prob_win={prob_win:.2f}, {n_tr} trades historicos.") else: out["note"] = (f"sin entrada hoy: custom '{strategy}' no disparo en " f"la ultima barra de {timeframe}.") return out def evaluate_promoted(p: dict, db_path: Path = DB_PATH) -> dict: """Evalua UNA estrategia promovida sobre la ultima barra. Devuelve un dict de diagnostico con `live` (bool): True si el trigger Y el filtro de la estrategia disparan en la barra mas reciente — es decir, hoy es un dia de ENTRADA para esa estrategia. `prob_win` es el win-rate empirico de la estrategia sobre todo su historico (estimador honesto de P(ganar) que el Executor usa para el sizing Kelly).""" ticker = p["ticker"] strategy = p["strategy"] timeframe = p.get("timeframe", "1d") dsr = p.get("dsr", 0.0) out = { "ticker": ticker, "strategy": strategy, "timeframe": timeframe, "dsr": round(float(dsr), 4), "live": False, "prob_win": 0.0, "rsi_14": 0.0, "momentum_20": 0.0, "regime": "", "last_bar": "", "note": "", } parts = strategy.split("|") if len(parts) != 3: # No es trigger|filter|exit → puede ser una estrategia CUSTOM (single # name registrada en custom_strategies.py). Si esta disponible la # cargamos y la evaluamos con su exit_config en porcentaje + EOD. if _CUSTOM_AVAILABLE and strategy in CUSTOM_STRATEGIES_BY_NAME: return _evaluate_custom_promoted(p, strategy, out, db_path) out["note"] = (f"unknown custom strategy '{strategy}' (no esta en " f"CUSTOM_STRATEGIES) — comprueba imports y nombre.") \ if _CUSTOM_AVAILABLE else ( "strategy name is not trigger|filter|exit and custom " "strategies module not available") return out trg_name, flt_name, exc_name = parts out["regime"] = flt_name # el filtro de regimen es el contexto de la entrada df = load_prices(ticker, timeframe, db_path) if df.empty or len(df) < MIN_BARS: out["note"] = f"insufficient_data ({len(df)} bars)" return out df = _clean_ohlcv(df) if len(df) < MIN_BARS: out["note"] = "insufficient_data (post-clean)" return out triggers = build_triggers(df) filters = build_filters(df) if (trg_name not in triggers or flt_name not in filters or exc_name not in ALL_EXIT_CONFIGS): out["note"] = "unknown trigger/filter/exit (stale catalog?)" return out dates = df.index out["last_bar"] = str(dates[-1])[:16] close = df["close"] atr = _atr(df) rsi = _rsi(close, 14) mom20 = close / close.shift(20) - 1.0 out["rsi_14"] = round(float(rsi.iloc[-1]) if pd.notna(rsi.iloc[-1]) else 0.0, 2) out["momentum_20"] = round( float(mom20.iloc[-1]) if pd.notna(mom20.iloc[-1]) else 0.0, 4) # Win-rate empirico de la estrategia sobre TODO su historico (mismo computo # de trades que el Tracker / el Lab). Lo usamos como prob_win para el Kelly. cfg = ALL_EXIT_CONFIGS[exc_name] trades = simulate_tb_trades( close.to_numpy(dtype=float), df["high"].to_numpy(dtype=float), df["low"].to_numpy(dtype=float), atr.to_numpy(dtype=float), triggers[trg_name].to_numpy(dtype=bool), filters[flt_name].to_numpy(dtype=bool), cfg["tp"], cfg["sl"], cfg["timeout"], ) n_tr = len(trades) wins = sum(1 for t in trades if t["return"] > 0) prob_win = (wins / n_tr) if n_tr else 0.0 out["prob_win"] = round(prob_win, 4) out["backtest_n_trades"] = n_tr # Entrada VIVA hoy: el trigger Y el filtro disparan en la ultima barra y el # ATR es valido (necesario para dimensionar y para el Triple Barrier). last_atr = float(atr.iloc[-1]) if pd.notna(atr.iloc[-1]) else 0.0 trig_on = bool(triggers[trg_name].iloc[-1]) filt_on = bool(filters[flt_name].iloc[-1]) out["trigger_on"] = trig_on out["filter_on"] = filt_on if trig_on and filt_on and last_atr > 0: out["live"] = True out["note"] = (f"LIVE entry today: {trg_name} confirmado en regimen " f"{flt_name} (salida {exc_name}). prob_win={prob_win:.2f}, " f"DSR={dsr:.2f}, {n_tr} trades historicos.") else: why = [] if not trig_on: why.append(f"trigger '{trg_name}' no dispara hoy") if not filt_on: why.append(f"regimen '{flt_name}' no activo hoy") if last_atr <= 0: why.append("ATR no valido") out["note"] = "sin entrada hoy: " + "; ".join(why) return out def build_promoted_plan_today( db_path: Path = DB_PATH, logger: Optional[logging.Logger] = None, ) -> tuple[list[dict], list[dict], str]: """Construye las entradas de plan de las estrategias promovidas vivas HOY. Returns: entries: list[dict] con las claves EXACTAS de PlanEntry del Executor (ticker, side, entry_date, signal_strategy, prob_win, regime, rsi_14, momentum_20, in_universe, note). El Executor las mergea en su daily plan y las dimensiona/manda. evaluated: list[dict] diagnostico de TODAS las promovidas (vivas y no). note: resumen en una linea para el log/dashboard. `in_universe=True` para las promovidas: son equities reales del S&P 500 que Alpaca paper opera; pasaron el gate DSR del Lab, asi que son operables. El Executor sigue deduplicando contra las posiciones abiertas reales en Alpaca. """ if logger is None: logger = logging.getLogger("agente_bridge_promoted") promoted = load_promoted(db_path) evaluated: list[dict] = [] entries: list[dict] = [] for p in promoted: try: ev = evaluate_promoted(p, db_path) except Exception as exc: ev = {"ticker": p.get("ticker"), "strategy": p.get("strategy"), "live": False, "note": f"eval_error: {exc!r}"} evaluated.append(ev) if not ev.get("live"): continue side_str = (ev.get("side") or "LONG").upper() side = SIDE_LONG if side_str == "LONG" else "short" entries.append({ "ticker": ev["ticker"], "side": side, "entry_date": (ev.get("last_bar") or "")[:10], "signal_strategy": ev["strategy"], "prob_win": round(float(ev.get("prob_win", 0.0)), 4), "regime": ev.get("regime", ""), "rsi_14": round(float(ev.get("rsi_14", 0.0)), 2), "momentum_20": round(float(ev.get("momentum_20", 0.0)), 4), "in_universe": True, "note": ev.get("note", "promoted_live"), # Marca explícita "soy del bridge". El executor usa esto en # _is_bridge_entry para no aplicar la pausa de drift al carril # promovido, sea 3-part o custom single-name. "is_bridge": True, }) n_live = len(entries) if not promoted: note = ("Sin estrategias promovidas (asset_strategies vacia). Promueve " "alguna desde el Strategy Lab con --promote para que el bridge " "la pueda mandar a Alpaca.") elif n_live == 0: note = (f"{len(promoted)} estrategias promovidas evaluadas; ninguna con " f"entrada viva hoy (sus triggers/regimenes no disparan en la " f"ultima barra). El sistema esta plano por diseno.") else: live_names = ", ".join(f"{e['ticker']}({e['signal_strategy'].split('|')[0]})" for e in entries) note = (f"{n_live}/{len(promoted)} estrategias promovidas con entrada " f"VIVA hoy: {live_names}. Pasadas al daily plan del Executor.") logger.info(note) # Persistir el diagnostico al dashboard. Cualquier caller (executor, # n8n cron, boton del dashboard, CLI) actualiza el report con cada run — # asi el usuario ve "ningun trigger dispara hoy" en vez de un plan vacio # mudo, y para cada promovida la razon concreta (trigger off / regimen # off / ATR no valido / datos insuficientes / catalogo viejo). try: DASHBOARD_DATA_DIR.mkdir(parents=True, exist_ok=True) REPORT_PATH.write_text( json.dumps({ "generated_at": datetime.now(UTC).isoformat(), "n_promoted": len(promoted), "n_live": n_live, "summary": note, "live_entries": entries, "evaluated": evaluated, }, indent=2, ensure_ascii=False, default=str), encoding="utf-8", ) except Exception as exc: # pragma: no cover logger.warning("Bridge: no pude escribir %s — %s", REPORT_PATH, exc) return entries, evaluated, note # ───────────────────────────────────────────────────────────────────────────── # Main standalone — inspeccion. NO envia ordenes ni toca Alpaca. # ───────────────────────────────────────────────────────────────────────────── def main() -> int: parser = argparse.ArgumentParser( description="Bridge de estrategias promovidas -> daily plan del Executor " "(solo inspeccion; no envia ordenes).") parser.add_argument("--json", action="store_true", help="Salida JSON con el diagnostico completo.") args = parser.parse_args() logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", handlers=[logging.StreamHandler()], ) log = logging.getLogger("agente_bridge_promoted") entries, evaluated, note = build_promoted_plan_today(DB_PATH, log) if args.json: print(json.dumps({ "generated_at": datetime.now(UTC).isoformat(), "n_promoted": len(evaluated), "n_live": len(entries), "summary": note, "live_entries": entries, "evaluated": evaluated, }, indent=2, ensure_ascii=False, default=str)) return 0 sep = "=" * 84 print(f"\n{sep}") print("Leonex -- Bridge de estrategias promovidas (entradas vivas hoy)") print(sep) print(note) print() if evaluated: print(f"{'ticker':<9}{'tf':<5}{'strategy':<40}{'prob':>7}{'live':>7}") for ev in evaluated: print(f"{ev.get('ticker',''):<9}{ev.get('timeframe',''):<5}" f"{str(ev.get('strategy',''))[:39]:<40}" f"{ev.get('prob_win',0.0):>7.2f}" f"{('YES' if ev.get('live') else '-'):>7}") print(sep) if entries: print("Entradas que el Executor recogeria en su proximo dry-run/execute:") for e in entries: print(f" + {e['ticker']:<6} {e['signal_strategy']} " f"prob_win={e['prob_win']:.2f}") else: print("Ninguna entrada viva hoy. Nada que pasar al Executor.") print(sep) return 0 if __name__ == "__main__": raise SystemExit(main())