""" Leonex — Pipeline completo cross-platform. Equivalente Python de `actualizar_datos.bat` (Windows). Funciona en Linux, macOS y Windows. Ejecuta los 14 agentes en orden, captura logs, y devuelve codigo de salida apropiado. Uso: python actualizar_datos.py # pipeline completo python actualizar_datos.py --skip-intraday # solo daily python actualizar_datos.py --only universe,datos,regimen python actualizar_datos.py --strategy regime_adaptive_v1_lo python actualizar_datos.py --continue-on-error Pensado para crons en VPS/container donde el .bat no sirve. Tambien para pruebas locales en Linux/macOS. """ from __future__ import annotations import argparse import subprocess import sys import time from pathlib import Path from typing import Optional ROOT = Path(__file__).resolve().parent AGENTS_DIR = ROOT / "agents" DEFAULT_STRATEGY = "regime_adaptive_v1_lo" DEFAULT_THRESHOLD = "0.55" # ───────────────────────────────────────────────────────────────────────────── # Definicion de los 14 pasos del pipeline # ───────────────────────────────────────────────────────────────────────────── # Cada paso es (name, args, fatal_on_fail). # fatal_on_fail=True → si falla, parar el pipeline (datos imprescindibles) # fatal_on_fail=False → si falla, continuar (modulo opcional / observabilidad) # ───────────────────────────────────────────────────────────────────────────── def build_pipeline(strategy: str, threshold: str, include_intraday: bool = True ) -> list[tuple[str, list[str], bool]]: py = sys.executable steps = [ # name args fatal ("Universe", [py, "agents/agente_universe.py"], False), ("Datos", [py, "agents/agente_datos.py"], True), ] if include_intraday: steps.append( ("Datos intraday", [py, "agents/agente_datos_intraday.py"], False), ) steps.extend([ ("Regimen", [py, "agents/agente_regimen.py"], True), ("Senales", [py, "agents/agente_senales.py"], True), ("Triple Barrier", [py, "agents/agente_triple_barrier.py", "--strategy", strategy], True), ("Causal", [py, "agents/agente_causal.py"], False), ("Meta-Labeling", [py, "agents/agente_meta.py", "--strategy", strategy], True), ("Validacion", [py, "agents/agente_validacion.py", "--strategy", strategy, "--triple-barrier", "--meta-filter", threshold], True), ("CPCV + PBO", [py, "agents/agente_cpcv.py", "--strategy", strategy], False), ("Drift Monitor", [py, "agents/agente_drift_monitor.py", "--window", "60"], False), ("Executor", [py, "agents/agente_executor.py", "--strategy", strategy, "--threshold", threshold], False), ("HRP", [py, "agents/agente_hrp.py", "--plan"], False), ("Close Monitor", [py, "agents/agente_close_monitor.py"], False), ("Risk Monitor", [py, "agents/agente_risk_monitor.py"], False), ]) return steps # ───────────────────────────────────────────────────────────────────────────── # Ejecucion # ───────────────────────────────────────────────────────────────────────────── def run_step(name: str, args: list[str], fatal: bool, idx: int, total: int, continue_on_error: bool) -> bool: """Ejecuta un paso. Devuelve True si OK, False si fallo (y debe parar).""" print() print(f"[{idx}/{total}] {name}...") print(f" cmd: {' '.join(args)}") start = time.time() try: result = subprocess.run(args, cwd=str(ROOT), check=False) elapsed = time.time() - start if result.returncode == 0: print(f" ✓ OK ({elapsed:.1f}s)") return True # Fallo print(f" ✗ FALLO returncode={result.returncode} ({elapsed:.1f}s)") if fatal and not continue_on_error: return False return True # continuar aunque falle except FileNotFoundError as exc: print(f" ✗ FALLO: archivo no encontrado: {exc}") if fatal and not continue_on_error: return False return True except KeyboardInterrupt: print(f" ⚠ Interrumpido por usuario") return False except Exception as exc: print(f" ✗ FALLO inesperado: {exc!r}") if fatal and not continue_on_error: return False return True def main() -> int: parser = argparse.ArgumentParser(description="Pipeline completo de Leonex") parser.add_argument("--strategy", default=DEFAULT_STRATEGY, help=f"Estrategia (default {DEFAULT_STRATEGY})") parser.add_argument("--threshold", default=DEFAULT_THRESHOLD, help=f"Umbral Meta (default {DEFAULT_THRESHOLD})") parser.add_argument("--skip-intraday", action="store_true", help="Saltar descarga de datos intraday (1h/4h).") parser.add_argument("--only", type=str, default="", help="Solo ejecutar pasos cuyos nombres incluyan estas " "palabras (separadas por coma). Ej: --only datos,senales") parser.add_argument("--continue-on-error", action="store_true", help="No parar si un paso fatal falla. Util para debug.") args = parser.parse_args() pipeline = build_pipeline(args.strategy, args.threshold, include_intraday=not args.skip_intraday) if args.only: wanted = [w.strip().lower() for w in args.only.split(",") if w.strip()] pipeline = [ step for step in pipeline if any(w in step[0].lower() for w in wanted) ] if not pipeline: print(f"[!] Ningun paso coincide con --only={args.only}") return 1 total = len(pipeline) print("=" * 60) print(f" Leonex - Pipeline completo ({total} pasos)") print("=" * 60) print(f" Strategy : {args.strategy}") print(f" Threshold: {args.threshold}") print(f" Python : {sys.executable}") print(f" Root : {ROOT}") print("=" * 60) start_total = time.time() failed = 0 for i, (name, agent_args, fatal) in enumerate(pipeline, 1): ok = run_step(name, agent_args, fatal, i, total, args.continue_on_error) if not ok: elapsed = time.time() - start_total print() print("=" * 60) print(f" Pipeline INTERRUMPIDO en '{name}' ({elapsed:.1f}s total)") print("=" * 60) print(f" Use --continue-on-error para seguir tras un fallo") return 2 # No fatal_fail no incrementa failed; solo contamos los OK elapsed = time.time() - start_total print() print("=" * 60) print(f" Pipeline COMPLETO en {elapsed:.1f}s ({total} pasos)") print("=" * 60) print(f" Dashboard JSONs actualizados en dashboard/data/") print(f" Recarga http://leonex.187.127.86.200.nip.io/dashboard/index.html") return 0 if __name__ == "__main__": raise SystemExit(main())