En tant qu'ingénieur quantitatif ayant développé et déployé plus de quarante stratégies d'arbitrage crypto depuis 2019, je peux vous affirmer sans hésitation que 80% des échecs de stratégies automatiques proviennent d'un prétraitement des données insuffisant. Les bruits de marché, les spikes anormaux et les données manquantes ruinent des stratégies autrement excellentes. Dans ce tutoriel complet, je vais vous expliquer ma méthodologie complète de清洗 et prétraitement des données, en intégrant l'API HolySheep AI pour l'analyse intelligente des patterns.
Comparatif des Coûts API pour l'Analyse de Données Crypto
Avant de plonger dans le code, comparons les coûts réels pour traiter 10 millions de tokens par mois dans vos pipelines de données. Les tarifs 2026 sont désormais stabilisés :
| Fournisseur | Modèle | Prix par Million de Tokens | Coût pour 10M tokens/mois | Latence moyenne |
|---|---|---|---|---|
| HolySheep AI | DeepSeek V3.2 | 0,42 $ | 4,20 $ | <50ms |
| HolySheep AI | Gemini 2.5 Flash | 2,50 $ | 25,00 $ | <50ms |
| HolySheep AI | GPT-4.1 | 8,00 $ | 80,00 $ | <50ms |
| Concurrents | Claude Sonnet 4.5 | 15,00 $ | 150,00 $ | Variable |
Économie annuelle avec HolySheep AI : jusqu'à 1 740 $ en utilisant DeepSeek V3.2 au lieu de Claude Sonnet 4.5 pour le même volume de traitement.
Architecture du Pipeline de Données
Mon pipeline de nettoyage des données d'arbitrage crypto se compose de cinq couches distinctes, chacune jouant un rôle critique dans la qualité du signal final.
# Architecture complète du pipeline de prétraitement
HolySheep AI - https://api.holysheep.ai/v1
import pandas as pd
import numpy as np
from holy_sheep import HolySheepClient
from scipy import stats
import ccxt
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
class DataQuality(Enum):
EXCELLENT = "excellent"
GOOD = "good"
DEGRADED = "degraded"
CORRUPTED = "corrupted"
@dataclass
class CleaningConfig:
max_z_score: float = 3.5
min_volume_usd: float = 10000
max_spread_bps: float = 50
outlier_window: int = 20
fill_method: str = "interpolate"
holiday_tolerance: int = 3
class CryptoDataPipeline:
"""
Pipeline industriel de nettoyage pour données d'arbitrage crypto.
Conçu pour fonctionner 24/7 avec auto-correction.
"""
def __init__(self, api_key: str, config: CleaningConfig = None):
self.client = HolySheepClient(api_key=api_key)
self.config = config or CleaningConfig()
self.exchanges = self._init_exchanges()
self.quality_metrics = {}
def _init_exchanges(self) -> Dict:
"""Initialise les connexions aux exchanges avec gestion des erreurs."""
return {
'binance': ccxt.binance({'enableRateLimit': True}),
'bybit': ccxt.bybit({'enableRateLimit': True}),
'okx': ccxt.okx({'enableRateLimit': True}),
'huobi': ccxt.huobi({'enableRateLimit': True}),
}
async def fetch_and_clean_pair(
self,
symbol: str,
exchanges: List[str],
timeframe: str = '1m',
limit: int = 1000
) -> Dict[str, pd.DataFrame]:
"""Récupère et nettoie les données pour une paire sur plusieurs exchanges."""
raw_data = await self._fetch_multi_exchange(
symbol, exchanges, timeframe, limit
)
cleaned_data = {}
for exchange_name, df in raw_data.items():
cleaned_data[exchange_name] = await self._clean_dataframe(
df, symbol, exchange_name
)
return cleaned_data
async def _fetch_multi_exchange(
self, symbol: str, exchanges: List[str],
timeframe: str, limit: int
) -> Dict[str, pd.DataFrame]:
"""Récupération parallèle depuis plusieurs exchanges."""
tasks = []
for exchange_name in exchanges:
if exchange_name in self.exchanges:
tasks.append(
self._fetch_single_exchange(
exchange_name, symbol, timeframe, limit
)
)
results = await asyncio.gather(*tasks, return_exceptions=True)
return {ex: df for ex, df in results if isinstance(df, pd.DataFrame)}
async def _fetch_single_exchange(
self, exchange_name: str, symbol: str,
timeframe: str, limit: int
) -> pd.DataFrame:
"""Récupère les données OHLCV depuis un exchange unique."""
exchange = self.exchanges[exchange_name]
try:
ohlcv = await exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(
ohlcv,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
df['exchange'] = exchange_name
return df
except Exception as e:
print(f"Erreur {exchange_name}: {e}")
return pd.DataFrame()
async def _clean_dataframe(
self, df: pd.DataFrame,
symbol: str, exchange: str
) -> pd.DataFrame:
"""Applique tous les nettoyages au DataFrame."""
if df.empty:
return df
# Étape 1: Supprimer les doublons temporels
df = df[~df.index.duplicated(keep='last')]
# Étape 2: Détecter et 处理 les anomalies
df = self._remove_statistical_outliers(df)
# Étape 3: Vérifier la cohérence des prix
df = self._validate_price_consistency(df)
# Étape 4: Nettoyer les volumes aberrants
df = self._clean_volume_data(df)
# Étape 5: Compléter les données manquantes
df = self._impute_missing_data(df)
# Étape 6: Calculer les métriques de qualité
self.quality_metrics[f"{symbol}_{exchange}"] = self._compute_quality_score(df)
return df
def _remove_statistical_outliers(self, df: pd.DataFrame) -> pd.DataFrame:
"""Supprime les outliers via méthode Z-score adaptative."""
price_columns = ['open', 'high', 'low', 'close']
for col in price_columns:
if col in df.columns:
# Z-score mobile avec fenêtre adaptative
rolling_mean = df[col].rolling(
window=self.config.outlier_window,
center=True
).mean()
rolling_std = df[col].rolling(
window=self.config.outlier_window,
center=True
).std()
z_scores = np.abs((df[col] - rolling_mean) / rolling_std)
df.loc[z_scores > self.config.max_z_score, col] = np.nan
return df
def _validate_price_consistency(self, df: pd.DataFrame) -> pd.DataFrame:
"""Vérifie que high >= close >= low pour chaque bougie."""
# High doit être le maximum
df['high'] = df[['open', 'high', 'low', 'close']].max(axis=1)
# Low doit être le minimum
df['low'] = df[['open', 'high', 'low', 'close']].min(axis=1)
# Marquer les bougies invalides
invalid = (df['high'] < df['low']) | \
(df['high'] < df['close']) | \
(df['low'] > df['close'])
df.loc[invalid, ['open', 'high', 'low', 'close']] = np.nan
return df
def _clean_volume_data(self, df: pd.DataFrame) -> pd.DataFrame:
"""Nettoie les données de volume et détecte les wash trades."""
if 'volume' not in df.columns:
return df
# Remplacer les volumes négatifs par NaN
df.loc[df['volume'] < 0, 'volume'] = np.nan
# Détecter les volumes suspects (égaux pendant longtemps)
volume_changes = df['volume'].pct_change().abs()
suspicious = (volume_changes == 0) & (volume_changes.shift(1) == 0)
# Vérifier si le volume est trop rond (possible wash trade)
is_round = (df['volume'] % 0.001 == 0) & (df['volume'] > 0)
suspiciously_round = is_round & is_round.shift(5)
df.loc[suspicious | suspiciously_round, 'volume'] = np.nan
return df
def _impute_missing_data(self, df: pd.DataFrame) -> pd.DataFrame:
"""Complète les données manquantes avec interpolation intelligente."""
method = self.config.fill_method
if method == "interpolate":
# Interpolation linéaire avec extrapolation aux bords
df = df.interpolate(method='linear', limit_direction='both')
df = df.fillna(method='bfill').fillna(method='ffill')
elif method == "ffill":
# Forward fill pour données de prix (dernière valeur connue)
df = df.fillna(method='ffill')
elif method == "spline":
# Spline cubique pour plus de smoothness
df = df.interpolate(method='spline', order=3, limit_direction='both')
return df
def _compute_quality_score(self, df: pd.DataFrame) -> float:
"""Calcule un score de qualité de 0 à 1 pour le DataFrame."""
total_rows = len(df)
if total_rows == 0:
return 0.0
# Proportion de données non-null
completeness = df.notna().sum().sum() / (total_rows * len(df.columns))
# Continuité temporelle
expected_freq = pd.Timedelta(minutes=1)
actual_freq = df.index.to_series().diff().median()
continuity = 1.0 - min(abs(actual_freq - expected_freq) / expected_freq, 1.0)
# Stabilité des prix (pas de sauts abrupts)
returns = df['close'].pct_change().dropna()
stability = 1.0 - min(returns.std() * 100, 1.0)
return (completeness * 0.4 + continuity * 0.3 + stability * 0.3)
async def detect_arbitrage_opportunities(
self,
symbol: str,
exchanges: List[str] = None
) -> List[Dict]:
"""Détecte les opportunités d'arbitrage cross-exchange."""
exchanges = exchanges or list(self.exchanges.keys())
data = await self.fetch_and_clean_pair(symbol, exchanges)
# Calculer les spreads normalisés
opportunities = []
for i, (ex1, df1) in enumerate(data.items()):
for ex2, df2 in list(data.items())[i+1:]:
if df1.empty or df2.empty:
continue
# Merge sur timestamp
merged = pd.merge(
df1[['close']],
df2[['close']],
left_index=True,
right_index=True,
suffixes=('_1', '_2')
)
# Calcul du spread en basis points
merged['spread_bps'] = (
(merged['close_1'] - merged['close_2']) /
merged['close_2'] * 10000
)
# Filtrer les opportunités significatives
sig_opps = merged[abs(merged['spread_bps']) > 5].dropna()
if not sig_opps.empty:
opportunities.append({
'pair': f"{ex1}/{ex2}",
'avg_spread_bps': sig_opps['spread_bps'].mean(),
'max_spread_bps': sig_opps['spread_bps'].abs().max(),
'count': len(sig_opps),
'confidence': self._compute_quality_score(
pd.merge(df1, df2, left_index=True, right_index=True)
)
})
return opportunities
Utilisation principale
async def main():
# Connexion à HolySheep AI
client = HolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
pipeline = CryptoDataPipeline(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=CleaningConfig(
max_z_score=3.5,
min_volume_usd=50000,
max_spread_bps=100
)
)
# Analyse BTC/USDT cross-exchange
opportunities = await pipeline.detect_arbitrage_opportunities(
symbol="BTC/USDT",
exchanges=['binance', 'bybit', 'okx']
)
print(f"Opportunités détectées: {len(opportunities)}")
for opp in opportunities:
print(f" {opp['pair']}: spread moyen {opp['avg_spread_bps']:.2f} bps")
Exécution
if __name__ == "__main__":
asyncio.run(main())Modules de Détection d'Anomalies Avancés
La détection d'anomalies est le cœur de toute stratégie d'arbitrage robuste. Je vais maintenant vous présenter mon module de détection basé sur l'apprentissage automatique via l'API HolySheep AI.
# Module de détection d'anomalies pour données crypto
HolySheep AI - https://api.holysheep.ai/v1
from holy_sheep import HolySheepClient
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np
import pandas as pd
from typing import Tuple, List
from datetime import datetime
class AnomalyDetector:
"""
Détection d'anomalies multi-stratégie pour données de marché crypto.
Combine méthodes statistiques et IA via HolySheep.
"""
def __init__(self, api_key: str):
self.client = HolySheepClient(api_key=api_key)
self.scaler = StandardScaler()
self.isolation_forest = IsolationForest(
contamination=0.01,
n_estimators=100,
random_state=42
)
def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""Crée un ensemble de features pour la détection d'anomalies."""
features = pd.DataFrame(index=df.index)
# Retours logarithmiques
features['log_return'] = np.log(df['close'] / df['close'].shift(1))
# Volatilité Rolling
features['volatility_1m'] = features['log_return'].rolling(1).std()
features['volatility_5m'] = features['log_return'].rolling(5).std()
features['volatility_15m'] = features['log_return'].rolling(15).std()
# Rapport volume/prix
features['volume_price_ratio'] = df['volume'] / df['close']
# Spread high-low normalisé
features['hl_spread'] = (df['high'] - df['low']) / df['close']
# Skewness du prix
features['price_skew'] = df['close'].rolling(10).skew()
# Momentum
features['momentum'] = df['close'] / df['close'].shift(10) - 1
# Taux de variation du volume
features['volume_change'] = df['volume'].pct_change()
return features.fillna(0)
def detect_statistical_outliers(
self,
df: pd.DataFrame,
n_std: float = 3.0
) -> Tuple[pd.DataFrame, pd.Series]:
"""
Détection par méthode statistique (Z-score modifié).
"""
price_data = df[['open', 'high', 'low', 'close']].copy()
# Z-score avec robusteMAD (Median Absolute Deviation)
median = price_data.median()
mad = (price_data - median).abs().median()
modified_z = 0.6745 * (price_data - median) / mad
is_outlier = modified_z.abs() > n_std
return price_data, is_outlier.any(axis=1)
def detect_isolation_forest(
self,
features: pd.DataFrame
) -> np.ndarray:
"""
Détection par Isolation Forest (non-supervisé).
"""
# Normalisation des features
scaled_features = self.scaler.fit_transform(features)
# Entraînement et prédiction
predictions = self.isolation_forest.fit_predict(scaled_features)
# -1 = anomalie, 1 = normal
return predictions == -1
async def detect_with_ai(
self,
df: pd.DataFrame,
context: str = "cryptocurrency_arbitrage"
) -> List[Dict]:
"""
Utilise l'IA HolySheep pour analyser les patterns suspects.
Coût: ~$0.42/M tokens avec DeepSeek V3.2
"""
# Préparer le contexte
summary = {
'timestamp_range': f"{df.index.min()} to {df.index.max()}",
'rows': len(df),
'price_stats': {
'mean': float(df['close'].mean()),
'std': float(df['close'].std()),
'min': float(df['close'].min()),
'max': float(df['close'].max()),
},
'volume_stats': {
'mean': float(df['volume'].mean()),
'max': float(df['volume'].max()),
}
}
# Prompt pour analyse IA
prompt = f"""Analyse cette série de données crypto pour détecter des anomalies:
Données:
{summary}
Identifie:
1. Les périodes avec activité suspecte
2. Les patterns de wash trading possibles
3. Les correlations anormales
4. Les gaps de données problématiques
Réponds en JSON avec 'anomalies' (liste de timestamps suspects) et 'confiance' (0-1)."""
# Appel API HolySheep
response = await self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{
"role": "system",
"content": "Tu es un expert en analyse de données financières."
},
{"role": "user", "content": prompt}
],
temperature=0.1,
max_tokens=500
)
import json
result = json.loads(response.choices[0].message.content)
return result
def full_analysis(
self,
df: pd.DataFrame,
use_ai: bool = True
) -> pd.DataFrame:
"""
Analyse complète combinant toutes les méthodes.
"""
results = pd.DataFrame(index=df.index)
# Méthode 1: Z-score modifié
_, zscore_outliers = self.detect_statistical_outliers(df)
results['zscore_anomaly'] = zscore_outliers
# Méthode 2: Isolation Forest
features = self.create_features(df)
results['iforest_anomaly'] = self.detect_isolation_forest(features)
# Méthode 3: Volume anormal
volume_ma = df['volume'].rolling(20).mean()
volume_std = df['volume'].rolling(20).std()
results['volume_anomaly'] = (
(df['volume'] - volume_ma).abs() > 3 * volume_std
) | (df['volume'] == 0)
# Score composite (pondération)
results['anomaly_score'] = (
results['zscore_anomaly'].astype(int) * 0.3 +
results['iforest_anomaly'].astype(int) * 0.4 +
results['volume_anomaly'].astype(int) * 0.3
)
return results
async def analyze_and_clean(
self,
df: pd.DataFrame,
symbol: str
) -> Tuple[pd.DataFrame, Dict]:
"""
Pipeline complet: analyse + nettoyage + rapport.
"""
# Analyse des anomalies
analysis = self.full_analysis(df, use_ai=False)
# Analyse IA optionnelle (plus coûteuse mais plus précise)
if len(df) > 100:
ai_results = await self.detect_with_ai(df)
print(f"Analyse IA: {len(ai_results.get('anomalies', []))} anomalies détectées")
# Nettoyage sélectif
df_cleaned = df.copy()
# Remplacer les anomalies par NaN (à interpoler ensuite)
anomaly_mask = analysis['anomaly_score'] >= 0.7
df_cleaned.loc[anomaly_mask, ['open', 'high', 'low', 'close', 'volume']] = np.nan
# Rapport
report = {
'total_rows': len(df),
'anomalies_detected': anomaly_mask.sum(),
'anomaly_rate': f"{anomaly_mask.mean()*100:.2f}%",
'cleaned_rows': df_cleaned.notna().all(axis=1).sum(),
'quality_before': self._compute_quality(df),
'quality_after': self._compute_quality(df_cleaned),
}
return df_cleaned, report
def _compute_quality(self, df: pd.DataFrame) -> float:
"""Calcule le score de qualité global."""
completeness = df.notna().mean().mean()
return completeness
Intégration avec le pipeline principal
async def advanced_cleaning_pipeline():
"""Exemple d'intégration complète."""
detector = AnomalyDetector(api_key="YOUR_HOLYSHEEP_API_KEY")
# Simuler des données
dates = pd.date_range('2024-01-01', periods=1000, freq='1min')
np.random.seed(42)
# Données normales avec quelques anomalies injectées
df = pd.DataFrame({
'open': 50000 + np.cumsum(np.random.randn(1000) * 10),
'high': 50000 + np.cumsum(np.random.randn(1000) * 10) + 50,
'low': 50000 + np.cumsum(np.random.randn(1000) * 10) - 50,
'close': 50000 + np.cumsum(np.random.randn(1000) * 10),
'volume': np.random.lognormal(10, 1, 1000)
}, index=dates)
# Injecter des anomalies
df.loc[df.index[500], 'volume'] = 0 # Volume zéro
df.loc[df.index[600:605], 'close'] *= 1.5 # Spike de prix
# Analyse complète
df_cleaned, report = await detector.analyze_and_clean(df, "BTC/USDT")
print(f"Rapport de nettoyage:")
print(f" - Lignes totales: {report['total_rows']}")
print(f" - Anomalies détectées: {report['anomalies_detected']}")
print(f" - Taux d'anomalies: {report['anomaly_rate']}")
print(f" - Qualité avant: {report['quality_before']:.2%}")
print(f" - Qualité après: {report['quality_after']:.2%}")
return df_cleaned
if __name__ == "__main__":
import asyncio
result = asyncio.run(advanced_cleaning_pipeline())Gestion des Données Multi-Sources et Fusion
Dans une stratégie d'arbitrage crypto, vous devez fusionner des données provenant de multiples exchanges en temps réel. Voici mon système de fusion-tolerant aux pannes.
# Système de fusion de données multi-sources avec gestion des conflits
HolySheep AI - https://api.holysheep.ai/v1
import pandas as pd
import numpy as np
from holy_sheep import HolySheepClient
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import asyncio
import ccxt
class DataFusionEngine:
"""
Moteur de fusion de données multi-exchanges avec:
- Résolution de conflits temporels
- Pondération par qualité/latence
- Détection deManipulation
- Cache intelligent
"""
def __init__(self, api_key: str):
self.client = HolySheepClient(api_key=api_key)
self.exchanges = {}
self.weights = {}
self.quality_scores = {}
self._init_exchanges()
def _init_exchanges(self):
"""Initialise les connexions avec configuration."""
configs = {
'binance': {'rateLimit': 1200, 'options': {'defaultType': 'spot'}},
'bybit': {'rateLimit': 1500},
'okx': {'rateLimit': 1000},
'huobi': {'rateLimit': 2000},
}
for name, config in configs.items():
try:
exchange_class = getattr(ccxt, name)
self.exchanges[name] = exchange_class(config)
# Pondération initiale (à ajuster dynamiquement)
self.weights[name] = 1.0
except Exception as e:
print(f"Exchange {name} non disponible: {e}")
async def fetch_all_exchanges(
self,
symbol: str,
limit: int = 100
) -> Dict[str, pd.DataFrame]:
"""Récupère les données depuis tous les exchanges disponibles."""
tasks = []
for name, exchange in self.exchanges.items():
tasks.append(self._safe_fetch(exchange, name, symbol, limit))
results = await asyncio.gather(*tasks, return_exceptions=True)
data = {}
for name, result in results:
if isinstance(result, pd.DataFrame) and not result.empty:
data[name] = result
self.quality_scores[name] = self._assess_quality(result)
return data
async def _safe_fetch(
self,
exchange,
name: str,
symbol: str,
limit: int
) -> Tuple[str, pd.DataFrame]:
"""Récupération sécurisée avec timeout."""
try:
ohlcv = await asyncio.wait_for(
asyncio.to_thread(
exchange.fetch_ohlcv, symbol, '1m', limit=limit
),
timeout=10
)
df = pd.DataFrame(
ohlcv,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return name, df
except asyncio.TimeoutError:
print(f"Timeout {name} pour {symbol}")
self.weights[name] *= 0.8 # Pénalité
return name, pd.DataFrame()
except Exception as e:
print(f"Erreur {name}: {e}")
self.weights[name] *= 0.5
return name, pd.DataFrame()
def _assess_quality(self, df: pd.DataFrame) -> float:
"""Évalue la qualité d'un DataFrame."""
if df.empty or len(df) < 10:
return 0.0
score = 1.0
# Pénalité si gaps temporels
time_gaps = df.index.to_series().diff().dropna()
max_gap = timedelta(minutes=5)
gap_ratio = (time_gaps > max_gap).mean()
score *= (1 - gap_ratio * 0.5)
# Pénalité siOHLC invalides
invalid_ohlc = (df['high'] < df['low']) | \
(df['close'] > df['high']) | \
(df['close'] < df['low'])
score *= (1 - invalid_ohlc.mean() * 0.5)
# Pénalité si volume suspect
zero_volume = (df['volume'] == 0).mean()
extreme_volume = (df['volume'] > df['volume'].quantile(0.99)).mean()
score *= (1 - (zero_volume + extreme_volume) * 0.3)
return max(score, 0.1)
def fuse_data(
self,
data: Dict[str, pd.DataFrame],
method: str = 'weighted_average'
) -> pd.DataFrame:
"""
Fusionne les données de plusieurs exchanges.
Méthodes disponibles:
- weighted_average: Moyenne pondérée par qualité/latence
- median: Médiane robuste aux outliers
- latest_first: Priorise l'exchange le plus récent
- consensus: Valeur acceptée par la majorité
"""
if not data:
return pd.DataFrame()
# Trouver l'index commun (union de tous)
all_timestamps = set()
for df in data.values():
all_timestamps.update(df.index)
common_index = sorted(all_timestamps)
if method == 'weighted_average':
return self._weighted_fusion(data, common_index)
elif method == 'median':
return self._median_fusion(data, common_index)
elif method == 'consensus':
return self._consensus_fusion(data, common_index)
else:
return self._weighted_fusion(data, common_index)
def _weighted_fusion(
self,
data: Dict[str, pd.DataFrame],
index: pd.DatetimeIndex
) -> pd.DataFrame:
"""Fusion par moyenne pondérée."""
result = pd.DataFrame(index=index)
columns = ['open', 'high', 'low', 'close', 'volume']
for col in columns:
weighted_sum = None
total_weight = 0
for name, df in data.items():
weight = self.weights.get(name, 1.0) * \
self.quality_scores.get(name, 1.0)
# Réindexer vers l'index commun
aligned = df[col].reindex(index).fillna(method='ffill')
if weighted_sum is None:
weighted_sum = aligned * weight
else:
weighted_sum += aligned * weight
total_weight += weight
result[col] = weighted_sum / total_weight if total_weight > 0 else np.nan
return result
def _median_fusion(
self,
data: Dict[str, pd.DataFrame],
index: pd.DatetimeIndex
) -> pd.DataFrame:
"""Fusion par médiane (robuste aux Manipulations)."""
result = pd.DataFrame(index=index)
columns = ['open', 'high', 'low', 'close', 'volume']
for col in columns:
values = []
for name, df in data.items():
aligned = df[col].reindex(index).fillna(method='ffill')
values.append(aligned)
if values:
result[col] = np.nanmedian(values, axis=0)
return result
def _consensus_fusion(
self,
data: Dict[str, pd.DataFrame],
index: pd.DatetimeIndex
) -> pd.DataFrame:
"""Fusion par consensus (mode des valeurs proches)."""
result = pd.DataFrame(index=index)
columns = ['open', 'high', 'low', 'close', 'volume']
tolerance_bps = 5 # Tolérance en basis points
for ts in index:
row_data = {}
for col in columns:
values = []
for name, df in data.items():
if ts in df.index and not pd.isna(df.loc[ts, col]):
values.append((name, df.loc[ts, col]))
if len(values) >= 2:
# Trouver le consensus
base_value = values[0][1]
consensus_values = [
v for _, v in values
if abs(v - base_value) / base_value < tolerance_bps / 10000
]
if len(consensus_values) >= len(values) / 2:
row_data[col] = np.mean(consensus_values)
else:
# Pas de consensus, prendre la médiane
row_data[col] = np.median([v for _, v in values])
elif values:
row_data[col] = values[0][1]
for col in columns:
if col not in row_data:
row_data[col] = np.nan
result.loc[ts] = row_data
return result
async def detect_manipulation(
self,
fused_data: pd.DataFrame,
window: int = 60
) -> List[Dict]:
"""Détecte les patterns de manipulation dans les données fusionnées."""
alerts = []
# 1. Détecter les spoofing (gros ordres fictifs)
volume_std = fused_data['volume'].rolling(window).std()
volume_mean = fused_data['volume'].rolling(window).mean()
suspicious_volume = fused_data['volume'] > volume_mean + 5 * volume_std
# 2. Détecter leswash trading (volume sans mouvement)
price_change = fused_data['close'].pct_change()
volume_spike = fused_data['volume'] > volume_mean * 3
no_price_move = price_change.abs() < 0.0001
wash_trading = volume_spike & no_price_move
# 3. Détecter le painting (manipulation de prix)
for idx in fused_data.index:
if wash_trading.loc[idx]:
alerts.append({