Assistente de Codificação com IA: Raspador e Preditor do Mercado de Ações

Aqui está um script Python otimizado que coleta dados do mercado de ações em tempo real do Yahoo Finance, aplica aprendizado de máquina para prever tendências de curto prazo e visualiza os resultados com Matplotlib: ```python import yfinance as yf import pandas as pd import numpy as np from sklearn.ensemble import RandomForestRegressor from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split import matplotlib.pyplot as plt import matplotlib.dates as mdates from datetime import datetime, timedelta import warnings warnings.filterwarnings('ignore') # Configurações globais TICKER = "AAPL" PERIOD = "6mo" INTERVAL = "1h" LOOKBACK = 10 # Janela de lookback para features PREDICT_HORIZON = 5 # Passos à frente para previsão def fetch_stock_data(ticker, period, interval): """Busca dados em tempo real do Yahoo Finance""" stock = yf.Ticker(ticker) df = stock.history(period=period, interval=interval) df = df.reset_index() df['Date'] = pd.to_datetime(df['Datetime'] if 'Datetime' in df.columns else df['Date']) return df def create_features(df, lookback): """Cria features técnicas para ML""" df = df.copy() # Retornos e médias móveis df['Returns'] = df['Close'].pct_change() df['MA_5'] = df['Close'].rolling(window=5).mean() df['MA_10'] = df['Close'].rolling(window=10).mean() df['MA_20'] = df['Close'].rolling(window=20).mean() # Volatilidade df['Volatility'] = df['Returns'].rolling(window=10).std() # RSI (Relative Strength Index) delta = df['Close'].diff() gain = delta.where(delta > 0, 0).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss df['RSI'] = 100 - (100 / (1 + rs)) # Volume e momentum df['Volume_MA'] = df['Volume'].rolling(window=10).mean() df['Momentum'] = df['Close'] - df['Close'].shift(lookback) # Features de lag para o modelo for i in range(1, lookback + 1): df[f'Close_lag_{i}'] = df['Close'].shift(i) df[f'Volume_lag_{i}'] = df['Volume'].shift(i) # Target: preço futuro df['Target'] = df['Close'].shift(-PREDICT_HORIZON) # Remove linhas com NaN df = df.dropna() return df def prepare_data(df): """Prepara dados para treinamento""" feature_cols = [col for col in df.columns if col not in ['Date', 'Datetime', 'Target', 'Close']] X = df[feature_cols].values y = df['Target'].values # Normalização scaler = StandardScaler() X_scaled = scaler.fit_transform(X) return X_scaled, y, scaler, feature_cols def train_model(X, y): """Treina modelo Random Forest otimizado""" X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False) model = RandomForestRegressor( n_estimators=200, max_depth=15, min_samples_split=5, min_samples_leaf=2, random_state=42, n_jobs=-1 ) model.fit(X_train, y_train) # Métricas train_score = model.score(X_train, y_train) test_score = model.score(X_test, y_test) return model, train_score, test_score def predict_future(model, scaler, df, feature_cols, steps): """Faz previsões iterativas para o futuro""" last_data = df.iloc[-1][feature_cols].values.reshape(1, -1) predictions = [] current_df = df.copy() for _ in range(steps): # Escala e prevê scaled = scaler.transform(last_data) pred = model.predict(scaled)[0] predictions.append(pred) # Atualiza features para próxima iteração new_row = current_df.iloc[-1].copy() new_row['Close'] = pred new_row['Returns'] = pred / new_row['Close_lag_1'] - 1 if new_row['Close_lag_1'] != 0 else 0 new_row['MA_5'] = (current_df['Close'].tail(4).sum() + pred) / 5 new_row['MA_10'] = (current_df['Close'].tail(9).sum() + pred) / 10 new_row['MA_20'] = (current_df['Close'].tail(19).sum() + pred) / 20 # Atualiza lags for i in range(LOOKBACK, 1, -1): new_row[f'Close_lag_{i}'] = new_row[f'Close_lag_{i-1}'] new_row['Close_lag_1'] = current_df['Close'].iloc[-1] # Adiciona à base current_df = pd.concat([current_df, pd.DataFrame([new_row])], ignore_index=True) last_data = new_row[feature_cols].values.reshape(1, -1) return predictions def visualize_results(df, predictions, ticker): """Visualiza dados históricos e previsões""" fig, axes = plt.subplots(3, 1, figsize=(14, 12), sharex=True) # Preço e previsão ax1 = axes[0] ax1.plot(df['Date'], df['Close'], label='Preço Real', color='blue', linewidth=1.5) # Previsões futuras future_dates = pd.date_range(start=df['Date'].iloc[-1] + timedelta(hours=1), periods=len(predictions), freq='h') ax1.plot(future_dates, predictions, label='Previsão', color='red', linestyle='--', marker='o', markersize=4) # Médias móveis ax1.plot(df['Date'], df['MA_10'], label='MA 10', color='orange', alpha=0.7) ax1.plot(df['Date'], df['MA_20'], label='MA 20', color='green', alpha=0.7) ax1.set_ylabel('Preço (USD)') ax1.set_title(f'{ticker} - Previsão de Tendência de Curto Prazo') ax1.legend(loc='upper left') ax1.grid(True, alpha=0.3) # Volume ax2 = axes[1] ax2.bar(df['Date'], df['Volume'], color='gray', alpha=0.5, width=0.02) ax2.set_ylabel('Volume') ax2.grid(True, alpha=0.3) # RSI ax3 = axes[2] ax3.plot(df['Date'], df['RSI'], color='purple', linewidth=1.2) ax3.axhline(y=70, color='red', linestyle='--', alpha=0.5) ax3.axhline(y=30, color='green', linestyle='--', alpha=0.5) ax3.set_ylabel('RSI') ax3.set_ylim(0, 100) ax3.grid(True, alpha=0.3) # Formatação do eixo x for ax in axes: ax.xaxis.set_major_formatter(mdates.DateFormatter('%d/%m')) ax.xaxis.set_major_locator(mdates.WeekdayLocator(interval=1)) plt.xlabel('Data') plt.tight_layout() plt.savefig(f'{ticker}_prediction.png', dpi=150, bbox_inches='tight') plt.show() def main(): print(f"Buscando dados em tempo real para {TICKER}...") # Busca dados df = fetch_stock_data(TICKER, PERIOD, INTERVAL) print(f"Dados carregados: {len(df)} registros") # Cria features df_features = create_features(df, LOOKBACK) print(f"Features criadas: {len(df_features.columns)} colunas") # Prepara dados X, y, scaler, feature_cols = prepare_data(df_features) print(f"Dados preparados: {X.shape[0]} amostras, {X.shape[1]} features") # Treina modelo model, train_score, test_score = train_model(X, y) print(f"Score treino: {train_score:.4f}") print(f"Score teste: {test_score:.4f}") # Faz previsões predictions = predict_future(model, scaler, df_features, feature_cols, PREDICT_HORIZON) print(f"Previsões para os próximos {PREDICT_HORIZON} períodos:") for i, pred in enumerate(predictions, 1): print(f" Período {i}: ${pred:.2f}") # Visualiza visualize_results(df_features, predictions, TICKER) # Análise de tendência last_price = df_features['Close'].iloc[-1] avg_pred = np.mean(predictions) trend = "ALTA" if avg_pred > last_price else "BAIXA" print(f"\nPreço atual: ${last_price:.2f}") print(f"Preço médio previsto: ${avg_pred:.2f}") print(f"Tendência prevista: {trend}") if __name__ == "__main__": main() ``` **Instalação das dependências:** ```bash pip install yfinance scikit-learn pandas numpy matplotlib ``` **Principais características do script:** 1. **Otimização**: Uso de RandomForestRegressor com parâmetros otimizados, processamento vetorizado com pandas/numpy, e n_jobs=-1 para paralelização. 2. **Features técnicas**: Inclui médias móveis, RSI, volatilidade, momentum e lags de preço/volume. 3. **Previsão iterativa**: Gera previsões múltiplas passos à frente atualizando as features dinamicamente. 4. **Visualização completa**: Mostra preço real + previsão, volume e RSI em gráficos separados. 5. **Métricas de desempenho**: Exibe scores de treino e teste para avaliar a qualidade do modelo. 6. **Análise de tendência**: Calcula se a tendência prevista é de alta ou baixa. Para usar, basta executar o script. Você pode modificar TICKER, PERIOD, INTERVAL e outros parâmetros conforme necessário.