Machine Learning para Prever Ações com Python 2026: Tutorial Completo

┌─────────────────────────────────────────────────────────────┐
│                    ROADMAP DO TUTORIAL                       │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  1. Fundamentos                                              │
│     └─ Por que ML para ações é difícil                      │
│                                                              │
│  2. Preparação de Dados                                      │
│     └─ Buscar dados da brapi                                │
│     └─ Feature engineering                                  │
│                                                              │
│  3. Modelos Clássicos                                        │
│     └─ Random Forest                                        │
│     └─ Gradient Boosting                                    │
│                                                              │
│  4. Deep Learning                                            │
│     └─ LSTM para séries temporais                           │
│                                                              │
│  5. Avaliação                                                │
│     └─ Métricas e backtesting                               │
│                                                              │
│  6. Limitações                                               │
│     └─ Por que modelos falham                               │
│                                                              │
└─────────────────────────────────────────────────────────────┘

ML PODE:
├─ Identificar padrões em dados históricos
├─ Automatizar análise de indicadores
├─ Classificar ações por características
└─ Gerar features para análise

ML NÃO PODE:
├─ Prever com certeza movimentos futuros
├─ Antecipar notícias e eventos
├─ Garantir lucros consistentes
└─ Substituir análise fundamentalista

pip install requests pandas numpy scikit-learn tensorflow matplotlib seaborn

ml-stocks/
├── src/
│   ├── data/
│   │   ├── fetcher.py           # Buscar dados
│   │   └── preprocessing.py     # Preparação
│   ├── features/
│   │   └── technical.py         # Indicadores técnicos
│   ├── models/
│   │   ├── random_forest.py     # Modelo RF
│   │   ├── xgboost_model.py     # Modelo XGB
│   │   └── lstm.py              # Modelo LSTM
│   └── evaluation/
│       └── metrics.py           # Avaliação
├── notebooks/
│   └── exploration.ipynb
├── main.py
└── requirements.txt

# src/data/fetcher.py
import requests
import pandas as pd
from typing import Optional

class BrapiMLDataFetcher:
    """Busca dados para Machine Learning"""
    
    BASE_URL = "https://brapi.dev/api"
    
    def __init__(self, token: str):
        self.token = token
    
    def get_stock_data(
        self,
        ticker: str,
        range_period: str = "5y",
        include_fundamentals: bool = True
    ) -> pd.DataFrame:
        """Busca dados completos para ML"""
        
        url = f"{self.BASE_URL}/quote/{ticker}"
        params = {
            "token": self.token,
            "range": range_period,
            "interval": "1d",
            "fundamental": str(include_fundamentals).lower()
        }
        
        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}")
        
        result = data["results"][0]
        
        # Dados históricos
        historical = result.get("historicalDataPrice", [])
        df = pd.DataFrame(historical)
        df["date"] = pd.to_datetime(df["date"], unit="s")
        df = df.sort_values("date").reset_index(drop=True)
        
        # Adicionar ticker
        df["ticker"] = ticker
        
        # Adicionar dados fundamentais se disponíveis
        if include_fundamentals:
            if "defaultKeyStatistics" in result:
                stats = result["defaultKeyStatistics"]
                df["pe_ratio"] = stats.get("forwardPE", None)
                df["pb_ratio"] = stats.get("priceToBook", None)
                df["beta"] = stats.get("beta", None)
        
        return df
    
    def get_multiple_stocks(
        self,
        tickers: list[str],
        range_period: str = "5y"
    ) -> pd.DataFrame:
        """Busca dados de múltiplas ações"""
        
        all_data = []
        for ticker in tickers:
            try:
                df = self.get_stock_data(ticker, range_period)
                all_data.append(df)
                print(f"✓ {ticker}: {len(df)} registros")
            except Exception as e:
                print(f"✗ {ticker}: {e}")
        
        return pd.concat(all_data, ignore_index=True) if all_data else pd.DataFrame()


# Uso
if __name__ == "__main__":
    fetcher = BrapiMLDataFetcher("seu_token")
    df = fetcher.get_stock_data("PETR4", "3y")
    print(df.head())

# src/features/technical.py
import pandas as pd
import numpy as np

class TechnicalFeatures:
    """Cria features técnicas para ML"""
    
    @staticmethod
    def add_all_features(df: pd.DataFrame) -> pd.DataFrame:
        """Adiciona todas as features técnicas"""
        df = df.copy()
        
        # Retornos
        df = TechnicalFeatures.add_returns(df)
        
        # Médias móveis
        df = TechnicalFeatures.add_moving_averages(df)
        
        # Volatilidade
        df = TechnicalFeatures.add_volatility(df)
        
        # RSI
        df = TechnicalFeatures.add_rsi(df)
        
        # MACD
        df = TechnicalFeatures.add_macd(df)
        
        # Bollinger Bands
        df = TechnicalFeatures.add_bollinger_bands(df)
        
        # Volume features
        df = TechnicalFeatures.add_volume_features(df)
        
        # Lag features
        df = TechnicalFeatures.add_lag_features(df)
        
        # Target
        df = TechnicalFeatures.add_target(df)
        
        return df
    
    @staticmethod
    def add_returns(df: pd.DataFrame) -> pd.DataFrame:
        """Retornos em diferentes períodos"""
        df["return_1d"] = df["close"].pct_change(1)
        df["return_5d"] = df["close"].pct_change(5)
        df["return_10d"] = df["close"].pct_change(10)
        df["return_20d"] = df["close"].pct_change(20)
        return df
    
    @staticmethod
    def add_moving_averages(df: pd.DataFrame) -> pd.DataFrame:
        """Médias móveis e relações"""
        for window in [5, 10, 20, 50, 100, 200]:
            df[f"sma_{window}"] = df["close"].rolling(window).mean()
            df[f"ema_{window}"] = df["close"].ewm(span=window, adjust=False).mean()
        
        # Posição relativa às médias
        df["close_vs_sma20"] = df["close"] / df["sma_20"] - 1
        df["close_vs_sma50"] = df["close"] / df["sma_50"] - 1
        df["close_vs_sma200"] = df["close"] / df["sma_200"] - 1
        
        # Cruzamentos
        df["sma20_vs_sma50"] = df["sma_20"] / df["sma_50"] - 1
        df["sma50_vs_sma200"] = df["sma_50"] / df["sma_200"] - 1
        
        return df
    
    @staticmethod
    def add_volatility(df: pd.DataFrame) -> pd.DataFrame:
        """Volatilidade histórica"""
        df["volatility_10d"] = df["return_1d"].rolling(10).std() * np.sqrt(252)
        df["volatility_20d"] = df["return_1d"].rolling(20).std() * np.sqrt(252)
        df["volatility_60d"] = df["return_1d"].rolling(60).std() * np.sqrt(252)
        
        # ATR (Average True Range)
        df["tr"] = np.maximum(
            df["high"] - df["low"],
            np.maximum(
                abs(df["high"] - df["close"].shift(1)),
                abs(df["low"] - df["close"].shift(1))
            )
        )
        df["atr_14"] = df["tr"].rolling(14).mean()
        
        return df
    
    @staticmethod
    def add_rsi(df: pd.DataFrame, periods: list = [7, 14, 21]) -> pd.DataFrame:
        """RSI em diferentes períodos"""
        for period in periods:
            delta = df["close"].diff()
            gain = delta.where(delta > 0, 0)
            loss = (-delta).where(delta < 0, 0)
            avg_gain = gain.rolling(window=period).mean()
            avg_loss = loss.rolling(window=period).mean()
            rs = avg_gain / avg_loss
            df[f"rsi_{period}"] = 100 - (100 / (1 + rs))
        return df
    
    @staticmethod
    def add_macd(df: pd.DataFrame) -> pd.DataFrame:
        """MACD e histograma"""
        df["ema_12"] = df["close"].ewm(span=12, adjust=False).mean()
        df["ema_26"] = df["close"].ewm(span=26, adjust=False).mean()
        df["macd"] = df["ema_12"] - df["ema_26"]
        df["macd_signal"] = df["macd"].ewm(span=9, adjust=False).mean()
        df["macd_hist"] = df["macd"] - df["macd_signal"]
        return df
    
    @staticmethod
    def add_bollinger_bands(df: pd.DataFrame, window: int = 20) -> pd.DataFrame:
        """Bollinger Bands"""
        df["bb_middle"] = df["close"].rolling(window).mean()
        df["bb_std"] = df["close"].rolling(window).std()
        df["bb_upper"] = df["bb_middle"] + 2 * df["bb_std"]
        df["bb_lower"] = df["bb_middle"] - 2 * df["bb_std"]
        df["bb_width"] = (df["bb_upper"] - df["bb_lower"]) / df["bb_middle"]
        df["bb_position"] = (df["close"] - df["bb_lower"]) / (df["bb_upper"] - df["bb_lower"])
        return df
    
    @staticmethod
    def add_volume_features(df: pd.DataFrame) -> pd.DataFrame:
        """Features de volume"""
        df["volume_sma_20"] = df["volume"].rolling(20).mean()
        df["volume_ratio"] = df["volume"] / df["volume_sma_20"]
        df["obv"] = (np.sign(df["close"].diff()) * df["volume"]).cumsum()
        return df
    
    @staticmethod
    def add_lag_features(df: pd.DataFrame, lags: list = [1, 2, 3, 5, 10]) -> pd.DataFrame:
        """Features com lag"""
        for lag in lags:
            df[f"close_lag_{lag}"] = df["close"].shift(lag)
            df[f"return_lag_{lag}"] = df["return_1d"].shift(lag)
        return df
    
    @staticmethod
    def add_target(
        df: pd.DataFrame,
        horizon: int = 5,
        target_type: str = "classification"
    ) -> pd.DataFrame:
        """
        Adiciona variável target.
        
        target_type:
        - "classification": 1 se subiu, 0 se caiu
        - "regression": retorno percentual
        """
        future_return = df["close"].shift(-horizon) / df["close"] - 1
        
        if target_type == "classification":
            df["target"] = (future_return > 0).astype(int)
        else:
            df["target"] = future_return
        
        return df


# Uso
if __name__ == "__main__":
    from src.data.fetcher import BrapiMLDataFetcher
    
    fetcher = BrapiMLDataFetcher("seu_token")
    df = fetcher.get_stock_data("PETR4")
    
    df = TechnicalFeatures.add_all_features(df)
    print(f"Features criadas: {len(df.columns)}")
    print(df.columns.tolist())

# src/models/random_forest.py
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, confusion_matrix
from sklearn.preprocessing import StandardScaler

class RandomForestStockModel:
    """Random Forest para classificação de ações"""
    
    def __init__(
        self,
        n_estimators: int = 100,
        max_depth: int = 10,
        min_samples_split: int = 10,
        random_state: int = 42
    ):
        self.model = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            min_samples_split=min_samples_split,
            random_state=random_state,
            n_jobs=-1
        )
        self.scaler = StandardScaler()
        self.feature_columns = None
    
    def prepare_data(self, df: pd.DataFrame) -> tuple:
        """Prepara dados para treino"""
        
        # Remover linhas com NaN
        df = df.dropna()
        
        # Definir features (excluir colunas não-features)
        exclude_cols = ["date", "ticker", "target", "open", "high", "low", "close", "volume"]
        self.feature_columns = [c for c in df.columns if c not in exclude_cols]
        
        X = df[self.feature_columns].values
        y = df["target"].values
        
        return X, y, df["date"].values
    
    def train_test_split_temporal(
        self,
        X: np.ndarray,
        y: np.ndarray,
        dates: np.ndarray,
        test_size: float = 0.2
    ) -> tuple:
        """Split temporal (não aleatório)"""
        
        split_idx = int(len(X) * (1 - test_size))
        
        X_train, X_test = X[:split_idx], X[split_idx:]
        y_train, y_test = y[:split_idx], y[split_idx:]
        dates_test = dates[split_idx:]
        
        return X_train, X_test, y_train, y_test, dates_test
    
    def fit(self, X_train: np.ndarray, y_train: np.ndarray):
        """Treina o modelo"""
        
        # Normalizar features
        X_train_scaled = self.scaler.fit_transform(X_train)
        
        # Treinar
        self.model.fit(X_train_scaled, y_train)
        
        return self
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Faz previsões"""
        X_scaled = self.scaler.transform(X)
        return self.model.predict(X_scaled)
    
    def predict_proba(self, X: np.ndarray) -> np.ndarray:
        """Retorna probabilidades"""
        X_scaled = self.scaler.transform(X)
        return self.model.predict_proba(X_scaled)
    
    def evaluate(self, X_test: np.ndarray, y_test: np.ndarray) -> dict:
        """Avalia o modelo"""
        
        y_pred = self.predict(X_test)
        
        return {
            "accuracy": accuracy_score(y_test, y_pred),
            "precision": precision_score(y_test, y_pred, zero_division=0),
            "recall": recall_score(y_test, y_pred, zero_division=0),
            "f1": f1_score(y_test, y_pred, zero_division=0),
            "confusion_matrix": confusion_matrix(y_test, y_pred)
        }
    
    def get_feature_importance(self) -> pd.DataFrame:
        """Retorna importância das features"""
        
        importance = pd.DataFrame({
            "feature": self.feature_columns,
            "importance": self.model.feature_importances_
        })
        
        return importance.sort_values("importance", ascending=False)
    
    def cross_validate_temporal(
        self,
        X: np.ndarray,
        y: np.ndarray,
        n_splits: int = 5
    ) -> dict:
        """Validação cruzada temporal"""
        
        tscv = TimeSeriesSplit(n_splits=n_splits)
        
        scores = {
            "accuracy": [],
            "precision": [],
            "recall": [],
            "f1": []
        }
        
        for train_idx, test_idx in tscv.split(X):
            X_train, X_test = X[train_idx], X[test_idx]
            y_train, y_test = y[train_idx], y[test_idx]
            
            # Treinar
            X_train_scaled = self.scaler.fit_transform(X_train)
            self.model.fit(X_train_scaled, y_train)
            
            # Avaliar
            metrics = self.evaluate(X_test, y_test)
            
            for key in scores:
                scores[key].append(metrics[key])
        
        # Médias
        return {k: np.mean(v) for k, v in scores.items()}


# Exemplo de uso
if __name__ == "__main__":
    from src.data.fetcher import BrapiMLDataFetcher
    from src.features.technical import TechnicalFeatures
    
    # Buscar dados
    fetcher = BrapiMLDataFetcher("seu_token")
    df = fetcher.get_stock_data("PETR4", "5y")
    
    # Criar features
    df = TechnicalFeatures.add_all_features(df)
    
    # Preparar modelo
    model = RandomForestStockModel(n_estimators=200, max_depth=8)
    X, y, dates = model.prepare_data(df)
    
    # Split temporal
    X_train, X_test, y_train, y_test, dates_test = model.train_test_split_temporal(X, y, dates)
    
    # Treinar
    model.fit(X_train, y_train)
    
    # Avaliar
    metrics = model.evaluate(X_test, y_test)
    print("\n=== Métricas ===")
    print(f"Accuracy: {metrics['accuracy']:.4f}")
    print(f"Precision: {metrics['precision']:.4f}")
    print(f"Recall: {metrics['recall']:.4f}")
    print(f"F1 Score: {metrics['f1']:.4f}")
    
    # Feature importance
    print("\n=== Top 10 Features ===")
    print(model.get_feature_importance().head(10))

# src/models/lstm.py
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.callbacks import EarlyStopping

class LSTMStockModel:
    """LSTM para previsão de séries temporais"""
    
    def __init__(
        self,
        sequence_length: int = 60,
        n_features: int = 1,
        lstm_units: int = 50,
        dropout: float = 0.2
    ):
        self.sequence_length = sequence_length
        self.n_features = n_features
        self.lstm_units = lstm_units
        self.dropout = dropout
        self.scaler = MinMaxScaler()
        self.model = None
    
    def build_model(self):
        """Constrói a arquitetura LSTM"""
        
        model = Sequential([
            LSTM(
                self.lstm_units,
                return_sequences=True,
                input_shape=(self.sequence_length, self.n_features)
            ),
            Dropout(self.dropout),
            
            LSTM(self.lstm_units, return_sequences=False),
            Dropout(self.dropout),
            
            Dense(25),
            Dense(1)
        ])
        
        model.compile(
            optimizer="adam",
            loss="mse",
            metrics=["mae"]
        )
        
        self.model = model
        return model
    
    def create_sequences(
        self,
        data: np.ndarray,
        target: np.ndarray = None
    ) -> tuple:
        """Cria sequências para LSTM"""
        
        X, y = [], []
        
        for i in range(self.sequence_length, len(data)):
            X.append(data[i - self.sequence_length:i])
            if target is not None:
                y.append(target[i])
        
        X = np.array(X)
        
        if target is not None:
            y = np.array(y)
            return X, y
        
        return X
    
    def prepare_data(
        self,
        df: pd.DataFrame,
        feature_cols: list = ["close"],
        target_col: str = "close"
    ) -> tuple:
        """Prepara dados para LSTM"""
        
        # Selecionar features
        data = df[feature_cols].values
        
        # Normalizar
        data_scaled = self.scaler.fit_transform(data)
        
        # Criar target (próximo valor)
        target = data_scaled[1:, 0] if len(feature_cols) == 1 else data_scaled[1:, feature_cols.index(target_col)]
        data_scaled = data_scaled[:-1]
        
        # Criar sequências
        X, y = self.create_sequences(data_scaled, target)
        
        self.n_features = len(feature_cols)
        
        return X, y
    
    def train(
        self,
        X_train: np.ndarray,
        y_train: np.ndarray,
        X_val: np.ndarray = None,
        y_val: np.ndarray = None,
        epochs: int = 50,
        batch_size: int = 32
    ):
        """Treina o modelo"""
        
        if self.model is None:
            self.build_model()
        
        callbacks = [
            EarlyStopping(
                monitor="val_loss" if X_val is not None else "loss",
                patience=10,
                restore_best_weights=True
            )
        ]
        
        validation_data = (X_val, y_val) if X_val is not None else None
        
        history = self.model.fit(
            X_train, y_train,
            epochs=epochs,
            batch_size=batch_size,
            validation_data=validation_data,
            callbacks=callbacks,
            verbose=1
        )
        
        return history
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Faz previsões"""
        predictions = self.model.predict(X)
        
        # Inverter normalização
        predictions_rescaled = self.scaler.inverse_transform(
            np.concatenate([predictions, np.zeros((len(predictions), self.n_features - 1))], axis=1)
        )[:, 0]
        
        return predictions_rescaled
    
    def evaluate(
        self,
        X_test: np.ndarray,
        y_test: np.ndarray,
        y_test_original: np.ndarray = None
    ) -> dict:
        """Avalia o modelo"""
        
        predictions = self.model.predict(X_test)
        
        # MSE e MAE normalizados
        mse = np.mean((predictions.flatten() - y_test) ** 2)
        mae = np.mean(np.abs(predictions.flatten() - y_test))
        
        # Direção correta (classificação implícita)
        if y_test_original is not None:
            pred_direction = np.sign(predictions.flatten()[1:] - predictions.flatten()[:-1])
            actual_direction = np.sign(y_test_original[1:] - y_test_original[:-1])
            direction_accuracy = np.mean(pred_direction == actual_direction)
        else:
            pred_direction = np.sign(predictions.flatten()[1:] - predictions.flatten()[:-1])
            actual_direction = np.sign(y_test[1:] - y_test[:-1])
            direction_accuracy = np.mean(pred_direction == actual_direction)
        
        return {
            "mse": mse,
            "mae": mae,
            "rmse": np.sqrt(mse),
            "direction_accuracy": direction_accuracy
        }


# Exemplo de uso
if __name__ == "__main__":
    from src.data.fetcher import BrapiMLDataFetcher
    
    # Buscar dados
    fetcher = BrapiMLDataFetcher("seu_token")
    df = fetcher.get_stock_data("PETR4", "5y")
    
    # Preparar modelo
    lstm = LSTMStockModel(sequence_length=60, n_features=1, lstm_units=50)
    
    # Preparar dados (usando apenas preço de fechamento)
    X, y = lstm.prepare_data(df, feature_cols=["close"])
    
    # Split
    split = int(len(X) * 0.8)
    X_train, X_test = X[:split], X[split:]
    y_train, y_test = y[:split], y[split:]
    
    # Treinar
    history = lstm.train(X_train, y_train, X_test, y_test, epochs=50)
    
    # Avaliar
    metrics = lstm.evaluate(X_test, y_test)
    print("\n=== Métricas LSTM ===")
    print(f"RMSE: {metrics['rmse']:.4f}")
    print(f"MAE: {metrics['mae']:.4f}")
    print(f"Acurácia de Direção: {metrics['direction_accuracy']:.2%}")

# main.py
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

from src.data.fetcher import BrapiMLDataFetcher
from src.features.technical import TechnicalFeatures
from src.models.random_forest import RandomForestStockModel
from src.models.lstm import LSTMStockModel

def run_complete_pipeline(ticker: str, token: str):
    """Pipeline completo de ML para ações"""
    
    print(f"\n{'='*60}")
    print(f"MACHINE LEARNING PARA {ticker}")
    print(f"{'='*60}")
    
    # 1. Buscar dados
    print("\n[1/5] Buscando dados...")
    fetcher = BrapiMLDataFetcher(token)
    df = fetcher.get_stock_data(ticker, "5y")
    print(f"    Registros: {len(df)}")
    print(f"    Período: {df['date'].min()} a {df['date'].max()}")
    
    # 2. Feature engineering
    print("\n[2/5] Criando features...")
    df = TechnicalFeatures.add_all_features(df)
    print(f"    Features criadas: {len(df.columns)}")
    
    # 3. Treinar Random Forest
    print("\n[3/5] Treinando Random Forest...")
    rf_model = RandomForestStockModel(n_estimators=200, max_depth=8)
    X, y, dates = rf_model.prepare_data(df)
    X_train, X_test, y_train, y_test, dates_test = rf_model.train_test_split_temporal(X, y, dates)
    
    rf_model.fit(X_train, y_train)
    rf_metrics = rf_model.evaluate(X_test, y_test)
    
    print(f"    Accuracy: {rf_metrics['accuracy']:.4f}")
    print(f"    F1 Score: {rf_metrics['f1']:.4f}")
    
    # 4. Treinar LSTM
    print("\n[4/5] Treinando LSTM...")
    lstm_model = LSTMStockModel(sequence_length=60, lstm_units=50)
    X_lstm, y_lstm = lstm_model.prepare_data(df[["close"]])
    
    split = int(len(X_lstm) * 0.8)
    lstm_model.train(X_lstm[:split], y_lstm[:split], X_lstm[split:], y_lstm[split:], epochs=30)
    lstm_metrics = lstm_model.evaluate(X_lstm[split:], y_lstm[split:])
    
    print(f"    RMSE: {lstm_metrics['rmse']:.4f}")
    print(f"    Acurácia Direção: {lstm_metrics['direction_accuracy']:.2%}")
    
    # 5. Comparação
    print("\n[5/5] Resumo Comparativo...")
    print(f"""
    ┌─────────────────────────────────────────────────┐
    │              COMPARAÇÃO DE MODELOS               │
    ├─────────────────────────────────────────────────┤
    │  Random Forest                                   │
    │    - Accuracy: {rf_metrics['accuracy']:.2%}                          │
    │    - Precision: {rf_metrics['precision']:.2%}                         │
    │    - F1 Score: {rf_metrics['f1']:.2%}                          │
    │                                                  │
    │  LSTM                                            │
    │    - RMSE: {lstm_metrics['rmse']:.4f}                              │
    │    - Acurácia Direção: {lstm_metrics['direction_accuracy']:.2%}               │
    └─────────────────────────────────────────────────┘
    """)
    
    # Importância das features
    print("\n=== Top 10 Features (Random Forest) ===")
    importance = rf_model.get_feature_importance()
    print(importance.head(10).to_string(index=False))
    
    return {
        "rf_metrics": rf_metrics,
        "lstm_metrics": lstm_metrics,
        "feature_importance": importance
    }


def baseline_comparison(ticker: str, token: str):
    """Compara ML com estratégia baseline (buy and hold)"""
    
    fetcher = BrapiMLDataFetcher(token)
    df = fetcher.get_stock_data(ticker, "5y")
    
    # Buy and hold return
    buy_hold_return = (df["close"].iloc[-1] / df["close"].iloc[0] - 1) * 100
    
    print(f"\n=== Baseline: Buy and Hold ===")
    print(f"Retorno total: {buy_hold_return:.2f}%")
    
    # Estratégia "sempre subir" (naive)
    df["naive_pred"] = 1  # sempre prevê alta
    df["actual"] = (df["close"].shift(-5) > df["close"]).astype(int)
    naive_accuracy = (df["naive_pred"] == df["actual"]).mean()
    
    print(f"\n=== Baseline: Sempre Prever Alta ===")
    print(f"Accuracy: {naive_accuracy:.2%}")
    
    return buy_hold_return, naive_accuracy


if __name__ == "__main__":
    TOKEN = "seu_token_brapi"
    TICKER = "PETR4"
    
    # Comparar com baseline
    baseline_comparison(TICKER, TOKEN)
    
    # Pipeline completo
    results = run_complete_pipeline(TICKER, TOKEN)

# src/evaluation/backtest_ml.py
import pandas as pd
import numpy as np

def backtest_ml_model(
    model,
    df: pd.DataFrame,
    initial_capital: float = 100000,
    commission: float = 0.001,
    threshold: float = 0.6  # Confiança mínima para operar
) -> dict:
    """
    Backtesta modelo de ML.
    
    Estratégia:
    - Compra quando modelo prevê alta com confiança > threshold
    - Vende quando modelo prevê baixa ou posição aberta há muito tempo
    """
    
    # Preparar dados
    X, y, dates = model.prepare_data(df)
    split = int(len(X) * 0.8)
    
    # Treinar em dados de treino
    model.fit(X[:split], y[:split])
    
    # Previsões no período de teste
    X_test = X[split:]
    y_test = y[split:]
    dates_test = dates[split:]
    
    predictions = model.predict(X_test)
    probas = model.predict_proba(X_test)[:, 1]  # Probabilidade de alta
    
    # Simular trading
    capital = initial_capital
    position = 0
    shares = 0
    
    # Preços de fechamento do período de teste
    close_prices = df["close"].values[split + len(df) - len(X):][:len(predictions)]
    
    equity_curve = []
    trades = []
    
    for i in range(len(predictions)):
        price = close_prices[i]
        proba = probas[i]
        pred = predictions[i]
        
        # Decisão de trading
        if position == 0 and proba > threshold:
            # Comprar
            investment = capital * 0.95  # 95% do capital
            shares = int(investment / price)
            cost = shares * price * (1 + commission)
            capital -= cost
            position = 1
            trades.append({
                "type": "buy",
                "date": dates_test[i],
                "price": price,
                "shares": shares
            })
        
        elif position == 1 and proba < (1 - threshold):
            # Vender
            revenue = shares * price * (1 - commission)
            capital += revenue
            trades.append({
                "type": "sell",
                "date": dates_test[i],
                "price": price,
                "shares": shares,
                "pnl": revenue - (shares * trades[-1]["price"])
            })
            position = 0
            shares = 0
        
        # Calcular equity
        equity = capital + (shares * price if position == 1 else 0)
        equity_curve.append(equity)
    
    # Fechar posição aberta
    if position == 1:
        final_price = close_prices[-1]
        revenue = shares * final_price * (1 - commission)
        capital += revenue
    
    final_equity = capital
    
    # Calcular métricas
    total_return = (final_equity / initial_capital - 1) * 100
    
    # Buy and hold para comparação
    buy_hold_return = (close_prices[-1] / close_prices[0] - 1) * 100
    
    # Sharpe ratio
    returns = pd.Series(equity_curve).pct_change().dropna()
    sharpe = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
    
    # Max drawdown
    peak = pd.Series(equity_curve).expanding().max()
    drawdown = (pd.Series(equity_curve) - peak) / peak
    max_dd = drawdown.min() * 100
    
    return {
        "total_return": total_return,
        "buy_hold_return": buy_hold_return,
        "excess_return": total_return - buy_hold_return,
        "sharpe_ratio": sharpe,
        "max_drawdown": max_dd,
        "n_trades": len([t for t in trades if t["type"] == "buy"]),
        "final_equity": final_equity,
        "equity_curve": equity_curve
    }


# Uso
if __name__ == "__main__":
    from src.data.fetcher import BrapiMLDataFetcher
    from src.features.technical import TechnicalFeatures
    from src.models.random_forest import RandomForestStockModel
    
    fetcher = BrapiMLDataFetcher("seu_token")
    df = fetcher.get_stock_data("PETR4", "5y")
    df = TechnicalFeatures.add_all_features(df)
    
    model = RandomForestStockModel()
    results = backtest_ml_model(model, df, threshold=0.55)
    
    print("\n=== Backtest ML ===")
    print(f"Retorno Total: {results['total_return']:.2f}%")
    print(f"Buy & Hold: {results['buy_hold_return']:.2f}%")
    print(f"Excess Return: {results['excess_return']:.2f}%")
    print(f"Sharpe Ratio: {results['sharpe_ratio']:.2f}")
    print(f"Max Drawdown: {results['max_drawdown']:.2f}%")
    print(f"Número de Trades: {results['n_trades']}")

┌─────────────────────────────────────────────────────────────┐
│                 POR QUE ML FALHA EM AÇÕES                   │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  1. OVERFITTING                                              │
│     └─ Modelo memoriza padrões do passado                   │
│     └─ Não generaliza para novos dados                      │
│     └─ Solução: regularização, cross-validation temporal    │
│                                                              │
│  2. DATA LEAKAGE                                             │
│     └─ Usar informação futura no treino                     │
│     └─ Exemplo: normalizar antes de split                   │
│     └─ Solução: pipeline cuidadoso                          │
│                                                              │
│  3. REGIME CHANGES                                           │
│     └─ Mercado muda ao longo do tempo                       │
│     └─ Modelo treinado em bull market falha em bear         │
│     └─ Solução: retreino periódico, features adaptativas    │
│                                                              │
│  4. SURVIVORSHIP BIAS                                        │
│     └─ Dados não incluem empresas que faliram               │
│     └─ Backtests otimistas                                  │
│     └─ Solução: dados com delisted stocks                   │
│                                                              │
│  5. TRANSACTION COSTS                                        │
│     └─ Ignorar corretagem, slippage, spread                 │
│     └─ Lucro vira prejuízo                                  │
│     └─ Solução: simular custos realistas                    │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Antes de confiar no seu modelo:

□ Usou split temporal (não aleatório)?
□ Features não usam dados futuros?
□ Testou em múltiplos ativos?
□ Comparou com baseline simples?
□ Incluiu custos de transação?
□ Fez walk-forward validation?
□ Modelo é simples o suficiente?
□ Resultados fazem sentido econômico?
□ Testou em diferentes regimes de mercado?
□ Tem margem de segurança nos resultados?