La volatilité du Bitcoin fascine autant qu'elle effraie les investisseurs. Comment anticiper les pics de volatilité qui font varier les prix de 5% en quelques minutes ? Dans ce tutoriel technique, je vais vous montrer comment construire deux modèles complémentaires avec les données de qualité professionnelle de Tardis, puis vous révéler pourquoi j'utilise HolySheep AI pour analyser les résultats de ces modèles avec une latence inférieure à 50ms.
Comparatif : HolySheep vs API Tardis vs Services Relais
| Critère | HolySheep AI | API Tardis Direct | Services Relais (CoinGecko+) |
|---|---|---|---|
| Prix typique/requête | $0.0002 (modèles rapides) | $0.0015-0.005 | $0.001-0.003 |
| Latence moyenne | <50ms | 200-500ms | 500-2000ms |
| Devises acceptées | ¥ Yuan + $ USD + WeChat/Alipay | $ USD uniquement (Stripe) | $ USD uniquement |
| Économie vs officiel | -85% | Référence | -30% à -50% |
| Analyse IA des données | ✅ Native (GPT-4.1, Claude Sonnet) | ❌ Non disponible | ⚠️ Via intégration tierce |
| Crédits gratuits | ✅ Oui | ❌ Trial limité | ⚠️ Rarement |
S'inscrire ici pour accéder à ces tarifs avantageux avec des crédits gratuits dès l'inscription.
Prérequis et Installation de l'Environnement
Pour suivre ce tutoriel, vous aurez besoin de Python 3.9+, des bibliothèques de calcul scientifique, et d'un accès aux données Tardis. Voici comment configurer votre environnement complet :
# Installation des dépendances
pip install tardis-client pandas numpy scipy arch scikit-learn
pip install plotly matplotlib requests python-dotenv
Structure du projet
mkdir btc_volatility && cd btc_volatility
mkdir data models notebooks src
touch .env config.py requirements.txtConnexion à l'API Tardis et Récupération des Données OHLCV
# config.py
import os
from dataclasses import dataclass
@dataclass
class Config:
# Configuration Tardis - données historiques BTC/USD
TARDIS_EXCHANGE = "binance"
TARDIS_SYMBOL = "btcusdt"
TARDIS_START = "2024-01-01"
TARDIS_END = "2024-12-31"
TARDIS_INTERVAL = "1m" # Candles 1 minute
# Configuration HolySheep AI pour analyse des résultats
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# Configuration modèle
GARCH_P = 1
GARCH_Q = 1
GARCH_MODEL = "GARCH"
# Réduction de la fenêtre d'entraînement (optimisé pour la vitesse)
TRAIN_WINDOW_DAYS = 30 # 30 jours au lieu de 365 pour tester rapidement# src/data_fetcher.py
from tardis_client import TardisClient, Binance
import pandas as pd
from datetime import datetime, timedelta
from config import Config
class TardisDataFetcher:
"""Récupère les données OHLCV depuis Tardis pour BTC/USDT"""
def __init__(self, exchange: str = Config.TARDIS_EXCHANGE,
symbol: str = Config.TARDIS_SYMBOL):
self.exchange = exchange
self.symbol = symbol
self.client = None
def fetch_ohlcv(self, start_date: str, end_date: str,
interval: str = "1m") -> pd.DataFrame:
"""
Récupère les données OHLCV de Binance via Tardis
Args:
start_date: Date de début (YYYY-MM-DD)
end_date: Date de fin (YYYY-MM-DD)
interval: Intervalle des bougies (1m, 5m, 1h, 1d)
Returns:
DataFrame avec colonnes: timestamp, open, high, low, close, volume
"""
print(f"📡 Connexion à Tardis: {self.exchange}/{self.symbol}")
print(f"📅 Période: {start_date} → {end_date}")
# Conversion des dates en timestamps
start_ts = int(pd.Timestamp(start_date).timestamp() * 1000)
end_ts = int(pd.Timestamp(end_date).timestamp() * 1000)
candles = []
# Utilisation du client synchrone de Tardis
client = TardisClient()
for candle in client.get_historical_candles(
exchange=self.exchange,
symbol=self.symbol,
from_timestamp=start_ts,
to_timestamp=end_ts,
interval=interval
):
candles.append({
'timestamp': pd.to_datetime(candle.timestamp, unit='ms'),
'open': float(candle.open),
'high': float(candle.high),
'low': float(candle.low),
'close': float(candle.close),
'volume': float(candle.volume)
})
df = pd.DataFrame(candles)
df.set_index('timestamp', inplace=True)
print(f"✅ {len(df):,} bougies récupérées")
print(f" Prix BTC: ${df['close'].iloc[0]:,.2f} → ${df['close'].iloc[-1]:,.2f}")
return df
def calculate_returns(self, df: pd.DataFrame) -> pd.DataFrame:
"""Calcule les rendements logarithmiques et la volatilité"""
df['log_return'] = np.log(df['close'] / df['close'].shift(1))
df['return_pct'] = df['close'].pct_change() * 100
df['volatility'] = df['log_return'].rolling(window=60).std() * np.sqrt(525600)
# Volatilité annualisée (minutes → année)
# 525600 = minutes dans une année
return df.dropna()
import numpy as np
Test de récupération
if __name__ == "__main__":
fetcher = TardisDataFetcher()
# Test avec 7 jours de données pour validation rapide
df = fetcher.fetch_ohlcv(
start_date="2024-12-01",
end_date="2024-12-08",
interval="5m"
)
df = fetcher.calculate_returns(df)
print(f"\n📊 Statistiques de volatilité:")
print(f" Volatilité moyenne: {df['volatility'].mean()*100:.2f}%")
print(f" Volatilité max: {df['volatility'].max()*100:.2f}%")Modèle 1 : GARCH (Generalized Autoregressive Conditional Heteroskedasticity)
Le modèle GARCH est le standard de l'industrie financière pour la modélisation de la volatilité. Il capture parfaitement la "clusterisation de volatilité" observe sur les cryptomonnaies : les périodes de forte volatilité ont tendance à se regrouper.
# src/garch_model.py
import pandas as pd
import numpy as np
from arch import arch_model
from scipy import stats
import warnings
warnings.filterwarnings('ignore')
class GARCHVolatilityModel:
"""
Implémente un modèle GARCH(1,1) pour prédire la volatilité du BTC.
GARCH(1,1) = σ²(t) = ω + α·ε²(t-1) + β·σ²(t-1)
où:
- ω = constante (long-run variance)
- α = impact du choc précédent (ARCH)
- β = persistance de la volatilité (GARCH)
- α + β ≈ 1 = forte persistance (typique du BTC)
"""
def __init__(self, p: int = 1, q: int = 1, dist: str = 'normal'):
self.p = p
self.q = q
self.dist = dist
self.model = None
self.results = None
self.forecast = None
def fit(self, returns: pd.Series, verbose: bool = True) -> dict:
"""
Entraîne le modèle GARCH sur les rendements
Args:
returns: Série des rendements logarithmiques (en fractions, pas pourcentages)
Returns:
Dict avec les paramètres et métriques
"""
# Le modèle GARCH travaille avec des rendements × 100 dans arch
scaled_returns = returns * 100
# Construction du modèle GARCH(p,q) avec moyenne constante
self.model = arch_model(
scaled_returns,
vol='GARCH',
p=self.p,
q=self.q,
dist=self.dist # 'normal', 't', 'skewt'
)
if verbose:
print("🔧 Entraînement du modèle GARCH({},{})...".format(self.p, self.q))
# Ajustement avec méthode 'L-BFGS' (rapide et stable)
self.results = self.model.fit(disp='off', method='L-BFGS')
if verbose:
print("\n" + "="*60)
print("📊 RÉSULTATS GARCH({},{})".format(self.p, self.q))
print("="*60)
print(self.results.summary().tables[1])
# Extraction des paramètres
params = self.results.params
omega = params['omega']
alpha = params['alpha[1]'] if 'alpha[1]' in params.index else params['alpha[1]']
beta = params['beta[1]'] if 'beta[1]' in params.index else params['beta[1]']
print(f"\n📐 Paramètres du modèle:")
print(f" ω (omega) = {omega:.6f}")
print(f" α (alpha) = {alpha:.4f}")
print(f" β (beta) = {beta:.4f}")
print(f" α + β = {alpha + beta:.4f} (persistance)")
# Half-life de la volatilité
if alpha + beta < 1:
half_life = np.log(0.5) / np.log(alpha + beta)
print(f" Half-life = {half_life:.1f} périodes")
return self._extract_metrics()
def _extract_metrics(self) -> dict:
"""Extrait les métriques clés du modèle"""
params = self.results.params
cond_var = self.results.conditional_volatility / 100
metrics = {
'omega': float(params['omega']),
'alpha': float(params.get('alpha[1]', params['alpha[1]'])),
'beta': float(params.get('beta[1]', params['beta[1]'])),
'persistence': float(params.get('alpha[1]', params['alpha[1]'])) +
float(params.get('beta[1]', params['beta[1]'])),
'avg_volatility': float(cond_var.mean()),
'aic': float(self.results.aic),
'bic': float(self.results.bic),
'log_likelihood': float(self.results.loglikelihood)
}
return metrics
def predict_volatility(self, horizon: int = 60) -> np.ndarray:
"""
Prédit la volatilité future sur 'horizon' périodes
Args:
horizon: Nombre de périodes à prédire (ex: 60 = 60 minutes)
Returns:
Array des volatilités prédites (annualisées)
"""
forecast = self.results.forecast(horizon=horizon)
variance_forecast = forecast.variance.values[-1, :]
# Conversion: variance → écart-type → volatilité annualisée
# Échelle: les rendements étaient ×100, on ramène à l'échelle originale
vol_forecast = np.sqrt(variance_forecast) / 100
# Annualisation selon la fréquence des données
# Si données 5min: 525600/5 = 105120 périodes par an
periods_per_year = 105120 # Pour candles 5min
vol_annualized = vol_forecast * np.sqrt(periods_per_year)
self.forecast = vol_annualized
return vol_annualized
def rolling_forecast(self, returns: pd.Series,
window: int = 2520,
horizon: int = 60) -> pd.DataFrame:
"""
Prévision roulante pour backtesting
Args:
returns: Toutes les données de rendements
window: Taille de la fenêtre d'entraînement (2520 = 1 semaine de 5min)
horizon: Horizon de prédiction
Returns:
DataFrame avec predictions vs réalisations
"""
predictions = []
actuals = []
dates = []
for i in range(window, len(returns) - horizon, horizon):
train = returns.iloc[i-window:i] * 100
test_return = returns.iloc[i:i+horizon] * 100
try:
# Entraînement sur la fenêtre
model = arch_model(train, vol='GARCH', p=1, q=1, dist='normal')
res = model.fit(disp='off')
# Prédiction
fc = res.forecast(horizon=horizon)
pred_vol = np.sqrt(fc.variance.mean(axis=1).values[-1]) / 100
# Volatilité réalisée sur la période de test
actual_vol = test_return.std()
predictions.append(pred_vol)
actuals.append(actual_vol)
dates.append(returns.index[i])
except Exception as e:
continue
return pd.DataFrame({
'predicted_vol': predictions,
'actual_vol': actuals
}, index=dates)Modèle 2 : Machine Learning avec LSTM et Random Forest
# src/ml_models.py
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_squared_error, mean_absolute_error
import warnings
warnings.filterwarnings('ignore')
Optionnel: TensorFlow/Keras pour LSTM
try:
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.callbacks import EarlyStopping
HAS_TF = True
except ImportError:
HAS_TF = False
print("⚠️ TensorFlow non disponible, LSTM désactivé")
class VolatilityFeatureEngineer:
"""Ingénierie des features pour la prédiction de volatilité"""
@staticmethod
def create_features(df: pd.DataFrame) -> pd.DataFrame:
"""Crée les features pour les modèles ML"""
data = df.copy()
# Features de prix
data['price_momentum_5'] = data['close'].pct_change(5)
data['price_momentum_15'] = data['close'].pct_change(15)
data['price_momentum_60'] = data['close'].pct_change(60)
# Features de volatilité historique
for window in [10, 30, 60, 120]:
data[f'vol_realized_{window}'] = data['log_return'].rolling(window).std()
data[f'vol_range_{window}'] = (data['high'] - data['low']).rolling(window).mean()
# Volume
data['volume_change'] = data['volume'].pct_change()
data['volume_ma_ratio'] = data['volume'] / data['volume'].rolling(30).mean()
# Indicateurs techniques
data['bb_width'] = (
data['close'].rolling(20).mean() + 2*data['close'].rolling(20).std() -
(data['close'].rolling(20).mean() - 2*data['close'].rolling(20).std())
) / data['close'].rolling(20).mean()
data['rsi'] = VolatilityFeatureEngineer._calculate_rsi(data['close'])
# Targets: volatilité future (à prédire)
for horizon in [15, 60, 180]:
data[f'target_vol_{horizon}'] = data['log_return'].shift(-horizon).rolling(horizon).std() * np.sqrt(525600)
return data.dropna()
@staticmethod
def _calculate_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
"""Calcule le RSI"""
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
class MLVolatilityModel:
"""
Modèle de prédiction de volatilité par Machine Learning
Utilise Random Forest et LSTM pour capturer les patterns non-linéaires
"""
def __init__(self, model_type: str = 'random_forest'):
self.model_type = model_type
self.model = None
self.scaler = StandardScaler()
self.feature_cols = None
def _get_model(self):
"""Retourne une instance du modèle choisi"""
if self.model_type == 'random_forest':
return RandomForestRegressor(
n_estimators=100,
max_depth=10,
min_samples_split=20,
random_state=42,
n_jobs=-1
)
elif self.model_type == 'gradient_boosting':
return GradientBoostingRegressor(
n_estimators=100,
max_depth=5,
learning_rate=0.1,
random_state=42
)
elif self.model_type == 'lstm' and HAS_TF:
return self._build_lstm()
else:
raise ValueError(f"Modèle {self.model_type} non reconnu")
def _build_lstm(self) -> Sequential:
"""Construit un modèle LSTM pour séries temporelles"""
model = Sequential([
LSTM(64, return_sequences=True, input_shape=(60, 15)),
Dropout(0.2),
LSTM(32, return_sequences=False),
Dropout(0.2),
Dense(16, activation='relu'),
Dense(1) # Prédiction de volatilité
])
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
return model
def prepare_data(self, df: pd.DataFrame, target_col: str = 'target_vol_60',
lookback: int = 60) -> tuple:
"""
Prépare les données pour l'entraînement
Args:
df: DataFrame avec les features
target_col: Colonne cible à prédire
lookback: Nombre de périodes passées à utiliser
Returns:
(X_train, y_train, X_test, y_test, scaler)
"""
self.feature_cols = [col for col in df.columns
if col not in ['target_vol_15', 'target_vol_60', 'target_vol_180',
'open', 'high', 'low', 'close', 'volume', 'log_return']]
X = df[self.feature_cols].values
y = df[target_col].values
# Normalisation
X_scaled = self.scaler.fit_transform(X)
# Train/test split temporel (80/20)
split_idx = int(len(X) * 0.8)
X_train, X_test = X_scaled[:split_idx], X_scaled[split_idx:]
y_train, y_test = y[:split_idx], y[split_idx:]
# Pour LSTM: reshape en (samples, timesteps, features)
if self.model_type == 'lstm':
X_train = X_train.reshape(-1, lookback, X_train.shape[1] // lookback)
# Note: simplification, en pratique utiliser une vraie structure temporelle
print(f"📊 Données préparées: {len(X_train)} train / {len(X_test)} test")
print(f" Features: {len(self.feature_cols)}")
return X_train, y_train, X_test, y_test
def train(self, X_train, y_train, X_test, y_test) -> dict:
"""Entraîne le modèle ML"""
print(f"\n🤖 Entraînement du modèle: {self.model_type}")
self.model = self._get_model()
if self.model_type == 'lstm' and HAS_TF:
early_stop = EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True)
self.model.fit(
X_train, y_train,
epochs=50,
batch_size=32,
validation_split=0.1,
callbacks=[early_stop],
verbose=1
)
else:
self.model.fit(X_train, y_train)
# Évaluation
y_pred_train = self.model.predict(X_train[:1000]) # Sample pour speed
y_pred_test = self.model.predict(X_test)
metrics = {
'train_rmse': np.sqrt(mean_squared_error(y_train[:1000], y_pred_train)),
'test_rmse': np.sqrt(mean_squared_error(y_test, y_pred_test)),
'test_mae': mean_absolute_error(y_test, y_pred_test),
'feature_importance': self._get_feature_importance()
}
print(f"\n📊 Métriques {self.model_type}:")
print(f" Train RMSE: {metrics['train_rmse']:.4f}")
print(f" Test RMSE: {metrics['test_rmse']:.4f}")
print(f" Test MAE: {metrics['test_mae']:.4f}")
return metrics
def _get_feature_importance(self) -> dict:
"""Retourne l'importance des features (pour Random Forest/GB)"""
if hasattr(self.model, 'feature_importances_'):
importance = self.model.feature_importances_
return dict(zip(self.feature_cols, importance))
return {}
def predict(self, X) -> np.ndarray:
"""Prédit la volatilité future"""
X_scaled = self.scaler.transform(X)
return self.model.predict(X_scaled)Analyse des Résultats avec HolySheep AI
Une fois vos modèles GARCH et ML entraînés, l'analyse des résultats devient critique. J'utilise HolySheep AI pour générer des insights actionnables à partir des prédictions de volatilité. Avec leur API offrant moins de 50ms de latence et des prix jusqu'à 85% inférieurs aux APIs officielles, c'est devenu mon outil préféré pour l'analyse financière.
# src/analysis_with_holysheep.py
import requests
import json
from config import Config
from typing import List, Dict
class HolySheepAnalyzer:
"""
Utilise HolySheep AI pour analyser les prédictions de volatilité BTC
HolySheep offre des tarifs imbattables:
- GPT-4.1: $8/MTok (vs $15 officiel)
- Claude Sonnet 4.5: $15/MTok (vs $30 officiel)
- Gemini 2.5 Flash: $2.50/MTok (vs $15 officiel)
- DeepSeek V3.2: $0.42/MTok (excellent rapport qualité/prix)
Latence moyenne: <50ms
"""
def __init__(self, api_key: str = Config.HOLYSHEEP_API_KEY):
self.base_url = Config.HOLYSHEEP_BASE_URL
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_volatility_prediction(self,
garch_pred: float,
ml_pred: float,
actual_vol: float,
btc_price: float) -> Dict:
"""
Demande à l'IA d'analyser et comparer les prédictions
Args:
garch_pred: Prédiction GARCH (volatilité annualisée)
ml_pred: Prédiction ML (volatilité annualisée)
actual_vol: Volatilité réalisée (pour contexte)
btc_price: Prix actuel du BTC en USD
Returns:
Analyse structurée du modèle
"""
prompt = f"""Analyse technique de prédiction de volatilité BTC:
Contexte:
- Prix BTC actuel: ${btc_price:,.2f}
- Prédiction GARCH (volatilité annualisée): {garch_pred*100:.2f}%
- Prédiction ML (volatilité annualisée): {ml_pred*100:.2f}%
- Volatilité réalisée récemment: {actual_vol*100:.2f}%
Instructions:
1. Calculer l'erreur de chaque modèle
2. Identifier lequel est le plus fiable actuellement
3. Donner une recommandation de trading (en supposant un horizon de 1h)
4. Indiquer le niveau de confiance (0-100%)
Répondre en JSON avec les clés: garch_error, ml_error, winner,
recommendation, confidence, reasoning (max 200 caractères)."""
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "gpt-4.1", # Modèle performant pour analyse financière
"messages": [
{"role": "system", "content": "Tu es un analyste financier expert en cryptomonnaies."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"response_format": {"type": "json_object"}
},
timeout=10
)
if response.status_code == 200:
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
else:
return {"error": f"HTTP {response.status_code}", "message": response.text}
def batch_analyze_predictions(self, predictions_df: pd.DataFrame,
sample_size: int = 10) -> List[Dict]:
"""
Analyse un échantillon de prédictions en lot
Utilise DeepSeek V3.2 pour les analyses de masse ($0.42/MTok!)
"""
sample = predictions_df.sample(min(sample_size, len(predictions_df)))
results = []
for _, row in sample.iterrows():
try:
analysis = self.analyze_volatility_prediction(
garch_pred=row['garch_pred'],
ml_pred=row['ml_pred'],
actual_vol=row['actual_vol'],
btc_price=row['btc_price']
)
results.append(analysis)
except Exception as e:
print(f"⚠️ Erreur analyse: {e}")
return results
def generate_report(self, all_results: List[Dict],
garch_mape: float, ml_mape: float) -> str:
"""
Génère un rapport complet avec HolySheep (utilise Claude Sonnet 4.5)
"""
prompt = f"""Génère un rapport d'analyse de modèle de volatilité BTC:
Performance:
- GARCH MAPE: {garch_mape:.2f}%
- ML MAPE: {ml_mape:.2f}%
Analyses individuelles:
{json.dumps(all_results[:5], indent=2)}
Format: Markdown avec sections:
1. Résumé Exécutif
2. Performance des Modèles
3. Cas d'Usage Recommandés
4. Limitations et Risques
5. Conclusion"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "claude-sonnet-4.5", # Excellent pour génération de rapports
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.5,
"max_tokens": 2000
},
timeout=30
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
return "Erreur génération rapport"
Exemple d'utilisation
if __name__ == "__main__":
analyzer = HolySheepAnalyzer()
# Test avec des données fictives
result = analyzer.analyze_volatility_prediction(
garch_pred=0.45, # 45% annualisée
ml_pred=0.52,
actual_vol=0.48,
btc_price=67500.00
)
print("📊 Analyse HolySheep AI:")
print(json.dumps(result, indent=2))Pipeline Complet : Du Raw Data à la Prédiction
# main.py - Pipeline complet
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from config import Config
from src.data_fetcher import TardisDataFetcher
from src.garch_model import GARCHVolatilityModel
from src.ml_models import MLVolatilityModel, VolatilityFeatureEngineer
from src.analysis_with_holysheep import HolySheepAnalyzer
def main():
"""Pipeline complet de prédiction de volatilité BTC"""
print("="*70)
print("🚀 BTC VOLATILITY PREDICTION PIPELINE")
print(" Data: Tardis | Models: GARCH + ML | Analysis: HolySheep AI")
print("="*70)
# ============================================
# ÉTAPE 1: Récupération des données Tardis
# ============================================
print("\n[1/5] 📡 Récupération des données depuis Tardis...")
fetcher = TardisDataFetcher()
df = fetcher.fetch_ohlcv(
start_date=Config.TARDIS_START,
end_date=Config.TARDIS_END,
interval=Config.TARDIS_INTERVAL
)
# Calcul des rendements
df = fetcher.calculate_returns(df)
returns = df['log_return'].dropna()
print(f" Dataset: {len(df):,} bougies")
print(f" Période: {df.index.min()} → {df.index.max()}")
# ============================================
# ÉTAPE 2: Modèle GARCH
# ============================================
print("\n[2/5] 📊 Entraînement GARCH...")
garch = GARCHVolatilityModel(p=1, q=1)
garch_metrics = garch.fit(returns)
# Prédiction de volatilité
garch_vol_pred = garch.predict_volatility(horizon=60)
print(f" Volatilité prédite (prochaines 60 périodes): {garch_vol_pred.mean()*100:.2f}%")
# ============================================
# ÉTAPE 3: Modèle ML
# ============================================
print("\n[3/5] 🤖 Entraînement modèle ML...")
# Création des features
features_df = VolatilityFeatureEngineer.create_features(df)
ml_model = MLVolatilityModel(model_type='random_forest')
X_train, y_train, X_test, y_test = ml_model.prepare_data(
features_df,
target_col='target_vol_60'
)
ml_metrics = ml_model.train(X_train, y_train, X_test, y_test)
# ============================================
# ÉTAPE 4: Backtesting comparatif
# ============================================
print("\n[4/5] 🔄 Backtesting comparatif...")
# Rolling forecast pour GARCH
garch_rolling = garch.rolling_forecast(
returns,
window=2520, # ~1 semaine de données 5min
horizon=60
)
# Résultats comparatifs
garch_mape = np.abs(
(garch_rolling['predicted_vol'] - garch_rolling['actual_vol']) /
garch_rolling['actual_vol']
).mean() * 100
print(f" GARCH MAPE: {garch_mape:.2f}%")
print(f" ML RMSE: {ml_metrics['test_rmse']:.4f}")
# ============================================
# ÉTAPE 5: Analyse HolySheep
# ============================================
print("\n[5/5] 🐑 Analyse avec HolySheep AI...")
analyzer = HolySheepAnalyzer()
# Analyse de la dernière prédiction
latest_analysis = analyzer.analyze_volatility_prediction(
garch_pred=garch_vol_pred[-1],
ml_pred=ml_model.predict(X_test[-1:].reshape(1, -1))[0],
actual_vol=returns.iloc[-60:].std() * np.sqrt(525600),
btc_price=df['close'].iloc[-1]
)
print(f"\n{'='*70}")
print("📋 RÉSULTAT DE L'ANALYSE HOLYSHEEP AI")
print("="*70)
print(f"🏆 Modèle recommandé: {latest_analysis.get('winner', 'N/A')}")
print(f"📌 Confidence: {latest_analysis.get('confidence', 'N/A')}%")
print(f"💡 {latest_analysis.get('recommendation', 'N/A')}")
print("\n✅ Pipeline terminé avec succès!")
print(f" Total candles traitées: {len(df):,}")
print(f" Modèles entraînés: GARCH(1,1) + RandomForest")
print(f" Analyse IA: HolySheep API (<50ms latence)")
return {
'garch_metrics': garch_metrics