#!/usr/bin/env python3 """ Position Bias Analyzer - Enrichit les signaux avec le contexte des positions ouvertes Utilise les positions actuelles pour confirmer ou ajuster les signaux de trading """ from typing import Dict, List, Tuple import logging logger = logging.getLogger(__name__) class PositionBiasAnalyzer: """ Analyse le biais des positions ouvertes pour enrichir les signaux """ def __init__(self): self.min_positions_for_analysis = 2 # Minimum de positions pour analyse def analyze_position_bias(self, positions: List[Dict], new_signal: Dict) -> Dict: """ Analyse le biais des positions et ajuste le signal en conséquence Args: positions: Liste des positions ouvertes new_signal: Signal à analyser et potentiellement ajuster Returns: Signal ajusté avec contexte de position """ if not positions or len(positions) < self.min_positions_for_analysis: return new_signal # Calculer les métriques de base bias_metrics = self._calculate_bias_metrics(positions) # Ajuster le signal selon le contexte adjusted_signal = self._adjust_signal_with_bias(new_signal, bias_metrics) return adjusted_signal def _calculate_bias_metrics(self, positions: List[Dict]) -> Dict: """Calcule les métriques de biais du portfolio""" long_positions = [p for p in positions if p['side'] == 'LONG'] short_positions = [p for p in positions if p['side'] == 'SHORT'] total_count = len(positions) long_count = len(long_positions) short_count = len(short_positions) # Performance moyenne par side long_pnl = sum(p['pnl_percent'] for p in long_positions) / long_count if long_count > 0 else 0 short_pnl = sum(p['pnl_percent'] for p in short_positions) / short_count if short_count > 0 else 0 total_pnl = sum(p['pnl_percent'] for p in positions) / total_count # Biais directionnel basé sur la PERFORMANCE, pas juste le nombre # Si les shorts performent mieux que les longs -> biais SHORT # Si les longs performent mieux que les shorts -> biais LONG performance_differential = short_pnl - long_pnl if abs(performance_differential) < 1.0: # Moins de 1% de différence directional_bias = 'NEUTRAL' elif short_pnl > long_pnl: directional_bias = 'SHORT' else: directional_bias = 'LONG' # Force du biais basée sur l'écart de performance bias_strength = min(1.0, abs(performance_differential) / 10.0) # Normalise sur 10% # Performance du biais dominant dominant_side_performance = short_pnl if directional_bias == 'SHORT' else long_pnl # Concentration (risque de sur-exposition) max_side_ratio = max(long_count, short_count) / total_count return { 'total_positions': total_count, 'long_count': long_count, 'short_count': short_count, 'directional_bias': directional_bias, 'bias_strength': bias_strength, 'long_pnl_avg': long_pnl, 'short_pnl_avg': short_pnl, 'total_pnl_avg': total_pnl, 'dominant_side_performance': dominant_side_performance, 'concentration_ratio': max_side_ratio, 'is_concentrated': max_side_ratio > 0.75, # Plus de 75% dans une direction 'is_profitable': total_pnl > 0 } def _adjust_signal_with_bias(self, signal: Dict, bias_metrics: Dict) -> Dict: """Ajuste le signal selon les métriques de biais""" adjusted_signal = signal.copy() original_confidence = signal.get('confidence', 0.5) reasoning_additions = [] signal_type = signal.get('signal_type', 'HOLD') signal_side = 'LONG' if signal_type == 'BUY' else 'SHORT' if signal_type == 'SELL' else 'NEUTRAL' if signal_side == 'NEUTRAL': return adjusted_signal # 1. STRONG MOMENTUM - Signal aligné avec biais très profitable (>5%) if (signal_side == bias_metrics['directional_bias'] and bias_metrics['dominant_side_performance'] > 5.0): boost = 0.25 + (bias_metrics['bias_strength'] * 0.15) # 25-40% boost selon force adjusted_signal['confidence'] = min(1.0, original_confidence + boost) reasoning_additions.append( f"STRONG_MOMENTUM: {bias_metrics['directional_bias']} bias very profitable " f"({bias_metrics['dominant_side_performance']:+.1f}%) +{boost*100:.0f}%" ) # 2. PROFITABLE TREND - Signal aligné avec biais profitable (>1%) elif (signal_side == bias_metrics['directional_bias'] and bias_metrics['dominant_side_performance'] > 1.0): boost = 0.15 + (bias_metrics['bias_strength'] * 0.1) # 15-25% boost selon force adjusted_signal['confidence'] = min(1.0, original_confidence + boost) reasoning_additions.append( f"PROFITABLE_TREND: {bias_metrics['directional_bias']} bias profitable " f"({bias_metrics['dominant_side_performance']:+.1f}%) +{boost*100:.0f}%" ) # 3. LOSING TREND STOP - Signal aligné avec biais perdant (<-2%) elif (signal_side == bias_metrics['directional_bias'] and bias_metrics['dominant_side_performance'] < -2.0): reduction = 0.2 + (bias_metrics['bias_strength'] * 0.15) # 20-35% réduction adjusted_signal['confidence'] = max(0.1, original_confidence - reduction) reasoning_additions.append( f"LOSING_TREND_STOP: {bias_metrics['directional_bias']} bias losing " f"({bias_metrics['dominant_side_performance']:+.1f}%) -{reduction*100:.0f}%" ) # 4. DIVERSIFICATION OPPORTUNITY - Signal opposé quand concentration élevée sur bias neutre/faible elif (bias_metrics['is_concentrated'] and signal_side != bias_metrics['directional_bias'] and abs(bias_metrics['dominant_side_performance']) < 2.0): # Bias neutre diversification_boost = 0.2 adjusted_signal['confidence'] = min(1.0, original_confidence + diversification_boost) reasoning_additions.append( f"DIVERSIFICATION: High {bias_metrics['directional_bias']} concentration " f"but neutral performance, diversify +{diversification_boost*100:.0f}%" ) # 5. CAPITULATION SIGNAL - Portfolio très perdant, signal opposé au bias elif (bias_metrics['total_pnl_avg'] < -5.0 and signal_side != bias_metrics['directional_bias']): capitulation_boost = 0.25 adjusted_signal['confidence'] = min(1.0, original_confidence + capitulation_boost) reasoning_additions.append( f"CAPITULATION_SIGNAL: Portfolio losing ({bias_metrics['total_pnl_avg']:+.1f}%), " f"potential reversal +{capitulation_boost*100:.0f}%" ) # 6. CONTRE-TREND RISK - Signal opposé à un bias profitable fort elif (signal_side != bias_metrics['directional_bias'] and bias_metrics['dominant_side_performance'] > 3.0): reduction = 0.1 + (bias_metrics['bias_strength'] * 0.1) # 10-20% réduction adjusted_signal['confidence'] = max(0.2, original_confidence - reduction) reasoning_additions.append( f"COUNTER_TREND_RISK: Against profitable {bias_metrics['directional_bias']} " f"({bias_metrics['dominant_side_performance']:+.1f}%) -{reduction*100:.0f}%" ) # Ajouter le raisonnement if reasoning_additions: existing_reasoning = adjusted_signal.get('reasoning', '') bias_reasoning = ' | '.join(reasoning_additions) adjusted_signal['reasoning'] = f"{existing_reasoning} [POSITION_BIAS: {bias_reasoning}]" return adjusted_signal def get_portfolio_summary(self, positions: List[Dict]) -> Dict: """Résumé du portfolio pour logging et debug""" if not positions: return {'status': 'empty'} bias_metrics = self._calculate_bias_metrics(positions) return { 'total_positions': bias_metrics['total_positions'], 'directional_split': f"{bias_metrics['long_count']}L/{bias_metrics['short_count']}S", 'bias': bias_metrics['directional_bias'], 'bias_strength': f"{bias_metrics['bias_strength']*100:.0f}%", 'performance': f"{bias_metrics['total_pnl_avg']:+.1f}%", 'concentration': f"{bias_metrics['concentration_ratio']*100:.0f}%", 'risk_level': self._assess_risk_level(bias_metrics) } def _assess_risk_level(self, bias_metrics: Dict) -> str: """Évalue le niveau de risque du portfolio""" if bias_metrics['is_concentrated'] and bias_metrics['dominant_side_performance'] < -5: return "HIGH" elif bias_metrics['is_concentrated']: return "MEDIUM" elif abs(bias_metrics['total_pnl_avg']) > 10: return "MEDIUM" else: return "LOW" # Instance globale position_bias_analyzer = PositionBiasAnalyzer() def analyze_signal_with_positions(positions: List[Dict], signal: Dict) -> Dict: """Fonction helper pour analyser un signal avec contexte positions""" return position_bias_analyzer.analyze_position_bias(positions, signal) def get_portfolio_context(positions: List[Dict]) -> Dict: """Fonction helper pour obtenir le contexte portfolio""" return position_bias_analyzer.get_portfolio_summary(positions)