Backtesting de Estratégias com Python e brapi 2026: Guia Completo

┌─────────────────────────────────────────────────────────────┐
│                    FLUXO DO BACKTESTING                      │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  1. Definir       2. Coletar         3. Simular             │
│     Estratégia →     Dados       →      Trades              │
│     (regras)         Históricos        (compra/venda)       │
│                                                              │
│                            ↓                                 │
│                                                              │
│  4. Calcular      5. Analisar        6. Otimizar            │
│     Métricas   ←     Resultados   ←     Parâmetros          │
│     (Sharpe, DD)     (gráficos)        (se necessário)      │
│                                                              │
└─────────────────────────────────────────────────────────────┘

pip install requests pandas numpy matplotlib

backtesting-brapi/
├── src/
│   ├── data/
│   │   └── fetcher.py         # Buscar dados da brapi
│   ├── strategies/
│   │   ├── base.py            # Classe base de estratégia
│   │   ├── moving_average.py  # Cruzamento de médias
│   │   └── rsi.py             # RSI oversold/overbought
│   ├── backtest/
│   │   ├── engine.py          # Motor de backtesting
│   │   └── metrics.py         # Cálculo de métricas
│   └── utils/
│       └── plots.py           # Visualização
├── main.py
└── requirements.txt

import requests
import pandas as pd
from typing import Optional

class BrapiDataFetcher:
    """Busca dados históricos da brapi"""
    
    BASE_URL = "https://brapi.dev/api"
    
    def __init__(self, token: str):
        self.token = token
    
    def get_historical_data(
        self,
        ticker: str,
        range_period: str = "5y",
        interval: str = "1d"
    ) -> pd.DataFrame:
        """
        Busca dados históricos de um ticker.
        
        Args:
            ticker: Código da ação (ex: PETR4)
            range_period: Período (1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, max)
            interval: Intervalo (1d, 1wk, 1mo)
        
        Returns:
            DataFrame com colunas: date, open, high, low, close, volume
        """
        url = f"{self.BASE_URL}/quote/{ticker}"
        params = {
            "token": self.token,
            "range": range_period,
            "interval": interval,
            "fundamental": "false"
        }
        
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()
        
        if not data.get("results"):
            raise ValueError(f"Sem dados para {ticker}")
        
        historical = data["results"][0].get("historicalDataPrice", [])
        
        if not historical:
            raise ValueError(f"Sem dados históricos para {ticker}")
        
        df = pd.DataFrame(historical)
        df["date"] = pd.to_datetime(df["date"], unit="s")
        df = df.sort_values("date").reset_index(drop=True)
        
        # Renomear colunas para padrão
        df = df.rename(columns={
            "open": "open",
            "high": "high", 
            "low": "low",
            "close": "close",
            "volume": "volume"
        })
        
        return df[["date", "open", "high", "low", "close", "volume"]]
    
    def get_multiple_tickers(
        self,
        tickers: list[str],
        range_period: str = "5y",
        interval: str = "1d"
    ) -> dict[str, pd.DataFrame]:
        """Busca dados de múltiplos tickers"""
        data = {}
        for ticker in tickers:
            try:
                data[ticker] = self.get_historical_data(
                    ticker, range_period, interval
                )
            except Exception as e:
                print(f"Erro ao buscar {ticker}: {e}")
        return data


# Exemplo de uso
if __name__ == "__main__":
    fetcher = BrapiDataFetcher("seu_token_brapi")
    df = fetcher.get_historical_data("PETR4", "2y", "1d")
    print(df.head())
    print(f"Total de registros: {len(df)}")

# src/strategies/base.py
from abc import ABC, abstractmethod
import pandas as pd
import numpy as np

class Strategy(ABC):
    """Classe base para estratégias de trading"""
    
    def __init__(self, name: str):
        self.name = name
    
    @abstractmethod
    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        Gera sinais de compra/venda.
        
        Args:
            data: DataFrame com OHLCV
        
        Returns:
            DataFrame com coluna 'signal':
            - 1: compra
            - -1: venda
            - 0: manter posição
        """
        pass
    
    def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame:
        """Calcula indicadores técnicos (opcional)"""
        return data
    
    def __repr__(self):
        return f"Strategy({self.name})"

# src/strategies/moving_average.py
import pandas as pd
import numpy as np
from .base import Strategy

class MovingAverageCrossover(Strategy):
    """
    Estratégia de cruzamento de médias móveis.
    
    Regras:
    - Compra quando média curta cruza acima da média longa
    - Vende quando média curta cruza abaixo da média longa
    """
    
    def __init__(self, short_window: int = 20, long_window: int = 50):
        super().__init__(f"MA_Crossover_{short_window}_{long_window}")
        self.short_window = short_window
        self.long_window = long_window
    
    def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame:
        """Calcula médias móveis"""
        df = data.copy()
        df["sma_short"] = df["close"].rolling(window=self.short_window).mean()
        df["sma_long"] = df["close"].rolling(window=self.long_window).mean()
        return df
    
    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        """Gera sinais de cruzamento"""
        df = self.calculate_indicators(data)
        
        # Inicializa coluna de sinais
        df["signal"] = 0
        
        # Detecta cruzamentos
        # Cruzamento para cima: short cruza acima de long
        df.loc[
            (df["sma_short"] > df["sma_long"]) & 
            (df["sma_short"].shift(1) <= df["sma_long"].shift(1)),
            "signal"
        ] = 1
        
        # Cruzamento para baixo: short cruza abaixo de long
        df.loc[
            (df["sma_short"] < df["sma_long"]) & 
            (df["sma_short"].shift(1) >= df["sma_long"].shift(1)),
            "signal"
        ] = -1
        
        return df


class ExponentialMovingAverageCrossover(Strategy):
    """Cruzamento de médias móveis exponenciais (EMA)"""
    
    def __init__(self, short_window: int = 12, long_window: int = 26):
        super().__init__(f"EMA_Crossover_{short_window}_{long_window}")
        self.short_window = short_window
        self.long_window = long_window
    
    def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame:
        df = data.copy()
        df["ema_short"] = df["close"].ewm(span=self.short_window, adjust=False).mean()
        df["ema_long"] = df["close"].ewm(span=self.long_window, adjust=False).mean()
        return df
    
    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        df = self.calculate_indicators(data)
        df["signal"] = 0
        
        df.loc[
            (df["ema_short"] > df["ema_long"]) & 
            (df["ema_short"].shift(1) <= df["ema_long"].shift(1)),
            "signal"
        ] = 1
        
        df.loc[
            (df["ema_short"] < df["ema_long"]) & 
            (df["ema_short"].shift(1) >= df["ema_long"].shift(1)),
            "signal"
        ] = -1
        
        return df

# src/strategies/rsi.py
import pandas as pd
import numpy as np
from .base import Strategy

class RSIStrategy(Strategy):
    """
    Estratégia baseada no RSI (Relative Strength Index).
    
    Regras:
    - Compra quando RSI cruza acima do nível oversold (sobrevendido)
    - Vende quando RSI cruza abaixo do nível overbought (sobrecomprado)
    """
    
    def __init__(
        self,
        period: int = 14,
        oversold: float = 30,
        overbought: float = 70
    ):
        super().__init__(f"RSI_{period}_{oversold}_{overbought}")
        self.period = period
        self.oversold = oversold
        self.overbought = overbought
    
    def calculate_rsi(self, prices: pd.Series) -> pd.Series:
        """Calcula o RSI"""
        delta = prices.diff()
        
        gain = delta.where(delta > 0, 0)
        loss = (-delta).where(delta < 0, 0)
        
        avg_gain = gain.rolling(window=self.period).mean()
        avg_loss = loss.rolling(window=self.period).mean()
        
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        
        return rsi
    
    def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame:
        df = data.copy()
        df["rsi"] = self.calculate_rsi(df["close"])
        return df
    
    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        df = self.calculate_indicators(data)
        df["signal"] = 0
        
        # Compra: RSI cruza acima de oversold
        df.loc[
            (df["rsi"] > self.oversold) & 
            (df["rsi"].shift(1) <= self.oversold),
            "signal"
        ] = 1
        
        # Venda: RSI cruza abaixo de overbought
        df.loc[
            (df["rsi"] < self.overbought) & 
            (df["rsi"].shift(1) >= self.overbought),
            "signal"
        ] = -1
        
        return df


class RSIMeanReversion(Strategy):
    """
    Estratégia de reversão à média com RSI.
    
    Compra em oversold extremo, vende em overbought extremo.
    """
    
    def __init__(
        self,
        period: int = 14,
        oversold: float = 25,
        overbought: float = 75,
        exit_level: float = 50
    ):
        super().__init__(f"RSI_MeanRev_{period}")
        self.period = period
        self.oversold = oversold
        self.overbought = overbought
        self.exit_level = exit_level
    
    def calculate_rsi(self, prices: pd.Series) -> pd.Series:
        delta = prices.diff()
        gain = delta.where(delta > 0, 0)
        loss = (-delta).where(delta < 0, 0)
        avg_gain = gain.rolling(window=self.period).mean()
        avg_loss = loss.rolling(window=self.period).mean()
        rs = avg_gain / avg_loss
        return 100 - (100 / (1 + rs))
    
    def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame:
        df = data.copy()
        df["rsi"] = self.calculate_rsi(df["close"])
        return df
    
    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        df = self.calculate_indicators(data)
        df["signal"] = 0
        
        # Compra em oversold extremo
        df.loc[df["rsi"] < self.oversold, "signal"] = 1
        
        # Venda em overbought extremo
        df.loc[df["rsi"] > self.overbought, "signal"] = -1
        
        return df

# src/backtest/engine.py
import pandas as pd
import numpy as np
from typing import Optional
from dataclasses import dataclass

@dataclass
class Trade:
    """Representa uma operação"""
    entry_date: pd.Timestamp
    entry_price: float
    exit_date: Optional[pd.Timestamp] = None
    exit_price: Optional[float] = None
    shares: int = 0
    pnl: float = 0
    pnl_percent: float = 0
    
    @property
    def is_closed(self) -> bool:
        return self.exit_date is not None


class BacktestEngine:
    """
    Motor de backtesting.
    
    Simula execução de estratégia em dados históricos.
    """
    
    def __init__(
        self,
        initial_capital: float = 100000,
        commission: float = 0.0,        # Percentual por trade
        slippage: float = 0.0,          # Percentual de slippage
        position_size: float = 1.0,     # Fração do capital por trade
        allow_short: bool = False       # Permitir venda a descoberto
    ):
        self.initial_capital = initial_capital
        self.commission = commission
        self.slippage = slippage
        self.position_size = position_size
        self.allow_short = allow_short
        
        self.reset()
    
    def reset(self):
        """Reseta estado do backtest"""
        self.capital = self.initial_capital
        self.position = 0
        self.trades: list[Trade] = []
        self.equity_curve: list[float] = []
        self.current_trade: Optional[Trade] = None
    
    def run(self, data: pd.DataFrame, strategy) -> pd.DataFrame:
        """
        Executa o backtest.
        
        Args:
            data: DataFrame com OHLCV
            strategy: Instância de Strategy
        
        Returns:
            DataFrame com resultados do backtest
        """
        self.reset()
        
        # Gera sinais
        signals_df = strategy.generate_signals(data)
        
        # Simula trades
        results = []
        
        for i, row in signals_df.iterrows():
            signal = row["signal"]
            price = row["close"]
            date = row["date"]
            
            # Aplica slippage
            if signal == 1:
                execution_price = price * (1 + self.slippage)
            elif signal == -1:
                execution_price = price * (1 - self.slippage)
            else:
                execution_price = price
            
            # Processa sinais
            if signal == 1 and self.position == 0:
                # Abre posição comprada
                self._open_position(date, execution_price)
            
            elif signal == -1 and self.position > 0:
                # Fecha posição comprada
                self._close_position(date, execution_price)
            
            elif signal == -1 and self.position == 0 and self.allow_short:
                # Abre posição vendida (short)
                self._open_short(date, execution_price)
            
            elif signal == 1 and self.position < 0:
                # Fecha posição vendida
                self._close_short(date, execution_price)
            
            # Calcula equity
            if self.position > 0:
                equity = self.capital + (self.position * price)
            elif self.position < 0:
                equity = self.capital - (abs(self.position) * price) + (abs(self.position) * self.current_trade.entry_price * 2)
            else:
                equity = self.capital
            
            self.equity_curve.append(equity)
            
            results.append({
                "date": date,
                "close": price,
                "signal": signal,
                "position": self.position,
                "capital": self.capital,
                "equity": equity
            })
        
        # Fecha posição aberta no final
        if self.position != 0:
            last_row = signals_df.iloc[-1]
            self._close_position(last_row["date"], last_row["close"])
        
        return pd.DataFrame(results)
    
    def _open_position(self, date, price):
        """Abre posição comprada"""
        investment = self.capital * self.position_size
        commission_cost = investment * self.commission
        shares = int((investment - commission_cost) / price)
        
        if shares > 0:
            cost = shares * price + commission_cost
            self.capital -= cost
            self.position = shares
            self.current_trade = Trade(
                entry_date=date,
                entry_price=price,
                shares=shares
            )
    
    def _close_position(self, date, price):
        """Fecha posição comprada"""
        if self.position > 0 and self.current_trade:
            revenue = self.position * price
            commission_cost = revenue * self.commission
            net_revenue = revenue - commission_cost
            
            self.capital += net_revenue
            
            self.current_trade.exit_date = date
            self.current_trade.exit_price = price
            self.current_trade.pnl = net_revenue - (self.current_trade.shares * self.current_trade.entry_price)
            self.current_trade.pnl_percent = (price / self.current_trade.entry_price - 1) * 100
            
            self.trades.append(self.current_trade)
            self.position = 0
            self.current_trade = None
    
    def _open_short(self, date, price):
        """Abre posição vendida"""
        investment = self.capital * self.position_size
        commission_cost = investment * self.commission
        shares = int((investment - commission_cost) / price)
        
        if shares > 0:
            self.position = -shares
            self.current_trade = Trade(
                entry_date=date,
                entry_price=price,
                shares=shares
            )
    
    def _close_short(self, date, price):
        """Fecha posição vendida"""
        if self.position < 0 and self.current_trade:
            cost = abs(self.position) * price
            commission_cost = cost * self.commission
            
            pnl = (self.current_trade.entry_price - price) * abs(self.position) - commission_cost
            self.capital += pnl
            
            self.current_trade.exit_date = date
            self.current_trade.exit_price = price
            self.current_trade.pnl = pnl
            self.current_trade.pnl_percent = (self.current_trade.entry_price / price - 1) * 100
            
            self.trades.append(self.current_trade)
            self.position = 0
            self.current_trade = None
    
    def get_trades_df(self) -> pd.DataFrame:
        """Retorna DataFrame com todos os trades"""
        if not self.trades:
            return pd.DataFrame()
        
        trades_data = []
        for t in self.trades:
            trades_data.append({
                "entry_date": t.entry_date,
                "entry_price": t.entry_price,
                "exit_date": t.exit_date,
                "exit_price": t.exit_price,
                "shares": t.shares,
                "pnl": t.pnl,
                "pnl_percent": t.pnl_percent
            })
        
        return pd.DataFrame(trades_data)

# src/backtest/metrics.py
import pandas as pd
import numpy as np
from typing import Optional

class BacktestMetrics:
    """Calcula métricas de performance do backtest"""
    
    def __init__(
        self,
        results_df: pd.DataFrame,
        trades_df: pd.DataFrame,
        initial_capital: float,
        risk_free_rate: float = 0.10  # Taxa livre de risco anual (SELIC)
    ):
        self.results = results_df
        self.trades = trades_df
        self.initial_capital = initial_capital
        self.risk_free_rate = risk_free_rate
    
    def calculate_all(self) -> dict:
        """Calcula todas as métricas"""
        return {
            "total_return": self.total_return(),
            "total_return_pct": self.total_return_pct(),
            "annualized_return": self.annualized_return(),
            "volatility": self.volatility(),
            "sharpe_ratio": self.sharpe_ratio(),
            "sortino_ratio": self.sortino_ratio(),
            "max_drawdown": self.max_drawdown(),
            "max_drawdown_duration": self.max_drawdown_duration(),
            "total_trades": self.total_trades(),
            "winning_trades": self.winning_trades(),
            "losing_trades": self.losing_trades(),
            "win_rate": self.win_rate(),
            "avg_win": self.avg_win(),
            "avg_loss": self.avg_loss(),
            "profit_factor": self.profit_factor(),
            "avg_trade_duration": self.avg_trade_duration(),
            "calmar_ratio": self.calmar_ratio(),
        }
    
    def total_return(self) -> float:
        """Retorno total em R$"""
        final_equity = self.results["equity"].iloc[-1]
        return final_equity - self.initial_capital
    
    def total_return_pct(self) -> float:
        """Retorno total percentual"""
        return (self.total_return() / self.initial_capital) * 100
    
    def annualized_return(self) -> float:
        """Retorno anualizado"""
        total_days = (self.results["date"].iloc[-1] - self.results["date"].iloc[0]).days
        total_years = total_days / 365.25
        
        if total_years <= 0:
            return 0
        
        final_equity = self.results["equity"].iloc[-1]
        return ((final_equity / self.initial_capital) ** (1 / total_years) - 1) * 100
    
    def volatility(self) -> float:
        """Volatilidade anualizada dos retornos"""
        returns = self.results["equity"].pct_change().dropna()
        return returns.std() * np.sqrt(252) * 100
    
    def sharpe_ratio(self) -> float:
        """Índice de Sharpe"""
        annualized_ret = self.annualized_return() / 100
        vol = self.volatility() / 100
        
        if vol == 0:
            return 0
        
        return (annualized_ret - self.risk_free_rate) / vol
    
    def sortino_ratio(self) -> float:
        """Índice de Sortino (considera apenas volatilidade negativa)"""
        returns = self.results["equity"].pct_change().dropna()
        negative_returns = returns[returns < 0]
        
        downside_std = negative_returns.std() * np.sqrt(252)
        
        if downside_std == 0:
            return 0
        
        annualized_ret = self.annualized_return() / 100
        return (annualized_ret - self.risk_free_rate) / downside_std
    
    def max_drawdown(self) -> float:
        """Máximo drawdown percentual"""
        equity = self.results["equity"]
        peak = equity.expanding().max()
        drawdown = (equity - peak) / peak
        return drawdown.min() * 100
    
    def max_drawdown_duration(self) -> int:
        """Duração do máximo drawdown em dias"""
        equity = self.results["equity"]
        peak = equity.expanding().max()
        
        in_drawdown = equity < peak
        
        if not in_drawdown.any():
            return 0
        
        # Encontra períodos de drawdown
        drawdown_periods = []
        current_start = None
        
        for i, is_dd in enumerate(in_drawdown):
            if is_dd and current_start is None:
                current_start = i
            elif not is_dd and current_start is not None:
                drawdown_periods.append(i - current_start)
                current_start = None
        
        if current_start is not None:
            drawdown_periods.append(len(in_drawdown) - current_start)
        
        return max(drawdown_periods) if drawdown_periods else 0
    
    def total_trades(self) -> int:
        """Número total de trades"""
        return len(self.trades)
    
    def winning_trades(self) -> int:
        """Número de trades vencedores"""
        if self.trades.empty:
            return 0
        return len(self.trades[self.trades["pnl"] > 0])
    
    def losing_trades(self) -> int:
        """Número de trades perdedores"""
        if self.trades.empty:
            return 0
        return len(self.trades[self.trades["pnl"] <= 0])
    
    def win_rate(self) -> float:
        """Taxa de acerto percentual"""
        total = self.total_trades()
        if total == 0:
            return 0
        return (self.winning_trades() / total) * 100
    
    def avg_win(self) -> float:
        """Lucro médio dos trades vencedores"""
        if self.trades.empty:
            return 0
        winners = self.trades[self.trades["pnl"] > 0]
        return winners["pnl"].mean() if not winners.empty else 0
    
    def avg_loss(self) -> float:
        """Prejuízo médio dos trades perdedores"""
        if self.trades.empty:
            return 0
        losers = self.trades[self.trades["pnl"] <= 0]
        return losers["pnl"].mean() if not losers.empty else 0
    
    def profit_factor(self) -> float:
        """Fator de lucro (ganhos / perdas)"""
        if self.trades.empty:
            return 0
        
        winners = self.trades[self.trades["pnl"] > 0]["pnl"].sum()
        losers = abs(self.trades[self.trades["pnl"] <= 0]["pnl"].sum())
        
        if losers == 0:
            return float("inf") if winners > 0 else 0
        
        return winners / losers
    
    def avg_trade_duration(self) -> float:
        """Duração média dos trades em dias"""
        if self.trades.empty:
            return 0
        
        durations = (self.trades["exit_date"] - self.trades["entry_date"]).dt.days
        return durations.mean()
    
    def calmar_ratio(self) -> float:
        """Índice Calmar (retorno anualizado / max drawdown)"""
        max_dd = abs(self.max_drawdown())
        if max_dd == 0:
            return 0
        return self.annualized_return() / max_dd
    
    def print_report(self):
        """Imprime relatório formatado"""
        metrics = self.calculate_all()
        
        print("=" * 60)
        print("RELATÓRIO DE BACKTEST")
        print("=" * 60)
        
        print("\n--- Retorno ---")
        print(f"Retorno Total: R$ {metrics['total_return']:,.2f} ({metrics['total_return_pct']:.2f}%)")
        print(f"Retorno Anualizado: {metrics['annualized_return']:.2f}%")
        
        print("\n--- Risco ---")
        print(f"Volatilidade Anualizada: {metrics['volatility']:.2f}%")
        print(f"Máximo Drawdown: {metrics['max_drawdown']:.2f}%")
        print(f"Duração Max Drawdown: {metrics['max_drawdown_duration']} dias")
        
        print("\n--- Índices ---")
        print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
        print(f"Sortino Ratio: {metrics['sortino_ratio']:.2f}")
        print(f"Calmar Ratio: {metrics['calmar_ratio']:.2f}")
        
        print("\n--- Trades ---")
        print(f"Total de Trades: {metrics['total_trades']}")
        print(f"Trades Vencedores: {metrics['winning_trades']}")
        print(f"Trades Perdedores: {metrics['losing_trades']}")
        print(f"Win Rate: {metrics['win_rate']:.1f}%")
        print(f"Profit Factor: {metrics['profit_factor']:.2f}")
        print(f"Lucro Médio: R$ {metrics['avg_win']:,.2f}")
        print(f"Prejuízo Médio: R$ {metrics['avg_loss']:,.2f}")
        print(f"Duração Média: {metrics['avg_trade_duration']:.1f} dias")
        
        print("=" * 60)

# src/utils/plots.py
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np

def plot_backtest_results(
    results_df: pd.DataFrame,
    trades_df: pd.DataFrame,
    strategy_name: str,
    save_path: str = None
):
    """Plota resultados completos do backtest"""
    
    fig, axes = plt.subplots(3, 1, figsize=(14, 12))
    
    # Gráfico 1: Equity Curve
    ax1 = axes[0]
    ax1.plot(results_df["date"], results_df["equity"], label="Equity", color="blue")
    ax1.axhline(y=results_df["equity"].iloc[0], color="gray", linestyle="--", alpha=0.5)
    ax1.set_title(f"Equity Curve - {strategy_name}")
    ax1.set_xlabel("Data")
    ax1.set_ylabel("Equity (R$)")
    ax1.legend()
    ax1.grid(True, alpha=0.3)
    
    # Marcar trades
    if not trades_df.empty:
        for _, trade in trades_df.iterrows():
            color = "green" if trade["pnl"] > 0 else "red"
            ax1.axvspan(trade["entry_date"], trade["exit_date"], alpha=0.1, color=color)
    
    # Gráfico 2: Drawdown
    ax2 = axes[1]
    equity = results_df["equity"]
    peak = equity.expanding().max()
    drawdown = (equity - peak) / peak * 100
    
    ax2.fill_between(results_df["date"], drawdown, 0, alpha=0.3, color="red")
    ax2.plot(results_df["date"], drawdown, color="red", linewidth=0.5)
    ax2.set_title("Drawdown")
    ax2.set_xlabel("Data")
    ax2.set_ylabel("Drawdown (%)")
    ax2.grid(True, alpha=0.3)
    
    # Gráfico 3: Preço + Sinais
    ax3 = axes[2]
    ax3.plot(results_df["date"], results_df["close"], label="Preço", color="black", alpha=0.7)
    
    # Sinais de compra
    buys = results_df[results_df["signal"] == 1]
    ax3.scatter(buys["date"], buys["close"], marker="^", color="green", s=100, label="Compra", zorder=5)
    
    # Sinais de venda
    sells = results_df[results_df["signal"] == -1]
    ax3.scatter(sells["date"], sells["close"], marker="v", color="red", s=100, label="Venda", zorder=5)
    
    ax3.set_title("Preço e Sinais")
    ax3.set_xlabel("Data")
    ax3.set_ylabel("Preço (R$)")
    ax3.legend()
    ax3.grid(True, alpha=0.3)
    
    plt.tight_layout()
    
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches="tight")
    
    plt.show()


def plot_trades_distribution(trades_df: pd.DataFrame, save_path: str = None):
    """Plota distribuição dos resultados dos trades"""
    
    if trades_df.empty:
        print("Sem trades para plotar")
        return
    
    fig, axes = plt.subplots(1, 2, figsize=(12, 5))
    
    # Histograma de PnL
    ax1 = axes[0]
    colors = ["green" if x > 0 else "red" for x in trades_df["pnl"]]
    ax1.bar(range(len(trades_df)), trades_df["pnl"], color=colors)
    ax1.axhline(y=0, color="black", linestyle="-", linewidth=0.5)
    ax1.set_title("PnL por Trade")
    ax1.set_xlabel("Trade #")
    ax1.set_ylabel("PnL (R$)")
    
    # Histograma de retornos
    ax2 = axes[1]
    ax2.hist(trades_df["pnl_percent"], bins=20, edgecolor="black", alpha=0.7)
    ax2.axvline(x=0, color="red", linestyle="--")
    ax2.axvline(x=trades_df["pnl_percent"].mean(), color="blue", linestyle="--", label=f"Média: {trades_df['pnl_percent'].mean():.1f}%")
    ax2.set_title("Distribuição de Retornos")
    ax2.set_xlabel("Retorno (%)")
    ax2.set_ylabel("Frequência")
    ax2.legend()
    
    plt.tight_layout()
    
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches="tight")
    
    plt.show()

# main.py
from src.data.fetcher import BrapiDataFetcher
from src.strategies.moving_average import MovingAverageCrossover, ExponentialMovingAverageCrossover
from src.strategies.rsi import RSIStrategy
from src.backtest.engine import BacktestEngine
from src.backtest.metrics import BacktestMetrics
from src.utils.plots import plot_backtest_results, plot_trades_distribution

def run_backtest(
    ticker: str,
    token: str,
    strategy,
    initial_capital: float = 100000,
    commission: float = 0.001
):
    """Executa backtest completo"""
    
    print(f"\n{'='*60}")
    print(f"Backtesting: {strategy.name}")
    print(f"Ticker: {ticker}")
    print(f"Capital Inicial: R$ {initial_capital:,.2f}")
    print(f"{'='*60}")
    
    # 1. Buscar dados
    print("\n[1/4] Buscando dados históricos...")
    fetcher = BrapiDataFetcher(token)
    data = fetcher.get_historical_data(ticker, "5y", "1d")
    print(f"    Período: {data['date'].min()} a {data['date'].max()}")
    print(f"    Total de registros: {len(data)}")
    
    # 2. Executar backtest
    print("\n[2/4] Executando backtest...")
    engine = BacktestEngine(
        initial_capital=initial_capital,
        commission=commission,
        slippage=0.001
    )
    results = engine.run(data, strategy)
    trades = engine.get_trades_df()
    print(f"    Total de trades: {len(trades)}")
    
    # 3. Calcular métricas
    print("\n[3/4] Calculando métricas...")
    metrics = BacktestMetrics(results, trades, initial_capital)
    metrics.print_report()
    
    # 4. Plotar resultados
    print("\n[4/4] Gerando gráficos...")
    plot_backtest_results(results, trades, f"{strategy.name} - {ticker}")
    plot_trades_distribution(trades)
    
    return results, trades, metrics


def compare_strategies(
    ticker: str,
    token: str,
    strategies: list,
    initial_capital: float = 100000
):
    """Compara múltiplas estratégias"""
    
    print("\n" + "="*80)
    print("COMPARAÇÃO DE ESTRATÉGIAS")
    print("="*80)
    
    results = []
    
    for strategy in strategies:
        fetcher = BrapiDataFetcher(token)
        data = fetcher.get_historical_data(ticker, "5y", "1d")
        
        engine = BacktestEngine(initial_capital=initial_capital)
        bt_results = engine.run(data, strategy)
        trades = engine.get_trades_df()
        
        metrics = BacktestMetrics(bt_results, trades, initial_capital)
        metrics_dict = metrics.calculate_all()
        metrics_dict["strategy"] = strategy.name
        results.append(metrics_dict)
    
    # Tabela comparativa
    import pandas as pd
    comparison = pd.DataFrame(results)
    comparison = comparison.set_index("strategy")
    
    print("\n--- Comparação de Retorno ---")
    print(comparison[["total_return_pct", "annualized_return", "sharpe_ratio", "max_drawdown"]].round(2))
    
    print("\n--- Comparação de Trades ---")
    print(comparison[["total_trades", "win_rate", "profit_factor"]].round(2))
    
    return comparison


# Exemplo de uso
if __name__ == "__main__":
    TOKEN = "seu_token_brapi"
    TICKER = "PETR4"
    
    # Testar estratégia de médias móveis
    strategy_ma = MovingAverageCrossover(short_window=20, long_window=50)
    run_backtest(TICKER, TOKEN, strategy_ma)
    
    # Testar estratégia RSI
    strategy_rsi = RSIStrategy(period=14, oversold=30, overbought=70)
    run_backtest(TICKER, TOKEN, strategy_rsi)
    
    # Comparar estratégias
    strategies = [
        MovingAverageCrossover(20, 50),
        MovingAverageCrossover(10, 30),
        ExponentialMovingAverageCrossover(12, 26),
        RSIStrategy(14, 30, 70),
        RSIStrategy(7, 25, 75),
    ]
    
    comparison = compare_strategies(TICKER, TOKEN, strategies)

# src/optimization/grid_search.py
import itertools
from typing import Callable
import pandas as pd

def grid_search_optimization(
    data: pd.DataFrame,
    strategy_class: type,
    param_grid: dict,
    initial_capital: float = 100000,
    metric: str = "sharpe_ratio"
) -> pd.DataFrame:
    """
    Busca os melhores parâmetros via grid search.
    
    Args:
        data: DataFrame com dados históricos
        strategy_class: Classe da estratégia
        param_grid: Dicionário com parâmetros e valores a testar
        initial_capital: Capital inicial
        metric: Métrica para otimização
    
    Returns:
        DataFrame com resultados ordenados pela métrica
    """
    from src.backtest.engine import BacktestEngine
    from src.backtest.metrics import BacktestMetrics
    
    # Gerar todas as combinações
    param_names = list(param_grid.keys())
    param_values = list(param_grid.values())
    combinations = list(itertools.product(*param_values))
    
    print(f"Testando {len(combinations)} combinações...")
    
    results = []
    
    for i, combo in enumerate(combinations):
        params = dict(zip(param_names, combo))
        
        # Criar estratégia com parâmetros
        strategy = strategy_class(**params)
        
        # Executar backtest
        engine = BacktestEngine(initial_capital=initial_capital)
        bt_results = engine.run(data, strategy)
        trades = engine.get_trades_df()
        
        # Calcular métricas
        metrics = BacktestMetrics(bt_results, trades, initial_capital)
        metrics_dict = metrics.calculate_all()
        metrics_dict.update(params)
        
        results.append(metrics_dict)
        
        # Progresso
        if (i + 1) % 10 == 0:
            print(f"  {i + 1}/{len(combinations)} completo")
    
    # Ordenar por métrica
    results_df = pd.DataFrame(results)
    results_df = results_df.sort_values(metric, ascending=False)
    
    return results_df


# Exemplo de uso
if __name__ == "__main__":
    from src.data.fetcher import BrapiDataFetcher
    from src.strategies.moving_average import MovingAverageCrossover
    
    TOKEN = "seu_token_brapi"
    
    # Buscar dados
    fetcher = BrapiDataFetcher(TOKEN)
    data = fetcher.get_historical_data("PETR4", "5y", "1d")
    
    # Definir grid de parâmetros
    param_grid = {
        "short_window": [5, 10, 15, 20, 25],
        "long_window": [30, 40, 50, 60, 80, 100]
    }
    
    # Executar otimização
    results = grid_search_optimization(
        data=data,
        strategy_class=MovingAverageCrossover,
        param_grid=param_grid,
        metric="sharpe_ratio"
    )
    
    print("\n--- Top 10 Combinações ---")
    print(results[["short_window", "long_window", "sharpe_ratio", "total_return_pct", "win_rate"]].head(10))

# Técnica: Walk-Forward Analysis

def walk_forward_analysis(
    data: pd.DataFrame,
    strategy_class: type,
    param_grid: dict,
    train_size: float = 0.7,
    n_splits: int = 5
):
    """
    Validação walk-forward para evitar overfitting.
    
    Divide dados em janelas de treino/teste deslizantes.
    """
    results = []
    
    split_size = len(data) // n_splits
    
    for i in range(n_splits - 1):
        # Definir janelas
        train_start = i * split_size
        train_end = train_start + int(split_size * train_size)
        test_end = (i + 2) * split_size
        
        train_data = data.iloc[train_start:train_end].copy()
        test_data = data.iloc[train_end:test_end].copy()
        
        # Otimizar em treino
        best_params = optimize_on_data(train_data, strategy_class, param_grid)
        
        # Testar em dados fora da amostra
        test_metrics = test_with_params(test_data, strategy_class, best_params)
        
        results.append({
            "split": i,
            "train_period": f"{train_data['date'].min()} - {train_data['date'].max()}",
            "test_period": f"{test_data['date'].min()} - {test_data['date'].max()}",
            **test_metrics
        })
    
    return pd.DataFrame(results)

Antes de confiar no backtest:

□ Dados suficientes (mínimo 2-3 anos)
□ Incluiu custos de transação
□ Considerou slippage
□ Testou em múltiplos ativos
□ Fez walk-forward analysis
□ Verificou se faz sentido econômico
□ Comparou com buy-and-hold
□ Analisou drawdowns detalhadamente