""" Agente Strategy Tracker de Leonex. Seguimiento PAPER FORWARD de las estrategias promovidas con --promote del Strategy Lab. NO toca Alpaca: cada ejecucion re-simula cada estrategia promovida sobre todos sus datos y registra el track record OUT-OF-SAMPLE hacia delante — solo cuentan los trades cuya ENTRADA es posterior al momento en que se empezo a seguir la estrategia (tracking_start). Es la prueba honesta de verdad: el backtest del Strategy Lab eligio estas estrategias mirando el pasado; este tracker mide si funcionan sobre datos NUEVOS que no se usaron para elegirlas. Si una estrategia aguanta aqui, su edge es real; si se cae, era overfitting que se filtro. Salidas: - Tabla SQLite strategy_tracker (ticker, strategy, tracking_start). - JSON dashboard/data/strategy_tracker_report.json. Uso: python agents/agente_strategy_tracker.py """ from __future__ import annotations import argparse import json import logging import sqlite3 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 pandas as pd # Reutilizamos toda la maquinaria del Strategy Lab (mismas senales/salidas). sys.path.insert(0, str(Path(__file__).resolve().parent)) from agente_strategy_lab import ( # noqa: E402 DB_PATH, DASHBOARD_DATA_DIR, LOGS_DIR, DEFAULT_INITIAL_CAPITAL, EXIT_CONFIGS, build_triggers, build_filters, load_prices, _atr, simulate_tb_trades, ) REPORT_PATH = DASHBOARD_DATA_DIR / "strategy_tracker_report.json" MIN_BARS = 60 # ───────────────────────────────────────────────────────────────────────────── # Schema # ───────────────────────────────────────────────────────────────────────────── def ensure_schema(db_path: Path = DB_PATH) -> None: with sqlite3.connect(db_path) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS strategy_tracker ( ticker TEXT NOT NULL, strategy TEXT NOT NULL, timeframe TEXT, tracking_start TEXT, created_at TEXT, PRIMARY KEY (ticker, strategy) ) """ ) conn.commit() def load_promoted(db_path: Path = DB_PATH) -> list[dict]: """Lee las estrategias promovidas de asset_strategies.""" with sqlite3.connect(db_path) as conn: try: cols = [r[1] for r in conn.execute("PRAGMA table_info(asset_strategies)")] has_tf = "timeframe" in cols sql = ("SELECT ticker, strategy, " + ("timeframe" if has_tf else "'1d'") + " FROM asset_strategies") rows = conn.execute(sql).fetchall() except Exception: return [] return [{"ticker": r[0], "strategy": r[1], "timeframe": r[2] or "1d"} for r in rows] def get_or_create_tracking_start(conn, ticker: str, strategy: str, timeframe: str, last_data_date: str) -> str: row = conn.execute( "SELECT tracking_start FROM strategy_tracker " "WHERE ticker=? AND strategy=?", (ticker, strategy), ).fetchone() if row and row[0]: return row[0] conn.execute( "INSERT OR REPLACE INTO strategy_tracker " "(ticker, strategy, timeframe, tracking_start, created_at) " "VALUES (?, ?, ?, ?, ?)", (ticker, strategy, timeframe, last_data_date, datetime.now(UTC).isoformat()), ) conn.commit() return last_data_date # ───────────────────────────────────────────────────────────────────────────── # Seguimiento de una estrategia # ───────────────────────────────────────────────────────────────────────────── @dataclass class TrackedStrategy: ticker: str strategy: str timeframe: str tracking_start: str = "" n_bars: int = 0 backtest_n_trades: int = 0 backtest_win_rate: float = 0.0 fwd_n_trades: int = 0 fwd_win_rate: float = 0.0 fwd_total_return_pct: float = 0.0 fwd_final_capital: float = 0.0 fwd_avg_trade_pct: float = 0.0 fwd_best_trade_pct: float = 0.0 fwd_worst_trade_pct: float = 0.0 in_position: bool = False last_entry_date: str = "" fwd_trades: list = field(default_factory=list) note: str = "" def track_strategy(ticker: str, strategy: str, timeframe: str, conn, db_path: Path, initial_capital: float = DEFAULT_INITIAL_CAPITAL ) -> TrackedStrategy: parts = strategy.split("|") if len(parts) != 3: return TrackedStrategy(ticker, strategy, timeframe, note="strategy name is not trigger|filter|exit") trg_name, flt_name, exc_name = parts df = load_prices(ticker, timeframe, db_path) if df.empty or len(df) < MIN_BARS: return TrackedStrategy(ticker, strategy, timeframe, n_bars=int(len(df)), note=f"insufficient_data ({len(df)} bars)") 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 TrackedStrategy(ticker, strategy, timeframe, n_bars=int(len(df)), note="insufficient_data (post-clean)") dates = df.index last_data_date = str(dates[-1]) tracking_start = get_or_create_tracking_start( conn, ticker, strategy, timeframe, last_data_date) triggers = build_triggers(df) filters = build_filters(df) if (trg_name not in triggers or flt_name not in filters or exc_name not in EXIT_CONFIGS): return TrackedStrategy(ticker, strategy, timeframe, tracking_start=tracking_start, n_bars=int(len(df)), note="unknown trigger/filter/exit in strategy") cfg = EXIT_CONFIGS[exc_name] close_a = df["close"].to_numpy(dtype=float) high_a = df["high"].to_numpy(dtype=float) low_a = df["low"].to_numpy(dtype=float) atr_a = _atr(df).to_numpy(dtype=float) trades = simulate_tb_trades( close_a, high_a, low_a, atr_a, triggers[trg_name].to_numpy(dtype=bool), filters[flt_name].to_numpy(dtype=bool), cfg["tp"], cfg["sl"], cfg["timeout"], ) n = len(df) bt_wins = sum(1 for t in trades if t["return"] > 0) bt_n = len(trades) # Trades FORWARD: los que ENTRAN despues del inicio del seguimiento. fwd: list[dict] = [] for t in trades: entry_date = str(dates[t["entry_i"]]) if entry_date > tracking_start: fwd.append({ "entry_date": entry_date[:16], "exit_date": str(dates[t["exit_i"]])[:16], "return_pct": round(t["return"] * 100, 3), "reason": t["reason"], "bars_held": t["bars_held"], "win": bool(t["return"] > 0), }) # Posicion abierta ahora mismo: el ultimo trade cerro en la ultima barra # por timeout sin que el timeout se cumpliera (lo corto el fin de datos). in_position = False last_entry = "" if trades: last = trades[-1] if (last["exit_i"] == n - 1 and last["reason"] == "timeout" and last["bars_held"] < cfg["timeout"]): in_position = True last_entry = str(dates[last["entry_i"]])[:16] # Capital forward compuesto capital = float(initial_capital) rets = [t["return_pct"] for t in fwd] for r in rets: capital *= (1.0 + r / 100.0) fwd_wins = sum(1 for t in fwd if t["win"]) res = TrackedStrategy( ticker=ticker, strategy=strategy, timeframe=timeframe, tracking_start=tracking_start[:16], n_bars=n, backtest_n_trades=bt_n, backtest_win_rate=round(bt_wins / bt_n, 4) if bt_n else 0.0, fwd_n_trades=len(fwd), fwd_win_rate=round(fwd_wins / len(fwd), 4) if fwd else 0.0, fwd_total_return_pct=round((capital / initial_capital - 1.0) * 100, 2), fwd_final_capital=round(capital, 2), fwd_avg_trade_pct=round(sum(rets) / len(rets), 3) if rets else 0.0, fwd_best_trade_pct=round(max(rets), 3) if rets else 0.0, fwd_worst_trade_pct=round(min(rets), 3) if rets else 0.0, in_position=in_position, last_entry_date=last_entry, fwd_trades=fwd[-25:], ) if res.fwd_n_trades == 0: res.note = ("Tracking started. No forward signal yet — the forward " "paper record begins from here.") else: res.note = (f"{res.fwd_n_trades} forward trades since tracking " f"started. This is genuine out-of-sample performance.") return res # ───────────────────────────────────────────────────────────────────────────── # Main # ───────────────────────────────────────────────────────────────────────────── def main() -> int: parser = argparse.ArgumentParser(description="Agente Strategy Tracker") parser.add_argument("--initial-capital", type=float, default=DEFAULT_INITIAL_CAPITAL) parser.parse_args() LOGS_DIR.mkdir(parents=True, exist_ok=True) 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(LOGS_DIR / "agente_strategy_tracker.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_strategy_tracker") ensure_schema(DB_PATH) promoted = load_promoted(DB_PATH) log.info("Estrategias promovidas a seguir: %d", len(promoted)) results: list[TrackedStrategy] = [] with sqlite3.connect(DB_PATH) as conn: for p in promoted: try: res = track_strategy(p["ticker"], p["strategy"], p["timeframe"], conn, DB_PATH) results.append(res) log.info(" %s @%s [%s] fwd_trades=%d fwd_return=%.1f%% " "in_position=%s", res.ticker, res.timeframe, res.strategy, res.fwd_n_trades, res.fwd_total_return_pct, res.in_position) except Exception as exc: log.warning("Fallo %s: %s", p.get("ticker"), exc) n_active = sum(1 for r in results if r.fwd_n_trades > 0) total_fwd = sum(r.fwd_n_trades for r in results) if not promoted: summary = ("No promoted strategies yet. Use the Strategy Lab and run " "--promote TICKER:strategy to start a forward paper test.") else: summary = (f"Forward paper test of {len(results)} promoted strategies. " f"{n_active} have produced forward signals " f"({total_fwd} forward trades total). This is true " f"out-of-sample performance — data not used to select them.") log.info(summary) payload = { "generated_at": datetime.now(UTC).isoformat(), "n_strategies": len(results), "n_active": n_active, "total_forward_trades": total_fwd, "initial_capital": DEFAULT_INITIAL_CAPITAL, "summary_note": summary, "strategies": [asdict(r) 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 = "=" * 86 print(f"\n{sep}") print("Leonex -- Strategy Tracker (forward paper test de estrategias " "promovidas)") print(sep) print(summary) print() if results: print(f"{'ticker':<9}{'tf':<5}{'strategy':<38}{'fwd trades':>11}" f"{'fwd return':>12}{'pos':>6}") for r in results: print(f"{r.ticker:<9}{r.timeframe:<5}{r.strategy[:37]:<38}" f"{r.fwd_n_trades:>11}{r.fwd_total_return_pct:>+11.1f}%" f"{('YES' if r.in_position else '-'):>6}") print(sep) return 0 if __name__ == "__main__": raise SystemExit(main())