""" Agente Top-20 Rotation Selector de Leonex. ROTACION DINAMICA del bridge promovido: en vez de operar para siempre las estrategias que un dia fueron GOLD/SILVER, este agente rankea TODAS las estrategias de los 3 Labs (swing + scalping + custom) y selecciona el top-N (default 20). Las que entran al top quedan `active=1` en `asset_strategies` y el bridge las opera. Las que salen pasan a `active=0` (no se borran — quedan en histórico para análisis posterior). PROBLEMA QUE RESUELVE: "Strategy decay": una estrategia que fue GOLD hace 3 meses puede haber perdido su edge tras un cambio de regimen, pero sigue en asset_strategies operando inversiones perdedoras. La promocion estatica es fragil. LA SOLUCION: 1. Cada semana (recomendado lunes 12:00 UTC, tras el pipeline diario), este agente lee los 3 reports JSON. 2. Filtra por gates de calidad (tier in {GOLD, SILVER}, min_trades, etc.). 3. Rankea por (tier, DSR) — GOLD siempre primero, dentro de cada tier el DSR como tiebreak. 4. ANTI-FLICKER: una estrategia que YA esta activa Y lleva menos de MIN_WEEKS_IN_TOP semanas en el top, mantiene su slot aunque caiga en el ranking. Esto evita "intercambios cosmeticos" semana a semana. 5. Update de asset_strategies: - Top-N elegidas → active=1, last_top20_at = ahora, first_top20_at si era nueva. - Las que estaban active=1 pero salen del top → active=0. 6. Exporta `dashboard/data/top20_rotation_report.json` con entradas/ salidas/permanencias para visibilidad en el dashboard. NO ROMPE NADA SI NO SE ACTIVA: Si nunca se ejecuta este agente, la columna `active` no existe y load_promoted del bridge devuelve TODAS — comportamiento original. Activarlo es opt-in. REFERENCIAS: - Lopez de Prado (2018) AFML cap.16-17. Decay de estrategias. - Grinold & Kahn (1999) Active Portfolio Management cap.6. Ley fundamental: combinar muchas decisiones pequeñas e independientes. - Kahn & Lemmon (2016). "The Asset Manager's Dilemma". Por que el rebalanceo demasiado frecuente es peor que el insuficiente. USO: python agents/agente_top20_selector.py # dry-run python agents/agente_top20_selector.py --execute # aplica python agents/agente_top20_selector.py --execute --top-n 30 python agents/agente_top20_selector.py --execute --min-weeks 2 """ from __future__ import annotations import argparse import json import logging import sqlite3 import sys from dataclasses import asdict, dataclass, field from datetime import datetime try: from datetime import UTC except ImportError: from datetime import timezone UTC = timezone.utc from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] DATA_DIR = PROJECT_ROOT / "data" LOGS_DIR = PROJECT_ROOT / "logs" DASHBOARD_DATA_DIR = PROJECT_ROOT / "dashboard" / "data" DB_PATH = DATA_DIR / "Leonex.sqlite" SWING_REPORT_PATH = DASHBOARD_DATA_DIR / "strategy_lab_report.json" SCALPING_REPORT_PATH = DASHBOARD_DATA_DIR / "strategy_lab_scalping_report.json" CUSTOM_REPORT_PATH = DASHBOARD_DATA_DIR / "custom_strategies_report.json" OUTPUT_REPORT_PATH = DASHBOARD_DATA_DIR / "top20_rotation_report.json" # Gates de admision al ranking — solo strategies que pasan estos minimos # entran a competir por uno de los 20 slots. Sin esto, una BRONZE con DSR 0.3 # podria entrar al top-20 si el dia es flojo en GOLDs/SILVERs. MIN_DSR = 0.65 # bajo esto, irrelevante MIN_TRADES = 25 MIN_PROFIT_FACTOR = 1.10 # Tier rank: cuanto mas bajo, mejor. GOLD gana siempre a SILVER aunque DSR sea # mayor en SILVER. TIER_RANK = {"GOLD": 0, "SILVER": 1, "BRONZE": 2, "REJECTED": 3, "UNKNOWN": 4} @dataclass class Candidate: ticker: str strategy: str timeframe: str tier: str dsr: float sharpe: float n_trades: int profit_factor: float source: str # "swing" | "scalping" | "custom" # Estado en BD antes de la rotacion (para diff/anti-flicker) was_active: bool = False weeks_in_top: int = 0 first_top20_at: str = "" def rank_key(self) -> tuple: """Tuple para sorted(): menor es mejor. Tier primero, DSR descendente como tiebreak.""" return (TIER_RANK.get(self.tier, 99), -self.dsr, -self.sharpe) @dataclass class RotationReport: generated_at: str mode: str # "dry_run" | "execute" top_n: int min_weeks_in_top: int n_candidates_eligible: int n_entered: int # nuevas al top que no estaban n_held: int # ya activas que continuan n_protected: int # caidas en ranking pero mantenidas por anti-flicker n_exited: int # estaban activas y salen summary_note: str = "" new_top: list = field(default_factory=list) entered: list = field(default_factory=list) held: list = field(default_factory=list) protected: list = field(default_factory=list) exited: list = field(default_factory=list) def _build_logger() -> logging.Logger: LOGS_DIR.mkdir(parents=True, exist_ok=True) logger = logging.getLogger("agente_top20_selector") logger.setLevel(logging.INFO) logger.handlers.clear() fmt = logging.Formatter("%(asctime)s | %(levelname)s | %(name)s | %(message)s") fh = logging.FileHandler(LOGS_DIR / "agente_top20_selector.log", encoding="utf-8") fh.setFormatter(fmt) sh = logging.StreamHandler() sh.setFormatter(fmt) logger.addHandler(fh) logger.addHandler(sh) return logger def _safe_load_json(path: Path, logger: logging.Logger) -> dict: if not path.exists(): logger.warning("Report no existe: %s", path.name) return {} try: return json.loads(path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("No pude leer %s: %s", path.name, exc) return {} def ensure_rotation_schema(db_path: Path = DB_PATH) -> None: """Anade columnas `active`, `first_top20_at`, `last_top20_at` a asset_strategies si no existen. Idempotente.""" with sqlite3.connect(db_path) as conn: cur = conn.execute("PRAGMA table_info(asset_strategies)") cols = {r[1] for r in cur.fetchall()} if "active" not in cols: conn.execute( "ALTER TABLE asset_strategies ADD COLUMN active INTEGER DEFAULT 1") if "first_top20_at" not in cols: conn.execute( "ALTER TABLE asset_strategies ADD COLUMN first_top20_at TEXT") if "last_top20_at" not in cols: conn.execute( "ALTER TABLE asset_strategies ADD COLUMN last_top20_at TEXT") conn.commit() def _gather_candidates(swing: dict, scalping: dict, custom: dict, logger: logging.Logger) -> list[Candidate]: """Genera candidatos desde los 3 reports aplicando gates de calidad.""" cands: list[Candidate] = [] # 1) Swing y scalping comparten estructura for src_label, report in (("swing", swing), ("scalping", scalping)): for a in report.get("assets", []): ticker = a.get("ticker", "") tf = a.get("timeframe", "1d") for r in a.get("results", []): tier = r.get("tier", "UNKNOWN") if tier not in ("GOLD", "SILVER"): continue dsr = float(r.get("dsr", 0.0) or 0.0) n_trades = int(r.get("n_trades", 0) or 0) pf = float(r.get("profit_factor", 0.0) or 0.0) if dsr < MIN_DSR or n_trades < MIN_TRADES or pf < MIN_PROFIT_FACTOR: continue cands.append(Candidate( ticker=ticker, strategy=r.get("strategy", ""), timeframe=tf, tier=tier, dsr=dsr, sharpe=float(r.get("sharpe", 0.0) or 0.0), n_trades=n_trades, profit_factor=pf, source=src_label, )) # 2) Custom: strategies[].per_asset[].ticker/tier (estructura diferente). # PSR juega el rol del DSR a nivel ticker (DSR no aplica por ticker # individual segun la implementacion del Custom Lab). for s in custom.get("strategies", []): name = s.get("name", "") tf = s.get("timeframe", "1h") for pa in s.get("per_asset", []): tier = pa.get("tier", "UNKNOWN") if tier not in ("GOLD", "SILVER"): continue psr = float(pa.get("psr", 0.0) or 0.0) n_trades = int(pa.get("n_trades", 0) or 0) pf = float(pa.get("profit_factor", 0.0) or 0.0) if psr < MIN_DSR or n_trades < MIN_TRADES or pf < MIN_PROFIT_FACTOR: continue cands.append(Candidate( ticker=pa.get("ticker", ""), strategy=name, timeframe=tf, tier=tier, dsr=psr, sharpe=float(pa.get("sharpe_per_trade", 0.0) or 0.0), n_trades=n_trades, profit_factor=pf, source="custom", )) logger.info("Candidatos elegibles (tier in GOLD/SILVER + gates): %d", len(cands)) return cands def _enrich_with_db_state(cands: list[Candidate], db_path: Path, now: datetime) -> dict[tuple, dict]: """Para cada candidato, anade was_active y first_top20_at desde BD. Devuelve tambien dict de estado de las CURRENTLY-ACTIVE (para detectar salientes). Key = (ticker, strategy).""" current_active: dict[tuple, dict] = {} with sqlite3.connect(db_path) as conn: cols = {r[1] for r in conn.execute("PRAGMA table_info(asset_strategies)")} has_active = "active" in cols has_first = "first_top20_at" in cols sql = "SELECT ticker, strategy, " sql += "active" if has_active else "1" sql += ", " sql += "first_top20_at" if has_first else "''" sql += " FROM asset_strategies" for ticker, strategy, active, first_at in conn.execute(sql).fetchall(): key = (ticker, strategy) entry = {"active": bool(active), "first_top20_at": first_at or ""} if entry["active"]: current_active[key] = entry # Enriquecer candidato for c in cands: if c.ticker == ticker and c.strategy == strategy: c.was_active = entry["active"] c.first_top20_at = entry["first_top20_at"] # Calcular weeks_in_top desde first_top20_at for c in cands: if c.first_top20_at: try: dt = datetime.fromisoformat(c.first_top20_at.replace("Z", "+00:00")) delta = now - dt c.weeks_in_top = max(0, delta.days // 7) except Exception: c.weeks_in_top = 0 return current_active def select_top_n(cands: list[Candidate], current_active: dict[tuple, dict], top_n: int, min_weeks_in_top: int, logger: logging.Logger) -> tuple[list[Candidate], list[Candidate], list[Candidate], list[Candidate]]: """Aplica ranking + anti-flicker. Devuelve: new_top, entered, held, protected, exited.""" # Ordenar por (tier_rank, -dsr, -sharpe) — menor key = mejor sorted_cands = sorted(cands, key=lambda c: c.rank_key()) # Pasada 1: top-N "natural" segun ranking natural_top = sorted_cands[:top_n] natural_top_keys = {(c.ticker, c.strategy) for c in natural_top} # Pasada 2: anti-flicker. Las que YA estaban activas con weeks_in_top < # min_weeks_in_top reclaman su slot aunque hayan caido del top. protected: list[Candidate] = [] for c in sorted_cands[top_n:]: if c.was_active and 0 < c.weeks_in_top < min_weeks_in_top: protected.append(c) if protected: # Para cada protected, expulsamos al peor del top natural que NO sea # tambien protected. natural_top_mutable = list(natural_top) for p in protected: # Encontrar al peor del top natural que no este ya en protected worst_idx = None for i in range(len(natural_top_mutable) - 1, -1, -1): cand = natural_top_mutable[i] k = (cand.ticker, cand.strategy) if (cand.ticker, cand.strategy) in {(x.ticker, x.strategy) for x in protected}: continue worst_idx = i break if worst_idx is not None and len(natural_top_mutable) >= top_n: natural_top_mutable.pop(worst_idx) natural_top_mutable.append(p) new_top = natural_top_mutable[:top_n] else: new_top = list(natural_top) new_top_keys = {(c.ticker, c.strategy) for c in new_top} active_keys = set(current_active.keys()) entered = [c for c in new_top if (c.ticker, c.strategy) not in active_keys] held = [c for c in new_top if (c.ticker, c.strategy) in active_keys and c not in protected] exited_keys = active_keys - new_top_keys exited_list: list[Candidate] = [] for (t, s) in exited_keys: # Reconstruir minimo: ticker+strategy desde info que tengamos exited_list.append(Candidate( ticker=t, strategy=s, timeframe="?", tier="?", dsr=0.0, sharpe=0.0, n_trades=0, profit_factor=0.0, source="db_only", was_active=True, first_top20_at=current_active[(t, s)].get("first_top20_at", ""), )) logger.info( "Ranking aplicado: %d natural-top, %d protected (anti-flicker), " "final top=%d. Entered=%d Held=%d Exited=%d", len(natural_top), len(protected), len(new_top), len(entered), len(held), len(exited_list), ) return new_top, entered, held, protected, exited_list def apply_rotation(new_top: list[Candidate], exited: list[Candidate], db_path: Path, now_iso: str, logger: logging.Logger) -> int: """Aplica cambios en BD. Devuelve numero de registros tocados.""" n = 0 with sqlite3.connect(db_path) as conn: for c in new_top: # UPSERT: si la fila existe (mismo ticker), actualiza; si no, inserta # Note: la PK actual de asset_strategies es solo ticker, no # (ticker, strategy). Si el ticker cambia de estrategia, sobreescribe. existing = conn.execute( "SELECT first_top20_at FROM asset_strategies WHERE ticker = ?", (c.ticker,) ).fetchone() if existing: first_at = existing[0] or now_iso conn.execute( "UPDATE asset_strategies SET " "strategy = ?, timeframe = ?, dsr = ?, sharpe = ?, " "robust = ?, promoted_at = ?, active = 1, " "first_top20_at = COALESCE(first_top20_at, ?), " "last_top20_at = ? " "WHERE ticker = ?", (c.strategy, c.timeframe, c.dsr, c.sharpe, 1 if c.tier == "GOLD" else 0, now_iso, first_at, now_iso, c.ticker), ) else: conn.execute( "INSERT INTO asset_strategies " "(ticker, strategy, timeframe, dsr, sharpe, robust, " "promoted_at, active, first_top20_at, last_top20_at) " "VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?, ?)", (c.ticker, c.strategy, c.timeframe, c.dsr, c.sharpe, 1 if c.tier == "GOLD" else 0, now_iso, now_iso, now_iso), ) n += 1 for c in exited: conn.execute( "UPDATE asset_strategies SET active = 0 WHERE ticker = ? " "AND strategy = ?", (c.ticker, c.strategy), ) n += 1 conn.commit() logger.info("BD actualizada: %d filas (top + exited).", n) return n def export_report(report: RotationReport, path: Path, logger: logging.Logger) -> None: DASHBOARD_DATA_DIR.mkdir(parents=True, exist_ok=True) def _cand_dict(c: Candidate) -> dict: return { "ticker": c.ticker, "strategy": c.strategy, "timeframe": c.timeframe, "tier": c.tier, "dsr": round(c.dsr, 4), "sharpe": round(c.sharpe, 4), "n_trades": c.n_trades, "profit_factor": round(c.profit_factor, 4), "source": c.source, "weeks_in_top": c.weeks_in_top, "was_active": c.was_active, } payload = asdict(report) payload["new_top"] = [_cand_dict(c) for c in report.new_top] payload["entered"] = [_cand_dict(c) for c in report.entered] payload["held"] = [_cand_dict(c) for c in report.held] payload["protected"] = [_cand_dict(c) for c in report.protected] payload["exited"] = [_cand_dict(c) for c in report.exited] path.write_text( json.dumps(payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8", ) logger.info("Report exportado → %s", path) def main() -> int: parser = argparse.ArgumentParser( description="Top-N Rotation Selector de Leonex") parser.add_argument("--top-n", type=int, default=20, help="Cuantas estrategias en el bridge activo " "(default 20)") parser.add_argument("--min-weeks", type=int, default=4, help="Anti-flicker: una activa con menos de X semanas " "en el top no se expulsa aunque caiga en " "ranking (default 4)") parser.add_argument("--execute", action="store_true", help="Aplica cambios en BD. Sin esta bandera = dry-run.") parser.add_argument("--db", type=str, default=str(DB_PATH)) args = parser.parse_args() logger = _build_logger() db_path = Path(args.db) if not db_path.exists(): logger.error("DB no existe: %s", db_path) return 2 mode = "execute" if args.execute else "dry_run" logger.info( "Rotation start | mode=%s | top_n=%d | min_weeks=%d", mode, args.top_n, args.min_weeks, ) # Asegurar schema (idempotente) if args.execute: ensure_rotation_schema(db_path) swing = _safe_load_json(SWING_REPORT_PATH, logger) scalping = _safe_load_json(SCALPING_REPORT_PATH, logger) custom = _safe_load_json(CUSTOM_REPORT_PATH, logger) cands = _gather_candidates(swing, scalping, custom, logger) if not cands: logger.warning("Sin candidatos elegibles tras gates. Abort.") return 1 now = datetime.now(UTC) now_iso = now.isoformat() current_active = _enrich_with_db_state(cands, db_path, now) new_top, entered, held, protected, exited = select_top_n( cands, current_active, args.top_n, args.min_weeks, logger, ) if args.execute: apply_rotation(new_top, exited, db_path, now_iso, logger) note_parts = [ f"{len(cands)} elegibles", f"top={len(new_top)}", f"entered={len(entered)}", f"held={len(held)}", f"protected={len(protected)}", f"exited={len(exited)}", ] note = " · ".join(note_parts) + "." report = RotationReport( generated_at=now_iso, mode=mode, top_n=args.top_n, min_weeks_in_top=args.min_weeks, n_candidates_eligible=len(cands), n_entered=len(entered), n_held=len(held), n_protected=len(protected), n_exited=len(exited), new_top=new_top, entered=entered, held=held, protected=protected, exited=exited, summary_note=note, ) export_report(report, OUTPUT_REPORT_PATH, logger) # Pretty print humano sep = "=" * 78 print() print(sep) print(f"Top-{args.top_n} Rotation — mode={mode} · " f"anti-flicker={args.min_weeks}w") print(sep) print(note) print() if entered: print(f"ENTERED ({len(entered)}) — nuevas al top:") for c in entered: print(f" + {c.ticker:<6} {c.strategy[:50]:<50} " f"{c.tier:<6} DSR={c.dsr:.3f} ({c.source})") print() if exited: print(f"EXITED ({len(exited)}) — salen del top, pasan a active=0:") for c in exited: print(f" - {c.ticker:<6} {c.strategy[:50]:<50}") print() if protected: print(f"PROTECTED ({len(protected)}) — caidas en ranking pero " f"mantenidas por anti-flicker:") for c in protected: print(f" * {c.ticker:<6} {c.strategy[:50]:<50} " f"weeks={c.weeks_in_top}/{args.min_weeks}") print() if not args.execute and (entered or exited or protected): print(f"DRY-RUN: ninguna cambio aplicado en BD. " f"Relanza con --execute para confirmar.") return 0 if __name__ == "__main__": sys.exit(main())