""" Agente Strategy Lab Scalping de Leonex. El mismo lab v3 (trigger + filtro + salida, scoring por TRADES, DSR anti-overfitting, walk-forward out-of-sample) pero sobre timeframes de SCALPING — 30m, 15m, 5m — y CON MODELO DE COSTES DE TRANSACCION. POR QUE EL MODELO DE COSTES ES OBLIGATORIO: A 5min una estrategia hace cientos de trades, cada uno con un edge diminuto. El spread + slippage se comen ese edge. Sin descontar costes el backtest es FICCION: estrategias "ganadoras" que en real pierden. Aqui cada trade paga un coste round-trip ANTES de calcular Sharpe, DSR y tiers. Esa es la diferencia entre un lab de scalping honesto y humo. ADVERTENCIA HONESTA: yfinance solo da 60 dias de datos a 30/15/5min. 60 dias = UN regimen de mercado. El DSR tendra muchos trades (bien) pero poca cobertura temporal (mal). Trata los resultados como PROVISIONALES hasta que el historico acumulado (agente_datos_intraday lo engorda cada dia) cubra varios meses. El forward tracker es quien valida de verdad. DISENO: reutiliza toda la maquinaria de agente_strategy_lab via monkeypatch: - EXIT_CONFIGS -> configs de salida ajustadas a scalping. - TRANSACTION_COST_PCT -> fijado al coste round-trip por trade. Asi no se duplica una sola linea de logica de senales/DSR/tiers. Uso: python agents/agente_strategy_lab_scalping.py python agents/agente_strategy_lab_scalping.py --cost-pct 0.08 python agents/agente_strategy_lab_scalping.py --limit-tickers 10 """ from __future__ import annotations import argparse import json import logging import sqlite3 import sys from dataclasses import asdict from datetime import datetime try: from datetime import UTC except ImportError: from datetime import timezone UTC = timezone.utc from pathlib import Path import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent)) import agente_strategy_lab as lab # noqa: E402 REPORT_PATH = lab.DASHBOARD_DATA_DIR / "strategy_lab_scalping_report.json" SCALPING_TIMEFRAMES = ("30m", "15m", "5m") # Salidas Triple Barrier ajustadas a scalping: TP/SL cortos en multiplos de # ATR (el ATR ya es pequeno en barras de 5/15/30min) y timeouts breves. SCALPING_EXIT_CONFIGS = { "scalp_tight": {"tp": 1.0, "sl": 0.7, "timeout": 6}, "scalp_quick": {"tp": 1.5, "sl": 1.0, "timeout": 12}, "scalp_wide": {"tp": 2.5, "sl": 1.5, "timeout": 24}, } # Coste round-trip por trade, en %. IBKR Pro en acciones liquidas. Un punto # por encima del lab swing (0.03): el scalping entra/sale con ordenes agresivas # que cruzan el spread completo, no media horquilla como un swing paciente. DEFAULT_COST_PCT = 0.04 def available_scalping_timeframes(db_path: Path) -> list[str]: """Devuelve los timeframes de scalping con datos en prices_intraday.""" tfs: list[str] = [] try: with sqlite3.connect(db_path) as conn: if lab._table_exists(conn, "prices_intraday"): rows = {r[0] for r in conn.execute( "SELECT DISTINCT timeframe FROM prices_intraday")} for tf in SCALPING_TIMEFRAMES: if tf in rows: tfs.append(tf) except Exception: pass return tfs def main() -> int: parser = argparse.ArgumentParser( description="Agente Strategy Lab Scalping de Leonex") parser.add_argument("--limit-tickers", type=int, default=0) parser.add_argument("--top-k", type=int, default=lab.DEFAULT_TOP_K) parser.add_argument("--cost-pct", type=float, default=DEFAULT_COST_PCT, help="Coste round-trip por trade en %% (IBKR; " "default 0.04).") 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_strategy_lab_scalping.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_strategy_lab_scalping") # ── Monkeypatch 1: salidas de scalping ──────────────────────────────── lab.EXIT_CONFIGS = SCALPING_EXIT_CONFIGS # ── Monkeypatch 2: coste de transaccion (IBKR) por trade ────────────── # simulate_tb_trades ya descuenta lab.TRANSACTION_COST_PCT de cada trade; # basta con fijar ese valor al coste de scalping pedido por CLI. Asi el # coste se descuenta UNA sola vez y no se duplica. lab.TRANSACTION_COST_PCT = max(args.cost_pct, 0.0) # Catalogo con las salidas de scalping (build_catalog itera EXIT_CONFIGS) dummy = pd.DataFrame({ "open": [1.0], "high": [1.0], "low": [1.0], "close": [1.0], "volume": [1.0], }) trigger_names = list(lab.build_triggers(dummy).keys()) filter_names = list(lab.build_filters(dummy).keys()) catalog = lab.build_catalog(trigger_names, filter_names) n_total = len(catalog) timeframes = available_scalping_timeframes(lab.DB_PATH) if not timeframes: log.warning("No hay datos de scalping en prices_intraday. Ejecuta " "agente_datos_intraday.py --timeframes 30m,15m,5m primero.") universe = lab.load_universe(lab.DB_PATH) if args.limit_tickers and len(universe) > args.limit_tickers: universe = universe[: args.limit_tickers] log.info("Strategy Lab SCALPING: %d tickers x %d timeframes %s | %d " "strategies/asset | coste=%.3f%%/trade", len(universe), len(timeframes), timeframes, n_total, args.cost_pct) assets: list = [] total = len(universe) * max(len(timeframes), 1) done = 0 for tf in timeframes: for ticker in universe: done += 1 try: rep = lab.evaluate_asset(ticker, catalog, lab.DB_PATH, top_k=args.top_k, timeframe=tf) assets.append(rep) log.info(" %5d/%d [%s] %s @%s -> %s (Sharpe=%.2f DSR=%.3f " "trades=%d)", done, total, rep.best_tier, ticker, tf, rep.best_strategy, rep.best_sharpe, rep.best_dsr, rep.best_n_trades) except Exception as exc: log.warning("Fallo %s @%s: %s", ticker, tf, exc) n_gold = sum(1 for a in assets if a.best_tier == lab.TIER_GOLD) n_silver = sum(1 for a in assets if a.best_tier == lab.TIER_SILVER) n_bronze = sum(1 for a in assets if a.best_tier == lab.TIER_BRONZE) n_rejected = sum(1 for a in assets if a.best_tier == lab.TIER_REJECTED) n_robust = sum(1 for a in assets if a.best_is_robust) summary = (f"SCALPING lab: {len(universe)} tickers x {len(timeframes)} " f"timeframes ({', '.join(timeframes) or '-'}) x {n_total} " f"strategies. Trade returns are NET of a {args.cost_pct:.3f}% " f"round-trip cost (spread+slippage) — without this a 5min " f"backtest is fiction. Tiers: {n_gold} GOLD, {n_silver} SILVER, " f"{n_bronze} BRONZE, {n_rejected} REJECTED. NOTE: only ~60 days " f"of intraday data — one market regime; treat as provisional, " f"the forward tracker validates over time.") log.info(summary) payload = { "generated_at": datetime.now(UTC).isoformat(), "version": 3, "scoring": "trade_based_scalping", "mode": "scalping", "timeframes": timeframes, "transaction_cost_pct": args.cost_pct, "n_assets": len(assets), "n_strategies_in_catalog": n_total, "n_strategies_tested": n_total, "n_triggers": len(trigger_names), "n_filters": len(filter_names), "n_exit_configs": len(SCALPING_EXIT_CONFIGS), "triggers": trigger_names, "filters": filter_names, "exit_configs": SCALPING_EXIT_CONFIGS, "catalog": trigger_names, "dsr_robust_threshold": lab.DSR_ROBUST_THRESHOLD, "initial_capital": lab.DEFAULT_INITIAL_CAPITAL, "n_robust": n_robust, "n_gold": n_gold, "n_silver": n_silver, "n_bronze": n_bronze, "n_rejected": n_rejected, "tier_thresholds": lab.TIER_THRESHOLDS, "summary_note": summary, "assets": [ { **{k: v for k, v in asdict(a).items() if k != "results"}, "results": [asdict(r) for r in a.results], } for a in assets if a.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 = "=" * 80 print(f"\n{sep}") print("Leonex -- Strategy Lab SCALPING (30m/15m/5m, trade-based, " "con coste de transaccion)") print(sep) print(summary) print(f"\n{n_gold} GOLD / {n_silver} SILVER / {n_bronze} BRONZE / " f"{n_rejected} REJECTED") print(sep) return 0 if __name__ == "__main__": raise SystemExit(main())