""" Agente Strategy Lab de Leonex — v3 (evaluacion basada en TRADES, multi-timeframe). POR QUE v3 — el fallo de v2: v2 puntuaba una estrategia por sus retornos DIARIOS sobre todas las barras. Consecuencia: una estrategia que compraba un activo y lo mantenia a traves de UNA sola tendencia gigante mostraba un Sharpe y un DSR preciosos... con 1 o 2 trades. Eso no es una estrategia: es una apuesta unica que salio bien. LA SOLUCION v3 — cada estrategia se juzga por sus TRADES: Cada senal de entrada abre un trade DISCRETO que se cierra con take-profit, stop-loss o timeout (metodo Triple Barrier). Sharpe, DSR, win rate y profit factor se calculan sobre los retornos de los TRADES. El tamano muestral del DSR = numero de trades. Gate: sin 25 trades no hay SILVER, sin 40 no hay GOLD. Las ilusiones de "comprar y aguantar" quedan fuera solas. MULTI-TIMEFRAME: El lab evalua cada estrategia en 1d, 4h y 1h (estos ultimos desde la tabla prices_intraday). Los timeframes intradia generan MUCHOS mas trades que el diario. Cada estrategia del reporte lleva su periodicidad. ESTRATEGIA v3 = TRIGGER de entrada + FILTRO de contexto + CONFIG de salida. 24 triggers x 5 filtros x 4 configs de salida = 480 estrategias por activo. Los 6 ultimos triggers son de CONFLUENCIA: exigen que varios indicadores esten de acuerdo a la vez (la forma disciplinada de combinar indicadores). Uso: python agents/agente_strategy_lab.py python agents/agente_strategy_lab.py --limit-tickers 10 python agents/agente_strategy_lab.py --promote AAPL:rsi_cross_30|uptrend|balanced REFERENCIAS BIBLIOGRAFICAS - Kaufman, P. (2020). Trading Systems and Methods, 6th ed. Wiley. Catalogo de triggers (RSI cross, MACD cross, Bollinger touch, etc.) y filtros (ADX, SMA, regime classification) que pueblan los 24 triggers x 5 filtros de v3. - Chan, E. (2013). Algorithmic Trading: Winning Strategies and Their Rationale. Wiley. Capitulo 4 (mean reversion en SIDEWAYS) y capitulo 5 (momentum en UPTREND) — la logica regime-aware del lab proviene de aqui. - Lopez de Prado, M. (2018). Advances in Financial Machine Learning. Wiley. Capitulo 3 (Triple Barrier — TP/SL/timeout que cierran cada trade discreto) y capitulo 11 (DSR como gate de calidad). - Aronson, D. (2007). Evidence-Based Technical Analysis. Wiley. Por que el gate "min 25 trades para SILVER, 40 para GOLD" no es arbitrario — viene de su capitulo 8 sobre tamano muestral necesario para edge estadisticamente significativo. - Bandy, H. (2011). Quantitative Trading Systems. Walk-forward analysis cap. 9-10 — base de la validacion OOS del Strategy Lab (consistency entre folds). - Carver, R. (2015). Systematic Trading. Harriman House. La salida balanced/ quick/wide/trend_ride se inspira en la "diversificacion por horizonte" del cap. 14 de Carver — un mismo trigger con varias salidas explota distintos modos del mercado. """ from __future__ import annotations import argparse import json import logging import math import sqlite3 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 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 / "strategy_lab_report.json" DEFAULT_MIN_BARS = 400 # descarta activos/timeframes demasiado jovenes DSR_ROBUST_THRESHOLD = 0.95 EULER_MASCHERONI = 0.5772156649 DEFAULT_INITIAL_CAPITAL = 10000.0 DEFAULT_TOP_K = 5 # estrategias guardadas por activo-timeframe MIN_TRADES_FOR_TRIAL = 5 # con menos no cuenta como "trial" del DSR ATR_PERIOD = 14 # ───────────────────────────────────────────────────────────────────────────── # Tiers — con GATE DE NUMERO DE TRADES. # ───────────────────────────────────────────────────────────────────────────── TIER_GOLD = "GOLD" TIER_SILVER = "SILVER" TIER_BRONZE = "BRONZE" TIER_REJECTED = "REJECTED" TIER_ORDER = [TIER_GOLD, TIER_SILVER, TIER_BRONZE, TIER_REJECTED] TIER_RANK = {t: i for i, t in enumerate(TIER_ORDER)} TIER_THRESHOLDS = { TIER_GOLD: {"dsr": 0.95, "min_trades": 40, "profit_factor": 1.40}, TIER_SILVER: {"dsr": 0.80, "min_trades": 25, "profit_factor": 1.20}, TIER_BRONZE: {"dsr": 0.00, "min_trades": 12, "profit_factor": 1.05}, } def demote_tier(tier: str) -> str: return TIER_ORDER[min(TIER_RANK.get(tier, 3) + 1, 3)] # ───────────────────────────────────────────────────────────────────────────── # Indicadores # ───────────────────────────────────────────────────────────────────────────── def _sma(s: pd.Series, period: int) -> pd.Series: return s.rolling(period, min_periods=period).mean() def _ema(s: pd.Series, span: int) -> pd.Series: return s.ewm(span=span, adjust=False).mean() def _rsi(close: pd.Series, period: int = 14) -> pd.Series: delta = close.diff() gain = delta.clip(lower=0.0) loss = -delta.clip(upper=0.0) avg_gain = gain.ewm(alpha=1.0 / period, adjust=False).mean() avg_loss = loss.ewm(alpha=1.0 / period, adjust=False).mean() rs = avg_gain / avg_loss.replace(0.0, np.nan) return (100.0 - (100.0 / (1.0 + rs))).fillna(50.0) def _atr(df: pd.DataFrame, period: int = ATR_PERIOD) -> pd.Series: high, low, close = df["high"], df["low"], df["close"] prev = close.shift(1) tr = pd.concat([high - low, (high - prev).abs(), (low - prev).abs()], axis=1).max(axis=1) return tr.ewm(alpha=1.0 / period, adjust=False).mean() def _vwap_rolling(df: pd.DataFrame, period: int = 20) -> pd.Series: typical = (df["high"] + df["low"] + df["close"]) / 3.0 vol = df["volume"].replace(0.0, np.nan) pv = (typical * vol).rolling(period, min_periods=period).sum() vv = vol.rolling(period, min_periods=period).sum() return pv / vv.replace(0.0, np.nan) def _macd(close: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9): macd_line = _ema(close, fast) - _ema(close, slow) return macd_line, _ema(macd_line, signal) def _bollinger(close: pd.Series, period: int = 20, n_std: float = 2.0): ma = close.rolling(period, min_periods=period).mean() sd = close.rolling(period, min_periods=period).std() return ma - n_std * sd, ma, ma + n_std * sd def _adx(df: pd.DataFrame, period: int = 14): high, low = df["high"], df["low"] up = high.diff() down = -low.diff() plus_dm = up.where((up > down) & (up > 0.0), 0.0) minus_dm = down.where((down > up) & (down > 0.0), 0.0) atr = _atr(df, period).replace(0.0, np.nan) plus_di = 100.0 * plus_dm.ewm(alpha=1.0 / period, adjust=False).mean() / atr minus_di = 100.0 * minus_dm.ewm(alpha=1.0 / period, adjust=False).mean() / atr dx = 100.0 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0.0, np.nan) return dx.ewm(alpha=1.0 / period, adjust=False).mean().fillna(0.0) def _cross_up(a: pd.Series, b) -> pd.Series: """Evento: la serie a cruza por ENCIMA de b (b serie o escalar).""" if np.isscalar(b): return ((a > b) & (a.shift(1) <= b)).fillna(False) return ((a > b) & (a.shift(1) <= b.shift(1))).fillna(False) def _cross_down(a: pd.Series, b) -> pd.Series: if np.isscalar(b): return ((a < b) & (a.shift(1) >= b)).fillna(False) return ((a < b) & (a.shift(1) >= b.shift(1))).fillna(False) # ───────────────────────────────────────────────────────────────────────────── # Velas japonesas (eventos) # ───────────────────────────────────────────────────────────────────────────── def _candle_parts(df: pd.DataFrame): o, h, l, c = df["open"], df["high"], df["low"], df["close"] body = (c - o).abs() rng = (h - l) upper = h - np.maximum(o, c) lower = np.minimum(o, c) - l return o, h, l, c, body, rng, upper, lower def ev_hammer(df: pd.DataFrame) -> pd.Series: o, h, l, c, body, rng, upper, lower = _candle_parts(df) is_hammer = ((body > 0) & (rng > 0) & (lower >= 2.0 * body) & (lower >= 0.5 * rng) & (upper <= 0.6 * body)) downtrend = c < _sma(c, 10) return (is_hammer & downtrend).fillna(False) def ev_bullish_engulfing(df: pd.DataFrame) -> pd.Series: o, c = df["open"], df["close"] o1, c1 = o.shift(1), c.shift(1) return ((c1 < o1) & (c > o) & (o <= c1) & (c >= o1)).fillna(False) def ev_piercing_line(df: pd.DataFrame) -> pd.Series: o, c = df["open"], df["close"] o1, c1 = o.shift(1), c.shift(1) mid1 = (o1 + c1) / 2.0 return ((c1 < o1) & (c > o) & (o < c1) & (c > mid1) & (c < o1)).fillna(False) def ev_morning_star(df: pd.DataFrame) -> pd.Series: o, c = df["open"], df["close"] o1, c1 = o.shift(1), c.shift(1) o2, c2 = o.shift(2), c.shift(2) body0 = (c - o).abs() body1 = (c1 - o1).abs() body2 = (c2 - o2).abs() cond = ((c2 < o2) & (body2 > 0) & (body1 <= 0.5 * body2) & (c > o) & (c > (o2 + c2) / 2.0) & (body0 > 0) & (c2 < _sma(c, 10).shift(2))) return cond.fillna(False) # ───────────────────────────────────────────────────────────────────────────── # TRIGGERS de entrada — serie de EVENTOS booleana (True en la barra de entrada) # ───────────────────────────────────────────────────────────────────────────── def build_triggers(df: pd.DataFrame) -> dict[str, pd.Series]: """Eventos de ENTRADA (long). Cada trigger es un evento discreto y puntual. El REGIMEN en el que conviene operarlo NO va aqui: lo aporta el filtro (build_filters), y el lab cruza cada trigger con cada regimen para descubrir que combinacion tiene edge real. Dos familias: - MOMENTUM / TENDENCIA: rupturas y cruces que siguen la inercia. - REVERSION A LA MEDIA: caidas cortas que tienden a rebotar. """ close = df["close"] high = df["high"] trig: dict[str, pd.Series] = {} ema20, ema50 = _ema(close, 20), _ema(close, 50) rsi = _rsi(close, 14) atr = _atr(df) macd_line, macd_sig = _macd(close) vol = pd.to_numeric(df["volume"], errors="coerce").fillna(0.0) # ── Familia MOMENTUM / TENDENCIA ────────────────────────────────────── trig["macd_cross_up"] = _cross_up(macd_line, macd_sig) trig["ma_golden_20_100"] = _cross_up(_sma(close, 20), _sma(close, 100)) # Recupera la EMA20 desde abajo con EMA20 ya por encima de EMA50: es un # pullback comprado dentro de una estructura alcista, no un rebote ciego. trig["ema20_reclaim_aligned"] = (_cross_up(close, ema20) & (ema20 > ema50)).fillna(False) for p in (20, 55): chan = high.rolling(p, min_periods=p).max().shift(1) trig[f"donchian_break_{p}"] = _cross_up(close, chan) # Ruptura del maximo de 20 barras CONFIRMADA por volumen (> 1.3x su media): # descarta las rupturas flojas sin participacion detras. vol_avg = vol.rolling(20, min_periods=10).mean() chan20 = high.rolling(20, min_periods=20).max().shift(1) trig["breakout_vol_confirmed"] = (_cross_up(close, chan20) & (vol > 1.3 * vol_avg)).fillna(False) mom50 = close / close.shift(50) - 1.0 trig["momentum_pos_50"] = _cross_up(mom50, 0.0) cond = close > (close.shift(1) + 1.5 * atr) trig["atr_breakout"] = (cond & (~cond.shift(1).fillna(False))).fillna(False) trig["vwap_reclaim"] = _cross_up(close, _vwap_rolling(df, 20)) # ── Familia REVERSION A LA MEDIA ────────────────────────────────────── for th in (30, 25, 20): trig[f"rsi_cross_{th}"] = _cross_down(rsi, float(th)) lower, _, _ = _bollinger(close) trig["bb_lower_touch"] = _cross_down(close, lower) # Racha bajista de 3 barras seguidas: dip corto que suele rebotar. down = (close < close.shift(1)).fillna(False) trig["down_streak_3"] = (down & down.shift(1).fillna(False) & down.shift(2).fillna(False) & ~down.shift(3).fillna(False)).fillna(False) # ── Familia VELAS de reversion ──────────────────────────────────────── trig["hammer"] = ev_hammer(df) trig["morning_star"] = ev_morning_star(df) trig["bullish_engulfing"] = ev_bullish_engulfing(df) trig["piercing_line"] = ev_piercing_line(df) # ── Familia CONFLUENCIA (multi-indicador) ───────────────────────────── # Cada trigger de aqui exige que VARIOS indicadores esten de acuerdo a la # vez. Disparan menos veces, pero cada entrada llega mas confirmada: es la # forma DISCIPLINADA de "combinar mas indicadores" — confluencia, no un # grid a lo bruto. El DSR sigue siendo el juez: si una confluencia es # casualidad, la castiga igual que a cualquier otra. sma200 = _sma(close, 200) # Momentum confirmado por TRES frentes: el MACD gira al alza, el RSI sube # pero sin sobrecompra (50-72) y el precio esta sobre su EMA50. trig["macd_rsi_trend_sync"] = ( _cross_up(macd_line, macd_sig) & (rsi > 50.0) & (rsi < 72.0) & (close > ema50)).fillna(False) # Ruptura de canal con DOS confirmaciones distintas: volumen fuerte # (>1.3x su media) y momentum sano (RSI>55) — descarta rupturas flojas y # las que ocurren sin empuje detras. trig["breakout_volume_trend"] = ( _cross_up(close, chan20) & (vol > 1.3 * vol_avg) & (rsi > 55.0)).fillna(False) # Pullback comprado dentro de una estructura alcista (EMA20>EMA50) que # ademas habia corregido (RSI bajo 45 en las ultimas 5 barras) y ahora # reconquista la EMA20 desde abajo. pulled_back = (rsi.rolling(5, min_periods=1).min() < 45.0) trig["golden_pullback_thrust"] = ( (ema20 > ema50) & pulled_back & _cross_up(close, ema20)).fillna(False) # Reversion confirmada: el precio toco la banda inferior de Bollinger en # las ultimas 3 barras, el RSI esta en sobreventa (<35) y hoy ya cierra # al alza — el rebote ha empezado, no es un cuchillo cayendo. touched_bb = (close < lower).astype(float).rolling( 3, min_periods=1).max() > 0.0 trig["bb_rsi_reversal"] = ( touched_bb & (rsi < 35.0) & (close > close.shift(1))).fillna(False) # Sobreventa con suelo: RSI<40, una vela alcista de reversion (martillo, # envolvente o piercing) y el precio AUN sobre su SMA200 — un dip dentro # de una tendencia de fondo alcista, no una caida libre. bull_candle = (ev_hammer(df) | ev_bullish_engulfing(df) | ev_piercing_line(df)) trig["oversold_candle_support"] = ( (rsi < 40.0) & bull_candle & (close > sma200)).fillna(False) # Capitulacion y reconquista: 3 barras bajistas seguidas y, a la siguiente, # el precio rebota con fuerza (>0.5 ATR) y con volumen (>1.3x su media). streak3 = (down & down.shift(1).fillna(False) & down.shift(2).fillna(False)).fillna(False) reclaim = close > (close.shift(1) + 0.5 * atr) trig["capitulation_reclaim"] = ( streak3.shift(1).fillna(False) & reclaim & (vol > 1.3 * vol_avg)).fillna(False) return {k: v.fillna(False).astype(bool) for k, v in trig.items()} # ───────────────────────────────────────────────────────────────────────────── # FILTROS de REGIMEN — mascara booleana (entrada permitida solo en esa barra) # ───────────────────────────────────────────────────────────────────────────── def build_filters(df: pd.DataFrame) -> dict[str, pd.Series]: """Filtros de REGIMEN. Cada uno define un contexto de mercado distinto y la entrada de un trigger solo se permite donde el regimen es favorable. Cruzar trigger x regimen es lo que convierte un indicador suelto en una estrategia con sentido economico: un breakout en tendencia no es lo mismo que un breakout en un lateral (ahi suele ser una trampa).""" close = df["close"] idx = df.index sma200 = _sma(close, 200) ema20 = _ema(close, 20) rsi = _rsi(close, 14) adx = _adx(df) atr = _atr(df) atr_med = atr.rolling(50, min_periods=20).median() filt: dict[str, pd.Series] = {} # Baseline sin condicionar: deja ver la version cruda del trigger. filt["none"] = pd.Series(True, index=idx) # Tendencia alcista fuerte: precio sobre la SMA200, SMA200 subiendo y # direccionalidad alta (ADX>20). El terreno de los triggers de momentum. filt["trend_up"] = ((close > sma200) & (sma200 > sma200.shift(20)) & (adx > 20.0)).fillna(False) # Rango / lateral: poca direccionalidad (ADX<20). El terreno natural de la # reversion a la media — ahi los rebotes funcionan y las rupturas no. filt["range_bound"] = (adx < 20.0).fillna(False) # Pullback dentro de tendencia: alcista de fondo (>SMA200) pero con una # correccion corta en curso ( sma200) & (close < ema20) & (rsi < 50.0)).fillna(False) # Expansion de volatilidad: ATR sobre su mediana y subiendo. Las rupturas # tienen mas recorrido cuando la volatilidad se esta expandiendo. filt["vol_expansion"] = ((atr > atr_med) & (atr > atr.shift(5))).fillna(False) return {k: v.astype(bool) for k, v in filt.items()} # Configuraciones de SALIDA (Triple Barrier en multiplos de ATR) EXIT_CONFIGS: dict[str, dict] = { "quick": {"tp": 1.5, "sl": 1.0, "timeout": 8}, "balanced": {"tp": 2.5, "sl": 1.5, "timeout": 15}, "wide": {"tp": 4.0, "sl": 2.0, "timeout": 30}, "trend_ride": {"tp": 6.0, "sl": 2.5, "timeout": 50}, } # ───────────────────────────────────────────────────────────────────────────── # Modelo de costes de transaccion — Interactive Brokers # # Cada trade redondo (entrada + salida) paga un coste que se DESCUENTA del # retorno antes de calcular Sharpe / DSR / tiers. Un backtest sin costes # miente: estrategias "ganadoras" en papel que en real pierden. # # TRANSACTION_COST_PCT es una estimacion para Interactive Brokers Pro sobre # acciones liquidas del S&P 500 (USD). Cubre tres componentes round-trip: # - Comision IBKR Pro Tiered: ~0.0035 USD/accion, minimo 0.35 USD/orden. # - Spread bid-ask efectivo: IBKR rutea por SMART y suele conseguir price # improvement, asi que en large caps liquidas el spread efectivo ronda # ~0.5-1.5 pb por lado. Es mas barato que Alpaca: Alpaca no cobra # comision, pero su ejecucion / spread efectivo es peor. # - Slippage / impacto de mercado para tamano moderado. # 0.03% round-trip es una cifra honesta para IBKR Pro en nombres liquidos. # Es una APROXIMACION: la comision real es por accion con minimo por orden, # asi que posiciones pequenas o de precio bajo pagan proporcionalmente mas. # # Es un valor de modulo: el lab de scalping lo sobreescribe con su propio # coste; el Custom Lab lo pone a 0 porque descuenta el coste por su cuenta. # Se puede ajustar por CLI con --cost-pct en cada lab. TRANSACTION_COST_PCT = 0.03 # ───────────────────────────────────────────────────────────────────────────── # Salida por REGIMEN — cortar en beneficio cuando el mercado gira # # Un CUARTO criterio de salida, ademas de TP / SL / timeout: si una posicion # esta EN BENEFICIO y el regimen se da la vuelta (el precio pierde su media de # tendencia), se cierra esa barra y se asegura la ganancia, en vez de esperar # al take-profit y comerse la reversion. Solo cierra operaciones GANADORAS — # nunca adelanta un stop-loss. El lab puntua las estrategias YA con esta salida. REGIME_EXIT_ENABLED = True REGIME_SMA_WINDOW = 50 # media movil que define la tendencia de fondo # ───────────────────────────────────────────────────────────────────────────── # Simulador de trades — Triple Barrier, posiciones NO solapadas # ───────────────────────────────────────────────────────────────────────────── def simulate_tb_trades(close: np.ndarray, high: np.ndarray, low: np.ndarray, atr: np.ndarray, entry: np.ndarray, filt: np.ndarray, tp_mult: float, sl_mult: float, timeout: int) -> list[dict]: """Para cada evento de entrada permitido por el filtro abre un trade y lo cierra en take-profit, stop-loss o timeout. No solapa posiciones.""" n = len(close) trades: list[dict] = [] # Media de tendencia para la salida por regimen (NaN durante el warm-up). if REGIME_EXIT_ENABLED and n >= REGIME_SMA_WINDOW: trend = (pd.Series(close).rolling( REGIME_SMA_WINDOW, min_periods=REGIME_SMA_WINDOW).mean().to_numpy()) else: trend = np.full(n, np.nan) i = 0 while i < n - 1: if (entry[i] and filt[i] and close[i] > 0 and atr[i] > 0 and np.isfinite(atr[i])): ep = close[i] tp = ep + tp_mult * atr[i] sl = ep - sl_mult * atr[i] end = min(i + timeout, n - 1) exit_i = -1 xp = 0.0 reason = "" j = i + 1 while j <= end: if low[j] <= sl: # conservador: SL antes que TP exit_i, xp, reason = j, sl, "SL" break if high[j] >= tp: exit_i, xp, reason = j, tp, "TP" break # Salida por REGIMEN: en beneficio y el precio pierde su media # de tendencia -> cerrar la barra y asegurar la ganancia. if (close[j] > ep and np.isfinite(trend[j]) and close[j] < trend[j]): exit_i, xp, reason = j, close[j], "regime" break j += 1 if exit_i < 0: exit_i, xp, reason = end, close[end], "timeout" ret = (xp - ep) / ep if ep > 0 else 0.0 ret -= TRANSACTION_COST_PCT / 100.0 # coste round-trip (IBKR) 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 # ───────────────────────────────────────────────────────────────────────────── # Resultado de una estrategia # ───────────────────────────────────────────────────────────────────────────── @dataclass class StrategyResult: strategy: str timeframe: str = "1d" trigger: str = "" filter: str = "" exit_config: str = "" n_trades: int = 0 win_rate: float = 0.0 profit_factor: float = 0.0 expectancy_pct: float = 0.0 avg_trade_pct: float = 0.0 best_trade_pct: float = 0.0 worst_trade_pct: float = 0.0 avg_bars_held: float = 0.0 max_drawdown: float = 0.0 sharpe: float = 0.0 # anualizado (para mostrar) sharpe_per_trade: float = 0.0 # por trade (base del DSR) skew: float = 0.0 kurtosis: float = 3.0 dsr: float = 0.0 robust: bool = False 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 sim_n_trades: int = 0 sim_win_rate: float = 0.0 trades: list = field(default_factory=list) def _sharpe_per_trade(rets: np.ndarray) -> float: if len(rets) < 3: return 0.0 sd = float(np.std(rets, ddof=1)) if sd <= 1e-12: return 0.0 return float(np.mean(rets) / sd) def compute_trade_metrics(trades: list[dict], dates: pd.DatetimeIndex) -> dict: """Metricas a partir de la lista de trades.""" n = len(trades) if n == 0: return {"n_trades": 0} rets = np.array([t["return"] for t in trades], dtype=float) wins = rets[rets > 0] losses = rets[rets < 0] win_rate = float(len(wins) / n) gross_win = float(wins.sum()) gross_loss = float(-losses.sum()) profit_factor = gross_win / gross_loss if gross_loss > 1e-12 else ( gross_win if gross_win > 0 else 0.0) sr_trade = _sharpe_per_trade(rets) span_days = max((dates[trades[-1]["exit_i"]] - dates[trades[0]["entry_i"]]).days, 1) years = span_days / 365.25 trades_per_year = n / years if years > 0 else float(n) sharpe_ann = sr_trade * math.sqrt(trades_per_year) if trades_per_year > 0 else 0.0 equity = np.cumprod(1.0 + rets) running_max = np.maximum.accumulate(equity) drawdown = equity / running_max - 1.0 max_dd = float(drawdown.min()) 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 return { "n_trades": n, "win_rate": win_rate, "profit_factor": profit_factor, "expectancy_pct": float(rets.mean() * 100.0), "avg_trade_pct": float(rets.mean() * 100.0), "best_trade_pct": float(rets.max() * 100.0), "worst_trade_pct": float(rets.min() * 100.0), "avg_bars_held": float(np.mean([t["bars_held"] for t in trades])), "max_drawdown": max_dd, "sharpe": sharpe_ann, "sharpe_per_trade": sr_trade, "skew": skew, "kurtosis": kurt, "trades_per_year": trades_per_year, "_rets": rets, } def walk_forward_trades(rets: np.ndarray) -> dict: """Parte los trades 70/30 en orden cronologico: Sharpe por trade IS vs OOS, y consistencia entre 4 bloques.""" n = len(rets) if n < 12: return {"is_sharpe": 0.0, "oos_sharpe": 0.0, "regime_consistency": 0.0} split = int(n * 0.70) is_sr = _sharpe_per_trade(rets[:split]) oos_sr = _sharpe_per_trade(rets[split:]) blocks = np.array_split(rets, 4) pos = sum(1 for b in blocks if len(b) >= 2 and float(np.mean(b)) > 0) return { "is_sharpe": round(is_sr, 4), "oos_sharpe": round(oos_sr, 4), "regime_consistency": round(pos / 4.0, 2), } # ───────────────────────────────────────────────────────────────────────────── # Deflated Sharpe Ratio — en espacio POR TRADE (n_observations = n_trades) # ───────────────────────────────────────────────────────────────────────────── def _norm_cdf(x: float) -> float: return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0))) def _norm_ppf(p: float) -> float: if p <= 0.0: return -8.0 if p >= 1.0: return 8.0 a = [-3.969683028665376e+01, 2.209460984245205e+02, -2.759285104469687e+02, 1.383577518672690e+02, -3.066479806614716e+01, 2.506628277459239e+00] b = [-5.447609879822406e+01, 1.615858368580409e+02, -1.556989798598866e+02, 6.680131188771972e+01, -1.328068155288572e+01] c = [-7.784894002430293e-03, -3.223964580411365e-01, -2.400758277161838e+00, -2.549732539343734e+00, 4.374664141464968e+00, 2.938163982698783e+00] d = [7.784695709041462e-03, 3.224671290700398e-01, 2.445134137142996e+00, 3.754408661907416e+00] p_low = 0.02425 p_high = 1.0 - p_low if p < p_low: q = math.sqrt(-2.0 * math.log(p)) return (((((c[0]*q+c[1])*q+c[2])*q+c[3])*q+c[4])*q+c[5]) / \ ((((d[0]*q+d[1])*q+d[2])*q+d[3])*q+1.0) if p > p_high: q = math.sqrt(-2.0 * math.log(1.0 - p)) return -(((((c[0]*q+c[1])*q+c[2])*q+c[3])*q+c[4])*q+c[5]) / \ ((((d[0]*q+d[1])*q+d[2])*q+d[3])*q+1.0) q = p - 0.5 r = q * q return (((((a[0]*r+a[1])*r+a[2])*r+a[3])*r+a[4])*r+a[5])*q / \ (((((b[0]*r+b[1])*r+b[2])*r+b[3])*r+b[4])*r+1.0) def expected_max_sharpe(sharpe_variance: float, n_trials: int) -> float: if n_trials < 2 or sharpe_variance <= 0: return 0.0 std = math.sqrt(sharpe_variance) e = math.e term1 = (1.0 - EULER_MASCHERONI) * _norm_ppf(1.0 - 1.0 / n_trials) term2 = EULER_MASCHERONI * _norm_ppf(1.0 - 1.0 / (n_trials * e)) return float(std * (term1 + term2)) def deflated_sharpe_ratio(sr_per_trade: float, all_sr: list[float], n_trades: int, skew: float, kurtosis: float) -> float: """DSR en espacio por-trade. n_trades es el tamano muestral real.""" n_trials = len(all_sr) if n_trials < 2 or n_trades < 5: return 0.0 sr_var = float(np.var(all_sr, ddof=1)) if n_trials > 1 else 0.0 sr_max = expected_max_sharpe(sr_var, n_trials) denom = (1.0 - skew * sr_per_trade + (kurtosis - 1.0) / 4.0 * sr_per_trade ** 2) if denom <= 0: denom = 1e-6 numerator = (sr_per_trade - sr_max) * math.sqrt(n_trades - 1) return float(_norm_cdf(numerator / math.sqrt(denom))) def classify_tier(dsr: float, n_trades: int, profit_factor: float, expectancy_pct: float) -> str: """Clasifica con GATE de numero de trades.""" for tier in (TIER_GOLD, TIER_SILVER, TIER_BRONZE): th = TIER_THRESHOLDS[tier] if (dsr >= th["dsr"] and n_trades >= th["min_trades"] and profit_factor >= th["profit_factor"] and expectancy_pct > 0.0): return tier return TIER_REJECTED # ───────────────────────────────────────────────────────────────────────────── # Catalogo de estrategias = trigger x filtro x salida # ───────────────────────────────────────────────────────────────────────────── def build_catalog(trigger_names: list[str], filter_names: list[str] ) -> list[tuple[str, str, str, str]]: catalog: list[tuple[str, str, str, str]] = [] for trg in trigger_names: for flt in filter_names: for exc in EXIT_CONFIGS: catalog.append((f"{trg}|{flt}|{exc}", trg, flt, exc)) return catalog # ───────────────────────────────────────────────────────────────────────────── # Runner por activo # ───────────────────────────────────────────────────────────────────────────── @dataclass class AssetStrategyReport: ticker: str n_bars: int n_strategies_tested: int best_strategy: str best_sharpe: float best_dsr: float best_is_robust: bool timeframe: str = "1d" best_tier: str = "REJECTED" best_final_capital: float = 0.0 best_sim_return_pct: float = 0.0 best_n_trades: int = 0 best_oos_sharpe: float = 0.0 n_trials_dsr: int = 0 results: list = field(default_factory=list) note: str = "" def _table_exists(conn, name: str) -> bool: return conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (name,) ).fetchone() is not None def available_timeframes(db_path: Path = DB_PATH) -> list[str]: """Timeframes con datos: '1d' siempre; '4h'/'1h' si prices_intraday existe y tiene filas para ese timeframe.""" tfs = ["1d"] try: with sqlite3.connect(db_path) as conn: if _table_exists(conn, "prices_intraday"): rows = {r[0] for r in conn.execute( "SELECT DISTINCT timeframe FROM prices_intraday")} for tf in ("4h", "1h"): if tf in rows: tfs.append(tf) except Exception: pass return tfs def load_prices(ticker: str, timeframe: str = "1d", db_path: Path = DB_PATH) -> pd.DataFrame: """Carga OHLCV. '1d' lee de prices; '4h'/'1h' leen de prices_intraday.""" with sqlite3.connect(db_path) as conn: if timeframe == "1d": df = pd.read_sql( "SELECT date AS ts, open, high, low, close, volume FROM prices " "WHERE ticker = ? ORDER BY date ASC", conn, params=(ticker,), parse_dates=["ts"], ) else: if not _table_exists(conn, "prices_intraday"): return pd.DataFrame() df = pd.read_sql( "SELECT ts, open, high, low, close, volume FROM prices_intraday " "WHERE ticker = ? AND timeframe = ? ORDER BY ts ASC", conn, params=(ticker, timeframe), parse_dates=["ts"], ) if df.empty: return df return df.set_index("ts") def evaluate_asset(ticker: str, catalog: list, db_path: Path = DB_PATH, top_k: int = DEFAULT_TOP_K, initial_capital: float = DEFAULT_INITIAL_CAPITAL, min_bars: int = DEFAULT_MIN_BARS, timeframe: str = "1d") -> AssetStrategyReport: df = load_prices(ticker, timeframe, db_path) if df.empty or len(df) < min_bars: return AssetStrategyReport( ticker=ticker, n_bars=int(len(df)), n_strategies_tested=0, best_strategy="none", best_sharpe=0.0, best_dsr=0.0, best_is_robust=False, timeframe=timeframe, note=f"insufficient_data ({len(df)} bars, min {min_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 AssetStrategyReport( ticker=ticker, n_bars=int(len(df)), n_strategies_tested=0, best_strategy="none", best_sharpe=0.0, best_dsr=0.0, best_is_robust=False, timeframe=timeframe, note="insufficient_data (post-clean)", ) 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) atr_a = _atr(df).to_numpy(dtype=float) triggers = build_triggers(df) filters = build_filters(df) trig_arr = {k: v.to_numpy(dtype=bool) for k, v in triggers.items()} filt_arr = {k: v.to_numpy(dtype=bool) for k, v in filters.items()} raw: dict[str, dict] = {} trades_map: dict[str, list] = {} for name, trg, flt, exc in catalog: cfg = EXIT_CONFIGS[exc] try: trades = simulate_tb_trades( close_a, high_a, low_a, atr_a, trig_arr[trg], filt_arr[flt], cfg["tp"], cfg["sl"], cfg["timeout"], ) except Exception: trades = [] trades_map[name] = trades raw[name] = compute_trade_metrics(trades, dates) traded_sr = [m["sharpe_per_trade"] for m in raw.values() if m.get("n_trades", 0) >= MIN_TRADES_FOR_TRIAL] if len(traded_sr) < 2: traded_sr = [m.get("sharpe_per_trade", 0.0) for m in raw.values()] n_trials = len(traded_sr) prelim: list[tuple] = [] for name, m in raw.items(): if m.get("n_trades", 0) == 0: prelim.append((name, m, 0.0, TIER_REJECTED)) continue dsr = deflated_sharpe_ratio( sr_per_trade=m["sharpe_per_trade"], all_sr=traded_sr, n_trades=m["n_trades"], skew=m["skew"], kurtosis=m["kurtosis"], ) tier = classify_tier(dsr, m["n_trades"], m["profit_factor"], m["expectancy_pct"]) prelim.append((name, m, dsr, tier)) prelim.sort(key=lambda x: x[2], reverse=True) cat_meta = {name: (trg, flt, exc) for name, trg, flt, exc in catalog} results: list[StrategyResult] = [] for name, m, dsr, tier in prelim: if tier == TIER_REJECTED: continue rets = m["_rets"] wf = walk_forward_trades(rets) final_tier = tier tier_note = "" if tier in (TIER_GOLD, TIER_SILVER): if wf["oos_sharpe"] <= 0.0: final_tier = demote_tier(tier) tier_note = "demoted: out-of-sample Sharpe <= 0" elif wf["regime_consistency"] < 0.5: final_tier = demote_tier(tier) tier_note = "demoted: inconsistent across regimes" equity = float(initial_capital) * float(np.prod(1.0 + rets)) trg, flt, exc = cat_meta[name] trade_list = trades_map[name] sample = [{ "entry_date": str(dates[t["entry_i"]])[:16], "exit_date": str(dates[t["exit_i"]])[:16], "return_pct": round(t["return"] * 100, 3), "reason": t["reason"], "win": bool(t["return"] > 0), } for t in trade_list[-15:]] results.append(StrategyResult( strategy=name, timeframe=timeframe, trigger=trg, filter=flt, exit_config=exc, n_trades=m["n_trades"], win_rate=round(m["win_rate"], 4), profit_factor=round(m["profit_factor"], 4), expectancy_pct=round(m["expectancy_pct"], 4), avg_trade_pct=round(m["avg_trade_pct"], 4), best_trade_pct=round(m["best_trade_pct"], 3), worst_trade_pct=round(m["worst_trade_pct"], 3), avg_bars_held=round(m["avg_bars_held"], 1), max_drawdown=round(m["max_drawdown"], 4), sharpe=round(m["sharpe"], 4), sharpe_per_trade=round(m["sharpe_per_trade"], 4), skew=round(m["skew"], 4), kurtosis=round(m["kurtosis"], 4), dsr=round(dsr, 4), robust=(dsr >= DSR_ROBUST_THRESHOLD and wf["oos_sharpe"] > 0), is_sharpe=wf["is_sharpe"], oos_sharpe=wf["oos_sharpe"], regime_consistency=wf["regime_consistency"], tier=final_tier, tier_note=tier_note, final_capital=round(equity, 2), sim_total_return_pct=round((equity / initial_capital - 1.0) * 100, 2), sim_n_trades=m["n_trades"], sim_win_rate=round(m["win_rate"], 4), trades=sample, )) results.sort(key=lambda r: (TIER_RANK.get(r.tier, 9), -r.dsr)) results = results[:top_k] for r in results[3:]: r.trades = [] if results: best = results[0] best_strategy, best_sharpe = best.strategy, best.sharpe best_dsr, best_tier = best.dsr, best.tier best_robust = best.robust best_final_capital = best.final_capital best_sim_return = best.sim_total_return_pct best_n_trades = best.n_trades best_oos = best.oos_sharpe else: name, m, dsr, _ = prelim[0] best_strategy = name best_sharpe = round(m.get("sharpe", 0.0), 4) best_dsr = round(dsr, 4) best_tier = TIER_REJECTED best_robust = False best_final_capital = 0.0 best_sim_return = 0.0 best_n_trades = int(m.get("n_trades", 0)) best_oos = 0.0 tier_msg = { TIER_GOLD: "GOLD — genuine edge, trades often, robust out-of-sample.", TIER_SILVER: "SILVER — good, trades often, watch before operating.", TIER_BRONZE: "BRONZE — marginal, doubtful.", TIER_REJECTED: "REJECTED — too few trades or no edge. Ignore.", } note = (f"Best by DSR over {n_trials} trials, scored on TRADE returns " f"(not bars). {tier_msg[best_tier]}") return AssetStrategyReport( ticker=ticker, n_bars=int(len(df)), n_strategies_tested=len(catalog), best_strategy=best_strategy, best_sharpe=best_sharpe, best_dsr=best_dsr, best_is_robust=best_robust, timeframe=timeframe, best_tier=best_tier, best_final_capital=best_final_capital, best_sim_return_pct=best_sim_return, best_n_trades=best_n_trades, best_oos_sharpe=best_oos, n_trials_dsr=n_trials, results=results, note=note, ) def load_universe(db_path: Path = DB_PATH) -> list[str]: """Data pool: TODOS los tickers con datos en prices (S&P 500 completo). El Strategy Lab estudia el pool entero, no solo el universo operativo (active_universe, top-N) que usa el executor.""" with sqlite3.connect(db_path) as conn: df = pd.read_sql("SELECT DISTINCT ticker FROM prices", conn) return sorted(df["ticker"].tolist()) # ───────────────────────────────────────────────────────────────────────────── # Schema + promote # ───────────────────────────────────────────────────────────────────────────── def ensure_schema(db_path: Path = DB_PATH) -> None: with sqlite3.connect(db_path) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS asset_strategies ( ticker TEXT PRIMARY KEY, strategy TEXT NOT NULL, timeframe TEXT, dsr REAL, sharpe REAL, robust INTEGER, promoted_at TEXT ) """ ) try: conn.execute("ALTER TABLE asset_strategies ADD COLUMN timeframe TEXT") except Exception: pass conn.commit() def promote_strategy(ticker: str, strategy: str, db_path: Path = DB_PATH, logger: Optional[logging.Logger] = None) -> bool: if logger is None: logger = logging.getLogger("strategy_lab") # Buscamos en TRES reportes: # swing (1d/4h/1h) → strategy_lab_report.json # scalping (30m/15m/5m) → strategy_lab_scalping_report.json # custom (PineScript) → custom_strategies_report.json (estructura distinta) # Las estrategias del Custom Lab evaluan UN nombre en TODOS los tickers y # exponen `strategies[].per_asset[].ticker/tier/psr/...`, por eso requiere # adaptador para que devuelva un `match` con la misma forma {dsr, sharpe, robust}. scalping_report_path = DASHBOARD_DATA_DIR / "strategy_lab_scalping_report.json" custom_report_path = DASHBOARD_DATA_DIR / "custom_strategies_report.json" match = None asset_tf = "1d" found_in: str = "" for report_path, label in [ (REPORT_PATH, "swing"), (scalping_report_path, "scalping"), ]: if not report_path.exists(): continue try: report = json.loads(report_path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("No se pudo leer %s: %s", report_path.name, exc) continue for a in report.get("assets", []): if a["ticker"] != ticker: continue for r in a.get("results", []): if r["strategy"] == strategy: match = r asset_tf = a.get("timeframe", "1d") found_in = label break if match: break if match: break # Si no estaba en swing/scalping, miramos el reporte del Custom Lab. # Estructura: strategies: [{name, timeframe, per_asset:[{ticker,tier,psr,...}]}] # Solo aceptamos si el per_asset[ticker].tier es GOLD o SILVER. if not match and custom_report_path.exists(): try: report = json.loads(custom_report_path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("No se pudo leer %s: %s", custom_report_path.name, exc) report = {} for s in report.get("strategies", []): if s.get("name") != strategy: continue for pa in s.get("per_asset", []): if pa.get("ticker") != ticker: continue tier_pa = pa.get("tier", "REJECTED") if tier_pa not in ("GOLD", "SILVER"): logger.warning( "Custom %s/%s en %s tiene tier %s — solo promovibles " "GOLD/SILVER.", strategy, ticker, s.get("timeframe", "?"), tier_pa, ) break # Adaptamos al shape esperado por la insercion en BD match = { "dsr": pa.get("psr", 0.0), # PSR como proxy de DSR a nivel ticker "sharpe": pa.get("sharpe_per_trade", 0.0), "robust": 1 if tier_pa == "GOLD" else 0, } asset_tf = s.get("timeframe", "1h") found_in = "custom" break if match: break if not match: logger.error( "Strategy %s not found for %s in swing OR scalping reports. " "Verifica que esa estrategia aparezca en el Strategy Lab o " "Scalping Lab tras el ultimo run.", strategy, ticker) return False logger.info("Promote: estrategia encontrada en reporte %s (tf=%s)", found_in, asset_tf) ensure_schema(db_path) with sqlite3.connect(db_path) as conn: conn.execute( """ INSERT INTO asset_strategies (ticker, strategy, timeframe, dsr, sharpe, robust, promoted_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(ticker) DO UPDATE SET strategy=excluded.strategy, timeframe=excluded.timeframe, dsr=excluded.dsr, sharpe=excluded.sharpe, robust=excluded.robust, promoted_at=excluded.promoted_at """, (ticker, strategy, asset_tf, match["dsr"], match["sharpe"], int(match["robust"]), datetime.now(UTC).isoformat()), ) conn.commit() logger.info("Promoted %s @%s -> %s (DSR=%.3f)", ticker, asset_tf, strategy, match["dsr"]) return True def cleanup_stale_promotions(db_path: Path = DB_PATH, logger: Optional[logging.Logger] = None) -> dict: """Quita de asset_strategies las entradas cuyo trigger, filtro o salida no exista YA en el catalogo actual (build_triggers / build_filters / EXIT_CONFIGS). Tras rediseñar el catalogo, las promociones del catalogo viejo quedan como zombies: el tracker no las puede recomputar y arrastran 0 trades forward para siempre. Aqui las eliminamos limpiamente.""" if logger is None: logger = logging.getLogger("strategy_lab") dummy = pd.DataFrame({ "open": [1.0], "high": [1.0], "low": [1.0], "close": [1.0], "volume": [1.0], }) valid_triggers = set(build_triggers(dummy).keys()) valid_filters = set(build_filters(dummy).keys()) valid_exits = set(EXIT_CONFIGS.keys()) # Tambien aceptamos las salidas de scalping: las promociones desde el # Scalping Lab tienen scalp_tight/scalp_quick/scalp_wide y son validas. try: sys_path_backup = list(sys.path) sys.path.insert(0, str(Path(__file__).resolve().parent)) import agente_strategy_lab_scalping as _scalp # noqa: F401 valid_exits |= set(_scalp.SCALPING_EXIT_CONFIGS.keys()) sys.path[:] = sys_path_backup except Exception: pass ensure_schema(db_path) removed: list[dict] = [] kept: list[dict] = [] with sqlite3.connect(db_path) as conn: rows = conn.execute( "SELECT ticker, strategy, timeframe FROM asset_strategies" ).fetchall() for ticker, strategy, tf in rows: parts = (strategy or "").split("|") entry = {"ticker": ticker, "strategy": strategy, "timeframe": tf} if len(parts) != 3: entry["reason"] = "malformed (expected trigger|filter|exit)" removed.append(entry) continue trg, flt, exc = parts reasons = [] if trg not in valid_triggers: reasons.append(f"trigger '{trg}' not in current catalog") if flt not in valid_filters: reasons.append(f"filter '{flt}' not in current catalog") if exc not in valid_exits: reasons.append(f"exit '{exc}' not in current catalog") if reasons: entry["reason"] = "; ".join(reasons) removed.append(entry) else: kept.append(entry) if removed: conn.executemany( "DELETE FROM asset_strategies WHERE ticker = ?", [(r["ticker"],) for r in removed]) conn.commit() for r in removed: logger.info("REMOVED stale: %-8s @%-3s %-40s -> %s", r["ticker"], r.get("timeframe") or "-", r["strategy"], r["reason"]) for k in kept: logger.info("KEPT current : %-8s @%-3s %s", k["ticker"], k.get("timeframe") or "-", k["strategy"]) logger.info( "Cleanup done: %d removed (stale catalog), %d kept (current catalog).", len(removed), len(kept)) return { "ok": True, "n_removed": len(removed), "n_kept": len(kept), "removed": removed, "kept": kept, "current_catalog": { "n_triggers": len(valid_triggers), "n_filters": len(valid_filters), "n_exit_configs": len(valid_exits), }, } # ───────────────────────────────────────────────────────────────────────────── # Main # ───────────────────────────────────────────────────────────────────────────── def main() -> int: global TRANSACTION_COST_PCT parser = argparse.ArgumentParser(description="Agente Strategy Lab de Leonex") parser.add_argument("--limit-tickers", type=int, default=0) parser.add_argument("--promote", type=str, default="") parser.add_argument("--cleanup-stale", action="store_true", help="Borra de asset_strategies las promociones cuyo " "trigger/filter/exit ya no exista en el catalogo " "actual (zombies del catalogo viejo).") parser.add_argument("--top-k", type=int, default=DEFAULT_TOP_K) parser.add_argument("--cost-pct", type=float, default=TRANSACTION_COST_PCT, help="Coste round-trip por trade en %% descontado de " "cada trade (IBKR; default 0.03).") args = parser.parse_args() TRANSACTION_COST_PCT = max(args.cost_pct, 0.0) 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_strategy_lab.log", encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger("agente_strategy_lab") if args.promote: if ":" not in args.promote: log.error("Invalid format. Use --promote TICKER:strategy_name") return 1 ticker, strategy = args.promote.split(":", 1) ok = promote_strategy(ticker.strip(), strategy.strip(), DB_PATH, log) return 0 if ok else 1 if args.cleanup_stale: result = cleanup_stale_promotions(DB_PATH, log) print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 dummy = pd.DataFrame({ "open": [1.0], "high": [1.0], "low": [1.0], "close": [1.0], "volume": [1.0], }) trigger_names = list(build_triggers(dummy).keys()) filter_names = list(build_filters(dummy).keys()) catalog = build_catalog(trigger_names, filter_names) n_total = len(catalog) universe = load_universe(DB_PATH) if args.limit_tickers and len(universe) > args.limit_tickers: universe = universe[: args.limit_tickers] timeframes = available_timeframes(DB_PATH) log.info("Strategy Lab v3: %d tickers x %d timeframes %s | %d strategies/" "asset (trigger x filter x exit, trade-based scoring)", len(universe), len(timeframes), timeframes, n_total) assets: list[AssetStrategyReport] = [] total = len(universe) * len(timeframes) done = 0 for tf in timeframes: for ticker in universe: done += 1 try: rep = evaluate_asset(ticker, catalog, DB_PATH, top_k=args.top_k, timeframe=tf) assets.append(rep) log.info(" %4d/%d [%s] %s @%s -> %s (Sharpe=%.2f DSR=%.3f " "trades=%d OOS=%.2f)", done, total, rep.best_tier, ticker, tf, rep.best_strategy, rep.best_sharpe, rep.best_dsr, rep.best_n_trades, rep.best_oos_sharpe) except Exception as exc: log.warning("Failed %s @%s: %s", ticker, tf, exc) n_gold = sum(1 for a in assets if a.best_tier == TIER_GOLD) n_silver = sum(1 for a in assets if a.best_tier == TIER_SILVER) n_bronze = sum(1 for a in assets if a.best_tier == TIER_BRONZE) n_rejected = sum(1 for a in assets if a.best_tier == TIER_REJECTED) n_robust = sum(1 for a in assets if a.best_is_robust) summary = (f"Evaluated {len(universe)} tickers across {len(timeframes)} " f"timeframes ({', '.join(timeframes)}) x {n_total} strategies " f"each (trigger x filter x exit). Every strategy is scored on " f"its TRADES, not on bars — intraday timeframes (1h/4h) produce " f"many more trades than daily. Trade returns are NET of a " f"{TRANSACTION_COST_PCT:.3f}% IBKR round-trip cost per trade " f"(commission + spread + slippage). Tiers: {n_gold} GOLD, " f"{n_silver} SILVER, {n_bronze} BRONZE, {n_rejected} REJECTED.") log.info(summary) payload = { "generated_at": datetime.now(UTC).isoformat(), "version": 3, "scoring": "trade_based", "timeframes": timeframes, "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(EXIT_CONFIGS), "triggers": trigger_names, "filters": filter_names, "exit_configs": EXIT_CONFIGS, "catalog": trigger_names, "dsr_robust_threshold": DSR_ROBUST_THRESHOLD, "initial_capital": DEFAULT_INITIAL_CAPITAL, "transaction_cost_pct": TRANSACTION_COST_PCT, "broker_cost_model": "Interactive Brokers Pro (Tiered) — commission " "+ spread + slippage, round-trip per trade", "n_robust": n_robust, "n_gold": n_gold, "n_silver": n_silver, "n_bronze": n_bronze, "n_rejected": n_rejected, "tier_thresholds": 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("Report exported -> %s", REPORT_PATH) sep = "=" * 96 print(f"\n{sep}") print("Leonex -- Strategy Lab v3 (trade-based scoring, multi-timeframe)") print(sep) print(summary) print() ordered = sorted(assets, key=lambda x: (TIER_RANK.get(x.best_tier, 9), -x.best_dsr)) print(f"{'tier':<10}{'ticker':<9}{'tf':<5}{'strategy':<40}{'Sharpe':>8}" f"{'DSR':>7}{'trades':>8}{'OOS':>7}") shown = 0 for a in ordered: if a.best_tier == TIER_REJECTED: continue shown += 1 print(f" {a.best_tier:<8}{a.ticker:<9}{a.timeframe:<5}" f"{a.best_strategy:<40}" f"{a.best_sharpe:>+8.2f}{a.best_dsr:>7.3f}" f"{a.best_n_trades:>8}{a.best_oos_sharpe:>+7.2f}") if shown == 0: print(" (no asset's best strategy passes the filter — see BRONZE in " "the dashboard with the tier filter)") if n_rejected: print(f"\n ({n_rejected} REJECTED asset/timeframe pairs hidden)") print(sep) print("To activate a winning strategy:") print(" python agents/agente_strategy_lab.py --promote TICKER:strategy_name") print(sep) return 0 if __name__ == "__main__": raise SystemExit(main())