""" Agente Custom Lab de Leonex. Evalua las estrategias PERSONALIZADAS registradas en custom_strategies.py (traducidas de PineScript) con el MISMO rigor que el Strategy Lab: - Cada entrada abre un trade discreto. Salida en multiplos de ATR (TP/SL/timeout) o en % fijo con cierre fin-de-dia (EOD), segun la estrategia. Soporta direccion LONG y SHORT. - Metricas sobre los TRADES, no sobre las barras. - PSR (Probabilistic Sharpe Ratio): probabilidad de que el Sharpe sea genuinamente > 0 dado el tamano muestral y las colas no normales. - DSR (Deflated Sharpe Ratio): con 2+ evaluaciones, descuenta ademas el multiple testing. - Walk-forward out-of-sample + tiers GOLD/SILVER/BRONZE/REJECTED. - Coste de transaccion descontado por trade. Salida: dashboard/data/custom_strategies_report.json Uso: python agents/agente_custom_lab.py """ from __future__ import annotations import argparse import json import logging import math 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 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 from custom_strategies import CUSTOM_STRATEGIES # noqa: E402 REPORT_PATH = lab.DASHBOARD_DATA_DIR / "custom_strategies_report.json" MIN_BARS = 250 # Coste round-trip por trade, en %. Estimacion Interactive Brokers Pro # (comision Tiered + spread efectivo SMART + slippage) en acciones liquidas # del S&P 500. Mismo valor que el lab swing para que sean comparables. DEFAULT_COST_PCT = 0.03 def psr(sr_per_trade: float, n_trades: int, skew: float, kurtosis: float) -> float: """Probabilistic Sharpe Ratio vs benchmark 0: probabilidad de que el Sharpe por trade sea genuinamente positivo, dado el numero de trades y la no-normalidad. NO descuenta multiple testing (eso es el DSR).""" if n_trades < 5: return 0.0 denom = (1.0 - skew * sr_per_trade + (kurtosis - 1.0) / 4.0 * sr_per_trade ** 2) if denom <= 0: denom = 1e-6 z = sr_per_trade * math.sqrt(n_trades - 1) / math.sqrt(denom) return float(lab._norm_cdf(z)) def simulate_pct_eod_trades(close: np.ndarray, high: np.ndarray, low: np.ndarray, entry: np.ndarray, filt: np.ndarray, tp_pct: float, sl_pct: float, day_codes: np.ndarray, direction: str = "long" ) -> list[dict]: """Simulador de trades con salida en % FIJO + cierre fin-de-dia (EOD). Replica la salida del PineScript: TP/SL como % del precio de entrada; si no salta ninguno antes del cambio de dia natural, cierra al cierre de la ultima barra del dia. Posiciones NO solapadas. Soporta long y short.""" n = len(close) trades: list[dict] = [] tp_f = tp_pct / 100.0 sl_f = sl_pct / 100.0 is_long = (direction != "short") i = 0 while i < n - 1: if entry[i] and filt[i] and close[i] > 0: ep = close[i] if is_long: tp = ep * (1.0 + tp_f) sl = ep * (1.0 - sl_f) else: tp = ep * (1.0 - tp_f) sl = ep * (1.0 + sl_f) exit_i = -1 xp = 0.0 reason = "" j = i + 1 while j < n: if day_codes[j] != day_codes[i]: # cierre fin de dia exit_i, xp, reason = j - 1, close[j - 1], "EOD" break if is_long: if low[j] <= sl: exit_i, xp, reason = j, sl, "SL" break if high[j] >= tp: exit_i, xp, reason = j, tp, "TP" break else: if high[j] >= sl: exit_i, xp, reason = j, sl, "SL" break if low[j] <= tp: exit_i, xp, reason = j, tp, "TP" break j += 1 if exit_i < 0: exit_i, xp, reason = n - 1, close[n - 1], "EOD" exit_i = max(exit_i, i) ret = (xp - ep) / ep if is_long else (ep - xp) / ep trades.append({ "entry_i": i, "exit_i": exit_i, "return": float(ret), "reason": reason, "bars_held": exit_i - i, }) i = exit_i + 1 else: i += 1 return trades @dataclass class CustomResult: name: str timeframe: str direction: str = "long" source: str = "" description: str = "" n_trials_hint: int = 1 n_bars: int = 0 n_trades: int = 0 win_rate: float = 0.0 profit_factor: float = 0.0 expectancy_pct: float = 0.0 sharpe: float = 0.0 sharpe_per_trade: float = 0.0 max_drawdown: float = 0.0 psr: float = 0.0 dsr: float = 0.0 confidence: float = 0.0 confidence_metric: str = "PSR" is_sharpe: float = 0.0 oos_sharpe: float = 0.0 regime_consistency: float = 0.0 tier: str = "REJECTED" tier_note: str = "" final_capital: float = 0.0 sim_total_return_pct: float = 0.0 trades: list = field(default_factory=list) # Breakdown POR ACTIVO: top 15 tickers ordenados por tier + PSR # (descendiente). Para cada ticker: n_trades, win_rate, profit_factor, # expectancy_pct, sharpe_per_trade, psr, tier y reason legible. per_asset: list = field(default_factory=list) note: str = "" def evaluate_strategy(strat: dict, timeframe: str, tickers: list[str], db_path: Path, cost_frac: float) -> CustomResult: """Evalua una estrategia custom en un timeframe sobre TODO el universo: agrega los trades de todos los tickers en una sola distribucion.""" direction = strat.get("direction", "long") res = CustomResult( name=strat["name"], timeframe=timeframe, direction=direction, source=strat["source"], description=strat["description"], n_trials_hint=strat["n_trials_hint"], ) cfg = strat["exit"] mode = cfg.get("mode", "atr") all_trades: list[dict] = [] total_bars = 0 for ticker in tickers: df = lab.load_prices(ticker, timeframe, db_path) if df.empty or len(df) < MIN_BARS: continue 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: continue total_bars += len(df) try: entry = strat["entry_fn"](df) entry = pd.Series(entry).reindex(df.index).fillna(False).astype(bool) except Exception as exc: res.note = f"entry_fn error: {exc!r}" continue dates = df.index close_a = df["close"].to_numpy(dtype=float) high_a = df["high"].to_numpy(dtype=float) low_a = df["low"].to_numpy(dtype=float) filt_a = np.ones(len(df), dtype=bool) try: if mode == "pct": day_codes = pd.factorize( pd.to_datetime(df.index).normalize())[0] trades = simulate_pct_eod_trades( close_a, high_a, low_a, entry.to_numpy(dtype=bool), filt_a, cfg.get("tp_pct", 3.0), cfg.get("sl_pct", 1.5), day_codes, direction) else: atr_a = lab._atr(df).to_numpy(dtype=float) trades = lab.simulate_tb_trades( close_a, high_a, low_a, atr_a, entry.to_numpy(dtype=bool), filt_a, cfg["tp"], cfg["sl"], cfg["timeout"]) except Exception as exc: res.note = f"simulate error: {exc!r}" continue for t in trades: t["return"] = t["return"] - cost_frac t["_entry_date"] = str(dates[t["entry_i"]])[:16] t["_exit_date"] = str(dates[t["exit_i"]])[:16] t["ticker"] = ticker all_trades.extend(trades) res.n_bars = total_bars n = len(all_trades) res.n_trades = n if n == 0: res.note = res.note or "0 trades — la regla no dispara en este timeframe." return res rets = np.array([t["return"] for t in all_trades], dtype=float) wins = rets[rets > 0] losses = rets[rets < 0] res.win_rate = round(float(len(wins) / n), 4) gross_win = float(wins.sum()) gross_loss = float(-losses.sum()) res.profit_factor = round(gross_win / gross_loss, 4) if gross_loss > 1e-12 \ else round(gross_win if gross_win > 0 else 0.0, 4) res.expectancy_pct = round(float(rets.mean() * 100.0), 4) sd = float(np.std(rets, ddof=1)) if n > 2 else 0.0 sr_trade = float(rets.mean() / sd) if sd > 1e-12 else 0.0 res.sharpe_per_trade = round(sr_trade, 4) equity = np.cumprod(1.0 + rets) running_max = np.maximum.accumulate(equity) res.max_drawdown = round(float((equity / running_max - 1.0).min()), 4) res.final_capital = round(lab.DEFAULT_INITIAL_CAPITAL * float(equity[-1]), 2) res.sim_total_return_pct = round(float((equity[-1] - 1.0) * 100.0), 2) res.sharpe = round(sr_trade * math.sqrt(min(n, 252)), 4) skew = float(pd.Series(rets).skew()) if n > 2 else 0.0 kurt = float(pd.Series(rets).kurtosis() + 3.0) if n > 3 else 3.0 res.psr = round(psr(sr_trade, n, skew, kurt), 4) wf = lab.walk_forward_trades(rets) res.is_sharpe = wf["is_sharpe"] res.oos_sharpe = wf["oos_sharpe"] res.regime_consistency = wf["regime_consistency"] res.trades = [{ "ticker": t["ticker"], "entry_date": t["_entry_date"], "exit_date": t["_exit_date"], "return_pct": round(t["return"] * 100, 3), "reason": t["reason"], "win": bool(t["return"] > 0), } for t in all_trades[-20:]] res._rets = rets # type: ignore[attr-defined] res._skew = skew # type: ignore[attr-defined] res._kurt = kurt # type: ignore[attr-defined] # ── Breakdown POR ACTIVO (top 15 tickers) ────────────────────────────── # Una regla puede tener edge solido en 8 tickers y ser puro ruido en # los otros 500. Calculamos PSR + tier por ticker INDEPENDIENTEMENTE # (PSR es la metrica honesta a nivel ticker — DSR no tiene sentido aqui # porque no hay multiple testing entre tickers de la misma regla). by_ticker: dict[str, list[dict]] = {} for t in all_trades: by_ticker.setdefault(t["ticker"], []).append(t) per_asset: list[dict] = [] for tk, trades_t in by_ticker.items(): n_t = len(trades_t) rets_t = np.array([t["return"] for t in trades_t], dtype=float) wins_t = rets_t[rets_t > 0] losses_t = rets_t[rets_t < 0] wr_t = float(len(wins_t) / n_t) if n_t else 0.0 gwin_t = float(wins_t.sum()) gloss_t = float(-losses_t.sum()) pf_t = (gwin_t / gloss_t if gloss_t > 1e-12 else (gwin_t if gwin_t > 0 else 0.0)) exp_t = float(rets_t.mean() * 100.0) if n_t else 0.0 sd_t = float(np.std(rets_t, ddof=1)) if n_t > 2 else 0.0 sr_t = float(rets_t.mean() / sd_t) if sd_t > 1e-12 else 0.0 skew_t = float(pd.Series(rets_t).skew()) if n_t > 2 else 0.0 kurt_t = float(pd.Series(rets_t).kurtosis() + 3.0) if n_t > 3 else 3.0 psr_t = psr(sr_t, n_t, skew_t, kurt_t) tier_t = lab.classify_tier(psr_t, n_t, pf_t, exp_t) # Razon legible: por que esa fila se queda en su tier if n_t < 12: reason_t = (f"REJECTED: solo {n_t} trade(s) (<12 minimo " f"para BRONZE)") elif exp_t <= 0: reason_t = (f"REJECTED: expectancy {exp_t:+.2f}% no positiva " f"sobre {n_t} trades") elif pf_t < 1.05: reason_t = (f"REJECTED: profit factor {pf_t:.2f} < 1.05 " f"sobre {n_t} trades") elif tier_t == lab.TIER_BRONZE: reason_t = (f"BRONZE: PSR {psr_t:.2f}, PF {pf_t:.2f}, " f"{n_t} trades — edge debil pero positivo") elif tier_t == lab.TIER_SILVER: reason_t = (f"SILVER: PSR {psr_t:.2f}, PF {pf_t:.2f}, " f"{n_t} trades — vigilar en forward") elif tier_t == lab.TIER_GOLD: reason_t = (f"GOLD: PSR {psr_t:.2f} fuerte, PF {pf_t:.2f}, " f"{n_t} trades — operable") else: reason_t = f"REJECTED: confianza PSR {psr_t:.2f} baja" per_asset.append({ "ticker": tk, "n_trades": n_t, "win_rate": round(wr_t, 4), "profit_factor": round(pf_t, 4), "expectancy_pct": round(exp_t, 4), "sharpe_per_trade": round(sr_t, 4), "psr": round(psr_t, 4), "tier": tier_t, "reason": reason_t, }) tier_rank_map = {lab.TIER_GOLD: 0, lab.TIER_SILVER: 1, lab.TIER_BRONZE: 2, lab.TIER_REJECTED: 3} per_asset.sort(key=lambda a: (tier_rank_map.get(a["tier"], 9), -a["psr"])) res.per_asset = per_asset[:15] return res def main() -> int: parser = argparse.ArgumentParser(description="Agente Custom Lab de Leonex") parser.add_argument("--limit-tickers", type=int, default=0, help="Limita tickers del universo (smoke test).") parser.add_argument("--cost-pct", type=float, default=DEFAULT_COST_PCT) 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_custom_lab.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_custom_lab") cost_frac = max(args.cost_pct, 0.0) / 100.0 # El Custom Lab descuenta el coste por su cuenta (en evaluate_strategy). # Ponemos a 0 el coste del lab base para que simulate_tb_trades — usado en # el modo de salida ATR — no lo descuente otra vez (evita doble conteo). lab.TRANSACTION_COST_PCT = 0.0 universe = lab.load_universe(lab.DB_PATH) if args.limit_tickers and len(universe) > args.limit_tickers: universe = universe[: args.limit_tickers] log.info("Custom Lab: %d estrategias registradas, universo %d tickers, " "coste %.3f%%/trade", len(CUSTOM_STRATEGIES), len(universe), args.cost_pct) results: list[CustomResult] = [] for strat in CUSTOM_STRATEGIES: for tf in strat["timeframes"]: try: res = evaluate_strategy(strat, tf, universe, lab.DB_PATH, cost_frac) results.append(res) log.info(" %s @%s [%s] -> trades=%d PSR=%.3f", res.name, tf, res.direction, res.n_trades, res.psr) except Exception as exc: log.warning("Fallo %s @%s: %s", strat["name"], tf, exc) traded = [r for r in results if r.n_trades >= 5] all_sr = [r.sharpe_per_trade for r in traded] declared = max([r.n_trials_hint for r in results], default=1) use_dsr = len(all_sr) >= 2 for r in results: if r.n_trades < 5: r.confidence = round(r.psr, 4) r.confidence_metric = "PSR" elif use_dsr: r.dsr = round(lab.deflated_sharpe_ratio( sr_per_trade=r.sharpe_per_trade, all_sr=all_sr, n_trades=r.n_trades, skew=getattr(r, "_skew", 0.0), kurtosis=getattr(r, "_kurt", 3.0)), 4) r.confidence = r.dsr r.confidence_metric = "DSR" else: r.confidence = round(r.psr, 4) r.confidence_metric = "PSR" tier = lab.classify_tier(r.confidence, r.n_trades, r.profit_factor, r.expectancy_pct) if tier in (lab.TIER_GOLD, lab.TIER_SILVER): if r.oos_sharpe <= 0.0: tier = lab.demote_tier(tier) r.tier_note = "demoted: out-of-sample Sharpe <= 0" elif r.regime_consistency < 0.5: tier = lab.demote_tier(tier) r.tier_note = "demoted: inconsistent across regimes" r.tier = tier if not r.note: r.note = (f"{r.confidence_metric}={r.confidence:.3f} sobre " f"{r.n_trades} trades. {r.tier}.") n_gold = sum(1 for r in results if r.tier == lab.TIER_GOLD) n_silver = sum(1 for r in results if r.tier == lab.TIER_SILVER) n_bronze = sum(1 for r in results if r.tier == lab.TIER_BRONZE) n_rejected = sum(1 for r in results if r.tier == lab.TIER_REJECTED) metric = "DSR (descuenta multiple testing)" if use_dsr else \ "PSR (sin descuento — sube mas estrategias)" summary = (f"{len(CUSTOM_STRATEGIES)} estrategias custom, " f"{len(results)} evaluaciones (estrategia x timeframe). " f"Confianza por {metric}. Coste {args.cost_pct:.3f}%/trade " f"descontado. Tiers: {n_gold} GOLD, {n_silver} SILVER, " f"{n_bronze} BRONZE, {n_rejected} REJECTED.") log.info(summary) payload = { "generated_at": datetime.now(UTC).isoformat(), "n_custom_strategies": len(CUSTOM_STRATEGIES), "n_evaluations": len(results), "confidence_metric": "DSR" if use_dsr else "PSR", "n_trials_declared": declared, "transaction_cost_pct": args.cost_pct, "initial_capital": lab.DEFAULT_INITIAL_CAPITAL, "n_gold": n_gold, "n_silver": n_silver, "n_bronze": n_bronze, "n_rejected": n_rejected, "tier_thresholds": lab.TIER_THRESHOLDS, "summary_note": summary, "strategies": [ {k: v for k, v in asdict(r).items() if not k.startswith("_")} for r in results ], } REPORT_PATH.write_text( json.dumps(payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8", ) log.info("Reporte exportado -> %s", REPORT_PATH) sep = "=" * 84 print(f"\n{sep}") print("Leonex -- Custom Lab (estrategias traducidas, evaluadas con rigor)") print(sep) print(summary) for r in sorted(results, key=lambda x: -x.confidence): print(f" [{r.tier:<8}] {r.name:<26} @{r.timeframe:<4} {r.direction:<5} " f"trades={r.n_trades:<5} {r.confidence_metric}={r.confidence:.3f} " f"PF={r.profit_factor:.2f} OOS={r.oos_sharpe:+.2f}") print(sep) return 0 if __name__ == "__main__": raise SystemExit(main())