# ============================================================ # backtest_engine.py — Trade Halley v3.1 # Motor de Backtest COMPLETO # Correções: volume médio por trades, drawdown negativo, # daily sem period/capital # Prioridade de dados: Supabase → yfinance → BRAPI # ============================================================ import pandas as pd import numpy as np import logging from datetime import datetime, timedelta logger = logging.getLogger("trade_halley.backtest") # Mapeamento de tickers BMF para Yahoo Finance BMF_TICKERS = { "IBOV_FUT": "^BVSP", "DOL_FUT": "USDBRL=X", "SP500": "ES=F", "NASDAQ": "NQ=F", "DOW": "YM=F", "CRUDE_OIL": "CL=F", "GOLD": "GC=F", "SILVER": "SI=F", "EURO_FX": "EURUSD=X", "BITCOIN": "BTC-USD", } def _load_data(ticker: str, period: str = "1y", interval: str = "1d", start_date: str = None, end_date: str = None) -> pd.DataFrame: """ Carrega dados OHLCV para o ticker. Tenta na ordem: 1. Supabase cache (via data_fetcher.get_stock_data_from_cache) 2. yfinance como fallback 3. BRAPI como último recurso Retorna DataFrame com colunas: Open, High, Low, Close, Volume (index = DatetimeIndex). """ df = pd.DataFrame() # --- Tentativa 1: Supabase cache (PRIORIDADE) --- try: from data_fetcher import get_stock_data_from_cache timeframe = "daily" if interval in ("1d", "1wk", "1mo") else "intraday" cached = get_stock_data_from_cache(ticker, timeframe=timeframe, start_date=start_date, end_date=end_date) if cached and len(cached) > 0: df = pd.DataFrame(cached) if "date" in df.columns: df["date"] = pd.to_datetime(df["date"]) df.set_index("date", inplace=True) elif "datetime" in df.columns: df["datetime"] = pd.to_datetime(df["datetime"]) df.set_index("datetime", inplace=True) logger.info(f"[supabase] {ticker} {interval}: {len(df)} registros") except Exception as e: logger.debug(f"Supabase cache miss para {ticker}: {e}") # --- Tentativa 2: yfinance fallback --- if df.empty: try: import yfinance as yf yf_ticker = ticker if ticker in BMF_TICKERS: yf_ticker = BMF_TICKERS[ticker] elif not ticker.endswith(".SA") and not any(c in ticker for c in ["^", "=", "-"]): yf_ticker = f"{ticker}.SA" if start_date and end_date: data = yf.download(yf_ticker, start=start_date, end=end_date, interval=interval, progress=False) else: data = yf.download(yf_ticker, period=period, interval=interval, progress=False) if data is not None and not data.empty: # Normaliza MultiIndex columns (yfinance >= 0.2.31) if isinstance(data.columns, pd.MultiIndex): data.columns = [col[0] if isinstance(col, tuple) else col for col in data.columns] df = data.copy() logger.info(f"[yfinance] {ticker} ({yf_ticker}) {interval}: {len(df)} registros") except Exception as e: logger.warning(f"yfinance falhou para {ticker}: {e}") # --- Tentativa 3: BRAPI como último recurso --- if df.empty: try: from data_fetcher import get_stock_data raw = get_stock_data(ticker, period=period, interval=interval, start_date=start_date, end_date=end_date) if raw and len(raw) > 0: df = pd.DataFrame(raw) if "date" in df.columns: df["date"] = pd.to_datetime(df["date"]) df.set_index("date", inplace=True) elif "datetime" in df.columns: df["datetime"] = pd.to_datetime(df["datetime"]) df.set_index("datetime", inplace=True) logger.info(f"[brapi] {ticker} {interval}: {len(df)} registros") except Exception as e: logger.warning(f"BRAPI falhou para {ticker}: {e}") if df.empty: logger.error(f"Sem dados para {ticker} (period={period}, interval={interval})") return pd.DataFrame() # --- Normaliza colunas --- col_map = {} for col in df.columns: cl = str(col).lower().strip() if cl in ("open", "abertura"): col_map[col] = "Open" elif cl in ("high", "máxima", "maxima", "alta"): col_map[col] = "High" elif cl in ("low", "mínima", "minima", "baixa"): col_map[col] = "Low" elif cl in ("close", "fechamento", "adj close", "adj_close", "adjustedclose"): col_map[col] = "Close" elif cl in ("volume", "vol"): col_map[col] = "Volume" if col_map: df.rename(columns=col_map, inplace=True) for col in ["Open", "High", "Low", "Close", "Volume"]: if col not in df.columns: df[col] = 0.0 for col in ["Open", "High", "Low", "Close", "Volume"]: df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0.0) df = df[df["Close"] > 0] if not isinstance(df.index, pd.DatetimeIndex): try: df.index = pd.to_datetime(df.index) except Exception: pass df.sort_index(inplace=True) df.dropna(subset=["Close"], inplace=True) # Filtra por datas se fornecidas (caso os dados vieram do yfinance/BRAPI sem filtro) if start_date: try: sd = pd.to_datetime(start_date) df = df[df.index >= sd] except Exception: pass if end_date: try: ed = pd.to_datetime(end_date) df = df[df.index <= ed] except Exception: pass return df def _calc_metrics(trades: list, initial_capital: float, df: pd.DataFrame) -> dict: """ Calcula métricas no formato Trade Certo. - resultado_pct: retorno composto (como o Trade Certo original) - max_drawdown_pct: pior trade individual (valor negativo) - ganho_medio_pct: total_return_pct / total_trades - Volume médio: média de TODOS os dias do período (não só dos dias com trade) """ if not trades: return { "total_gain": 0, "pct_gain": 0.0, "total_loss": 0, "pct_loss": 0.0, "total_trades": 0, "resultado_pct": 0.0, "max_drawdown_pct": 0.0, "ganho_maximo_pct": 0.0, "ganho_medio_pct": 0.0, "volume_medio": 0.0, "win_rate": 0.0, "profit_factor": 0.0, "sharpe_ratio": 0.0, "sortino_ratio": 0.0, "initial_capital": initial_capital, "final_capital": initial_capital, "total_return_pct": 0.0, "avg_win": 0.0, "avg_loss": 0.0, "gross_profit": 0.0, "gross_loss": 0.0, "equity_curve": [initial_capital], } pnls = [t["pnl_pct"] for t in trades] wins = [p for p in pnls if p > 0] losses = [p for p in pnls if p <= 0] total_trades = len(trades) total_gain = len(wins) total_loss = len(losses) pct_gain = (total_gain / total_trades * 100) if total_trades > 0 else 0.0 pct_loss = (total_loss / total_trades * 100) if total_trades > 0 else 0.0 gross_profit = sum(wins) if wins else 0.0 gross_loss = sum(losses) if losses else 0.0 avg_win = np.mean(wins) if wins else 0.0 avg_loss = np.mean(losses) if losses else 0.0 ganho_maximo_pct = max(pnls) if pnls else 0.0 win_rate = pct_gain profit_factor = abs(gross_profit / gross_loss) if gross_loss != 0 else ( float("inf") if gross_profit > 0 else 0.0 ) # Equity curve (retorno composto) equity = [initial_capital] capital = initial_capital for p in pnls: capital = capital * (1 + p / 100.0) equity.append(round(capital, 2)) final_capital = equity[-1] total_return_pct = ((final_capital - initial_capital) / initial_capital * 100) if initial_capital > 0 else 0.0 # resultado_pct = retorno COMPOSTO (como o Trade Certo original) resultado_pct = total_return_pct # ← FIX 2: Max Drawdown = pior trade individual (valor negativo), como Trade Certo max_drawdown_pct = min(pnls) if min(pnls) < 0 else 0.0 # ← FIX # ← FIX 3: Ganho Médio = retorno composto / número de trades, como Trade Certo ganho_medio_pct = total_return_pct / total_trades if total_trades > 0 else 0.0 # ← FIX # Sharpe / Sortino if len(pnls) > 1: pnl_arr = np.array(pnls) std_all = np.std(pnl_arr, ddof=1) sharpe = (np.mean(pnl_arr) / std_all) if std_all > 0 else 0.0 neg_returns = pnl_arr[pnl_arr < 0] std_neg = np.std(neg_returns, ddof=1) if len(neg_returns) > 1 else 0.0 sortino = (np.mean(pnl_arr) / std_neg) if std_neg > 0 else 0.0 else: sharpe = 0.0 sortino = 0.0 # Volume médio — média de TODOS os dias do período vol_medio = 0.0 if df is not None and "Volume" in df.columns and len(df) > 0: vol_values = df["Volume"] vol_values = vol_values[vol_values > 0] if len(vol_values) > 0: vol_medio = float(vol_values.mean()) return { "total_gain": total_gain, "pct_gain": round(pct_gain, 2), "total_loss": total_loss, "pct_loss": round(pct_loss, 2), "total_trades": total_trades, "resultado_pct": round(resultado_pct, 4), "max_drawdown_pct": round(max_drawdown_pct, 4), "ganho_maximo_pct": round(ganho_maximo_pct, 4), "ganho_medio_pct": round(ganho_medio_pct, 4), "volume_medio": round(vol_medio, 2), "win_rate": round(win_rate, 2), "profit_factor": round(profit_factor, 4) if profit_factor != float("inf") else 999.99, "sharpe_ratio": round(sharpe, 4), "sortino_ratio": round(sortino, 4), "initial_capital": initial_capital, "final_capital": round(final_capital, 2), "total_return_pct": round(total_return_pct, 4), "avg_win": round(avg_win, 4), "avg_loss": round(avg_loss, 4), "gross_profit": round(gross_profit, 4), "gross_loss": round(gross_loss, 4), "equity_curve": equity, } # ============================================================ # BACKTEST — INDICADORES TÉCNICOS (compatibilidade v2.2) # ============================================================ def run_backtest(ticker: str, strategy_id: str, period: str = "1y", interval: str = "1d", initial_capital: float = 10000.0, stop_loss: float = None, take_profit: float = None) -> dict: """ Backtest de indicadores técnicos (SMA cross, RSI, MACD, etc.). """ from strategies import STRATEGIES, add_indicators, get_strategy_signal if strategy_id not in STRATEGIES: logger.error(f"Estratégia '{strategy_id}' não encontrada") return None df = _load_data(ticker, period, interval) if df.empty: logger.error(f"Sem dados para {ticker}") return None df = add_indicators(df) signals = get_strategy_signal(df, strategy_id) if signals is None or signals.empty: logger.warning(f"Sem sinais para {ticker} com {strategy_id}") return None trades = [] in_position = False entry_price = 0.0 entry_date = None for i in range(1, len(df)): sig = signals.iloc[i] prev_sig = signals.iloc[i - 1] row = df.iloc[i] if not in_position: if sig == 1 and prev_sig <= 0: in_position = True entry_price = row["Open"] entry_date = df.index[i] else: if stop_loss and row["Low"] <= entry_price * (1 - stop_loss): exit_price = entry_price * (1 - stop_loss) pnl_pct = -stop_loss * 100 trades.append({ "entry_date": str(entry_date), "entry_price": round(entry_price, 4), "exit_date": str(df.index[i]), "exit_price": round(exit_price, 4), "pnl_pct": round(pnl_pct, 4), "reason": "stop_loss", }) in_position = False continue if take_profit and row["High"] >= entry_price * (1 + take_profit): exit_price = entry_price * (1 + take_profit) pnl_pct = take_profit * 100 trades.append({ "entry_date": str(entry_date), "entry_price": round(entry_price, 4), "exit_date": str(df.index[i]), "exit_price": round(exit_price, 4), "pnl_pct": round(pnl_pct, 4), "reason": "take_profit", }) in_position = False continue if sig <= 0 and prev_sig > 0: exit_price = row["Open"] pnl_pct = ((exit_price - entry_price) / entry_price) * 100 trades.append({ "entry_date": str(entry_date), "entry_price": round(entry_price, 4), "exit_date": str(df.index[i]), "exit_price": round(exit_price, 4), "pnl_pct": round(pnl_pct, 4), "reason": "signal", }) in_position = False if in_position and len(df) > 0: exit_price = df.iloc[-1]["Close"] pnl_pct = ((exit_price - entry_price) / entry_price) * 100 trades.append({ "entry_date": str(entry_date), "entry_price": round(entry_price, 4), "exit_date": str(df.index[-1]), "exit_price": round(exit_price, 4), "pnl_pct": round(pnl_pct, 4), "reason": "end_of_data", }) strat_info = STRATEGIES[strategy_id] metrics = _calc_metrics(trades, initial_capital, df) return { "ticker": ticker, "strategy": strategy_id, "strategy_name": strat_info["name"], "period": period, "interval": interval, "data_points": len(df), "start_date": str(df.index[0]) if len(df) > 0 else None, "end_date": str(df.index[-1]) if len(df) > 0 else None, "trades": trades, "metrics": metrics, } def run_bulk_backtest(tickers: list, strategy_id: str, period: str = "1y", interval: str = "1d", initial_capital: float = 10000.0) -> list: """Backtest de indicadores em múltiplos ativos.""" results = [] for ticker in tickers: try: r = run_backtest( ticker=ticker, strategy_id=strategy_id, period=period, interval=interval, initial_capital=initial_capital, ) if r and r.get("metrics"): m = r["metrics"] results.append({ "ticker": ticker, "strategy": strategy_id, "total_trades": m["total_trades"], "win_rate": m["win_rate"], "total_return_pct": m["total_return_pct"], "profit_factor": m["profit_factor"], "max_drawdown_pct": m["max_drawdown_pct"], "sharpe_ratio": m["sharpe_ratio"], "final_capital": m["final_capital"], }) except Exception as e: logger.warning(f"Bulk skip {ticker}: {e}") results.sort(key=lambda x: x.get("total_return_pct", 0), reverse=True) return results # ============================================================ # BACKTEST TRADE CERTO — DAILY # ============================================================ def run_backtest_daily(ticker: str, entry_strategy_id: str, exit_strategy_id: str, direction: str = "compra", variation_pct: float = 0.0, period: str = "1y", initial_capital: float = 10000.0, start_date: str = None, end_date: str = None) -> dict: """ Backtest Trade Certo — Timeframe Diário. """ from strategies import ENTRY_STRATEGIES_DAILY, EXIT_STRATEGIES_DAILY if entry_strategy_id not in ENTRY_STRATEGIES_DAILY: logger.error(f"Entrada daily '{entry_strategy_id}' não encontrada") return None if exit_strategy_id not in EXIT_STRATEGIES_DAILY: logger.error(f"Saída daily '{exit_strategy_id}' não encontrada") return None fetch_period = "1y" if start_date and end_date: try: sd = pd.to_datetime(start_date) ed = pd.to_datetime(end_date) diff_days = (ed - sd).days if diff_days <= 30: fetch_period = "3mo" elif diff_days <= 90: fetch_period = "3mo" elif diff_days <= 180: fetch_period = "6mo" else: fetch_period = "1y" except Exception: fetch_period = "1y" # Carrega dados SEM filtro de data (para ter o dia anterior ao start_date) df = _load_data(ticker, fetch_period, interval="1d") if df.empty: logger.error(f"Sem dados diários para {ticker}") return None if len(df) < 2: logger.error(f"Dados insuficientes para {ticker} (apenas {len(df)} registros)") return None df_full = df.copy() if start_date: try: sd = pd.to_datetime(start_date) df_full = df_full[df_full.index >= sd] except Exception: pass if end_date: try: ed = pd.to_datetime(end_date) df_full = df_full[df_full.index <= ed] except Exception: pass if len(df_full) < 1: logger.error(f"Nenhum dado no período para {ticker}") return None entry_strat = ENTRY_STRATEGIES_DAILY[entry_strategy_id] exit_strat = EXIT_STRATEGIES_DAILY[exit_strategy_id] entry_kwargs = {"direction": direction} entry_requires = entry_strat.get("requires", []) if "variation_pct" in entry_requires: entry_kwargs["variation_pct"] = variation_pct entry_func = entry_strat["function"] entry_signals = entry_func(df, **entry_kwargs) if entry_signals is None or entry_signals.empty: logger.info(f"Nenhum sinal de entrada para {ticker} com {entry_strategy_id}") metrics = _calc_metrics([], initial_capital, df_full) return { "ticker": ticker, "entry_strategy": entry_strategy_id, "entry_strategy_name": entry_strat["name"], "exit_strategy": exit_strategy_id, "exit_strategy_name": exit_strat["name"], "direction": direction, "variation_pct": variation_pct, "period": period, "start_date": str(df_full.index[0]) if len(df_full) > 0 else start_date, "end_date": str(df_full.index[-1]) if len(df_full) > 0 else end_date, "data_points": len(df_full), "trades": [], "metrics": metrics, } trades = [] exit_func = exit_strat["function"] for i in range(len(df)): idx = df.index[i] if start_date: try: if idx < pd.to_datetime(start_date): continue except Exception: pass if end_date: try: if idx > pd.to_datetime(end_date): continue except Exception: pass if idx not in entry_signals.index: continue sig = entry_signals.loc[idx] if isinstance(sig, pd.DataFrame): sig = sig.iloc[0] if not sig.get("entry", False): continue entry_price = sig.get("entry_price", np.nan) if pd.isna(entry_price) or entry_price <= 0: continue try: exit_result = exit_func(df, i, entry_price, direction=direction) except Exception as e: logger.warning(f"Exit error para {ticker} idx {i}: {e}") continue if exit_result is None: continue trades.append({ "entry_date": str(idx), "exit_date": str(exit_result["exit_date"]), "entry_price": round(float(entry_price), 4), "exit_price": round(float(exit_result["exit_price"]), 4), "pnl_pct": round(float(exit_result["pnl_pct"]), 4), "direction": direction, }) calc_capital = 10000.0 metrics = _calc_metrics(trades, calc_capital, df_full) return { "ticker": ticker, "entry_strategy": entry_strategy_id, "entry_strategy_name": entry_strat["name"], "exit_strategy": exit_strategy_id, "exit_strategy_name": exit_strat["name"], "direction": direction, "variation_pct": variation_pct, "period": period, "start_date": start_date or str(df_full.index[0]), "end_date": end_date or str(df_full.index[-1]), "data_points": len(df_full), "trades": trades, "metrics": metrics, } def run_bulk_backtest_daily(tickers: list, entry_strategy_id: str, exit_strategy_id: str, direction: str = "compra", variation_pct: float = 0.0, period: str = "1y", initial_capital: float = 10000.0, start_date: str = None, end_date: str = None) -> list: """Backtest Trade Certo Daily em múltiplos ativos.""" results = [] for ticker in tickers: try: r = run_backtest_daily( ticker=ticker, entry_strategy_id=entry_strategy_id, exit_strategy_id=exit_strategy_id, direction=direction, variation_pct=variation_pct, period=period, initial_capital=initial_capital, start_date=start_date, end_date=end_date, ) if r and r.get("metrics"): m = r["metrics"] results.append({ "acao": ticker, "total_gain": m["total_gain"], "pct_gain": m["pct_gain"], "total_loss": m["total_loss"], "pct_loss": m["pct_loss"], "total_trades": m["total_trades"], "resultado_pct": m["resultado_pct"], "max_drawdown_pct": m["max_drawdown_pct"], "ganho_maximo_pct": m["ganho_maximo_pct"], "ganho_medio_pct": m["ganho_medio_pct"], "volume_medio": m["volume_medio"], "total_return_pct": m["total_return_pct"], "win_rate": m["win_rate"], "profit_factor": m["profit_factor"], }) except Exception as e: logger.warning(f"Bulk daily skip {ticker}: {e}") results.sort(key=lambda x: x.get("resultado_pct", 0), reverse=True) return results # ============================================================ # BACKTEST TRADE CERTO — INTRADAY # ============================================================ def _get_daily_reference(df_daily: pd.DataFrame, current_date) -> dict: """Obtém valores de referência do dia anterior.""" if df_daily is None or df_daily.empty: return {"prev_close": None, "prev_open": None, "day_open": None} if hasattr(current_date, 'date'): target_date = current_date.date() else: target_date = current_date prev_rows = df_daily[df_daily.index.date < target_date] if hasattr(df_daily.index, 'date') else df_daily[df_daily.index < str(target_date)] prev_close = None prev_open = None if len(prev_rows) > 0: prev_close = prev_rows.iloc[-1]["Close"] prev_open = prev_rows.iloc[-1]["Open"] day_rows = df_daily[df_daily.index.date == target_date] if hasattr(df_daily.index, 'date') else None day_open = None if day_rows is not None and len(day_rows) > 0: day_open = day_rows.iloc[0]["Open"] return {"prev_close": prev_close, "prev_open": prev_open, "day_open": day_open} def run_backtest_intraday(ticker: str, entry_strategy_id: str, exit_strategy_id: str, direction: str = "compra", variation_pct: float = 0.0, hour_start: str = "09:00", hour_end: str = "17:00", hour_target: str = None, target_pct: float = None, stop_loss_pct: float = None, close_eod: bool = True, max_trades_per_day: int = 99, bb_window: int = 20, bb_std: float = 2.0, ma_period: int = 20, period: str = "3mo", initial_capital: float = 10000.0, start_date: str = None, end_date: str = None) -> dict: """Backtest Trade Certo — Intraday.""" from strategies import ENTRY_STRATEGIES_INTRADAY, EXIT_STRATEGIES_INTRADAY if entry_strategy_id not in ENTRY_STRATEGIES_INTRADAY: logger.error(f"Entrada intraday '{entry_strategy_id}' não encontrada") return None if exit_strategy_id not in EXIT_STRATEGIES_INTRADAY: logger.error(f"Saída intraday '{exit_strategy_id}' não encontrada") return None df_intra = _load_data(ticker, period, interval="5m") if df_intra.empty: df_intra = _load_data(ticker, period, interval="15m") if df_intra.empty: df_intra = _load_data(ticker, period, interval="1h") if df_intra.empty: logger.error(f"Sem dados intraday para {ticker}") return None df_daily = _load_data(ticker, period, interval="1d") if start_date: try: sd = pd.to_datetime(start_date) df_intra = df_intra[df_intra.index >= sd] except Exception: pass if end_date: try: ed = pd.to_datetime(end_date) df_intra = df_intra[df_intra.index <= ed] except Exception: pass if len(df_intra) < 2: logger.error(f"Dados intraday insuficientes para {ticker}") return None if hasattr(df_intra.index, 'date'): days = sorted(set(df_intra.index.date)) else: days = [None] entry_strat = ENTRY_STRATEGIES_INTRADAY[entry_strategy_id] exit_strat = EXIT_STRATEGIES_INTRADAY[exit_strategy_id] all_trades = [] for day in days: if day is None: day_df = df_intra else: day_mask = df_intra.index.date == day day_df = df_intra[day_mask] if day_df.empty: continue ref = _get_daily_reference(df_daily, day) if df_daily is not None and not df_daily.empty else { "prev_close": None, "prev_open": None, "day_open": None } if ref["prev_close"] is None and day is not None: prev_day_mask = df_intra.index.date < day prev_data = df_intra[prev_day_mask] if len(prev_data) > 0: ref["prev_close"] = prev_data.iloc[-1]["Close"] prev_days = sorted(set(prev_data.index.date)) if prev_days: last_prev_day = prev_days[-1] lpd_data = prev_data[prev_data.index.date == last_prev_day] if len(lpd_data) > 0: ref["prev_open"] = lpd_data.iloc[0]["Open"] if ref["day_open"] is None and len(day_df) > 0: ref["day_open"] = day_df.iloc[0]["Open"] entry_kwargs = {"direction": direction} entry_requires = entry_strat.get("requires", []) fixed_params = entry_strat.get("params_fixed", {}) if "variation_pct" in entry_requires: entry_kwargs["variation_pct"] = variation_pct if "hour_start" in entry_requires: entry_kwargs["hour_start"] = hour_start if "hour_end" in entry_requires: entry_kwargs["hour_end"] = hour_end if "hour_target" in entry_requires: entry_kwargs["hour_target"] = hour_target or hour_start if "prev_close_value" in entry_requires: entry_kwargs["prev_close_value"] = ref["prev_close"] if "prev_open_value" in entry_requires: entry_kwargs["prev_open_value"] = ref["prev_open"] if "day_open_value" in entry_requires: entry_kwargs["day_open_value"] = ref["day_open"] if "bb_window" in entry_requires: entry_kwargs["bb_window"] = bb_window if "bb_std" in entry_requires: entry_kwargs["bb_std"] = bb_std if "ma_period" in entry_requires: entry_kwargs["ma_period"] = ma_period entry_kwargs.update(fixed_params) entry_func = entry_strat["function"] entry_signals = entry_func(day_df, **entry_kwargs) if entry_signals is None or entry_signals.empty: continue day_trade_count = 0 for j in range(len(day_df)): if day_trade_count >= max_trades_per_day: break idx = day_df.index[j] if idx not in entry_signals.index: continue sig = entry_signals.loc[idx] if isinstance(sig, pd.DataFrame): sig = sig.iloc[0] if not sig.get("entry", False): continue entry_price = sig.get("entry_price", np.nan) if pd.isna(entry_price) or entry_price <= 0: continue try: intra_idx = df_intra.index.get_loc(idx) if isinstance(intra_idx, slice): intra_idx = intra_idx.start elif isinstance(intra_idx, np.ndarray): intra_idx = int(np.where(intra_idx)[0][0]) except Exception: continue exit_kwargs = {"direction": direction} exit_requires = exit_strat.get("requires", []) exit_fixed = exit_strat.get("params_fixed", {}) if "target_pct" in exit_requires: exit_kwargs["target_pct"] = target_pct or 1.0 if "stop_loss_pct" in exit_requires: exit_kwargs["stop_loss_pct"] = stop_loss_pct if "close_eod" in exit_requires: exit_kwargs["close_eod"] = close_eod if "hour_target" in exit_requires: exit_kwargs["hour_target"] = hour_target or "17:00" if "bb_window" in exit_requires: exit_kwargs["bb_window"] = bb_window if "bb_std" in exit_requires: exit_kwargs["bb_std"] = bb_std if "ma_period" in exit_requires: exit_kwargs["ma_period"] = ma_period exit_kwargs.update(exit_fixed) try: exit_func = exit_strat["function"] exit_result = exit_func(df_intra, intra_idx, entry_price, **exit_kwargs) except Exception as e: logger.warning(f"Intraday exit error {ticker} idx {intra_idx}: {e}") continue if exit_result is None: continue all_trades.append({ "entry_date": str(idx), "exit_date": str(exit_result["exit_date"]), "entry_price": round(float(entry_price), 4), "exit_price": round(float(exit_result["exit_price"]), 4), "pnl_pct": round(float(exit_result["pnl_pct"]), 4), "direction": direction, "day": str(day), }) day_trade_count += 1 metrics = _calc_metrics(all_trades, initial_capital, df_intra) return { "ticker": ticker, "entry_strategy": entry_strategy_id, "entry_strategy_name": entry_strat["name"], "exit_strategy": exit_strategy_id, "exit_strategy_name": exit_strat["name"], "direction": direction, "variation_pct": variation_pct, "hour_start": hour_start, "hour_end": hour_end, "period": period, "start_date": start_date or str(df_intra.index[0]), "end_date": end_date or str(df_intra.index[-1]), "data_points": len(df_intra), "max_trades_per_day": max_trades_per_day, "close_eod": close_eod, "target_pct": target_pct, "stop_loss_pct": stop_loss_pct, "trades": all_trades, "metrics": metrics, } # ============================================================ # BULK BACKTEST TRADE CERTO — INTRADAY # ============================================================ def run_bulk_backtest_intraday(tickers: list, entry_strategy_id: str, exit_strategy_id: str, direction: str = "compra", variation_pct: float = 0.0, hour_start: str = "09:00", hour_end: str = "17:00", hour_target: str = None, target_pct: float = None, stop_loss_pct: float = None, close_eod: bool = True, max_trades_per_day: int = 99, bb_window: int = 20, bb_std: float = 2.0, ma_period: int = 20, period: str = "3mo", initial_capital: float = 10000.0, start_date: str = None, end_date: str = None) -> list: """Backtest Trade Certo Intraday em múltiplos ativos.""" results = [] for ticker in tickers: try: r = run_backtest_intraday( ticker=ticker, entry_strategy_id=entry_strategy_id, exit_strategy_id=exit_strategy_id, direction=direction, variation_pct=variation_pct, hour_start=hour_start, hour_end=hour_end, hour_target=hour_target, target_pct=target_pct, stop_loss_pct=stop_loss_pct, close_eod=close_eod, max_trades_per_day=max_trades_per_day, bb_window=bb_window, bb_std=bb_std, ma_period=ma_period, period=period, initial_capital=initial_capital, start_date=start_date, end_date=end_date, ) if r and r.get("metrics"): m = r["metrics"] results.append({ "acao": ticker, "total_gain": m["total_gain"], "pct_gain": m["pct_gain"], "total_loss": m["total_loss"], "pct_loss": m["pct_loss"], "total_trades": m["total_trades"], "resultado_pct": m["resultado_pct"], "max_drawdown_pct": m["max_drawdown_pct"], "ganho_maximo_pct": m["ganho_maximo_pct"], "ganho_medio_pct": m["ganho_medio_pct"], "volume_medio": m["volume_medio"], "total_return_pct": m["total_return_pct"], "win_rate": m["win_rate"], "profit_factor": m["profit_factor"], }) except Exception as e: logger.warning(f"Bulk intraday skip {ticker}: {e}") results.sort(key=lambda x: x.get("resultado_pct", 0), reverse=True) return results