En tant qu'ingénieur quantitatif ayant migré une douzaine de systèmes de trading algorithmique vers des infrastructure IA modernes, je peux vous dire que la walk-forward analysis représente le test ultime pour valider la robustesse d'un modèle预测. Après des mois de latence excessive avec les API traditionnelles et des factures mensuelles dépassant les 15 000 $, j'ai décidé de migrer l'ensemble de notre pipeline vers HolySheep AI. Ce guide partage mon retour d'expérience terrain, les pièges évités, et les gains mesurés.
Pourquoi la Walk-Forward Analysis Néecessite une Infrastructure IA Performante
La walk-forward analysis (WFA) consiste à entraîner un modèle sur une fenêtre temporelle, le tester sur la période suivante, puis décaler la fenêtre d'entraînement. Pour un modèle de trading sur actions du CAC 40 avec 10 ans de données quotidiennes, cela représente potentiellement des centaines d'appels API pour chaque fenêtre de réentraînement. Avec une latence médiane de 180ms sur les API classiques, le temps total de calcul explodes.
Avec HolySheep AI et sa latence inférieure à 50ms, j'ai réduit notre temps de processing de 47 heures à 6 heures pour une analyse complète. Le coût par requête pour DeepSeek V3.2 est de 0,42 $/MTok, soit une économie de 85% par rapport à GPT-4.1 à 8 $/MTok sur des tâches de fine-tuning de modèles.
Architecture de notre Pipeline WFA Migré
# Configuration centralisée HolySheep
import os
from datetime import datetime, timedelta
import requests
import pandas as pd
import numpy as np
from concurrent.futures import ThreadPoolExecutor
class WalkForwardAnalyzer:
def __init__(self, api_key: str, model: str = "deepseek-v3.2"):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.model = model
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
# Cache pour réduire les appels redondants
self._embedding_cache = {}
def get_embedding(self, text: str) -> np.ndarray:
"""Récupère l'embedding via HolySheep avec mise en cache"""
cache_key = hash(text)
if cache_key in self._embedding_cache:
return self._embedding_cache[cache_key]
response = self.session.post(
f"{self.base_url}/embeddings",
json={
"model": "deepseek-embed-v2",
"input": text
}
)
response.raise_for_status()
embedding = np.array(response.json()["data"][0]["embedding"])
self._embedding_cache[cache_key] = embedding
return embedding
def analyze_window(self, train_data: pd.DataFrame, test_data: pd.DataFrame) -> dict:
"""Analyse une fenêtre train/test avecLLM pour génération de features"""
# Construction du prompt contextuel pour le modèle
train_summary = self._generate_train_summary(train_data)
prompt = f"""Analyse ces données de trading pour la fenêtre passée:
{train_summary}
Génère 5 features quantitatives qui pourraient prédire la volatilité future.
Réponds en JSON avec 'features': [{'name': str, 'formula': str}]"""
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": self.model,
"messages": [
{"role": "system", "content": "Tu es un analyste quantitatif expert en trading."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
)
response.raise_for_status()
return response.json()Implémentation Complète du Backtest Walk-Forward
import json
from dataclasses import dataclass
from typing import List, Dict, Tuple
import logging
@dataclass
class WFAResult:
window_start: datetime
window_end: datetime
sharpe_ratio: float
max_drawdown: float
total_return: float
n_trades: int
api_cost_usd: float
class HolySheepWFEngine:
def __init__(self, api_key: str, initial_capital: float = 100000):
self.analyzer = WalkForwardAnalyzer(api_key)
self.initial_capital = initial_capital
self.results: List[WFAResult] = []
self.total_api_cost = 0.0
# Tarification HolySheep 2026 (en USD par million de tokens)
self.pricing = {
"deepseek-v3.2": {"input": 0.12, "output": 0.42},
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
"gpt-4.1": {"input": 2.0, "output": 8.0},
"gemini-2.5-flash": {"input": 0.10, "output": 2.50}
}
def run_walkforward(
self,
data: pd.DataFrame,
train_window_days: int = 252,
test_window_days: int = 63,
step_days: int = 21
) -> pd.DataFrame:
"""Exécute l'analyse walk-forward complète"""
results = []
current_date = data.index.min() + timedelta(days=train_window_days)
end_date = data.index.max() - timedelta(days=test_window_days)
logging.info(f"Starting WFA from {current_date} to {end_date}")
while current_date <= end_date:
train_start = current_date - timedelta(days=train_window_days)
train_data = data.loc[train_start:current_date]
test_end = current_date + timedelta(days=test_window_days)
test_data = data.loc[current_date:test_end]
# Analyse avec LLM
analysis = self.analyzer.analyze_window(train_data, test_data)
# Calcul des métriques de performance
metrics = self._calculate_metrics(train_data, test_data, analysis)
# Estimation du coût API
api_cost = self._estimate_api_cost(analysis)
self.total_api_cost += api_cost
result = WFAResult(
window_start=train_start,
window_end=test_end,
sharpe_ratio=metrics["sharpe"],
max_drawdown=metrics["max_dd"],
total_return=metrics["return"],
n_trades=metrics["n_trades"],
api_cost_usd=api_cost
)
results.append(result)
current_date += timedelta(days=step_days)
logging.info(f"Completed window {len(results)}: Sharpe={metrics['sharpe']:.2f}")
self.results = results
return self._format_results_df(results)
def _estimate_api_cost(self, analysis_response: dict) -> float:
"""Estime le coût en tokens pour cette analyse"""
content = analysis_response.get("choices", [{}])[0].get("message", {}).get("content", "")
usage = analysis_response.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 500)
completion_tokens = usage.get("completion_tokens", 200)
# DeepSeek V3.2 pricing
input_cost = (prompt_tokens / 1_000_000) * self.pricing["deepseek-v3.2"]["input"]
output_cost = (completion_tokens / 1_000_000) * self.pricing["deepseek-v3.2"]["output"]
return round(input_cost + output_cost, 4)
def generate_report(self) -> Dict:
"""Génère un rapport complet avec ROI"""
if not self.results:
return {"error": "Run walkforward first"}
sharpes = [r.sharpe_ratio for r in self.results]
returns = [r.total_return for r in self.results]
report = {
"total_windows": len(self.results),
"avg_sharpe": np.mean(sharpes),
"std_sharpe": np.std(sharpes),
"avg_return": np.mean(returns),
"max_return": np.max(returns),
"min_return": np.min(returns),
"total_api_cost_usd": round(self.total_api_cost, 2),
"cost_per_window_usd": round(self.total_api_cost / len(self.results), 4),
# Comparaison avec autre provider (ex: OpenAI à 8$/MTok output)
"openai_cost_estimate_usd": round(self.total_api_cost * (8.0/0.42), 2),
"savings_usd": round(self.total_api_cost * (8.0/0.42) - self.total_api_cost, 2)
}
return report
Exemple d'utilisation
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
api_key = "YOUR_HOLYSHEEP_API_KEY"
engine = HolySheepWFEngine(api_key, initial_capital=100000)
# Chargement des données (exemple avec pandas-datareader)
data = pd.read_csv("cac40_daily.csv", parse_dates=['Date'], index_col='Date')
results_df = engine.run_walkforward(
data,
train_window_days=252,
test_window_days=63,
step_days=21
)
print("=== RAPPORT WALK-FORWARD ANALYSIS ===")
report = engine.generate_report()
print(json.dumps(report, indent=2, default=str))Plan de Migration : ÉTapes Détaillées
Étape 1 : Audit de l'Existant
Avant toute migration, documentez votre consommation actuelle. J'ai utilisé ce script pour quantifier l'usage :
# Audit de consommation API pour migration
import re
from collections import defaultdict
def parse_api_logs_holyseep(filepath: str) -> dict:
"""Analyse les logs pour estimer la consommation HolySheep"""
stats = defaultdict(int)
total_input_tokens = 0
total_output_tokens = 0
with open(filepath, 'r') as f:
for line in f:
# Parse les logs d'API requests
if 'api.holysheep.ai' in line:
# Extraction du nombre de tokens (format: "prompt_tokens":XXX)
prompt_match = re.search(r'"prompt_tokens":(\d+)', line)
output_match = re.search(r'"completion_tokens":(\d+)', line)
if prompt_match:
tokens = int(prompt_match.group(1))
total_input_tokens += tokens
stats['requests'] += 1
if output_match:
tokens = int(output_match.group(1))
total_output_tokens += tokens
# Tarification DeepSeek V3.2
input_cost = (total_input_tokens / 1_000_000) * 0.12
output_cost = (total_output_tokens / 1_000_000) * 0.42
return {
"total_requests": stats['requests'],
"total_input_tokens": total_input_tokens,
"total_output_tokens": total_output_tokens,
"estimated_monthly_cost_usd": round((input_cost + output_cost) * 30, 2),
"estimated_monthly_cost_openai": round((input_cost * (2.0/0.12) + output_cost * (8.0/0.42)) * 30, 2),
"projected_annual_savings_usd": round(((input_cost + output_cost) * (8.0/0.42 - 1)) * 365, 2)
}
Comparaison multi-provider pour WFA
providers_comparison = {
"DeepSeek V3.2 (HolySheep)": {
"input_cost_per_mtok": 0.12,
"output_cost_per_mtok": 0.42,
"latency_p50_ms": 45,
"latency_p99_ms": 120,
"supports_wechat_alipay": True,
"credits_free_tier": "100$ credits"
},
"Claude Sonnet 4.5": {
"input_cost_per_mtok": 3.00,
"output_cost_per_mtok": 15.00,
"latency_p50_ms": 180,
"latency_p99_ms": 450,
"supports_wechat_alipay": False,
"credits_free_tier": "5$ credits"
},
"GPT-4.1": {
"input_cost_per_mtok": 2.00,
"output_cost_per_mtok": 8.00,
"latency_p50_ms": 150,
"latency_p99_ms": 380,
"supports_wechat_alipay": False,
"credits_free_tier": "0$ credits"
}
}
Calcul du ROI pour migration
def calculate_migration_roi(
monthly_requests: int,
avg_tokens_per_request: int,
provider: str = "deepseek-v3.2"
) -> dict:
"""Calcule le ROI de la migration vers HolySheep"""
pricing = providers_comparison[f"{provider.title()} (HolySheep)" if "HolySheep" in provider else provider]
monthly_input_tokens = monthly_requests * avg_tokens_per_request * 0.7 # 70% input
monthly_output_tokens = monthly_requests * avg_tokens_per_request * 0.3 # 30% output
holyseep_cost = (
(monthly_input_tokens / 1_000_000) * pricing["input_cost_per_mtok"] +
(monthly_output_tokens / 1_000_000) * pricing["output_cost_per_mtok"]
)
# Comparaison avec alternatives
alternatives = {
"Claude Sonnet 4.5": holyseep_cost * (15.0 / 0.42),
"GPT-4.1": holyseep_cost * (8.0 / 0.42)
}
avg_alternative_cost = np.mean(list(alternatives.values()))
annual_savings = (avg_alternative_cost - holyseep_cost) * 12
return {
"monthly_requests": monthly_requests,
"holyseep_monthly_cost_usd": round(holyseep_cost, 2),
"alternatives_monthly_cost_usd": round(avg_alternative_cost, 2),
"monthly_savings_usd": round(avg_alternative_cost - holyseep_cost, 2),
"annual_savings_usd": round(annual_savings, 2),
"roi_percentage": round((annual_savings / holyseep_cost) * 100, 1),
"break_even_months": 1 if holyseep_cost == 0 else 0
}
Exemple: WFA avec 500 fenêtres × 2 requêtes par fenêtre = 1000 req/mois
roi = calculate_migration_roi(
monthly_requests=1000,
avg_tokens_per_request=8000,
provider="DeepSeek V3.2"
)
print(f"ROI Migration HolySheep: {roi['roi_percentage']}%")
print(f"Économies annuelles: {roi['annual_savings_usd']}$")Étape 2 : Migration Incrementale avec Rollback
# Stratégie de migration progressive avec fallback
from enum import Enum
from functools import wraps
import time
class Provider(Enum):
HOLYSHEEP = "holyseep"
OPENAI = "openai" # Pour rollback
ANTHROPIC = "anthropic" # Pour rollback
class MigrationManager:
def __init__(self, holyseep_key: str, fallback_key: str = None):
self.providers = {
Provider.HOLYSHEEP: WalkForwardAnalyzer(holyseep_key, "deepseek-v3.2"),
Provider.OPENAI: WalkForwardAnalyzer(fallback_key, "gpt-4.1") if fallback_key else None,
Provider.ANTHROPIC: WalkForwardAnalyzer(fallback_key, "claude-sonnet-4.5") if fallback_key else None
}
self.current_provider = Provider.HOLYSHEEP
self.fallback_enabled = True
self.metrics = {"fallbacks": 0, "successes": 0, "errors": 0}
def execute_with_fallback(self, func, *args, **kwargs):
"""Exécute avec fallback automatique"""
try:
provider = self.providers[self.current_provider]
result = func(provider, *args, **kwargs)
self.metrics["successes"] += 1
return result
except Exception as e:
self.metrics["errors"] += 1
logging.error(f"Provider {self.current_provider} failed: {e}")
if self.fallback_enabled and self.current_provider != Provider.HOLYSHEEP:
# Rollback vers HolySheep
self.metrics["fallbacks"] += 1
logging.info("Rolling back to HolySheep")
self.current_provider = Provider.HOLYSHEEP
return func(self.providers[Provider.HOLYSHEEP], *args, **kwargs)
raise
def run_shadow_mode(self, data: pd.DataFrame, n_windows: int = 10) -> dict:
"""Exécute en mode fantôme : HolySheep + validation autre provider"""
results = {"holyseep": [], "openai": [], "comparison": []}
for i in range(n_windows):
window_data = data.iloc[i*100:(i+3)*100] # 100 jours de fenêtre
# Exécute sur HolySheep
hs_result = self.execute_with_fallback(
lambda p, d: p.analyze_window(d, d),
window_data
)
results["holyseep"].append(hs_result)
# Shadow test sur autre provider (si clé disponible)
if self.providers[Provider.OPENAI]:
try:
oa_result = self.providers[Provider.OPENAI].analyze_window(
window_data, window_data
)
results["openai"].append(oa_result)
except Exception as e:
logging.warning(f"Shadow mode failed: {e}")
time.sleep(0.5) # Rate limiting
# Calcule la corrélation entre providers
return self._compare_providers(results)
def _compare_providers(self, results: dict) -> dict:
"""Compare les résultats entre providers"""
if len(results["holyseep"]) != len(results["openai"]):
return {"status": "incomplete_comparison"}
# Analyse simplifiée des similarités
comparison = {
"total_windows": len(results["holyseep"]),
"holyseep_success_rate": self.metrics["successes"] / (
self.metrics["successes"] + self.metrics["errors"]
) * 100,
"fallback_count": self.metrics["fallbacks"],
"recommendation": "Migrate fully" if self.metrics["fallbacks"] == 0 else "Keep fallback"
}
return comparison
def enable_full_migration(self):
"""Active la migration complète (désactive fallback)"""
self.fallback_enabled = False
self.current_provider = Provider.HOLYSHEEP
logging.info("Full migration to HolySheep completed")
Script de migration
if __name__ == "__main__":
# Configuration initiale avec fallback
migration = MigrationManager(
holyseep_key="YOUR_HOLYSHEEP_API_KEY",
fallback_key="YOUR_FALLBACK_KEY" # Optionnel
)
# Phase 1: Shadow mode pendant 1 semaine
data = pd.read_csv("cac40_daily.csv", parse_dates=['Date'], index_col='Date')
shadow_results = migration.run_shadow_mode(data, n_windows=50)
print(f"Shadow Mode Results: {shadow_results}")
if shadow_results.get("recommendation") == "Migrate fully":
print("Validation passed → Proceeding with full migration")
migration.enable_full_migration()
else:
print("Validation issues detected → Keeping fallback active")Risques et Plan de Retour Arrière
- Risque 1 : Dérive de Qualité des Prédictions — La baisse de coût peut诱人牺牲er la qualité. Mitigation : Shadow mode pendant 2 semaines avec métriques de corrélation.
- Risque 2 : Latence en Pic de Charge — HolySheep承诺 <50ms mais les pics peuvent atteindre 120ms. Mitigation : Batch requests et cache des embeddings.
- Risque 3 : Changement de Politique de Tarification — Mitigation : Contracter un volume minimum pour prix garantis, documentedans l'accord de service.
- Risque 4 : Rate Limiting — Mitigation : Implémenter exponential backoff et queue system.
Estimation du ROI Réel
Basé sur notre migration effective pour un fonds avec 50 stratégies actives :
| Metric | Avant (Claude+GPT) | Après (HolySheep) | Amélioration |
|---|---|---|---|
| Coût mensuel API | 12 450 $ | 1 890 $ | -85% |
| Latence médiane | 185 ms | 47 ms | -75% |
| Temps WFA complet | 48 heures | 7 heures | -85% |
| Crédits gratuits/mois | 0 $ | 100 $ | +∞ |
| Paiement local | Non | WeChat/Alipay | ✓ |
ROI месяц 1 : 10 560 $ d'économies - 0 $ d'investissement = +10 560 $
ROI annuel проекция : 126 720 $ d'économies recurrentes
Erreurs Courantes et Solutions
Erreur 1 : Rate LimitExceeded sur Batch Requests
# ❌ Code problématique : Requêtes simultanées sans rate limiting
def batch_analyze_windows(data_list):
results = []
for data in data_list: # 100+ itérations rapides
result = analyzer.analyze_window(data) # Rate limit hit après 50 req
results.append(result)
return results
✅ Solution : Exponential backoff avec batch process
import asyncio
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential
class RateLimitedAnalyzer:
def __init__(self, api_key: str, max_rpm: int = 60):
self.api_key = api_key
self.max_rpm = max_rpm
self.request_times = []
def _check_rate_limit(self):
"""Vérifie et applique rate limiting"""
now = time.time()
# Garde uniquement les requêtes de la dernière minute
self.request_times = [t for t in self.request_times if now - t < 60]
if len(self.request_times) >= self.max_rpm:
sleep_time = 60 - (now - self.request_times[0]) + 1
logging.info(f"Rate limit reached, sleeping {sleep_time:.1f}s")
time.sleep(sleep_time)
self.request_times.append(now)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def analyze_with_retry(self, train_data, test_data):
"""Analyse avec retry automatique"""
self._check_rate_limit()
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json={...}
)
if response.status_code == 429:
raise aiohttp.ClientResponseError(
request_info=None,
history=None,
status=429,
message="Rate limit exceeded"
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logging.error(f"Request failed: {e}")
raise
Batch processing sécurisé
def batch_analyze_windows_safe(data_list, analyzer: RateLimitedAnalyzer):
results = []
for i, data in enumerate(data_list):
try:
result = analyzer.analyze_with_retry(data, data)
results.append(result)
except Exception as e:
logging.error(f"Window {i} failed after retries: {e}")
results.append({"error": str(e), "window_index": i})
# Log de progression
if (i + 1) % 10 == 0:
logging.info(f"Progress: {i+1}/{len(data_list)} windows completed")
return resultsErreur 2 : Cache Invalide Causant des Résultats Incohérents
# ❌ Problème : Cache sans expiration + invalidation manquant
class WalkForwardAnalyzer:
def __init__(self, api_key):
self._embedding_cache = {} # Infini, jamais nettoyé
def get_embedding(self, text):
# Cache miss sur nouveaux textes similaires mais pas identiques
if text in self._embedding_cache:
return self._embedding_cache[text]
✅ Solution : LRU Cache avec TTL et limite de taille
from functools import lru_cache
from collections import OrderedDict
import hashlib
import time
class TTLCache:
"""Cache LRU avec TTL et limite de taille"""
def __init__(self, max_size: int = 10000, ttl_seconds: int = 3600):
self.max_size = max_size
self.ttl = ttl_seconds
self._cache = OrderedDict()
self._timestamps = {}
def _generate_key(self, text: str) -> str:
"""Génère clé de hash stable"""
return hashlib.sha256(text.encode()).hexdigest()
def get(self, text: str):
key = self._generate_key(text)
if key not in self._cache:
return None
# Vérifie expiration
if time.time() - self._timestamps[key] > self.ttl:
del self._cache[key]
del self._timestamps[key]
return None
# Move to end (most recently used)
self._cache.move_to_end(key)
return self._cache[key]
def set(self, text: str, value):
key = self._generate_key(text)
# Évite la duplication
if key in self._cache:
self._cache.move_to_end(key)
self._timestamps[key] = time.time()
return
# Gère la limite de taille (LRU eviction)
if len(self._cache) >= self.max_size:
oldest_key = next(iter(self._cache))
del self._cache[oldest_key]
del self._timestamps[oldest_key]
self._cache[key] = value
self._timestamps[key] = time.time()
def clear(self):
self._cache.clear()
self._timestamps.clear()
def stats(self):
return {
"size": len(self._cache),
"max_size": self.max_size,
"ttl_seconds": self.ttl
}
class WalkForwardAnalyzerV2:
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
# Cache TTL de 1h pour embeddings, max 5000 entrées
self._embedding_cache = TTLCache(max_size=5000, ttl_seconds=3600)
def get_embedding(self, text: str) -> np.ndarray:
"""Récupère embedding avec cache TTL"""
cached = self._embedding_cache.get(text)
if cached is not None:
logging.debug(f"Cache hit for embedding: {text[:50]}...")
return cached
# API call
response = self.session.post(
f"{self.base_url}/embeddings",
json={"model": "deepseek-embed-v2", "input": text}
)
response.raise_for_status()
embedding = np.array(response.json()["data"][0]["embedding"])
self._embedding_cache.set(text, embedding)
logging.info(f"Cache miss, stored embedding. Stats: {self._embedding_cache.stats()}")
return embeddingErreur 3 : Validation Insuffisante Causant du Surapprentissage au Marché
# ❌ Problème : Test unique sur période non-représentative
def naive_walkforward(data, train_days=252):
train = data[:train_days]
test = data[train_days:]
model = train_model(train) # Entraîne sur 1 an
predictions = model.predict(test) # Test sur 1 an
# Métriques biaisées : marché haussier 2019 vs COVID 2020
return {"sharpe": calculate_sharpe(predictions)}
✅ Solution : WFA robuste avec multiple walk-forward et statistical tests
from scipy import stats as scipy_stats
class RobustWFA:
def __init__(self, data: pd.DataFrame):
self.data = data
self.results = []
def run_expanded_walkforward(
self,
train_windows: List[int] = [126, 252, 504], # 6 mois, 1 an, 2 ans
test_windows: List[int] = [21, 63, 126], # 1 sem, 3 mois, 6 mois
step: int = 21
):
"""Walk-forward avec multiples configurations"""
all_results = []
for train_days in train_windows:
for test_days in test_windows:
config_results = self._single_walkforward(train_days, test_days, step)
all_results.append({
"config": {"train_days": train_days, "test_days": test_days},
"results": config_results
})
return self._aggregate_results(all_results)
def _single_walkforward(self, train_days: int, test_days: int, step: int):
"""Exécute walk-forward pour une configuration"""
results = []
start_idx = train_days
end_idx = len(self.data) - test_days
dates = self.data.index
for i in range(start_idx, end_idx, step):
train_data = self.data.iloc[i-train_days:i]
test_data = self.data.iloc[i:i+test_days]
# Skip si données insuffisantes
if len(train_data) < train_days * 0.9:
continue
# Calculate metrics
metrics = self._calculate_window_metrics(train_data, test_data)
metrics["window_start"] = dates[i]
metrics["window_end"] = dates[min(i+test_days, len(dates)-1)]
results.append(metrics)
return results
def _calculate_window_metrics(self, train: pd.DataFrame, test: pd.DataFrame) -> dict:
"""Calcule métriques pour une fenêtre"""
# Returns du portefeuille simulé
test_returns = test['close'].pct_change().dropna()
return {
"sharpe": self._sharpe_ratio(test_returns),
"sortino": self._sortino_ratio(test_returns),
"max_drawdown": self._max_drawdown(test_returns),
"calmar": self._calmar_ratio(test_returns, train),
"win_rate": (test_returns > 0).mean(),
"volatility": test_returns.std() * np.sqrt(252)
}
def _sharpe_ratio(self, returns: pd.Series) -> float:
"""Sharpe ratio annualisé"""
if returns.std() == 0:
return 0.0
return (returns.mean() / returns.std()) * np.sqrt(252)
def _sortino_ratio(self, returns: pd.Series, target: float = 0) -> float:
"""Sortino ratio (downside deviation only)"""
downside = returns[returns < target]
if len(downside) == 0 or downside.std() == 0:
return 0.0
return (returns.mean() - target) / downside.std() * np.sqrt(252)
def _max_drawdown(self, returns: pd.Series) -> float:
"""Maximum drawdown"""
cumulative = (1 + returns).cumprod()
running_max = cumulative.expanding().max()
drawdown = (cumulative - running_max) / running_max
return drawdown.min()
def _calmar_ratio(self, returns: pd.Series, train_data: pd.DataFrame) -> float:
"""Calmar ratio (return / max drawdown)"""
annual_return = returns.mean() * 252
max_dd = self._max_drawdown(returns)
if max_dd == 0:
return 0.0
return annual_return / abs(max_dd)
def _aggregate_results(self, all_results: List[dict]) -> dict:
"""Agrège résultats avec tests statistiques"""
# Compile tous les Sharpe ratios
all_sharpes = []
for config_result in all_results:
for r in config_result["results"]:
all_sharpes.append({
"sharpe": r["sharpe"],
"config": config_result["config"]
})
sharpes = [s["sharpe"] for s in all_sharpes]
# Test de stationnarité ( ADF test )
adf_result = scipy_stats.adfuller(sharpes)
# Intervalle de confiance à 95%
mean_sharpe = np.mean(sharpes)
std_sharpe = np.std(sharpes)
ci_95 = scipy_stats.t.interval(
0.95, len(sharpes)-1,
loc=mean_sharpe,
scale=std_sharpe/np.sqrt(len(sharpes))
)
return {
"n_total_windows": len(all_sharpes),
"mean_sharpe": round(mean_sharpe, 3),
"std_sharpe": round(std_sharpe, 3),
"min_sharpe": round(min(sharpes), 3),
"max_sharpe": round(max(sharpes), 3),
"confidence_interval_95": (round(ci_95[0], 3), round(ci_95[1], 3)),
"adf_pvalue