""" Agente Permutation Test de Leonex. EL TEST DE PERMUTACION MONTE-CARLO (David Aronson, "Evidence-Based Technical Analysis", Wiley 2007, cap. 6-7) es la herramienta clasica contra el data-mining. La pregunta que responde: el edge de esta estrategia, ¿podria haber salido por PURA SUERTE? REFERENCIAS BIBLIOGRAFICAS - Aronson, D. (2007). Evidence-Based Technical Analysis: Applying the Scientific Method and Statistical Inference to Trading Signals. Wiley. Capitulo 6: multiple testing problem. Capitulo 7: Monte Carlo permutation tests aplicados a senales de trading. Este agente implementa la "permutation of entry timing" descrita en cap. 7, seccion 3. - Aronson + Masters (2013). Statistically Sound Machine Learning for Algorithmic Trading. Continuacion practica del libro de 2007, anade ML. - Bandy, H. (2011). Quantitative Trading Systems, 2nd ed. Cap. 8 sobre randomization tests para robustez. Complementa a Aronson con foco walk-forward. - Efron, B., Tibshirani, R. (1993). An Introduction to the Bootstrap. Chapman & Hall. Base estadistica del muestreo con reemplazo / resampling usado en los tests de permutacion modernos. METODO: Se mantiene la serie de precios REAL, pero se baraja el TIMING de las entradas: se colocan las MISMAS N entradas en barras AL AZAR y se vuelve a medir, con la misma logica de salida (TP/SL/timeout) y el mismo coste. Se repite cientos de veces para construir la distribucion de "lo que hace la suerte" sobre ese mismo activo. El p-value es la fraccion de barajadas aleatorias que IGUALAN O SUPERAN a la estrategia real. p <= 0.05 -> el timing de las entradas tiene skill genuino; la suerte casi nunca lo iguala. p alto -> entrar al azar lo hace igual de bien -> probable humo. POR QUE IMPORTA: Es una tercera lente anti-overfitting, INDEPENDIENTE del DSR (que descuenta el multiple testing) y del CPCV/PBO. El test de permutacion aisla una cosa muy concreta: si el MOMENTO de entrada del trigger aporta algo, o si la estrategia solo esta cabalgando la deriva del activo (cosa que tambien consiguen las entradas al azar). Salida: dashboard/data/permutation_test_report.json Uso: python agents/agente_permutation_test.py python agents/agente_permutation_test.py --n-perms 500 --top-n 40 python agents/agente_permutation_test.py --report ruta.json --db ruta.sqlite """ from __future__ import annotations import argparse import json import logging 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 / "permutation_test_report.json" DEFAULT_TOP_N = 40 DEFAULT_N_PERMS = 200 MIN_BARS = 250 MIN_ENTRIES = 5 # con menos entradas el test no es informativo P_REAL = 0.05 # p-value para "edge real" P_WEAK = 0.20 # p-value para "debil" def load_candidates(report: dict, top_n: int) -> list[dict]: """Mejor estrategia no-REJECTED por TICKER, ordenadas por Sharpe.""" best: 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), reverse=True) for r in ranked: if str(r.get("tier", "REJECTED")).upper() == "REJECTED": continue trg, flt, exc = r.get("trigger"), r.get("filter"), 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), } prev = best.get(ticker) if prev is None or cand["lab_sharpe"] > prev["lab_sharpe"]: best[ticker] = cand break return sorted(best.values(), key=lambda c: c["lab_sharpe"], reverse=True)[:top_n] def _mean_return(trades: list[dict]) -> float | None: if not trades: return None return float(np.mean([t["return"] for t in trades])) def test_strategy(spec: dict, db_path: Path, n_perms: int, rng: np.random.Generator, exit_configs: dict) -> dict | None: """Test de permutacion de UNA estrategia. Devuelve su p-value y veredicto, o None si no hay datos / entradas suficientes.""" try: df = lab.load_prices(spec["ticker"], spec["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 trg, flt, exc = spec["trigger"], spec["filter"], spec["exit_config"] 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 # Senal de entrada REAL = trigger Y filtro combinados. combined = (triggers[trg].to_numpy(dtype=bool) & filters[flt].to_numpy(dtype=bool)) k = int(combined.sum()) n = len(df) if k < MIN_ENTRIES or k >= n: 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) ones = np.ones(n, dtype=bool) def _sim(entry: np.ndarray) -> list[dict]: return lab.simulate_tb_trades(close, high, low, atr, entry, ones, cfg["tp"], cfg["sl"], cfg["timeout"]) real_trades = _sim(combined) real_metric = _mean_return(real_trades) if real_metric is None or len(real_trades) < MIN_ENTRIES: return None # Distribucion nula: K entradas en barras al azar, misma logica de salida. null: list[float] = [] for _ in range(n_perms): perm = np.zeros(n, dtype=bool) perm[rng.choice(n, size=k, replace=False)] = True m = _mean_return(_sim(perm)) if m is not None: null.append(m) if len(null) < max(20, n_perms // 2): return None null_arr = np.array(null, dtype=float) ge = int((null_arr >= real_metric).sum()) p_value = (1 + ge) / (1 + len(null_arr)) # correccion conservadora +1 verdict = ("REAL EDGE" if p_value <= P_REAL else "WEAK" if p_value <= P_WEAK else "LIKELY LUCK") return { "ticker": spec["ticker"], "timeframe": spec["timeframe"], "strategy": spec["strategy"], "trigger": trg, "filter": flt, "exit_config": exc, "tier": spec.get("tier", "BRONZE"), "n_trades": len(real_trades), "n_perms": len(null_arr), "real_metric_pct": round(real_metric * 100.0, 4), "null_median_pct": round(float(np.median(null_arr)) * 100.0, 4), "null_p95_pct": round(float(np.percentile(null_arr, 95)) * 100.0, 4), "p_value": round(float(p_value), 4), "verdict": verdict, } def main() -> int: parser = argparse.ArgumentParser( description="Agente Permutation Test de Leonex") parser.add_argument("--top-n", type=int, default=DEFAULT_TOP_N, help="Cuantas estrategias testear (default 40).") parser.add_argument("--n-perms", type=int, default=DEFAULT_N_PERMS, help="Permutaciones por estrategia (default 200).") parser.add_argument("--swing-report", type=str, default=str(SWING_REPORT_IN)) parser.add_argument("--scalping-report", type=str, default=str(SCALPING_REPORT_IN)) parser.add_argument("--db", type=str, default=str(lab.DB_PATH)) parser.add_argument("--seed", type=int, default=20260524) 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_permutation_test.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_permutation_test") db_path = Path(args.db) n_perms = max(args.n_perms, 50) top_n = max(args.top_n, 1) rng = np.random.default_rng(args.seed) log.info("Permutation Test: %d permutaciones/estrategia. Procesando swing " "y scalping.", n_perms) def process_report(report_path: Path, label: str) -> dict: """Test de permutacion sobre las mejores estrategias de un reporte.""" empty = {"results": [], "n_real_edge": 0, "n_weak": 0, "n_likely_luck": 0} if not report_path.exists(): return {**empty, "summary_note": ( f"No existe {report_path.name} — ejecuta el Strategy Lab " f"{label} primero.")} try: report = json.loads(report_path.read_text(encoding="utf-8")) except Exception as exc: return {**empty, "summary_note": f"No se pudo leer {report_path.name}: {exc}"} # Las salidas las trae el propio reporte (swing y scalping difieren). exit_configs = report.get("exit_configs") or lab.EXIT_CONFIGS candidates = load_candidates(report, top_n) log.info("[%s] %d estrategias a testear", label, len(candidates)) results: list[dict] = [] for i, spec in enumerate(candidates, 1): try: res = test_strategy(spec, db_path, n_perms, rng, exit_configs) except Exception as exc: log.warning(" [%s] fallo %s: %s", label, spec.get("ticker"), exc) res = None if res is None: continue results.append(res) log.info(" [%s] %3d/%d %-8s %-36s p=%.3f %s", label, i, len(candidates), res["ticker"], res["strategy"], res["p_value"], res["verdict"]) results.sort(key=lambda r: r["p_value"]) n_real = sum(1 for r in results if r["verdict"] == "REAL EDGE") n_weak = sum(1 for r in results if r["verdict"] == "WEAK") n_luck = sum(1 for r in results if r["verdict"] == "LIKELY LUCK") if results: summary = ( f"[{label}] Test de permutacion Monte-Carlo (Aronson) sobre " f"{len(results)} estrategias, {n_perms} barajadas del timing " f"de entrada cada una. {n_real} con EDGE REAL (p<=0.05), " f"{n_weak} debiles, {n_luck} probablemente suerte. Un p-value " f"alto = entrar al azar lo hace igual de bien; el trigger no " f"aporta timing.") else: summary = (f"[{label}] Sin estrategias testables — ejecuta el " f"Strategy Lab {label} primero.") log.info(summary) return {"results": results, "n_real_edge": n_real, "n_weak": n_weak, "n_likely_luck": n_luck, "summary_note": summary} swing = process_report(Path(args.swing_report), "swing") scalping = process_report(Path(args.scalping_report), "scalping") payload = { "generated_at": datetime.now(UTC).isoformat(), "method": "Monte-Carlo permutation test (Aronson) — shuffled entry timing", "n_perms": n_perms, "p_real_threshold": P_REAL, "p_weak_threshold": P_WEAK, "swing": swing, "scalping": scalping, } 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 -- Permutation Test (Aronson: ¿edge real o suerte?)") print(sep) for label, blk in (("SWING", swing), ("SCALPING", scalping)): print(f" [{label:<8}] {blk['n_real_edge']} REAL EDGE / " f"{blk['n_weak']} WEAK / {blk['n_likely_luck']} LIKELY LUCK") print(sep) return 0 if __name__ == "__main__": raise SystemExit(main())