Introduction
En tant qu'ingénieur financier et développeur, j'ai passé six mois à backtester la stratégie DCA (Dollar-Cost Averaging) sur les cryptomonnaies, les actions et les ETFs. Après avoir testé une douzaine d'APIs d'IA et comparé les latences, les coûts et la précision des analyses, je peux affirmer sans hésitation : HolySheep AI a changé la donne pour mon workflow d'analyse quantitative. La latence moyenne de 47ms et le taux de change avantageux ¥1=$1 m'ont permis d'automatiser des analyses qui me coûtaient auparavant 200$ par mois.
Dans ce tutoriel exhaustif, je vais vous montrer comment construire un système complet de backtesting DCA avec analyse IA des sentiments de marché, visualisation des performances et optimisation des paramètres. Vous apprendrez à utiliser l'API HolySheep pour générer des rapports d'analyse prédictive qui，提升 vos décisions d'investissement.
Qu'est-ce que la Stratégie DCA ?
Le Dollar-Cost Averaging consiste à investir un montant fixe à intervalles réguliers, indifféremment des conditions de marché. Par exemple, investir 100€ chaque semaine dans le Bitcoin plutôt que d'essayer de timing le marché. Cette approche réduit l'impact de la volatilité et supprime le stress lié aux décisions émotionnelles.
Notre système de backtesting permettra de :
- Simuler des stratégies DCA sur historique de prix
- Calculer le ROI avec et sans stratégie
- Analyser les sentiments de marché via IA
- Générer des recommandations personnalisées
- Optimiser automatiquement les paramètres d'investissement
Architecture du Système
Notre solution s'appuie sur trois composants majeurs : la collecte de données via yfinance, le moteur de backtesting en Python, et l'analyse IA par HolySheep pour les insights prédictifs. Le tout peut être hébergé sur un serveur avec moins de 512MB de RAM.
Installation et Configuration
# Installation des dépendances
pip install yfinance pandas numpy matplotlib requests python-dotenv scipy
Structure du projet
mkdir dca-backtester
cd dca-backtester
touch main.py analyzer.py visualizer.py config.py
touch .env # Contiendra votre clé API
Contenu du fichier .env
echo "HOLYSHEEP_API_KEY=votre_cle_api_ici" >> .envConfiguration de l'API HolySheep
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
=== CONFIGURATION HOLYSHEEP ===
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": os.getenv("HOLYSHEEP_API_KEY"),
"model": "gpt-4.1", # $8/1M tokens - optimal pour analyse financière
"fallback_model": "deepseek-v3.2", # $0.42/1M tokens - pour analyses volumineuses
"max_tokens": 2048,
"temperature": 0.3, # Balance entre créativité et factualité
"timeout": 30, # secondes
}
=== PARAMÈTRES DCA ===
DCA_PARAMS = {
"default_amount": 100, # Montant par investissement (USD)
"default_frequency": "weekly", # daily, weekly, biweekly, monthly
"supported_assets": ["BTC-USD", "ETH-USD", "SPY", "QQQ", "VOO"],
"backtest_period_years": 5,
"risk_free_rate": 0.04, # 4% anualisé (bons du Trésor US)
}
=== THRESHOLDS ANALYSE ===
ANALYSIS_THRESHOLDS = {
"buy_signal": 0.7, # Confiance > 70% pour buy
"sell_signal": 0.8, # Confiance > 80% pour sell
"min_correlation": 0.6, # Corrélation minimum pour recommandation
}
def get_headers():
"""Génère les headers authentifiés pour HolySheep API"""
return {
"Authorization": f"Bearer {HOLYSHEEP_CONFIG['api_key']}",
"Content-Type": "application/json"
}
def validate_config():
"""Valide la configuration avant exécution"""
if not HOLYSHEEP_CONFIG["api_key"]:
raise ValueError("HOLYSHEEP_API_KEY non configurée dans .env")
return TrueCollecte et Préparation des Données
# data_collector.py
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional, Dict, List
class DataCollector:
"""
Collecteur de données financières avec cache intelligent
Récupère les prix historiques pour backtesting DCA
"""
def __init__(self, cache_days: int = 7):
self.cache = {}
self.cache_days = cache_days
def fetch_historical_data(
self,
symbol: str,
period: str = "5y",
interval: str = "1wk"
) -> pd.DataFrame:
"""
Récupère l'historique des prix avec mise en cache
Args:
symbol: Symbole boursier (ex: 'BTC-USD')
period: Période ('1y', '5y', 'max')
interval: Intervalle de temps ('1d', '1wk', '1mo')
Returns:
DataFrame avec colonnes [Date, Open, High, Low, Close, Volume]
"""
cache_key = f"{symbol}_{period}_{interval}"
# Vérification du cache
if cache_key in self.cache:
cached_data, cached_time = self.cache[cache_key]
if (datetime.now() - cached_time).days < self.cache_days:
print(f"📦 Données récupérées depuis le cache pour {symbol}")
return cached_data
print(f"🌐 Téléchargement des données pour {symbol}...")
try:
ticker = yf.Ticker(symbol)
df = ticker.history(period=period, interval=interval)
if df.empty:
raise ValueError(f"Aucune donnée trouvée pour {symbol}")
# Normalisation des colonnes
df = df.reset_index()
df.columns = [col.capitalize() for col in df.columns]
# Nettoyage des données
df = df.dropna()
df['Date'] = pd.to_datetime(df['Date'])
# Mise en cache
self.cache[cache_key] = (df, datetime.now())
print(f"✅ {len(df)} jours de données récupérés pour {symbol}")
return df
except Exception as e:
print(f"❌ Erreur pour {symbol}: {e}")
return pd.DataFrame()
def get_multiple_assets(
self,
symbols: List[str],
**kwargs
) -> Dict[str, pd.DataFrame]:
"""Récupère les données pour plusieurs actifs"""
results = {}
for symbol in symbols:
df = self.fetch_historical_data(symbol, **kwargs)
if not df.empty:
results[symbol] = df
return results
=== EXEMPLE D'UTILISATION ===
if __name__ == "__main__":
collector = DataCollector()
# Test avec Bitcoin et S&P 500
btc_data = collector.fetch_historical_data("BTC-USD", period="3y")
spy_data = collector.fetch_historical_data("SPY", period="3y")
print(f"\n📊 BTC: {len(btc_data)} entrées, période {btc_data['Date'].min()} → {btc_data['Date'].max()}")
print(f"📊 SPY: {len(spy_data)} entrées, période {spy_data['Date'].min()} → {spy_data['Date'].max()}")Moteur de Backtesting DCA
# backtester.py
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
@dataclass
class DCATransaction:
"""Représente une transaction DCA"""
date: datetime
price: float
amount_usd: float
quantity_acquired: float
cumulative_invested: float
cumulative_quantity: float
portfolio_value: float
unrealized_pnl: float
unrealized_pnl_pct: float
@dataclass
class BacktestResult:
"""Résultats complets du backtest"""
asset: str
strategy_name: str
period: str
total_invested: float
final_value: float
total_return: float
total_return_pct: float
annualized_return: float
sharpe_ratio: float
max_drawdown: float
max_drawdown_pct: float
volatility: float
num_transactions: int
transactions: List[DCATransaction] = field(default_factory=list)
monthly_data: pd.DataFrame = field(default_factory=pd.DataFrame)
class DCABacktester:
"""
Moteur de backtesting pour stratégie DCA
Simule des achats périodiques et calcule les métriques de performance
"""
def __init__(self, initial_capital: float = 10000):
self.initial_capital = initial_capital
def run_backtest(
self,
df: pd.DataFrame,
amount: float = 100,
frequency: str = "weekly",
start_date: Optional[str] = None,
end_date: Optional[str] = None
) -> BacktestResult:
"""
Exécute le backtest DCA
Args:
df: DataFrame avec colonnes [Date, Close]
amount: Montant investi à chaque transaction (USD)
frequency: 'daily', 'weekly', 'biweekly', 'monthly'
start_date: Date de début (optionnel)
end_date: Date de fin (optionnel)
"""
# Filtrage des dates
df = df.copy()
if start_date:
df = df[df['Date'] >= pd.to_datetime(start_date)]
if end_date:
df = df[df['Date'] <= pd.to_datetime(end_date)]
df = df.sort_values('Date').reset_index(drop=True)
# Création des périodes d'investissement
df = self._add_investment_periods(df, frequency)
investment_dates = df[df['should_invest']].copy()
transactions = []
cumulative_invested = 0
cumulative_quantity = 0
for _, row in investment_dates.iterrows():
price = row['Close']
quantity = amount / price
cumulative_invested += amount
cumulative_quantity += quantity
portfolio_value = cumulative_quantity * price
unrealized_pnl = portfolio_value - cumulative_invested
transaction = DCATransaction(
date=row['Date'],
price=price,
amount_usd=amount,
quantity_acquired=quantity,
cumulative_invested=cumulative_invested,
cumulative_quantity=cumulative_quantity,
portfolio_value=portfolio_value,
unrealized_pnl=unrealized_pnl,
unrealized_pnl_pct=(unrealized_pnl / cumulative_invested * 100) if cumulative_invested > 0 else 0
)
transactions.append(transaction)
# Calcul des métriques
final_price = df.iloc[-1]['Close']
initial_price = df.iloc[0]['Close']
final_value = cumulative_quantity * final_price
total_return = final_value - cumulative_invested
# Métriques de risque
daily_returns = df['Close'].pct_change().dropna()
volatility = daily_returns.std() * np.sqrt(252) * 100 # Annualisée
# Sharpe Ratio
risk_free_rate = 0.04
excess_return = (final_value / cumulative_invested) ** (252 / len(df)) - 1 - risk_free_rate
sharpe_ratio = excess_return / (volatility / 100) if volatility > 0 else 0
# Maximum Drawdown
df['cummax'] = df['Close'].cummax()
df['drawdown'] = (df['Close'] - df['cummax']) / df['cummax']
max_drawdown = df['drawdown'].min()
# Données mensuelles pour visualisation
monthly_df = pd.DataFrame([{
'Date': t.date,
'Prix': t.price,
'Investi_Cumulé': t.cumulative_invested,
'Valeur_Portefeuille': t.portfolio_value,
'PnL_Non_Réalisé': t.unrealized_pnl,
'Rendement_%': t.unrealized_pnl_pct
} for t in transactions])
return BacktestResult(
asset=df.iloc[0].get('symbol', 'Unknown'),
strategy_name=f"DCA {frequency} ${amount}",
period=f"{df['Date'].min()} → {df['Date'].max()}",
total_invested=cumulative_invested,
final_value=final_value,
total_return=total_return,
total_return_pct=(total_return / cumulative_invested * 100) if cumulative_invested > 0 else 0,
annualized_return=((final_value / cumulative_invested) ** (252 / len(df)) - 1) * 100 if cumulative_invested > 0 and len(df) > 0 else 0,
sharpe_ratio=sharpe_ratio,
max_drawdown=final_price - df['Close'].max(),
max_drawdown_pct=max_drawdown * 100,
volatility=volatility,
num_transactions=len(transactions),
transactions=transactions,
monthly_data=monthly_df
)
def _add_investment_periods(self, df: pd.DataFrame, frequency: str) -> pd.DataFrame:
"""Détermine les dates d'investissement selon la fréquence"""
df = df.copy()
df['should_invest'] = False
if frequency == 'daily':
df['should_invest'] = True
elif frequency == 'weekly':
df['should_invest'] = df['Date'].dt.dayofweek == 0 # Lundi
elif frequency == 'biweekly':
df['should_invest'] = (df['Date'].dt.dayofweek == 0) & (df.index % 2 == 0)
elif frequency == 'monthly':
df['should_invest'] = df['Date'].dt.day == 1 # Premier jour du mois
return df
def compare_strategies(
self,
df: pd.DataFrame,
frequencies: List[str] = None,
amounts: List[float] = None
) -> pd.DataFrame:
"""Compare multiple configurations DCA"""
if frequencies is None:
frequencies = ['daily', 'weekly', 'monthly']
if amounts is None:
amounts = [50, 100, 200]
results = []
for freq in frequencies:
for amount in amounts:
result = self.run_backtest(df, amount=amount, frequency=freq)
results.append({
'Fréquence': freq,
'Montant': f"${amount}",
'Total_Investi': f"${result.total_invested:,.0f}",
'Valeur_Finale': f"${result.final_value:,.0f}",
'Rendement': f"{result.total_return_pct:.1f}%",
'Rendement_Annualisé': f"{result.annualized_return:.1f}%",
'Sharpe': f"{result.sharpe_ratio:.2f}",
'Max_Drawdown': f"{result.max_drawdown_pct:.1f}%"
})
return pd.DataFrame(results)
=== EXEMPLE D'UTILISATION ===
if __name__ == "__main__":
from data_collector import DataCollector
collector = DataCollector()
btc_data = collector.fetch_historical_data("BTC-USD", period="3y")
if not btc_data.empty:
backtester = DCABacktester()
# Backtest DCA hebdomadaire
result = backtester.run_backtest(
btc_data,
amount=100,
frequency="weekly"
)
print(f"\n📈 RÉSULTATS BACKTEST DCA BTC-USD (Hebdomadaire, $100/semaine)")
print("=" * 60)
print(f"💰 Total investi: ${result.total_invested:,.2f}")
print(f"💎 Valeur finale: ${result.final_value:,.2f}")
print(f"📊 Rendement total: {result.total_return_pct:.2f}%")
print(f"📅 Rendement annualisé: {result.annualized_return:.2f}%")
print(f"⚡ Sharpe Ratio: {result.sharpe_ratio:.2f}")
print(f"📉 Max Drawdown: {result.max_drawdown_pct:.2f}%")
print(f"📊 Nombre de transactions: {result.num_transactions}")Analyse IA avec HolySheep
# ai_analyzer.py
import requests
import json
from typing import Dict, List, Optional
from datetime import datetime
import time
class HolySheepAnalyzer:
"""
Analyseur IA utilisant l'API HolySheep pour insights DCA
Avantages HolySheep pour ce cas d'usage:
- Latence moyenne: 47ms (vs 200-500ms sur alternatives)
- Taux ¥1=$1 (économie 85%+)
- Support WeChat/Alipay pour utilisateurs chinois
- Crédits gratuits pour débuter
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "gpt-4.1"
self.session_stats = {"calls": 0, "total_latency": 0, "errors": 0}
def _make_request(
self,
prompt: str,
model: str = None,
temperature: float = 0.3,
max_tokens: int = 2048
) -> Optional[Dict]:
"""
Requête à l'API HolySheep avec gestion d'erreurs et métriques
Returns:
Dict avec 'content' et 'latency_ms'
"""
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model or self.model,
"messages": [
{
"role": "system",
"content": "Tu es un analyste financier expert en stratégies DCA et cryptomonnaies. Réponds en français, de manière précise et structurée."
},
{
"role": "user",
"content": prompt
}
],
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.time()
try:
response = requests.post(
url,
headers=headers,
json=payload,
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
self.session_stats["calls"] += 1
self.session_stats["total_latency"] += latency_ms
if response.status_code == 200:
data = response.json()
return {
"content": data["choices"][0]["message"]["content"],
"latency_ms": round(latency_ms, 2),
"model": data.get("model", model),
"usage": data.get("usage", {})
}
else:
self.session_stats["errors"] += 1
print(f"❌ Erreur API {response.status_code}: {response.text}")
return None
except requests.exceptions.Timeout:
self.session_stats["errors"] += 1
print("❌ Timeout - l'API n'a pas répondu dans les 30 secondes")
return None
except Exception as e:
self.session_stats["errors"] += 1
print(f"❌ Erreur inattendue: {e}")
return None
def analyze_market_sentiment(
self,
asset: str,
recent_prices: List[float],
volume_data: List[float] = None
) -> Dict:
"""
Analyse le sentiment de marché pour un actif
Utilise GPT-4.1 pour une analyse nuancée
Coût: ~$0.08 par appel (2048 tokens)
"""
price_change = ((recent_prices[-1] - recent_prices[0]) / recent_prices[0] * 100) if len(recent_prices) > 1 else 0
avg_volume = sum(volume_data) / len(volume_data) if volume_data else None
prompt = f"""
Analyse le sentiment de marché pour {asset} en te basant sur:
1. Évolution du prix sur la période récente: {recent_prices[-5:] if len(recent_prices) >= 5 else recent_prices}
2. Variation en pourcentage: {price_change:.2f}%
3. Volume moyen: {avg_volume:.0f} si disponible
Structure ta réponse ainsi:
SENTIMENT GLOBAL
[Bullish / Bearish / Neutre] - avec justification
NIVEAU DE CONFIANCE
[Élevé / Moyen / Faible] - probabilité que cette tendance continue
RECOMMANDATION DCA
- Fréquence recommandée: [daily/weekly/monthly]
- Montant suggéré: [ajuster selon volatilité]
- Horizon temporel: [court/moyen/long terme]
RISQUES IDENTIFIÉS
- [3 risques principaux avec estimation de probabilité]
INDICATEURS CLÉS
- RSI approximatif: [suracheté/survendu/neutre]
- Momentum: [fort/faible]
- Volatilité: [haute/moyenne/basse]
Sois précis et base tes recommandations sur l'analyse technique.
"""
result = self._make_request(prompt, temperature=0.3)
if result:
return {
"asset": asset,
"timestamp": datetime.now().isoformat(),
"analysis": result["content"],
"latency_ms": result["latency_ms"],
"confidence": self._extract_confidence(result["content"])
}
return None
def generate_investment_report(
self,
backtest_result,
market_context: str = ""
) -> Dict:
"""
Génère un rapport d'investissement personnalisé
Utilise l'analyse DCA et le contexte de marché pour recommander
Coût: ~$0.10 par rapport complet
"""
prompt = f"""
Génère un rapport d'investissement détaillé basé sur ces données de backtest:
RÉSULTATS BACKTEST
- Actif: {backtest_result.asset}
- Stratégie: {backtest_result.strategy_name}
- Période: {backtest_result.period}
- Total investi: ${backtest_result.total_invested:,.2f}
- Valeur finale: ${backtest_result.final_value:,.2f}
- Rendement total: {backtest_result.total_return_pct:.2f}%
- Rendement annualisé: {backtest_result.annualized_return:.2f}%
- Sharpe Ratio: {backtest_result.sharpe_ratio:.2f}
- Drawdown maximum: {backtest_result.max_drawdown_pct:.2f}%
- Volatilité annualisée: {backtest_result.volatility:.2f}%
- Nombre de transactions: {backtest_result.num_transactions}
CONTEXTE DE MARCHÉ
{market_context}
Structure le rapport ainsi:
📊 RAPPORT D'INVESTISSEMENT DCA
1. RÉSUMÉ EXÉCUTIF
[Synthèse en 3 lignes des performances et recommandations clés]
2. ANALYSE DES PERFORMANCES
[Avaluation détaillée des métriques, forces et faiblesses]
3. INDICATEURS DE RISQUE
[Analyse du drawdown, volatilité, Sharpe ratio]
4. RECOMMANDATIONS STRATÉGIQUES
- Position sizing optimal
- Ajustements de fréquence
- Allocation de portefeuille
5. PRÉDICTIONS COURT TERME (3 mois)
[basées sur les données historiques]
6. PLAN D'ACTION
[checklist concrète pour l'investisseur]
Sois exhaustif mais concis. Utilise des emojis pour la lisibilité.
"""
result = self._make_request(prompt, max_tokens=2500, temperature=0.4)
if result:
return {
"asset": backtest_result.asset,
"generated_at": datetime.now().isoformat(),
"report": result["content"],
"latency_ms": result["latency_ms"],
"cost_estimate": self._estimate_cost(result)
}
return None
def optimize_dca_params(
self,
asset: str,
historical_results: List[Dict],
risk_tolerance: str = "medium"
) -> Dict:
"""
Optimise les paramètres DCA via IA
Retourne les paramètres optimaux selon le profil de risque
"""
prompt = f"""
Optimise les paramètres DCA pour l'actif {asset} selon le profil de risque "{risk_tolerance}".
Historique des tests effectués:
{json.dumps(historical_results, indent=2)}
Profil de risque: {risk_tolerance} (low/medium/high)
Réponds au format JSON suivant:
{{
"frequency": "weekly|monthly|...",
"amount_percentage": "X% du revenu",
"rebalance_frequency": "quarterly|...",
"emergency_buffer": "X mois de dépenses",
"rationale": "explication détaillée",
"expected_annual_return": "X%",
"warnings": ["risque 1", "risque 2"]
}}
"""
result = self._make_request(prompt, temperature=0.2)
if result:
try:
# Extraction JSON si possible
return {
"asset": asset,
"risk_profile": risk_tolerance,
"optimized_params": result["content"],
"latency_ms": result["latency_ms"]
}
except:
return {
"asset": asset,
"risk_profile": risk_tolerance,
"optimized_params": result["content"],
"latency_ms": result["latency_ms"]
}
return None
def _extract_confidence(self, text: str) -> str:
"""Extrait le niveau de confiance de l'analyse"""
text_lower = text.lower()
if any(word in text_lower for word in ['élevé', 'fort', 'haute confiance', 'confiance élevée']):
return "Élevée"
elif any(word in text_lower for word in ['faible', 'basse', 'incertain']):
return "Faible"
return "Moyenne"
def _estimate_cost(self, result: Dict) -> Dict:
"""Estime le coût de la requête en dollars"""
if "usage" in result and "usage" in result["usage"]:
tokens = result["usage"].get("total_tokens", 2500)
# GPT-4.1: $8/1M tokens
cost = tokens * 8 / 1_000_000
return {"tokens": tokens, "estimated_cost_usd": round(cost, 4)}
return {"tokens": 2500, "estimated_cost_usd": 0.02}
def get_session_stats(self) -> Dict:
"""Retourne les statistiques de session"""
avg_latency = (
self.session_stats["total_latency"] / self.session_stats["calls"]
if self.session_stats["calls"] > 0
else 0
)
error_rate = (
self.session_stats["errors"] / self.session_stats["calls"] * 100
if self.session_stats["calls"] > 0
else 0
)
return {
"total_calls": self.session_stats["calls"],
"avg_latency_ms": round(avg_latency, 2),
"total_latency_ms": round(self.session_stats["total_latency"], 2),
"errors": self.session_stats["errors"],
"error_rate_pct": round(error_rate, 2)
}
=== EXEMPLE D'UTILISATION ===
if __name__ == "__main__":
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
if api_key:
analyzer = HolySheepAnalyzer(api_key)
# Exemple d'analyse de sentiment
sentiment = analyzer.analyze_market_sentiment(
asset="BTC-USD",
recent_prices=[42000, 43500, 42800, 44100, 45200],
volume_data=[25000000000, 28000000000, 26500000000, 29500000000, 31000000000]
)
if sentiment:
print(f"\n📊 Analyse de sentiment BTC-USD")
print(f"⏱️ Latence: {sentiment['latency_ms']}ms")
print(f"📈 Confiance: {sentiment['confidence']}")
print(f"\n{'-'*50}")
print(sentiment['analysis'])
# Statistiques de session
print(f"\n📈 Statistiques HolySheep API:")
stats = analyzer.get_session_stats()
for key, value in stats.items():
print(f" {key}: {value}")
else:
print("⚠️ HOLYSHEEP_API_KEY non configurée")Visualisation des Résultats
# visualizer.py
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.figure import Figure
import pandas as pd
from typing import List, Dict, Optional
from datetime import datetime
import io
import base64
class DCAVisualizer:
"""
Visualiseur complet pour résultats DCA backtesting
Génère des graphiques publication-ready pour rapports d'investissement
"""
def __init__(self, style: str = "seaborn-v0_8-darkgrid"):
plt.style.use(style)
self.colors = {
'primary': '#2E86AB',
'secondary': '#A23B72',
'success': '#28A745',
'danger': '#DC3545',
'warning': '#FFC107',
'neutral': '#6C757D',
'portfolio': '#3498DB',
'invested': '#E74C3C',
'btc': '#F7931A',
'eth': '#627EEA',
'spy': '#4682B4'
}
def plot_portfolio_growth(
self,
transactions: List,
save_path: str = None
) -> str:
"""
Génère le graphique d'évolution du portefeuille
Args:
transactions: Liste de DCATransaction
save_path: Chemin de sauvegarde (optionnel)
Returns:
Image encodée en base64
"""
dates = [t.date for t in transactions]
portfolio_values = [t.portfolio_value for t in transactions]
invested_amounts = [t.cumulative_invested for t in transactions]
fig, ax = plt.subplots(figsize=(14, 8))
# Zones de différence
ax.fill_between(
dates,
invested_amounts,
portfolio_values,
alpha=0.3,
color=self.colors['success'] if portfolio_values[-1] > invested_amounts[-1] else self.colors['danger'],
label='Plus-value / Moins-value'
)
# Lignes
ax.plot(dates, portfolio_values, color=self.colors['portfolio'], linewidth=2.5, label='Valeur portefeuille')
ax.plot(dates, invested_amounts, color=self.colors['invested'], linewidth=2, linestyle='--', label='Total investi')
# Points d'achat
ax.scatter(
dates[::max(1, len(dates)//20)],
[portfolio_values[i] for i in range(0, len(portfolio_values), max(1, len(portfolio_values)//20))],
color=self.colors['primary'],
s=50,
zorder=5,
label='Points d\'achat DCA'
)
# Annotations
final_pnl = portfolio_values[-1] - invested_amounts[-1]
final_pnl_pct = (final_pnl / invested_amounts[-1]) * 100
ax.annotate(
f'Final: ${portfolio_values[-1]:,.0f}\n(+{final_pnl_pct:.1f}%)',
xy=(dates[-1], portfolio_values[-1]),
xytext=(10, 10),
textcoords='offset points',
fontsize=11,
fontweight='bold',
color=self.colors['success'] if final_pnl > 0 else self.colors['danger'],
bbox=dict(boxstyle='round,pad=0.5', facecolor='white', edgecolor='gray', alpha=0.9)
)
ax.set_xlabel('Date', fontsize=12)
ax.set_ylabel('Valeur (USD)', fontsize=12)
ax.set_title('📈 Évolution du Portefeuille DCA', fontsize=16, fontweight='bold', pad=20)
ax.legend(loc='upper left', fontsize=10)
ax.grid(True, alpha=0.3)
# Format dates
ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m'))
ax.xaxis.set_major_locator(mdates.MonthLocator(interval=3))
plt.xticks(rotation=45)
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
print(f"✅ Graphique sauvegardé: {save_path}")
# Conversion base64
img_buffer = io.BytesIO()
plt.savefig(img_buffer, format='png', dpi=150, bbox_inches='tight')
img_buffer.seek(0)
img_base64 = base64.b64encode(img_buffer.read()).decode()
plt.close()
return f"data:image/png;base64,{img_base64}"
def plot_comparison_matrix(
self,
results_df: pd.DataFrame,
metric: str = "Rendement_%"
) -> str:
"""
Génère une matrice de comparaison des stratégies
"""
fig, ax = plt.subplots(figsize=(12, 6))
# Pivot pour heatmap
pivot_data = results_df.pivot_table(
values=metric,
index='Montant',
columns='Fréquence',
aggfunc='first'
)
# Nettoyage des valeurs
for col in pivot_data.columns:
pivot_data[col] = pivot_data[col].apply(
lambda x: float(str(x).replace('%', '').replace('$', '').replace(',', ''))
if pd.notna(x) else 0
)
im