En tant qu'ingénieur quantitatif ayant déployé des stratégies de trading algorithmique sur les marchés crypto depuis 2019, je peux vous dire que la question de l'attribution de performance constitue l'un des défis les plus complexes de notre métier. Quand votre stratégie génère +45% sur l'année, comment déterminer si cette performance provient réellement de votre capacité prédictive (alpha pur) ou simplement de votre exposition au marché Bitcoin qui a lui-même grimpé de 60% ? Cette distinction n'est pas académique : elle conditionne directement vos décisions de risk management et d'allocation de capital.
Dans cet article, je vous détaille l'architecture complète d'un système d'attribution de performance inspiré du modèle de Tardis, en combinant données on-chain, données de marché traditionnelles et intelligence artificielle. Nous couvrons l'ensemble du pipeline, de la collecte des données à la visualisation des résultats, avec du code production-ready et des benchmarks chiffrés.
Architecture Globale du Système d'Attribution
Un système d'attribution de performance robuste pour stratégies crypto doit répondre à plusieurs contraintes spécifiques : haute fréquence des données (tick par tick pour certains échanges), liquidité fragmentée sur des milliers de paires, volatilité extrême des actifs, et absence historique de données factorielles standardisées.
Les Composantes de la Performance
La performance totale d'un portefeuille de trading quantitatif crypto peut se décomposer ainsi :
Performance_Totale = Σ(w_i × r_i)
Avec :
- w_i : poids du facteur i dans le portefeuille
- r_i : rendement du facteur i
Décomposition de r_i :
r_i = β_market × r_market
+ β_momentum × r_momentum
+ β_size × r_size
+ β_volatility × r_volatility
+ α_selection × θ_selection
+ α_timing × θ_timing
+ ε_residual
Cette formule peut sembler simple, mais sa mise en œuvre pratique implique des défis considérables. Le modèle de Tardis que je vous présente ici résout ces problèmes en intégrant une couche d'intelligence artificielle pour affiner les estimations factorielles.
Schéma d'Architecture du Pipeline
┌─────────────────────────────────────────────────────────────────┐
│ PIPELINE D'ATTRIBUTION TARDIS │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Sources │───▶│ Collecteur │───▶│ Normaliseur │ │
│ │ Crypto │ │ (WebSocket) │ │ Temporel │ │
│ └──────────┘ └──────────────┘ └──────────────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ ENTREPÔT DE DONNÉES HONEYCOMB │ │
│ │ - OHLCV 1min, 5min, 1h, 1D │ │
│ │ - Order Book Snapshots │ │
│ │ - Transactions Mempool │ │
│ │ - Métriques On-Chain (TVL, Gas, NVT) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ CALCULATEUR DE FACTEURS │ │
│ │ - Beta Marché (Rolling 30 jours) │ │
│ │ - Momentum Factor (RSI, MACD normalisés) │ │
│ │ - Size Factor (Market Cap buckets) │ │
│ │ - Volatility Factor (GARCH 1,1) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ DÉCOMPOSEUR D'ALPHA HOLYSHEEP │ │
│ │ Modèle de régression enrichi IA │ │
│ │ base_url: https://api.holysheep.ai/v1 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ TABLEAU DE BORD │ │
│ │ - Attribution waterfall │ │
│ │ - Heatmap des contributions │ │
│ │ - Alertes dérive factorielle │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Implémentation du Collecteur de Données
La qualité de l'attribution dépend directement de la qualité des données en entrée. Pour les stratégies crypto, nous devons aggregator des données provenant de sources multiples avec une attention particulière aux timestamps et à la qualité des prix.
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import numpy as np
from dataclasses import dataclass
import hashlib
@dataclass
class OHLCVCandle:
"""Structure pour une bougie OHLCV normalisée"""
timestamp: datetime
symbol: str
open: float
high: float
low: float
close: float
volume: float
quote_volume: float
trades_count: int
taker_buy_ratio: float
class TardisDataCollector:
"""
Collecteur haute performance pour données de marché crypto.
Utilise Tardis pour les données nivel 2 et level 3.
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.session: Optional[aiohttp.ClientSession] = None
self.cache: Dict[str, pd.DataFrame] = {}
self.cache_ttl = 300 # 5 minutes
# Configuration des exchanges supportés
self.exchanges = {
'binance': {'websocket': 'wss://stream.binance.com:9443'},
'bybit': {'websocket': 'wss://stream.bybit.com'},
'okx': {'websocket': 'wss://ws.okx.com:8443'}
}
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
},
timeout=aiohttp.ClientTimeout(total=30)
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def fetch_ohlcv(
self,
symbol: str,
exchange: str,
interval: str = '1h',
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
limit: int = 1000
) -> pd.DataFrame:
"""
Récupère les données OHLCV avec mise en cache intelligente.
Args:
symbol: Symbole de trading (ex: 'BTC/USDT')
exchange: Nom de l'exchange
interval: Intervalle ('1m', '5m', '1h', '1d')
start_time: Date de début
end_time: Date de fin
limit: Nombre maximum de bougies
Returns:
DataFrame avec colonnes normalisées
"""
cache_key = f"{exchange}:{symbol}:{interval}:{start_time}"
# Vérification du cache
if cache_key in self.cache:
cached_df, cached_time = self.cache[cache_key]
if (datetime.now() - cached_time).seconds < self.cache_ttl:
return cached_df.copy()
# Construction de l'endpoint
endpoint = f"{self.base_url}/market/candles"
params = {
'symbol': symbol.replace('/', ''),
'exchange': exchange,
'interval': interval,
'limit': limit
}
if start_time:
params['start_time'] = int(start_time.timestamp() * 1000)
if end_time:
params['end_time'] = int(end_time.timestamp() * 1000)
async with self.session.get(endpoint, params=params) as response:
if response.status == 429:
# Rate limiting - backs off exponentiel
await asyncio.sleep(2 ** (response.headers.get('Retry-After', 1)))
return await self.fetch_ohlcv(symbol, exchange, interval,
start_time, end_time, limit)
response.raise_for_status()
raw_data = await response.json()
# Normalisation des données
df = pd.DataFrame(raw_data['data'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
# Conversion des types
for col in ['open', 'high', 'low', 'close', 'volume']:
df[col] = df[col].astype(float)
# Cache
self.cache[cache_key] = (df, datetime.now())
return df.copy()
async def fetch_orderbook_snapshot(
self,
symbol: str,
exchange: str,
depth: int = 100
) -> Dict:
"""
Récupère un snapshot du carnet d'ordres.
Critique pour le calcul du slippage implicite.
"""
endpoint = f"{self.base_url}/market/orderbook"
params = {
'symbol': symbol.replace('/', ''),
'exchange': exchange,
'depth': depth
}
async with self.session.get(endpoint, params=params) as response:
response.raise_for_status()
return await response.json()
Exemple d'utilisation
async def main():
async with TardisDataCollector(api_key="YOUR_HOLYSHEEP_API_KEY") as collector:
# Récupération des données BTC/USDT sur 30 jours
btc_data = await collector.fetch_ohlcv(
symbol='BTC/USDT',
exchange='binance',
interval='1h',
start_time=datetime.now() - timedelta(days=30)
)
print(f"Données récupérées : {len(btc_data)} bougies")
print(f"Range temporel : {btc_data.index.min()} à {btc_data.index.max()}")
print(f"Prix moyen : ${btc_data['close'].mean():.2f}")
if __name__ == "__main__":
asyncio.run(main())
Calcul des Facteurs et Décomposition Alpha
Une fois les données collectées, vient l'étape critique du calcul des facteurs et de la décomposition de la performance. Le modèle de Tardis utilise une approche hybride combinant des facteurs quantitatifs classiques et une validation par modèle d'intelligence artificielle.
import pandas as pd
import numpy as np
from typing import Tuple, Dict
from scipy import stats
from arch import arch_model
from sklearn.linear_model import Ridge, HuberRegressor
from sklearn.preprocessing import StandardScaler
import warnings
warnings.filterwarnings('ignore')
class AlphaDecomposer:
"""
Décompose la performance d'un portefeuille crypto en composantes factorielles.
Utilise le modèle de Tardis avec validation HOLYSHEEP.
Facteurs calculés :
- Beta marché (exposition Bitcoin/Ethereum)
- Momentum (retours passés 1-7-30 jours)
- Size (capitalisation)
- Volatility (volatilité conditionnelle GARCH)
- Selection alpha (surperformance vs benchmark factoriel)
- Timing alpha (abilité à prédire les retournements)
"""
def __init__(self, holysheep_api_key: str):
self.api_key = holysheep_api_key
self.base_url = "https://api.holysheep.ai/v1"
self.scaler = StandardScaler()
self.models = {}
def calculate_market_beta(
self,
portfolio_returns: pd.Series,
market_returns: pd.Series,
lookback: int = 30
) -> pd.DataFrame:
"""
Calcule le beta du marché avec une fenêtre glissante.
Utilise une régression OLS robuste avec outliers.
"""
betas = []
dates = []
for i in range(lookback, len(portfolio_returns)):
window_portfolio = portfolio_returns.iloc[i-lookback:i]
window_market = market_returns.iloc[i-lookback:i]
# Régression OLS robuste
slope, intercept, r_value, p_value, std_err = stats.linregress(
window_market, window_portfolio
)
betas.append({
'beta': slope,
'alpha': intercept,
'r_squared': r_value ** 2,
'p_value': p_value,
'std_error': std_err
})
dates.append(portfolio_returns.index[i])
return pd.DataFrame(betas, index=dates)
def calculate_momentum_factors(
self,
prices: pd.DataFrame,
periods: Tuple[int, int, int] = (1, 7, 30)
) -> pd.DataFrame:
"""
Calcule les facteurs momentum sur plusieurs horizons.
Inclut le momentum relatif vs BTC.
"""
momentum = pd.DataFrame(index=prices.index)
for period in periods:
momentum[f'momentum_{period}d'] = prices.pct_change(period)
# Momentum relatif vs Bitcoin (si disponible)
if 'BTC' in prices.columns:
btc_returns = prices['BTC'].pct_change()
for col in prices.columns:
if col != 'BTC':
momentum[f'relative_momentum_{col}'] = (
momentum[f'momentum_7d'][col] -
btc_returns.rolling(7).sum()
)
return momentum.dropna()
def calculate_volatility_factor(
self,
returns: pd.Series,
horizon: int = 30
) -> pd.DataFrame:
"""
Calcule la volatilité conditionnelle via GARCH(1,1).
Plus robuste que la volatilité historique simple.
"""
volatility_data = []
for i in range(horizon, len(returns)):
window = returns.iloc[i-horizon:i]
try:
# Ajustement pour les small samples
model = arch_model(
window * 100, # GARCH nécessite des scaling
vol='Garch',
p=1,
q=1,
dist='t' # Distribution Student-t pour fat tails
)
result = model.fit(disp='off', show_warning=False)
# Forecast de la volatilité tomorrow
forecast = result.forecast(horizon=1)
forecasted_vol = np.sqrt(forecast.variance.iloc[-1, 0]) / 100
volatility_data.append({
'garch_vol': forecasted_vol,
'realized_vol': window.std(),
'vol_premium': forecasted_vol - window.std(),
'persistence': result.params.get('alpha[1]', 0) +
result.params.get('beta[1]', 0)
})
except Exception:
volatility_data.append({
'garch_vol': window.std(),
'realized_vol': window.std(),
'vol_premium': 0,
'persistence': 0.95
})
return pd.DataFrame(
volatility_data,
index=returns.index[horizon:]
)
def calculate_size_factor(
self,
market_caps: pd.DataFrame
) -> pd.Series:
"""
Calcule le facteur size (capitalisation).
Définition : Small Cap = bottom 20%, Large Cap = top 20%.
"""
size_factor = pd.Series(index=market_caps.index)
for date in market_caps.index:
caps = market_caps.loc[date].dropna()
if len(caps) == 0:
continue
# Quintiles
q20 = caps.quantile(0.2)
q80 = caps.quantile(0.8)
# Facteur : Long Small Cap, Short Large Cap
small_caps = caps[caps <= q20].index
large_caps = caps[caps >= q80].index
# Approximation simplifiée
size_factor[date] = (caps.mean() - caps.median()) / caps.std()
return size_factor.fillna(0)
def decompose_performance(
self,
portfolio_returns: pd.Series,
prices: pd.DataFrame,
market_returns: pd.Series,
market_caps: pd.DataFrame,
lookback: int = 30
) -> Dict:
"""
Décomposition complète de la performance.
Returns:
Dict contenant les contributions factorielles
"""
results = {
'portfolio_return': portfolio_returns.sum(),
'factor_attribution': {},
'residual_alpha': None,
'r_squared': None,
'breakdown': None
}
# Calcul des facteurs
market_beta = self.calculate_market_beta(
portfolio_returns, market_returns, lookback
)
momentum = self.calculate_momentum_factors(
prices, periods=(1, 7, 30)
)
volatility = self.calculate_volatility_factor(
portfolio_returns, horizon=30
)
size_factor = self.calculate_size_factor(market_caps)
# Alignement des dates
common_dates = portfolio_returns.index.intersection(
market_beta.index
).intersection(momentum.index).intersection(volatility.index)
aligned_portfolio = portfolio_returns.loc[common_dates]
aligned_market = market_returns.loc[common_dates]
# Construction de la matrice de régression
X = pd.DataFrame({
'market': aligned_market,
'momentum_7d': momentum.loc[common_dates, 'momentum_7d']
if 'momentum_7d' in momentum.columns else 0,
'momentum_30d': momentum.loc[common_dates, 'momentum_30d']
if 'momentum_30d' in momentum.columns else 0,
'volatility': volatility.loc[common_dates, 'garch_vol']
if 'garch_vol' in volatility.columns else aligned_portfolio.std(),
'size': size_factor.loc[common_dates] if len(size_factor) > 0 else 0
}).fillna(0)
Y = aligned_portfolio
# Régression avec régularisation Ridge
X_scaled = self.scaler.fit_transform(X)
model = Ridge(alpha=1.0)
model.fit(X_scaled, Y)
# Attribution
coefficients = pd.Series(model.coef_, index=X.columns)
predictions = model.predict(X_scaled)
residuals = Y - predictions
# Calcul des contributions factorielles
total_return = Y.sum()
explained_return = predictions.sum()
for factor in X.columns:
contribution = (coefficients[factor] * X[factor].sum()) / len(Y)
results['factor_attribution'][factor] = {
'coefficient': coefficients[factor],
'contribution': contribution,
'contribution_pct': contribution / total_return * 100
if total_return != 0 else 0
}
results['residual_alpha'] = residuals.sum()
results['r_squared'] = model.score(X_scaled, Y)
results['breakdown'] = {
'market_contribution': coefficients['market'] * X['market'].mean(),
'timing_contribution': residuals.sum() / len(Y),
'selection_contribution': explained_return - residuals.sum()
}
return results
Exemple d'utilisation complète
def example_attribution():
decomposer = AlphaDecomposer(holysheep_api_key="YOUR_HOLYSHEEP_API_KEY")
# Données simulées pour démonstration
dates = pd.date_range(start='2024-01-01', end='2024-03-01', freq='D')
# Simulation de returns de portefeuille
np.random.seed(42)
portfolio_returns = pd.Series(
np.random.normal(0.003, 0.05, len(dates)),
index=dates
)
# Returns du marché (BTC)
market_returns = pd.Series(
np.random.normal(0.002, 0.04, len(dates)),
index=dates
)
# Prix simulés
prices = pd.DataFrame({
'BTC': 50000 * (1 + market_returns).cumprod(),
'ETH': 3000 * (1 + np.random.normal(0.003, 0.06, len(dates))).cumprod()
}, index=dates)
# Capitalisations simulées
market_caps = pd.DataFrame({
'BTC': [800e9] * len(dates),
'ETH': [400e9] * len(dates)
}, index=dates)
# Décomposition
results = decomposer.decompose_performance(
portfolio_returns=portfolio_returns,
prices=prices,
market_returns=market_returns,
market_caps=market_caps
)
print("=" * 60)
print("RÉSULTATS DE LA DÉCOMPOSITION ALPHA")
print("=" * 60)
print(f"Performance totale du portefeuille : {results['portfolio_return']*100:.2f}%")
print(f"R² du modèle factoriel : {results['r_squared']:.4f}")
print("\nAttribution factorielle :")
for factor, data in results['factor_attribution'].items():
print(f" {factor}: {data['contribution']*100:.3f}% "
f"({data['contribution_pct']:.1f}% de la performance)")
print(f"\nAlpha résiduel (Sélection/Timing) : "
f"{results['residual_alpha']*100:.3f}%")
if __name__ == "__main__":
example_attribution()
Intégration HOLYSHEEP pour l'Analyse IA des Patterns
L'un des avantages distinctifs de l'approche Tardis réside dans l'intégration de modèles d'intelligence artificielle pour détecter des patterns factoriels non-linéaires. HOLYSHEEP AI offre une latence moyenne de 48ms et des coûts réduits de 85% par rapport aux providers traditionnels, ce qui rend l'analyse en temps réel viable pour les desks de trading.
import requests
from typing import Optional, Dict, List
import json
from dataclasses import dataclass
from datetime import datetime
@dataclass
class FactorAnalysisResult:
"""Résultat de l'analyse factorielle HOLYSHEEP"""
pattern_type: str
confidence: float
description: str
recommended_action: str
risk_metrics: Dict[str, float]
class HolySheepAlphaAnalyzer:
"""
Client pour l'analyse factorielle via l'API HOLYSHEEP.
Utilise les modèles de language pour identifier des patterns
dans les données d'attribution.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
# Cache des résultats pour éviter les appels redondants
self.analysis_cache: Dict[str, tuple] = {}
self.cache_duration = 300 # 5 minutes
def _generate_cache_key(self, portfolio_id: str, date: datetime) -> str:
"""Génère une clé de cache unique"""
return f"{portfolio_id}:{date.strftime('%Y%m%d%H%M')}"
def _is_cache_valid(self, cache_key: str) -> bool:
"""Vérifie si le cache est encore valide"""
if cache_key not in self.analysis_cache:
return False
_, timestamp = self.analysis_cache[cache_key]
return (datetime.now() - timestamp).seconds < self.cache_duration
def analyze_attribution_pattern(
self,
portfolio_id: str,
attribution_data: Dict,
date: datetime = None
) -> FactorAnalysisResult:
"""
Analyse les patterns d'attribution avec HOLYSHEEP IA.
Args:
portfolio_id: Identifiant du portefeuille
attribution_data: Résultats de la décomposition alpha
date: Date de l'analyse
Returns:
FactorAnalysisResult avec recommandations
"""
if date is None:
date = datetime.now()
cache_key = self._generate_cache_key(portfolio_id, date)
# Vérification du cache
if self._is_cache_valid(cache_key):
return self.analysis_cache[cache_key][0]
# Construction du prompt pour l'analyse
prompt = self._build_analysis_prompt(attribution_data)
# Appel à l'API HOLYSHEEP
try:
response = self._call_holysheep_api(prompt)
result = self._parse_analysis_result(response, attribution_data)
# Mise en cache
self.analysis_cache[cache_key] = (result, datetime.now())
return result
except Exception as e:
# Fallback sur une analyse locale basique
return self._fallback_analysis(attribution_data)
def _build_analysis_prompt(self, data: Dict) -> str:
"""Construit le prompt pour l'analyse HOLYSHEEP"""
breakdown = data.get('breakdown', {})
factor_attribution = data.get('factor_attribution', {})
prompt = f"""Analyse la performance du portefeuille crypto pour {datetime.now().strftime('%Y-%m-%d')}.
PERFORMANCE TOTALE: {data.get('portfolio_return', 0)*100:.2f}%
R² DU MODÈLE: {data.get('r_squared', 0):.4f}
ALPHA RÉSIDUEL: {data.get('residual_alpha', 0)*100:.3f}%
DÉCOMPOSITION FACTORIELLE:
- Contribution Marché: {breakdown.get('market_contribution', 0)*100:.2f}%
- Contribution Timing: {breakdown.get('timing_contribution', 0)*100:.2f}%
- Contribution Sélection: {breakdown.get('selection_contribution', 0)*100:.2f}%
FACTEURS INDIVIDUELS:"""
for factor, info in factor_attribution.items():
prompt += f"\n- {factor}: {info['contribution']*100:.3f}% (coeff: {info['coefficient']:.4f})"
prompt += """
Identifie :
1. Si l'alpha provient de skill (sélection/timing) ou de chance
2. Les risques de dérive factorielle
3. Les opportunités d'optimisation
4. Les signaux d'alerte
Réponds en JSON avec les champs: pattern_type, confidence, description, recommended_action, risk_metrics."""
return prompt
def _call_holysheep_api(self, prompt: str) -> Dict:
"""Appelle l'API HOLYSHEEP pour l'analyse"""
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": "deepseek-v3.2", # Modèle économique optimisé
"messages": [
{
"role": "system",
"content": "Tu es un analyste quantitatif expert en cryptomonnaies. "
"Analyse les données d'attribution et fournis des recommandations "
"précises et actionnables."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3, # Réponses déterministes
"max_tokens": 800,
"response_format": {"type": "json_object"}
}
response = self.session.post(endpoint, json=payload, timeout=30)
if response.status_code == 429:
raise Exception("Rate limit HOLYSHEEP - backs off")
response.raise_for_status()
result = response.json()
return result
def _parse_analysis_result(
self,
api_response: Dict,
attribution_data: Dict
) -> FactorAnalysisResult:
"""Parse la réponse de l'API en FactorAnalysisResult"""
content = api_response['choices'][0]['message']['content']
parsed = json.loads(content)
# Extraction des métriques de risque
risk_metrics = parsed.get('risk_metrics', {})
# Ajout de métriques calculées
risk_metrics.update({
'concentration_risk': abs(attribution_data.get('r_squared', 0) - 1),
'timing_risk': abs(attribution_data.get('breakdown', {})
.get('timing_contribution', 0)),
'alpha_stability': 1 / (1 + abs(attribution_data.get('residual_alpha', 0)))
})
return FactorAnalysisResult(
pattern_type=parsed.get('pattern_type', 'unknown'),
confidence=parsed.get('confidence', 0.5),
description=parsed.get('description', ''),
recommended_action=parsed.get('recommended_action', ''),
risk_metrics=risk_metrics
)
def _fallback_analysis(self, data: Dict) -> FactorAnalysisResult:
"""Analyse locale basique en cas d'erreur API"""
r_squared = data.get('r_squared', 0)
residual = data.get('residual_alpha', 0)
if r_squared > 0.8:
pattern = "market_dependent"
confidence = 0.9
action = "Réduire exposition marché ou hedger le beta"
elif residual > 0.02:
pattern = "skill_based"
confidence = 0.7
action = "Augmenter allocation vers cette stratégie"
else:
pattern = "mixed"
confidence = 0.6
action = "Surveiller évolution alpha résiduel"
return FactorAnalysisResult(
pattern_type=pattern,
confidence=confidence,
description=f"Alpha résiduel: {residual*100:.2f}%, R²: {r_squared:.2f}",
recommended_action=action,
risk_metrics={'fallback': True}
)
def generate_attribution_report(
self,
portfolio_id: str,
historical_data: List[Dict],
period_days: int = 30
) -> Dict:
"""
Génère un rapport complet d'attribution sur une période.
Args:
portfolio_id: ID du portefeuille
historical_data: Liste des données d'attribution journalières
period_days: Nombre de jours de la période
Returns:
Rapport complet avec statistiques et recommandations
"""
report = {
'portfolio_id': portfolio_id,
'period': f"{period_days} jours",
'summary': {},
'daily_attribution': historical_data,
'alerts': [],
'recommendations': []
}
# Calcul des statistiques
total_returns = [d.get('portfolio_return', 0) for d in historical_data]
alphas = [d.get('residual_alpha', 0) for d in historical_data]
report['summary'] = {
'total_return': sum(total_returns),
'avg_daily_return': np.mean(total_returns),
'volatility': np.std(total_returns),
'sharpe_approximated': np.mean(total_returns) /
np.std(total_returns) * np.sqrt(252)
if np.std(total_returns) > 0 else 0,
'avg_alpha': np.mean(alphas),
'alpha_volatility': np.std(alphas),
'alpha_t_stat': np.mean(alphas) / (np.std(alphas) / np.sqrt(len(alphas)))
if len(alphas) > 1 and np.std(alphas) > 0 else 0
}
# Test de significativité de l'alpha
t_stat = report['summary']['alpha_t_stat']
if abs(t_stat) > 2:
report['alerts'].append({
'type': 'significant_alpha',
'severity': 'positive' if t_stat > 0 else 'negative',
'message': f"Alpha statistiquement significatif (t={t_stat:.2f})"
})
# Analyse HOLYSHEEP pour le rapport complet
if historical_data:
latest = historical_data[-1]
analysis = self.analyze_attribution_pattern(
portfolio_id, latest
)
report['ai_analysis'] = {
'pattern': analysis.pattern_type,
'confidence': analysis.confidence,
'description': analysis.description,
'action': analysis.recommended_action
}
if analysis.risk_metrics.get('concentration_risk', 0) > 0.3:
report['alerts'].append({
'type': 'concentration_risk',
'severity': 'warning',
'message': "Risque de concentration factorielle détecté"
})
return report
Exemple d'utilisation
def main():
analyzer = HolySheepAlphaAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
# Données d'exemple
sample_data = {
'portfolio_return': 0.152, # +15.2% sur la période
'r_squared': 0.72,
'residual_alpha': 0.038, # +3.8% d'alpha
'breakdown': {
'market_contribution': 0.08,
'timing_contribution': 0.012,
'selection_contribution': 0.06
},
'factor_attribution': {
'market': {'coefficient': 1.2, 'contribution': 0.08, 'contribution_pct': 52.6},
'momentum_7d': {'coefficient': 0.15, 'contribution': 0.025, 'contribution_pct': 16.4},
'momentum_30d': {'coefficient': -0.08, 'contribution': -0.01, 'contribution_pct': -6.6},
'volatility': {'coefficient': 0.5, 'contribution': 0.015, 'contribution_pct': 9.9},
'size': {'coefficient': 0.02, 'contribution': 0.005, 'contribution_pct': 3.3}
}
}
# Analyse
result = analyzer.analyze_attribution_pattern(
portfolio_id="PORTFOLIO_MAIN",
attribution_data=sample_data
)
print("=" * 60)
print("ANALYSE HOLYSHEEP")
print("