Assistant de codage alimenté par IA : Scraper et prédicteur du marché boursier

Voici un script Python optimisé qui scrape les données boursières en temps réel depuis Yahoo Finance, applique un apprentissage automatique pour prédire les tendances à court terme, et visualise les résultats avec 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') # Configuration TICKER = 'AAPL' PERIOD = '1mo' INTERVAL = '15m' LOOKBACK = 10 # Nombre de périodes passées pour les caractéristiques PREDICT_HORIZON = 5 # Périodes futures à prédire def fetch_real_time_data(ticker, period=PERIOD, interval=INTERVAL): """Récupère les données boursières en temps réel depuis Yahoo Finance.""" stock = yf.Ticker(ticker) df = stock.history(period=period, interval=interval) if df.empty: raise ValueError(f"Aucune donnée trouvée pour {ticker}") return df def create_features(df, lookback=LOOKBACK): """Crée des caractéristiques techniques pour l'apprentissage automatique.""" df = df.copy() # Caractéristiques de base df['Returns'] = df['Close'].pct_change() df['Volume_Change'] = df['Volume'].pct_change() # Moyennes mobiles df['MA_5'] = df['Close'].rolling(window=5).mean() df['MA_10'] = df['Close'].rolling(window=10).mean() # Volatilité df['Volatility'] = df['Returns'].rolling(window=10).std() # RSI (Relative Strength Index) delta = df['Close'].diff() gain = delta.where(delta > 0, 0) loss = -delta.where(delta < 0, 0) avg_gain = gain.rolling(window=14).mean() avg_loss = loss.rolling(window=14).mean() rs = avg_gain / avg_loss df['RSI'] = 100 - (100 / (1 + rs)) # Caractéristiques décalées pour la prédiction 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) df[f'Returns_Lag_{i}'] = df['Returns'].shift(i) # Cible : prix futur df['Target'] = df['Close'].shift(-PREDICT_HORIZON) # Supprimer les lignes avec NaN df = df.dropna() return df def prepare_data(df): """Prépare les données pour l'entraînement et la prédiction.""" feature_cols = [col for col in df.columns if col not in ['Target', 'Dividends', 'Stock Splits']] X = df[feature_cols].values y = df['Target'].values # Normalisation scaler = StandardScaler() X_scaled = scaler.fit_transform(X) return X_scaled, y, scaler, feature_cols def train_model(X, y): """Entraîne un modèle Random Forest pour la prédiction.""" X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) model = RandomForestRegressor( n_estimators=100, max_depth=10, min_samples_split=5, min_samples_leaf=2, random_state=42, n_jobs=-1 ) model.fit(X_train, y_train) # Évaluation train_score = model.score(X_train, y_train) test_score = model.score(X_test, y_test) print(f"Score d'entraînement : {train_score:.4f}") print(f"Score de test : {test_score:.4f}") return model def predict_future(model, scaler, df, feature_cols, steps=PREDICT_HORIZON): """Prédit les prix futurs en utilisant le modèle entraîné.""" predictions = [] current_data = df[feature_cols].iloc[-1].values.reshape(1, -1) for _ in range(steps): current_scaled = scaler.transform(current_data) pred = model.predict(current_scaled)[0] predictions.append(pred) # Mise à jour des caractéristiques pour la prochaine prédiction # (simplification - dans un cas réel, il faudrait recalculer toutes les caractéristiques) current_data = np.roll(current_data, 1) current_data[0, 0] = pred # Mise à jour du prix le plus récent return predictions def visualize_results(df, predictions, ticker): """Visualise les données réelles et les prédictions.""" fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 10), sharex=True) # Graphique des prix ax1.plot(df.index, df['Close'], label='Prix réel', color='blue', linewidth=2) # Prédictions futures future_dates = [df.index[-1] + timedelta(minutes=15 * (i + 1)) for i in range(len(predictions))] ax1.plot(future_dates, predictions, label='Prédictions', color='red', linestyle='--', marker='o') ax1.set_title(f'Prédiction des tendances à court terme pour {ticker}', fontsize=16) ax1.set_ylabel('Prix (USD)', fontsize=12) ax1.legend() ax1.grid(True, alpha=0.3) # Graphique du volume ax2.bar(df.index, df['Volume'], color='green', alpha=0.6, label='Volume') ax2.set_ylabel('Volume', fontsize=12) ax2.set_xlabel('Date', fontsize=12) ax2.legend() ax2.grid(True, alpha=0.3) # Formatage des dates ax1.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d %H:%M')) ax1.xaxis.set_major_locator(mdates.AutoDateLocator()) plt.xticks(rotation=45) plt.tight_layout() plt.show() def main(): """Fonction principale du script.""" print(f"Récupération des données en temps réel pour {TICKER}...") try: # Récupération des données df = fetch_real_time_data(TICKER) print(f"Données récupérées : {len(df)} points") # Création des caractéristiques df_features = create_features(df) print(f"Caractéristiques créées : {len(df_features.columns)} colonnes") # Préparation des données X, y, scaler, feature_cols = prepare_data(df_features) # Entraînement du modèle print("Entraînement du modèle...") model = train_model(X, y) # Prédictions futures print(f"Prédiction des {PREDICT_HORIZON} prochaines périodes...") predictions = predict_future(model, scaler, df_features, feature_cols) # Affichage des prédictions print("\nPrédictions pour les prochaines périodes :") for i, pred in enumerate(predictions, 1): print(f"Période {i} : ${pred:.2f}") # Visualisation print("\nCréation de la visualisation...") visualize_results(df_features, predictions, TICKER) # Analyse des tendances last_price = df_features['Close'].iloc[-1] avg_prediction = np.mean(predictions) trend = "haussière" if avg_prediction > last_price else "baissière" print(f"\nAnalyse : Le prix actuel est ${last_price:.2f}") print(f"Tendance prédite : {trend} (moyenne des prédictions : ${avg_prediction:.2f})") except Exception as e: print(f"Erreur : {e}") print("Vérifiez votre connexion internet et le ticker utilisé.") if __name__ == "__main__": main() ``` Ce script optimisé comprend : 1. **Récupération des données** : Utilise `yfinance` pour obtenir des données en temps réel avec une période et un intervalle configurables. 2. **Ingénierie des caractéristiques** : Crée des indicateurs techniques (moyennes mobiles, RSI, volatilité, rendements) et des caractéristiques décalées pour améliorer la prédiction. 3. **Modèle d'apprentissage automatique** : Utilise un Random Forest Regressor optimisé avec validation croisée implicite via train/test split. 4. **Prédiction** : Génère des prédictions pour les prochaines périodes avec mise à jour itérative des caractéristiques. 5. **Visualisation** : Affiche deux graphiques - les prix réels avec les prédictions, et le volume des transactions. 6. **Optimisations** : - Utilisation de `n_jobs=-1` pour le parallélisme - Normalisation des données avec StandardScaler - Gestion des erreurs robuste - Configuration centralisée des paramètres Pour exécuter ce script, installez les dépendances : ```bash pip install yfinance pandas numpy scikit-learn matplotlib ``` Le script est conçu pour être facilement modifiable - changez simplement le ticker, la période, ou les paramètres du modèle selon vos besoins.