# ml_engine/processor.py (Updated to pass LearningHubManager) import pandas as pd import numpy as np from datetime import datetime import asyncio import json import re # (Import components from the same folder) from .indicators import AdvancedTechnicalAnalyzer from .monte_carlo import MonteCarloAnalyzer from .patterns import ChartPatternAnalyzer from .strategies import MultiStrategyEngine class MLProcessor: # 🔴 --- START OF CHANGE --- 🔴 def __init__(self, market_context, data_manager, learning_hub): # (Changed from learning_engine) self.market_context = market_context self.data_manager = data_manager self.learning_hub = learning_hub # (Changed from learning_engine) self.technical_analyzer = AdvancedTechnicalAnalyzer() # (Pass the hub to the strategy engine) self.strategy_engine = MultiStrategyEngine(data_manager, learning_hub) self.monte_carlo_analyzer = MonteCarloAnalyzer() self.pattern_analyzer = ChartPatternAnalyzer() self.whale_data_semaphore = asyncio.Semaphore(2) # 🔴 --- END OF CHANGE --- 🔴 async def process_and_score_symbol_enhanced(self, raw_data, preloaded_whale_data: dict = None): """ (Unchanged logic, but now self.strategy_engine uses the learning_hub) """ try: if not raw_data or not raw_data.get('ohlcv'): return None symbol = raw_data['symbol'] base_analysis = await self.process_and_score_symbol(raw_data) if not base_analysis: return None try: # Calculate Indicators advanced_indicators = {} ohlcv_available = raw_data.get('ohlcv', {}) for timeframe, candles in ohlcv_available.items(): if candles and len(candles) >= 20: dataframe = self._create_dataframe(candles) indicators = self.technical_analyzer.calculate_all_indicators(dataframe, timeframe) advanced_indicators[timeframe] = indicators base_analysis['advanced_indicators'] = advanced_indicators # Monte Carlo (Phase 1) monte_carlo_results = await self.monte_carlo_analyzer.generate_1h_price_distribution(ohlcv_available) if monte_carlo_results: base_analysis['monte_carlo_distribution'] = monte_carlo_results base_analysis['monte_carlo_probability'] = monte_carlo_results.get('probability_of_gain', 0) base_analysis['monte_carlo_details'] = self.monte_carlo_analyzer.simulation_results else: base_analysis['monte_carlo_distribution'] = None base_analysis['monte_carlo_probability'] = 0 base_analysis['monte_carlo_details'] = self.monte_carlo_analyzer.simulation_results # Pattern Analysis pattern_analysis = await self.pattern_analyzer.detect_chart_patterns(ohlcv_available) base_analysis['pattern_analysis'] = pattern_analysis # Whale Data if preloaded_whale_data: base_analysis['whale_data'] = preloaded_whale_data.get(symbol, {'data_available': False, 'reason': 'Not preloaded'}) else: base_analysis['whale_data'] = {'data_available': False, 'reason': 'Preloading disabled'} # 🔴 (This call now uses the Learning Hub via strategy_engine) strategy_scores, base_scores = await self.strategy_engine.evaluate_all_strategies(base_analysis, self.market_context) base_analysis['strategy_scores'] = strategy_scores base_analysis['base_strategy_scores'] = base_scores if base_scores: best_strategy = max(base_scores.items(), key=lambda x: x[1]) best_strategy_name = best_strategy[0] best_strategy_score = best_strategy[1] base_analysis['recommended_strategy'] = best_strategy_name base_analysis['strategy_confidence'] = best_strategy_score base_analysis['target_strategy'] = best_strategy_name if best_strategy_score > 0.3 else 'GENERIC' enhanced_score = self._calculate_enhanced_final_score(base_analysis) base_analysis['enhanced_final_score'] = enhanced_score return base_analysis except Exception as strategy_error: print(f"❌ Error in advanced analysis for {symbol}: {strategy_error}") return base_analysis except Exception as error: print(f"❌ Fatal error in enhanced processing for {raw_data.get('symbol', 'unknown')}: {error}") return None def _create_dataframe(self, candles): # (This function remains unchanged) try: if not candles: return pd.DataFrame() df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df[['open', 'high', 'low', 'close', 'volume']] = df[['open', 'high', 'low', 'close', 'volume']].astype(float) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') df.set_index('timestamp', inplace=True) df.sort_index(inplace=True) return df except Exception as e: print(f"❌ Error creating DataFrame: {e}") return pd.DataFrame() def _calculate_enhanced_final_score(self, analysis): # (This function remains unchanged) try: base_score = analysis.get('final_score', 0) pattern_confidence = analysis.get('pattern_analysis', {}).get('pattern_confidence', 0) strategy_confidence = analysis.get('strategy_confidence', 0) mc_distribution = analysis.get('monte_carlo_distribution') monte_carlo_score = 0 if mc_distribution: prob_gain = mc_distribution.get('probability_of_gain', 0) var_95_value = mc_distribution.get('risk_metrics', {}).get('VaR_95_value', 0) current_price = analysis.get('current_price', 1) if current_price > 0: normalized_var = var_95_value / current_price risk_penalty = 1.0 if normalized_var > 0.05: risk_penalty = 0.5 elif normalized_var > 0.03: risk_penalty = 0.8 monte_carlo_score = prob_gain * risk_penalty else: monte_carlo_score = 0 whale_confidence = 0 whale_data = analysis.get('whale_data') if whale_data and whale_data.get('data_available'): signal = whale_data.get('trading_signal', {}) if signal.get('action') != 'HOLD' and signal.get('confidence', 0) >= 0.5: whale_confidence = signal.get('confidence', 0) components = [] weights = [] if base_score > 0: components.append(base_score); weights.append(0.20) if monte_carlo_score > 0: components.append(monte_carlo_score); weights.append(0.25) if pattern_confidence > 0: components.append(pattern_confidence); weights.append(0.25) if strategy_confidence > 0: components.append(strategy_confidence); weights.append(0.15) if whale_confidence > 0: components.append(whale_confidence); weights.append(0.15) if not components: return 0 total_weight = sum(weights) if total_weight == 0: return 0 enhanced_score = sum(comp * weight for comp, weight in zip(components, weights)) / total_weight return min(max(enhanced_score, 0.0), 1.0) except Exception as e: print(f"❌ Error calculating enhanced score: {e}") return analysis.get('final_score', 0) async def process_and_score_symbol(self, raw_data): # (This function remains unchanged) try: symbol = raw_data['symbol'] ohlcv_data = raw_data.get('ohlcv') if not ohlcv_data: return None current_price = raw_data.get('current_price', 0) layer1_score = raw_data.get('layer1_score', 0) reasons = raw_data.get('reasons_for_candidacy', []) final_score = layer1_score successful_timeframes = raw_data.get('successful_timeframes', 0) return { 'symbol': symbol, 'current_price': current_price, 'final_score': final_score, 'enhanced_final_score': final_score, 'reasons_for_candidacy': reasons, 'layer1_score': layer1_score, 'ohlcv': ohlcv_data, 'successful_timeframes': successful_timeframes } except Exception as error: print(f"❌ Error in basic symbol processing {raw_data.get('symbol', 'unknown')}: {error}") return None def filter_top_candidates(self, candidates, number_of_candidates=10): # (This function remains unchanged) valid_candidates = [c for c in candidates if c is not None and isinstance(c, dict)] if not valid_candidates: print("❌ No valid candidates to filter"); return [] sorted_candidates = sorted(valid_candidates, key=lambda c: c.get('enhanced_final_score', 0), reverse=True) top_candidates = sorted_candidates[:number_of_candidates] print(f"🎖️ Top {len(top_candidates)} Candidates:") for i, c in enumerate(top_candidates): score = c.get('enhanced_final_score', 0); strategy = c.get('recommended_strategy', 'GENERIC'); mc_dist = c.get('monte_carlo_distribution'); pattern = c.get('pattern_analysis', {}).get('pattern_detected', 'no_pattern'); symbol = c.get('symbol', 'UNKNOWN'); timeframes = c.get('successful_timeframes', 0) print(f" {i+1}. {symbol}: 📊 {score:.3f} | TFs: {timeframes}/6") if mc_dist: mc_pi_90 = mc_dist.get('prediction_interval_90', [0,0]); mc_var = mc_dist.get('risk_metrics', {}).get('VaR_95_value', 0) print(f" 🎯 MonteCarlo: 90% PI [{mc_pi_90[0]:.4f} - {mc_pi_90[1]:.4f}] | VaR: ${mc_var:.4f}") print(f" 🎯 Strategy: {strategy} | Pattern: {pattern}") whale_data = c.get('whale_data') if whale_data and whale_data.get('data_available'): signal = whale_data.get('trading_signal', {}); print(f" 🐋 Whale: {signal.get('action', 'HOLD')} (Conf: {signal.get('confidence', 0):.2f})") return top_candidates async def process_multiple_symbols_parallel(self, symbols_data_list, preloaded_whale_data: dict, max_concurrent=5): # (This function remains unchanged) semaphore = asyncio.Semaphore(max_concurrent) tasks_results = [] async def process_symbol_with_semaphore(symbol_data): async with semaphore: try: return await self.process_and_score_symbol_enhanced(symbol_data, preloaded_whale_data) except Exception as e: return e try: batch_tasks = [asyncio.create_task(process_symbol_with_semaphore(sd)) for sd in symbols_data_list] batch_results = await asyncio.gather(*batch_tasks, return_exceptions=False) successful_results = [] for result in batch_results: if isinstance(result, Exception): raise result if isinstance(result, dict): successful_results.append(result) return successful_results except Exception as error: raise error def safe_json_parse(json_string): # (This function remains unchanged) if not json_string: return None try: return json.loads(json_string) except json.JSONDecodeError: try: s = str(json_string).replace("'", '"'); s = re.sub(r'\\"', '"', s); s = re.sub(r'[\n\t]', ' ', s); s = re.sub(r'(?