""" Agente Portfolio Lab de Leonex. LA LEY FUNDAMENTAL DE LA GESTION ACTIVA (Grinold & Kahn, Active Portfolio Management, McGraw-Hill 1999, cap. 6): Information Ratio ≈ IC × √Amplitud REFERENCIAS BIBLIOGRAFICAS - Grinold, R., Kahn, R. (1999). Active Portfolio Management, 2nd ed. McGraw-Hill. Capitulo 6: la ley fundamental que justifica diversificar entre muchos edges pequenos en vez de buscar un edge gigante. - Pedersen, L. (2015). Efficiently Inefficient. Princeton. Capitulos 8-12 sobre factor investing y por que las anomalias persisten — base economica para esperar que combinaciones de estrategias mantengan IR > 0 en OOS. - Ilmanen, A. (2011). Expected Returns. Wiley. Etiquetado economico de estrategias por factor: trend, value, carry, low-vol. El Portfolio Lab usa esta taxonomia para verificar que las estrategias combinadas no son todas expressing el mismo factor. - Carver, R. (2015). Systematic Trading. Harriman House. Capitulo 14 (Portfolio construction): forecasts continuos + diversificacion por horizonte. La amplitud efectiva del Portfolio Lab se calcula segun la formula de Carver. - Lopez de Prado, M. (2018). Advances in Financial Machine Learning. Wiley. Capitulo 16 (HRP) — alternativa a equal-weight cuando hay correlaciones fuertes entre estrategias. La idea, traducida a Leonex: no necesitas UNA estrategia GOLD genial. Si combinas MUCHAS estrategias con un edge pequeno pero genuino y POCO correlacionadas entre si, el conjunto puede tener un ratio riesgo/retorno muy superior al de cualquiera de ellas suelta. La amplitud (numero de apuestas independientes) le gana al talento aislado. QUE HACE ESTE AGENTE: 1. Lee strategy_lab_report.json (el Strategy Lab swing). 2. Coge la MEJOR estrategia no-REJECTED de cada ticker. 3. RE-SIMULA cada una para reconstruir su serie diaria de retornos (con la misma maquinaria y el mismo coste IBKR que el lab). 4. Las COMBINA en una cartera equiponderada entre las estrategias activas en cada momento. 5. Mide la cartera combinada (Sharpe, PSR, drawdown) y calcula la AMPLITUD EFECTIVA a partir de la matriz de correlacion — cuantas apuestas de verdad independientes hay detras. 6. Ensena la ganancia por diversificacion: una cesta de BRONCEs puede ser, en conjunto, de categoria oro. Salida: dashboard/data/portfolio_lab_report.json Uso: python agents/agente_portfolio_lab.py python agents/agente_portfolio_lab.py --top-n 30 python agents/agente_portfolio_lab.py --report ruta.json --db ruta.sqlite """ from __future__ import annotations import argparse import json import logging import math import sys from datetime import datetime try: from datetime import UTC except ImportError: # Python < 3.11 from datetime import timezone UTC = timezone.utc from pathlib import Path import numpy as np import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent)) import agente_strategy_lab as lab # noqa: E402 SWING_REPORT_IN = lab.DASHBOARD_DATA_DIR / "strategy_lab_report.json" SCALPING_REPORT_IN = lab.DASHBOARD_DATA_DIR / "strategy_lab_scalping_report.json" REPORT_OUT = lab.DASHBOARD_DATA_DIR / "portfolio_lab_report.json" DEFAULT_TOP_N = 40 MIN_BARS = 250 MIN_SERIES_DAYS = 20 TRADING_DAYS = 252 # ── Clasificacion por HORIZONTE ────────────────────────────────────────────── # El horizonte de diseno de una estrategia = timeout de salida (en barras) x # duracion de la barra, expresado en dias de mercado (un dia de trading ~6.5h). # Tres cestas: corto / medio / largo. Corto y largo son edges poco # correlacionados — combinarlos es diversificacion real. BAR_HOURS = {"1d": 6.5, "4h": 4.0, "1h": 1.0, "30m": 0.5, "15m": 0.25, "5m": 1.0 / 12.0} TOP_PER_HORIZON = 8 # mejores estrategias por cesta en la cartera def horizon_days(timeframe: str, exit_cfg: str, exit_configs: dict) -> float: """Periodo de tenencia de diseno de una estrategia, en dias de mercado.""" bh = BAR_HOURS.get(timeframe, 6.5) cfg = (exit_configs or {}).get(exit_cfg) or {} timeout = cfg.get("timeout") or 15 return (float(timeout) * bh) / 6.5 def horizon_bucket(days: float) -> str: return "SHORT" if days < 2.0 else ("MEDIUM" if days <= 15.0 else "LONG") def psr(sharpe_ann: float, n_periods: int, skew: float, kurt: float) -> float: """Probabilistic Sharpe Ratio de la cartera vs benchmark 0: probabilidad de que el Sharpe sea genuinamente positivo dado el tamano muestral y la no-normalidad de los retornos.""" if n_periods < 10: return 0.0 sr = sharpe_ann / math.sqrt(TRADING_DAYS) # Sharpe por dia denom = 1.0 - skew * sr + (kurt - 1.0) / 4.0 * sr ** 2 if denom <= 0: denom = 1e-9 z = sr * math.sqrt(n_periods - 1) / math.sqrt(denom) return float(lab._norm_cdf(z)) def resimulate(ticker: str, timeframe: str, trg: str, flt: str, exc: str, db_path: Path, exit_configs: dict): """Re-simula una estrategia (trigger + filtro + salida) y devuelve su serie DIARIA de retornos: 0 en dias sin cierre de trade, indexada solo en el rango en el que la estrategia estuvo activa. None si no se puede.""" try: df = lab.load_prices(ticker, timeframe, db_path) except Exception: return None if df is None or df.empty or len(df) < MIN_BARS: return None 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) df = df.dropna(subset=["open", "high", "low", "close"]) if len(df) < MIN_BARS: return None try: triggers = lab.build_triggers(df) filters = lab.build_filters(df) except Exception: return None if trg not in triggers or flt not in filters or exc not in exit_configs: return None cfg = exit_configs[exc] atr = lab._atr(df).to_numpy(dtype=float) close = df["close"].to_numpy(dtype=float) high = df["high"].to_numpy(dtype=float) low = df["low"].to_numpy(dtype=float) try: trades = lab.simulate_tb_trades( close, high, low, atr, triggers[trg].to_numpy(dtype=bool), filters[flt].to_numpy(dtype=bool), cfg["tp"], cfg["sl"], cfg["timeout"]) except Exception: return None if not trades: return None idx = pd.to_datetime(df.index) exits: dict[pd.Timestamp, float] = {} for t in trades: d = pd.Timestamp(idx[t["exit_i"]]).normalize() exits[d] = exits.get(d, 0.0) + float(t["return"]) if not exits: return None cal = pd.bdate_range(min(exits), max(exits)) s = pd.Series(0.0, index=cal, dtype=float) for d, r in exits.items(): if d in s.index: s.loc[d] += r else: # alinea al dia habil mas cercano pos = min(int(s.index.searchsorted(d)), len(s.index) - 1) s.iloc[pos] += r return s def _ann_sharpe(series: pd.Series) -> float: s = series.dropna() sd = float(s.std(ddof=1)) if len(s) > 2 else 0.0 if sd <= 1e-12: return 0.0 return float(s.mean() / sd * math.sqrt(TRADING_DAYS)) def effective_breadth(corr: pd.DataFrame) -> float: """Numero EFECTIVO de estrategias independientes a partir de la matriz de correlacion: N^2 / suma(C). Si todas estan perfectamente correlacionadas -> 1; si son independientes -> N.""" n = corr.shape[0] if n == 0: return 0.0 total = float(np.nansum(corr.to_numpy())) if total <= 0: return float(n) return max(1.0, min(float(n), n * n / total)) def build_portfolio(strats: list[dict], db_path: Path, exit_configs: dict, log: logging.Logger) -> tuple[dict | None, list[dict]]: """Re-simula y combina las estrategias en una cartera (inverse-vol).""" series: dict[str, pd.Series] = {} used: list[dict] = [] for s in strats: ser = resimulate(s["ticker"], s["timeframe"], s["trigger"], s["filter"], s["exit_config"], db_path, exit_configs) if ser is None or len(ser) < MIN_SERIES_DAYS: continue if not np.isfinite(ser.std()) or ser.std() <= 1e-12: continue key = (f'{s["ticker"]}·{s["trigger"]}|{s["filter"]}' f'|{s["exit_config"]}@{s["timeframe"]}') series[key] = ser rec = dict(s) rec["key"] = key rec["single_sharpe"] = round(_ann_sharpe(ser), 3) used.append(rec) log.info(" re-sim OK %-46s %d dias Sharpe=%.2f", key, len(ser), rec["single_sharpe"]) if len(series) < 2: return None, used # Indice comun = union de los rangos de actividad de cada estrategia. common = None for ser in series.values(): common = ser.index if common is None else common.union(ser.index) mat = pd.DataFrame({k: v.reindex(common) for k, v in series.items()}) # Ponderacion INVERSA A LA VOLATILIDAD (risk parity ingenuo): cada # estrategia aporta el MISMO riesgo a la cartera. Sin esto una estrategia # muy volatil (p.ej. cripto) domina la varianza y se come la ganancia por # diversificacion. Es la logica de Carver y de HRP: igualar riesgo, no # capital. vols = {k: float(series[k].std(ddof=1)) for k in series} inv = {k: (1.0 / v if v and v > 1e-12 else 0.0) for k, v in vols.items()} tot_inv = sum(inv.values()) or 1.0 weights = {k: inv[k] / tot_inv for k in series} for rec in used: rec["weight_pct"] = round(weights.get(rec["key"], 0.0) * 100.0, 2) # Media ponderada entre las estrategias ACTIVAS cada dia, renormalizando # los pesos a las que ya han "nacido": la que aun no opera no diluye. W = pd.Series(weights) active_w = mat.notna().mul(W, axis=1).sum(axis=1) port = (mat.fillna(0.0).mul(W, axis=1).sum(axis=1) / active_w.replace(0.0, np.nan)).dropna() if len(port) < MIN_SERIES_DAYS: return None, used n = len(port) mean_d = float(port.mean()) std_d = float(port.std(ddof=1)) sharpe = mean_d / std_d * math.sqrt(TRADING_DAYS) if std_d > 1e-12 else 0.0 equity = (1.0 + port).cumprod() run_max = equity.cummax() max_dd = float((equity / run_max - 1.0).min()) total_ret = float(equity.iloc[-1] - 1.0) ann_ret = float(equity.iloc[-1] ** (TRADING_DAYS / n) - 1.0) if n > 0 else 0.0 ann_vol = std_d * math.sqrt(TRADING_DAYS) skew = float(port.skew()) if n > 2 else 0.0 kurt = float(port.kurtosis() + 3.0) if n > 3 else 3.0 psr_val = psr(sharpe, n, skew, kurt) # Amplitud: correlaciones por pares -> numero efectivo de estrategias. corr = mat.corr(min_periods=10).fillna(0.0) n_strats = len(series) n_eff = effective_breadth(corr) off = corr.to_numpy().sum() - n_strats avg_corr = off / (n_strats * (n_strats - 1)) if n_strats > 1 else 0.0 indiv = [_ann_sharpe(s) for s in series.values()] avg_single = float(np.mean(indiv)) if indiv else 0.0 best_single = float(np.max(indiv)) if indiv else 0.0 multiplier = (sharpe / avg_single) if avg_single > 1e-9 else 0.0 # Comparacion apples-to-apples: cada estrategia evaluada en el MISMO rango # de dias que la cartera, rellenando con 0 los dias en que NO estaba activa. # Sin esto, cada Sharpe individual se mide en su rango denso de actividad # mientras la cartera se mide en el rango UNION (mucho mas disperso), lo # cual infla los Sharpe individuales y aplasta el multiplo realizado. Con # todas las series en el mismo rango, la comparacion es justa y el multiplo # significa de verdad lo que dice. indiv_aligned: list[float] = [] for k in series: s_aligned = mat[k].fillna(0.0) sd = float(s_aligned.std(ddof=1)) if len(s_aligned) > 2 else 0.0 if sd > 1e-12: indiv_aligned.append(_ann_sharpe(s_aligned)) aligned_avg = float(np.mean(indiv_aligned)) if indiv_aligned else 0.0 aligned_mult = (sharpe / aligned_avg) if aligned_avg > 1e-9 else 0.0 verdict = ("STRONG" if psr_val >= 0.95 else "DECENT" if psr_val >= 0.80 else "WEAK") portfolio = { "n_days": n, "sharpe": round(sharpe, 3), "psr": round(psr_val, 4), "verdict": verdict, "ann_return_pct": round(ann_ret * 100, 2), "ann_vol_pct": round(ann_vol * 100, 2), "max_drawdown_pct": round(max_dd * 100, 2), "total_return_pct": round(total_ret * 100, 2), "final_capital": round(lab.DEFAULT_INITIAL_CAPITAL * float(equity.iloc[-1]), 2), "start": str(port.index[0])[:10], "end": str(port.index[-1])[:10], } breadth = { "n_strategies": n_strats, "effective_breadth": round(n_eff, 2), "avg_pairwise_corr": round(float(avg_corr), 4), "avg_single_sharpe": round(avg_single, 3), "best_single_sharpe": round(best_single, 3), "breadth_multiplier": round(multiplier, 2), "theoretical_multiplier": round(math.sqrt(n_eff), 2), # Comparacion JUSTA (mismo rango que la cartera, 0 en dias inactivos): "aligned_avg_single_sharpe": round(aligned_avg, 3), "aligned_multiplier": round(aligned_mult, 2), } return {"portfolio": portfolio, "breadth": breadth}, used def load_candidates(report: dict, top_n: int) -> list[dict]: """Mejor estrategia no-REJECTED por TICKER, ordenadas por Sharpe.""" best_by_ticker: dict[str, dict] = {} for a in report.get("assets", []): ticker = a.get("ticker") results = a.get("results") or [] if not ticker or not results: continue ranked = sorted( results, key=lambda r: (r.get("sharpe", 0.0), r.get("dsr", 0.0)), reverse=True) for r in ranked: if str(r.get("tier", "REJECTED")).upper() == "REJECTED": continue trg = r.get("trigger") flt = r.get("filter") exc = r.get("exit_config") if not (trg and flt and exc) and r.get("strategy", "").count("|") == 2: trg, flt, exc = r["strategy"].split("|") if not (trg and flt and exc): continue cand = { "ticker": ticker, "timeframe": r.get("timeframe") or a.get("timeframe") or "1d", "trigger": trg, "filter": flt, "exit_config": exc, "strategy": r.get("strategy", f"{trg}|{flt}|{exc}"), "tier": r.get("tier", "BRONZE"), "lab_sharpe": round(float(r.get("sharpe", 0.0)), 3), "lab_dsr": round(float(r.get("dsr", 0.0)), 3), } prev = best_by_ticker.get(ticker) if prev is None or cand["lab_sharpe"] > prev["lab_sharpe"]: best_by_ticker[ticker] = cand break cands = sorted(best_by_ticker.values(), key=lambda c: c["lab_sharpe"], reverse=True) return cands[:top_n] def main() -> int: parser = argparse.ArgumentParser(description="Agente Portfolio Lab de Leonex") parser.add_argument("--top-n", type=int, default=DEFAULT_TOP_N, help="Maximo de estrategias a combinar (default 40).") parser.add_argument("--swing-report", type=str, default=str(SWING_REPORT_IN), help="Ruta a strategy_lab_report.json (swing).") parser.add_argument("--scalping-report", type=str, default=str(SCALPING_REPORT_IN), help="Ruta a strategy_lab_scalping_report.json.") parser.add_argument("--db", type=str, default=str(lab.DB_PATH), help="Ruta a la base de datos SQLite.") args = parser.parse_args() lab.LOGS_DIR.mkdir(parents=True, exist_ok=True) lab.DASHBOARD_DATA_DIR.mkdir(parents=True, exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", handlers=[ logging.FileHandler(lab.LOGS_DIR / "agente_portfolio_lab.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_portfolio_lab") db_path = Path(args.db) top_n = max(args.top_n, 2) log.info("Portfolio Lab: coste IBKR %.3f%%/trade. Procesando swing y " "scalping.", lab.TRANSACTION_COST_PCT) def process_report(report_path: Path, label: str) -> dict: """Procesa un reporte del Strategy Lab (swing o scalping) y devuelve el bloque de cartera: {portfolio, breadth, components, summary_note}.""" if not report_path.exists(): return {"portfolio": None, "breadth": None, "components": [], "summary_note": (f"No existe {report_path.name} — ejecuta " f"el Strategy Lab {label} primero.")} try: report = json.loads(report_path.read_text(encoding="utf-8")) except Exception as exc: return {"portfolio": None, "breadth": None, "components": [], "summary_note": f"No se pudo leer {report_path.name}: {exc}"} # Las salidas las trae el propio reporte: swing y scalping usan # configuraciones distintas (quick/balanced/... vs scalp_tight/...). exit_configs = report.get("exit_configs") or lab.EXIT_CONFIGS candidates = load_candidates(report, top_n) log.info("[%s] %d candidatas (mejor no-REJECTED por ticker)", label, len(candidates)) if len(candidates) < 2: return {"portfolio": None, "breadth": None, "components": [], "summary_note": (f"Hacen falta >=2 estrategias no-REJECTED " f"en el Strategy Lab {label}; hay " f"{len(candidates)}.")} result, used = build_portfolio(candidates, db_path, exit_configs, log) if result is None: return {"portfolio": None, "breadth": None, "components": used, "summary_note": (f"Solo {len(used)} estrategia(s) {label} " f"re-simulables — no alcanza para una " f"cartera.")} p, b = result["portfolio"], result["breadth"] summary = ( f"[{label}] Combinadas {b['n_strategies']} estrategias " f"(≈{b['effective_breadth']} efectivas tras descontar correlacion). " f"Sharpe cartera {p['sharpe']} frente a " f"{b['aligned_avg_single_sharpe']} de media individual medida en el " f"MISMO rango de dias -> multiplo realizado x{b['aligned_multiplier']}. " f"El techo teorico de la Ley Fundamental (~sqrt amplitud) seria " f"x{b['theoretical_multiplier']}, PERO exige estrategias activas " f"simultaneamente cada dia; las nuestras son entradas discretas y " f"dispersas, asi que ese techo no se alcanza — la diversificacion " f"real esta en el RIESGO: max drawdown {p['max_drawdown_pct']}% y " f"PSR {p['psr']} ({p['verdict']}). Una cesta de edges debiles pero " f"poco correlacionados sigue siendo, en conjunto, mas robusta que " f"cualquiera por separado (Grinold & Kahn).") log.info(summary) return {"portfolio": p, "breadth": b, "components": sorted( used, key=lambda c: -c.get("single_sharpe", 0.0)), "summary_note": summary} def process_horizon() -> dict: """Cartera diversificada por HORIZONTE: junta las candidatas de swing y scalping, las clasifica en corto / medio / largo por su periodo de tenencia, y combina las mejores de cada cesta. Corto y largo plazo son edges poco correlacionados — el horizonte es un eje de diversificacion.""" cands: list[dict] = [] merged_exits: dict = {} for rp in (Path(args.swing_report), Path(args.scalping_report)): if not rp.exists(): continue try: rep = json.loads(rp.read_text(encoding="utf-8")) except Exception: continue ex = rep.get("exit_configs") or {} merged_exits.update(ex) for c in load_candidates(rep, 10 ** 9): c["horizon"] = horizon_bucket( horizon_days(c["timeframe"], c["exit_config"], ex)) cands.append(c) if len(cands) < 2: return {"portfolio": None, "breadth": None, "components": [], "summary_note": "Hacen falta estrategias no-REJECTED en el " "Strategy Lab (swing y/o scalping)."} picked: list[dict] = [] per_bucket: dict[str, int] = {} for b in ("SHORT", "MEDIUM", "LONG"): inb = sorted([c for c in cands if c["horizon"] == b], key=lambda c: -c["lab_sharpe"])[:TOP_PER_HORIZON] per_bucket[b] = len(inb) picked += inb if len(picked) < 2: return {"portfolio": None, "breadth": None, "components": [], "summary_note": "No hay suficientes campeones por horizonte " "para construir una cartera."} result, used = build_portfolio(picked, db_path, merged_exits, log) if result is None: return {"portfolio": None, "breadth": None, "components": used, "summary_note": "No se pudieron re-simular suficientes " "campeones por horizonte."} p, hb = result["portfolio"], result["breadth"] summary = ( f"[horizonte] Cartera de campeones por horizonte: " f"{per_bucket.get('SHORT', 0)} corto + {per_bucket.get('MEDIUM', 0)} " f"medio + {per_bucket.get('LONG', 0)} largo. Sharpe cartera " f"{p['sharpe']} vs media individual {hb['aligned_avg_single_sharpe']} " f"en el mismo rango -> x{hb['aligned_multiplier']}. Max drawdown " f"{p['max_drawdown_pct']}%, PSR {p['psr']} ({p['verdict']}). " f"Corto y largo plazo son edges poco correlacionados: combinarlos " f"reduce el riesgo de concentrarse en un solo horizonte.") log.info(summary) return {"portfolio": p, "breadth": hb, "components": sorted(used, key=lambda c: -c.get("single_sharpe", 0.0)), "summary_note": summary} swing = process_report(Path(args.swing_report), "swing") scalping = process_report(Path(args.scalping_report), "scalping") horizon = process_horizon() payload = { "generated_at": datetime.now(UTC).isoformat(), "fundamental_law": "Information Ratio ≈ IC × √Amplitud (Grinold & Kahn)", "transaction_cost_pct": lab.TRANSACTION_COST_PCT, "swing": swing, "scalping": scalping, "horizon": horizon, } REPORT_OUT.write_text( json.dumps(payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8") log.info("Reporte exportado -> %s", REPORT_OUT) sep = "=" * 80 print(f"\n{sep}") print("Leonex -- Portfolio Lab (Ley Fundamental de la Gestion Activa)") print(sep) for label, blk in (("SWING", swing), ("SCALPING", scalping), ("HORIZON", horizon)): p = blk.get("portfolio") if p: print(f" [{label:<8}] Sharpe {p['sharpe']} PSR {p['psr']} " f"DD {p['max_drawdown_pct']}% -> {p['verdict']}") else: print(f" [{label:<8}] {blk.get('summary_note', '-')}") print(sep) return 0 if __name__ == "__main__": raise SystemExit(main())