""" Agente CPCV de Leonex — Combinatorial Purged Cross-Validation + PBO. Sigue Lopez de Prado "Advances in Financial Machine Learning" Wiley 2018, cap. 12. REFERENCIAS BIBLIOGRAFICAS - Lopez de Prado, M. (2018). Advances in Financial Machine Learning. Wiley. Capitulo 12 (CPCV). Capitulo 11 (Probability of Backtest Overfitting / PBO). - Bailey, D., Borwein, J., Lopez de Prado, M., Zhu, Q. (2017). "The Probability of Backtest Overfitting". Journal of Computational Finance 20(4). Paper original del PBO. Define la metrica que esta card surfacea como VERDE/RED. - Bailey, D., Lopez de Prado, M. (2014). "The Deflated Sharpe Ratio". Journal of Portfolio Management 40(5). Corrige el Sharpe IS por multiple testing — imprescindible cuando barres 480 estrategias en el Strategy Lab. - Aronson, D. (2007). Evidence-Based Technical Analysis. Wiley. Capitulos 5-7 formalizan multiple testing + data mining bias. El PBO RED 52% que ves en Leonex es exactamente el escenario del que Aronson advierte. PROBLEMA: Un Walk-Forward tradicional produce UN solo path de backtest. Si la estrategia parece buena, no sabes si es genuina o si tuviste suerte con el orden particular del split. SOLUCION: CPCV divide los trades en N grupos contiguos. De ahi se eligen k grupos como test y los demas como train. Eso produce C(N,k) combinaciones de splits, cada una con su Sharpe OOS. Tienes una DISTRIBUCION empirica de Sharpes en vez de un solo numero, lo que te dice si tu estrategia es robusta o ruido. Por construccion, cada observacion aparece en (k/N) de los paths como test, asi cubrimos todo el espacio. Por defecto N=8 y k=2 → C(8,2)=28 paths backtest distintos. PBO (Probability of Backtest Overfitting): Bailey, Borwein, Lopez de Prado, Zhu (2014). Procedimiento: 1) Defines S configuraciones de la estrategia (aqui: distintos umbrales Meta). 2) Para cada path CPCV, rankas las S configuraciones por Sharpe en train. Identificas la config #1 (mejor IS). 3) Miras el rank de esa misma config en test del mismo path. 4) PBO = fraccion de paths donde la mejor IS termino en la mitad inferior OOS. PBO < 0.30 → GREEN estrategia robusta PBO 0.30-0.50 → YELLOW vigilar PBO >= 0.50 → RED practicamente ruido Uso: python agents/agente_cpcv.py python agents/agente_cpcv.py --n-groups 8 --k-test 2 --embargo-pct 0.01 python agents/agente_cpcv.py --thresholds 0.50,0.55,0.60,0.65 """ from __future__ import annotations import argparse import json import logging import math import sqlite3 import sys from dataclasses import asdict, dataclass, field from datetime import datetime try: from datetime import UTC # Python 3.11+ except ImportError: # Python 3.10 from datetime import timezone UTC = timezone.utc from itertools import combinations from pathlib import Path from typing import Optional import numpy as np import pandas as pd 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" REPORT_PATH = DASHBOARD_DATA_DIR / "cpcv_report.json" DEFAULT_STRATEGY = "regime_adaptive_v1_lo" DEFAULT_N_GROUPS = 8 DEFAULT_K_TEST = 2 DEFAULT_EMBARGO_PCT = 0.01 # 1% de T (Lopez de Prado AFML cap.12) DEFAULT_THRESHOLDS = [0.50, 0.52, 0.55, 0.58, 0.60, 0.62, 0.65] DEFAULT_MIN_TRADES_PER_FOLD = 5 # un fold con menos trades se descarta DEFAULT_TRADES_PER_YEAR = 50 # aproximacion estandar para Sharpe anualizado # Un Sharpe anualizado (x sqrt(50)) calculado sobre 3-5 trades es ruido puro y # explota a valores absurdos (Sharpe 16, std 33 entre paths) que saturan el DSR. # Exigimos una muestra minima y capamos el valor a una banda razonable. SHARPE_MIN_TRADES = 10 SHARPE_CAP = 8.0 # |Sharpe anualizado| nunca por encima de esto PBO_THRESHOLDS = {"GREEN": 0.30, "YELLOW": 0.50} # >= YELLOW = RED @dataclass class ConfigStat: threshold: float n_paths_with_data: int mean_train_sharpe: float mean_test_sharpe: float std_test_sharpe: float pct_paths_top_train: float # % paths donde fue la mejor en train pct_paths_top_test: float # % paths donde fue la mejor en test overfit_gap: float # train - test (positivo = sobrefiteo) @dataclass class PathResult: path_id: int train_groups: list[int] test_groups: list[int] n_train_trades: int n_test_trades: int purged_count: int embargo_count: int best_train_threshold: float best_train_sharpe: float test_sharpe_of_best_train: float rank_oos_of_best_is: int # 1 = top, N = worst test_sharpes_by_config: dict[str, float] @dataclass class CPCVReport: generated_at: str strategy: str n_total_trades: int n_groups: int k_test: int n_paths: int embargo_pct: float thresholds: list[float] # Resumen Sharpe OOS (combinando todos los paths y configs) sharpe_dist_p05: float sharpe_dist_p25: float sharpe_dist_p50: float sharpe_dist_p75: float sharpe_dist_p95: float sharpe_dist_mean: float sharpe_dist_std: float # PBO global (usando la config #1 in-sample por path) pbo: float pbo_verdict: str # GREEN | YELLOW | RED pbo_note: str # Detalle por configuracion config_stats: list[ConfigStat] = field(default_factory=list) # Detalle por path (puede ser grande, recortable) paths: list[PathResult] = field(default_factory=list) # ───────────────────────────────────────────────────────────────────────────── # Carga de datos # ───────────────────────────────────────────────────────────────────────────── def load_trades_with_meta(strategy: str, db_path: Path = DB_PATH) -> pd.DataFrame: """Trades del Triple Barrier joineados con sus prob_win del Meta-Labeling. Devuelve un DataFrame ordenado por entry_date. Cada fila tiene: ticker, entry_date, exit_date, return_pct, days_held, prob_win. Trades sin prob_win quedan con NaN y se filtran luego. """ if not db_path.exists(): return pd.DataFrame() with sqlite3.connect(db_path) as conn: df = pd.read_sql( """ SELECT t.ticker, t.entry_date, t.exit_date, t.return_pct, t.days_held, m.prob_win FROM trades t LEFT JOIN meta_predictions m ON m.ticker = t.ticker AND m.entry_date = t.entry_date AND m.strategy = t.strategy WHERE t.strategy = ? ORDER BY t.entry_date ASC, t.ticker ASC """, conn, params=(strategy,), parse_dates=["entry_date", "exit_date"], ) df = df.dropna(subset=["return_pct", "prob_win"]).reset_index(drop=True) return df # ───────────────────────────────────────────────────────────────────────────── # Split + purge + embargo # ───────────────────────────────────────────────────────────────────────────── def make_groups(df: pd.DataFrame, n_groups: int) -> np.ndarray: """Asigna cada trade (ordenado por entry_date) a un grupo contiguo. Devuelve array shape (T,) con valores 0..n_groups-1. """ n = len(df) if n < n_groups: return np.zeros(n, dtype=int) # bins balanceados bins = np.array_split(np.arange(n), n_groups) groups = np.empty(n, dtype=int) for g, idx in enumerate(bins): groups[idx] = g return groups def apply_purge_embargo( df: pd.DataFrame, train_mask: np.ndarray, test_mask: np.ndarray, embargo_pct: float = DEFAULT_EMBARGO_PCT, ) -> tuple[np.ndarray, np.ndarray, int, int]: """Aplica purge (elimina overlaps temporales train↔test) + embargo. Devuelve (new_train_mask, test_mask, n_purged, n_embargoed). PURGE: un trade del train cuya ventana [entry_date, exit_date] se solapa con el rango temporal de cualquier trade del test se elimina. EMBARGO: tras cada ventana de test, se descarta del train un % de observaciones inmediatamente despues (corrige autocorrelacion). """ if not test_mask.any(): return train_mask, test_mask, 0, 0 test_dates = df.loc[test_mask, ["entry_date", "exit_date"]] test_min = test_dates["entry_date"].min() test_max = test_dates["exit_date"].max() # PURGE: eliminar trades del train cuyo [entry, exit] cae dentro del rango test train_idx = np.where(train_mask)[0] purge_drop = np.zeros(len(df), dtype=bool) for i in train_idx: entry = df.iloc[i]["entry_date"] exit_ = df.iloc[i]["exit_date"] # solape temporal con la ventana test if exit_ >= test_min and entry <= test_max: purge_drop[i] = True # EMBARGO: descartar las primeras embargo_pct*T filas inmediatamente # despues de cada bloque test contiguo embargo_drop = np.zeros(len(df), dtype=bool) n = len(df) embargo_n = max(1, int(round(embargo_pct * n))) # localizamos los indices test contiguos test_idx_sorted = np.where(test_mask)[0] if test_idx_sorted.size: # bloque test = ultimo indice test → embargo cubre [last+1 : last+1+embargo_n] groups_contig = [] cur = [test_idx_sorted[0]] for i in test_idx_sorted[1:]: if i == cur[-1] + 1: cur.append(i) else: groups_contig.append(cur) cur = [i] groups_contig.append(cur) for block in groups_contig: last = block[-1] for j in range(last + 1, min(last + 1 + embargo_n, n)): if train_mask[j]: embargo_drop[j] = True new_train_mask = train_mask & ~purge_drop & ~embargo_drop return new_train_mask, test_mask, int(purge_drop.sum()), int(embargo_drop.sum()) # ───────────────────────────────────────────────────────────────────────────── # Sharpe por configuracion # ───────────────────────────────────────────────────────────────────────────── def sharpe_annualized(returns: np.ndarray, periods_per_year: float = DEFAULT_TRADES_PER_YEAR ) -> float: """Sharpe ratio anualizado. Asume Rf=0. Protege contra std numericamente cero (errores de coma flotante en arrays constantes pueden dar std~1e-17 en vez de 0 exacto). """ if returns.size < SHARPE_MIN_TRADES: # Muestra demasiado pequena: el Sharpe anualizado no es fiable. return 0.0 mu = float(np.mean(returns)) sd = float(np.std(returns, ddof=1)) if sd <= 1e-9 or not math.isfinite(sd): return 0.0 sharpe = mu / sd * math.sqrt(periods_per_year) if not math.isfinite(sharpe): return 0.0 # Cap defensivo: evita que un fold ruidoso domine la distribucion OOS. return float(max(-SHARPE_CAP, min(SHARPE_CAP, sharpe))) def evaluate_config_on_subset(df: pd.DataFrame, mask: np.ndarray, threshold: float) -> tuple[float, int]: """Filtra trades del subset con prob_win >= threshold y devuelve (Sharpe, N). Si no hay trades que pasen el filtro, Sharpe = 0 y N = 0. """ sub = df.loc[mask] kept = sub[sub["prob_win"] >= threshold] if len(kept) < SHARPE_MIN_TRADES: return 0.0, len(kept) rets = kept["return_pct"].to_numpy() return sharpe_annualized(rets), len(kept) # ───────────────────────────────────────────────────────────────────────────── # PBO calculation # ───────────────────────────────────────────────────────────────────────────── def compute_pbo(paths: list[PathResult], thresholds: list[float]) -> tuple[float, str, str]: """PBO segun Bailey-Borwein-Lopez de Prado-Zhu (2014). Para cada path: identifica la config con mejor Sharpe en TRAIN, mira su rank en TEST. PBO = fraccion de paths donde el rank OOS esta en la mitad inferior (peor 50%). """ if not paths or len(thresholds) < 2: return 0.0, "YELLOW", "Insufficient data for PBO" n_configs = len(thresholds) median_rank = (n_configs + 1) / 2.0 # rank 1 = mejor; mediana = N+1/2 n_bad = 0 n_valid = 0 for p in paths: if p.rank_oos_of_best_is <= 0: continue n_valid += 1 if p.rank_oos_of_best_is > median_rank: n_bad += 1 if n_valid == 0: return 0.0, "YELLOW", "No valid paths for PBO" pbo = n_bad / n_valid if pbo < PBO_THRESHOLDS["GREEN"]: verdict = "GREEN" note = f"PBO {pbo*100:.0f}% < {PBO_THRESHOLDS['GREEN']*100:.0f}% — strategy appears robust." elif pbo < PBO_THRESHOLDS["YELLOW"]: verdict = "YELLOW" note = (f"PBO {pbo*100:.0f}% borderline. Some sign of overfitting; " f"monitor or reduce config space.") else: verdict = "RED" note = (f"PBO {pbo*100:.0f}% >= 50%. Strategy likely overfit. Best in-sample " f"configurations underperform out-of-sample more than half the time.") return float(pbo), verdict, note # ───────────────────────────────────────────────────────────────────────────── # Runner # ───────────────────────────────────────────────────────────────────────────── def run_cpcv( df: pd.DataFrame, n_groups: int = DEFAULT_N_GROUPS, k_test: int = DEFAULT_K_TEST, embargo_pct: float = DEFAULT_EMBARGO_PCT, thresholds: Optional[list[float]] = None, logger: Optional[logging.Logger] = None, ) -> tuple[list[PathResult], list[ConfigStat]]: """Genera todos los paths CPCV y evalua cada config en cada path.""" if logger is None: logger = logging.getLogger("cpcv") if thresholds is None: thresholds = DEFAULT_THRESHOLDS n = len(df) if n < n_groups * 2: logger.warning("Insuficientes trades (%d) para %d grupos. Aborto.", n, n_groups) return [], [] groups = make_groups(df, n_groups) combos = list(combinations(range(n_groups), k_test)) logger.info("CPCV: %d grupos, k=%d test, %d paths combinatoria, %d trades", n_groups, k_test, len(combos), n) paths: list[PathResult] = [] # Para PBO necesitamos matrix path × config con Sharpes train y test train_sh_matrix: list[list[float]] = [] # [path][config] test_sh_matrix: list[list[float]] = [] for path_id, test_groups in enumerate(combos): test_mask = np.isin(groups, test_groups) train_mask = ~test_mask if test_mask.sum() < DEFAULT_MIN_TRADES_PER_FOLD: continue train_mask_post, _, n_purged, n_embargo = apply_purge_embargo( df, train_mask, test_mask, embargo_pct=embargo_pct, ) if train_mask_post.sum() < DEFAULT_MIN_TRADES_PER_FOLD: continue train_sharpes = [] test_sharpes = [] test_sharpes_dict = {} for thr in thresholds: s_train, _ = evaluate_config_on_subset(df, train_mask_post, thr) s_test, _ = evaluate_config_on_subset(df, test_mask, thr) train_sharpes.append(s_train) test_sharpes.append(s_test) test_sharpes_dict[f"{thr:.2f}"] = round(float(s_test), 4) train_sh_matrix.append(train_sharpes) test_sh_matrix.append(test_sharpes) # Mejor train (mayor Sharpe IS) best_idx = int(np.argmax(train_sharpes)) best_thr = thresholds[best_idx] # Rank OOS de la mejor IS (1 = top) order = np.argsort(test_sharpes)[::-1] # decreasing rank_oos = int(np.where(order == best_idx)[0][0]) + 1 paths.append(PathResult( path_id=path_id, train_groups=[g for g in range(n_groups) if g not in test_groups], test_groups=list(test_groups), n_train_trades=int(train_mask_post.sum()), n_test_trades=int(test_mask.sum()), purged_count=n_purged, embargo_count=n_embargo, best_train_threshold=float(best_thr), best_train_sharpe=float(train_sharpes[best_idx]), test_sharpe_of_best_train=float(test_sharpes[best_idx]), rank_oos_of_best_is=rank_oos, test_sharpes_by_config=test_sharpes_dict, )) # Estadisticas por configuracion config_stats: list[ConfigStat] = [] if train_sh_matrix and test_sh_matrix: train_arr = np.array(train_sh_matrix) # shape (P, C) test_arr = np.array(test_sh_matrix) n_paths_eff = train_arr.shape[0] # % paths en los que cada config fue la mejor (train/test) best_train_per_path = np.argmax(train_arr, axis=1) best_test_per_path = np.argmax(test_arr, axis=1) for c, thr in enumerate(thresholds): mean_train = float(np.mean(train_arr[:, c])) mean_test = float(np.mean(test_arr[:, c])) std_test = float(np.std(test_arr[:, c], ddof=1)) if n_paths_eff > 1 else 0.0 pct_top_train = float(np.mean(best_train_per_path == c)) pct_top_test = float(np.mean(best_test_per_path == c)) config_stats.append(ConfigStat( threshold=float(thr), n_paths_with_data=int(n_paths_eff), mean_train_sharpe=round(mean_train, 4), mean_test_sharpe=round(mean_test, 4), std_test_sharpe=round(std_test, 4), pct_paths_top_train=round(pct_top_train, 4), pct_paths_top_test=round(pct_top_test, 4), overfit_gap=round(mean_train - mean_test, 4), )) return paths, config_stats # ───────────────────────────────────────────────────────────────────────────── # Schema + persistencia # ───────────────────────────────────────────────────────────────────────────── def ensure_schema(db_path: Path = DB_PATH) -> None: with sqlite3.connect(db_path) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS cpcv_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, strategy TEXT NOT NULL, n_trades INTEGER, n_groups INTEGER, k_test INTEGER, n_paths INTEGER, embargo_pct REAL, pbo REAL, pbo_verdict TEXT, sharpe_p50 REAL, best_config_overall REAL, summary_note TEXT ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS cpcv_paths ( run_id INTEGER NOT NULL, path_id INTEGER NOT NULL, n_train_trades INTEGER, n_test_trades INTEGER, purged INTEGER, embargoed INTEGER, best_train_threshold REAL, best_train_sharpe REAL, test_sharpe_of_best_train REAL, rank_oos_of_best_is INTEGER, PRIMARY KEY (run_id, path_id) ) """ ) conn.commit() def persist_run(report: CPCVReport, db_path: Path = DB_PATH) -> int: ensure_schema(db_path) best_overall = "" if report.config_stats: best = max(report.config_stats, key=lambda c: c.mean_test_sharpe) best_overall = best.threshold with sqlite3.connect(db_path) as conn: cur = conn.execute( """ INSERT INTO cpcv_runs (timestamp, strategy, n_trades, n_groups, k_test, n_paths, embargo_pct, pbo, pbo_verdict, sharpe_p50, best_config_overall, summary_note) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (report.generated_at, report.strategy, report.n_total_trades, report.n_groups, report.k_test, report.n_paths, float(report.embargo_pct), float(report.pbo), report.pbo_verdict, float(report.sharpe_dist_p50), float(best_overall) if best_overall != "" else None, report.pbo_note), ) run_id = cur.lastrowid conn.executemany( """ INSERT INTO cpcv_paths (run_id, path_id, n_train_trades, n_test_trades, purged, embargoed, best_train_threshold, best_train_sharpe, test_sharpe_of_best_train, rank_oos_of_best_is) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [(run_id, p.path_id, p.n_train_trades, p.n_test_trades, p.purged_count, p.embargo_count, p.best_train_threshold, p.best_train_sharpe, p.test_sharpe_of_best_train, p.rank_oos_of_best_is) for p in report.paths], ) conn.commit() return int(run_id) # ───────────────────────────────────────────────────────────────────────────── # Main # ───────────────────────────────────────────────────────────────────────────── def main() -> int: parser = argparse.ArgumentParser(description="Agente CPCV de Leonex") parser.add_argument("--strategy", default=DEFAULT_STRATEGY) parser.add_argument("--n-groups", type=int, default=DEFAULT_N_GROUPS) parser.add_argument("--k-test", type=int, default=DEFAULT_K_TEST) parser.add_argument("--embargo-pct", type=float, default=DEFAULT_EMBARGO_PCT) parser.add_argument("--thresholds", type=str, default=",".join(f"{t:.2f}" for t in DEFAULT_THRESHOLDS), help="Coma-separated lista de umbrales Meta a evaluar") args = parser.parse_args() LOGS_DIR.mkdir(parents=True, exist_ok=True) DASHBOARD_DATA_DIR.mkdir(parents=True, exist_ok=True) ensure_schema(DB_PATH) logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", handlers=[ logging.FileHandler(LOGS_DIR / "agente_cpcv.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_cpcv") thresholds = [float(x) for x in args.thresholds.split(",") if x.strip()] if len(thresholds) < 2: log.error("Necesito al menos 2 thresholds para PBO. Recibido: %s", thresholds) return 1 df = load_trades_with_meta(args.strategy, DB_PATH) if df.empty: log.warning("Sin trades con prob_win para %s. Ejecuta Triple Barrier + Meta antes.", args.strategy) report = CPCVReport( generated_at=datetime.now(UTC).isoformat(), strategy=args.strategy, n_total_trades=0, n_groups=args.n_groups, k_test=args.k_test, n_paths=0, embargo_pct=args.embargo_pct, thresholds=thresholds, sharpe_dist_p05=0.0, sharpe_dist_p25=0.0, sharpe_dist_p50=0.0, sharpe_dist_p75=0.0, sharpe_dist_p95=0.0, sharpe_dist_mean=0.0, sharpe_dist_std=0.0, pbo=0.0, pbo_verdict="YELLOW", pbo_note="No data", ) _write_report(report) return 0 paths, config_stats = run_cpcv( df, n_groups=args.n_groups, k_test=args.k_test, embargo_pct=args.embargo_pct, thresholds=thresholds, logger=log, ) # Distribucion de Sharpes OOS combinando todos los paths y todas las configs all_oos = [] for p in paths: all_oos.extend(p.test_sharpes_by_config.values()) arr = np.array(all_oos, dtype=float) if all_oos else np.zeros(0) if arr.size: p05, p25, p50, p75, p95 = np.percentile(arr, [5, 25, 50, 75, 95]) mean_, std_ = float(np.mean(arr)), float(np.std(arr, ddof=1)) if arr.size > 1 else 0.0 else: p05 = p25 = p50 = p75 = p95 = 0.0 mean_ = std_ = 0.0 pbo, verdict, note = compute_pbo(paths, thresholds) report = CPCVReport( generated_at=datetime.now(UTC).isoformat(), strategy=args.strategy, n_total_trades=len(df), n_groups=args.n_groups, k_test=args.k_test, n_paths=len(paths), embargo_pct=args.embargo_pct, thresholds=thresholds, sharpe_dist_p05=round(float(p05), 4), sharpe_dist_p25=round(float(p25), 4), sharpe_dist_p50=round(float(p50), 4), sharpe_dist_p75=round(float(p75), 4), sharpe_dist_p95=round(float(p95), 4), sharpe_dist_mean=round(mean_, 4), sharpe_dist_std=round(std_, 4), pbo=round(pbo, 4), pbo_verdict=verdict, pbo_note=note, config_stats=config_stats, paths=paths, ) run_id = persist_run(report, DB_PATH) _write_report(report) log.info("CPCV completo. run_id=%d | paths=%d | PBO=%.2f (%s) | Sharpe p50=%.2f", run_id, len(paths), pbo, verdict, p50) # Resumen consola sep = "=" * 64 print(f"\n{sep}") print(f"Leonex -- CPCV + PBO ({args.strategy})") print(sep) print(f"Trades total : {len(df)}") print(f"Groups / k : {args.n_groups} / {args.k_test}") print(f"Paths efectivos : {len(paths)} de {math.comb(args.n_groups, args.k_test)}") print(f"Embargo : {args.embargo_pct*100:.1f}% de T") print(f"Thresholds : {thresholds}") print(f"\nSharpe OOS distribution (todos paths × configs):") print(f" p05={p05:+.2f} p25={p25:+.2f} p50={p50:+.2f} p75={p75:+.2f} p95={p95:+.2f}") print(f" mean={mean_:+.2f} std={std_:.2f}") print(f"\nPBO = {pbo*100:.0f}% → {verdict}") print(f" {note}") if config_stats: print(f"\n[Por configuracion]") print(f" {'thr':>6} {'train':>8} {'test':>8} {'gap':>7} {'%best_IS':>9} {'%best_OOS':>10}") for cs in config_stats: print(f" {cs.threshold:>6.2f} {cs.mean_train_sharpe:>+8.2f} " f"{cs.mean_test_sharpe:>+8.2f} {cs.overfit_gap:>+7.2f} " f"{cs.pct_paths_top_train*100:>8.0f}% {cs.pct_paths_top_test*100:>9.0f}%") print(sep) return 0 def _write_report(report: CPCVReport) -> None: payload = {k: v for k, v in asdict(report).items() if k not in ("config_stats", "paths")} payload["config_stats"] = [asdict(c) for c in report.config_stats] # Limitamos los paths exportados al JSON para no inflar el dashboard payload["paths"] = [asdict(p) for p in report.paths[:50]] REPORT_PATH.write_text( json.dumps(payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8", ) if __name__ == "__main__": raise SystemExit(main())