""" Forex Signal Intelligence Platform - Signal Aggregator ======================================================== Combines technical analysis across multiple timeframes with fundamental/news context to produce final actionable signals. This is the brain — it decides what's actually worth paying attention to. Key principles: 1. Multi-timeframe agreement = higher confidence 2. High-impact news overrides technicals 3. Divergences between timeframes = caution 4. No signal is better than a bad signal """ import logging from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Dict, List, Optional import pandas as pd from config import ( PlatformConfig, SignalStrength, Timeframe, SYMBOL_DISPLAY, ) from data_provider import ForexDataProvider from engine import TechnicalEngine, TechnicalAnalysis from news_engine import NewsEngine, FundamentalContext logger = logging.getLogger(__name__) @dataclass class TradeSetup: """A complete trade setup recommendation.""" pair: str display_name: str direction: str # "BUY" or "SELL" entry_price: float stop_loss: float take_profit_1: float take_profit_2: float risk_reward: float risk_pips: float reward_pips: float pip_value: float @dataclass class SignalReport: """Complete signal report for a single pair.""" symbol: str display_name: str timestamp: str # Composite signals primary_signal: SignalStrength = SignalStrength.NEUTRAL primary_score: float = 0.0 primary_confidence: float = 0.0 # Multi-timeframe agreement mtf_agreement: bool = False mtf_signals: Dict[str, SignalStrength] = field(default_factory=dict) mtf_scores: Dict[str, float] = field(default_factory=dict) # Technical details primary_analysis: Optional[TechnicalAnalysis] = None confirmation_analyses: List[TechnicalAnalysis] = field(default_factory=list) # Fundamental context event_risk: str = "clear" event_description: str = "" news_sentiment: float = 0.0 news_description: str = "" # Final verdict final_signal: SignalStrength = SignalStrength.NEUTRAL final_score: float = 0.0 final_confidence: float = 0.0 action_recommendation: str = "" signal_quality: str = "" # "A+", "A", "B", "C", "D", "F" # Trade setup (only for actionable signals) trade_setup: Optional[TradeSetup] = None # Key reasons bullish_reasons: List[str] = field(default_factory=list) bearish_reasons: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) # Current price info current_price: float = 0.0 spread_estimate: float = 0.0 market_regime: str = "" volatility_regime: str = "" class SignalAggregator: """ Master signal aggregator. Orchestrates data fetching, technical analysis across timeframes, fundamental analysis, and produces final graded signals. """ def __init__(self, config: PlatformConfig): self.config = config self.data_provider = ForexDataProvider(config) self.tech_engine = TechnicalEngine(config.technical) self.news_engine = NewsEngine(config.news) def analyze_all_pairs(self) -> List[SignalReport]: """ Analyze all watched pairs and return sorted signal reports. Returns reports sorted by signal quality (best first). """ reports = [] context = self.news_engine.get_fundamental_context() # Batch-prefetch all pairs × all needed timeframes (3 API calls instead of 21) all_timeframes = ( [self.config.primary_timeframe] + self.config.confirmation_timeframes ) self.data_provider.prefetch_batch(self.config.watch_pairs, all_timeframes) for symbol in self.config.watch_pairs: try: report = self.analyze_pair(symbol, context) reports.append(report) except Exception as e: logger.error(f"Failed to analyze {symbol}: {e}") continue # Sort by absolute score (strongest signals first) reports.sort(key=lambda r: abs(r.final_score), reverse=True) return reports def analyze_pair( self, symbol: str, context: Optional[FundamentalContext] = None, ) -> SignalReport: """Complete analysis of a single pair across all layers.""" if context is None: context = self.news_engine.get_fundamental_context() pair_info = self.data_provider.get_pair_info(symbol) display = pair_info["display"] pip_value = pair_info["pip_value"] report = SignalReport( symbol=symbol, display_name=display, timestamp=str(datetime.now(timezone.utc)), ) # ── 1. Primary Timeframe Analysis ────────────────────────── try: primary_df = self.data_provider.fetch_ohlcv( symbol, self.config.primary_timeframe ) primary_analysis = self.tech_engine.analyze( primary_df, symbol, self.config.primary_timeframe.value ) report.primary_analysis = primary_analysis report.primary_signal = primary_analysis.composite_signal report.primary_score = primary_analysis.composite_score report.primary_confidence = primary_analysis.confidence report.market_regime = primary_analysis.market_regime report.volatility_regime = primary_analysis.volatility_regime report.current_price = float(primary_df["Close"].iloc[-1]) report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score except Exception as e: logger.error(f"Primary analysis failed for {symbol}: {e}") report.warnings.append(f"Primary timeframe data unavailable: {e}") report.signal_quality = "F" return report # ── 2. Confirmation Timeframe Analyses ───────────────────── for tf in self.config.confirmation_timeframes: try: cf_df = self.data_provider.fetch_ohlcv(symbol, tf) cf_analysis = self.tech_engine.analyze(cf_df, symbol, tf.value) report.confirmation_analyses.append(cf_analysis) report.mtf_signals[tf.value] = cf_analysis.composite_signal report.mtf_scores[tf.value] = cf_analysis.composite_score except Exception as e: logger.warning(f"Confirmation TF {tf.value} failed for {symbol}: {e}") # ── 3. Multi-Timeframe Agreement Check ───────────────────── report.mtf_agreement = self._check_mtf_agreement(report) # ── 4. Fundamental Context ───────────────────────────────── event_risk, event_desc = self.news_engine.get_event_risk_level(symbol, context) report.event_risk = event_risk report.event_description = event_desc sentiment, sent_desc = self.news_engine.get_pair_sentiment(symbol, context) report.news_sentiment = sentiment report.news_description = sent_desc # ── 5. Spread Estimate ───────────────────────────────────── report.spread_estimate = self.data_provider.get_spread_estimate(symbol) # ── 6. Compute Final Signal ──────────────────────────────── self._compute_final_signal(report) # ── 7. Generate Trade Setup (if signal is actionable) ────── if report.signal_quality in ("A+", "A", "B"): report.trade_setup = self._generate_trade_setup(report, pip_value) # ── 8. Collect Reasons ───────────────────────────────────── self._collect_reasons(report) return report def _check_mtf_agreement(self, report: SignalReport) -> bool: """Check if multiple timeframes agree on direction.""" signals = list(report.mtf_signals.values()) if len(signals) < 2: return False bullish = sum(1 for s in signals if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY)) bearish = sum(1 for s in signals if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL)) threshold = self.config.signals.mtf_agreement_threshold return bullish >= threshold or bearish >= threshold def _compute_final_signal(self, report: SignalReport): """ Compute the final signal by combining all layers. Priority: 1. Event risk can downgrade or block signals 2. MTF agreement boosts confidence 3. News sentiment adds a small bias 4. Quality grade combines everything """ score = report.primary_score confidence = report.primary_confidence # ── MTF adjustment ───────────────────────────────────────── if report.mtf_agreement: # All timeframes agree — boost confidence and blend scores confidence = min(confidence + 15, 95) all_scores = list(report.mtf_scores.values()) if all_scores: mtf_avg = sum(all_scores) / len(all_scores) score = score * 0.6 + mtf_avg * 0.4 else: # Timeframes disagree — reduce confidence confidence *= 0.7 if len(report.mtf_scores) >= 2: scores = list(report.mtf_scores.values()) has_bull = any(s > 10 for s in scores) has_bear = any(s < -10 for s in scores) if has_bull and has_bear: report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes") confidence *= 0.6 # ── News sentiment adjustment ────────────────────────────── if abs(report.news_sentiment) > 0.2: # Sentiment aligns with technical if (report.news_sentiment > 0 and score > 0) or \ (report.news_sentiment < 0 and score < 0): confidence = min(confidence + 5, 95) # Sentiment conflicts with technical elif (report.news_sentiment > 0 and score < -10) or \ (report.news_sentiment < 0 and score > 10): report.warnings.append("News sentiment CONFLICTS with technical signal") confidence *= 0.85 # ── Event risk adjustment ────────────────────────────────── if report.event_risk == "danger": report.warnings.append(f"EVENT RISK: {report.event_description}") confidence *= 0.3 # Heavily penalize — event will disrupt price action report.action_recommendation = f"WAIT — {report.event_description}" elif report.event_risk == "caution": report.warnings.append(f"Upcoming event: {report.event_description}") confidence *= 0.7 # ── Volatility regime adjustment ─────────────────────────── if report.volatility_regime == "extreme": report.warnings.append("EXTREME volatility — wider stops needed, reduced position size") confidence *= 0.8 elif report.volatility_regime == "low": report.warnings.append("Low volatility — potential for breakout, tight ranges") # ── Final signal determination ───────────────────────────── if score >= 40 and confidence >= 65: report.final_signal = SignalStrength.STRONG_BUY elif score >= 15 and confidence >= 50: report.final_signal = SignalStrength.BUY elif score <= -40 and confidence >= 65: report.final_signal = SignalStrength.STRONG_SELL elif score <= -15 and confidence >= 50: report.final_signal = SignalStrength.SELL else: report.final_signal = SignalStrength.NEUTRAL report.final_score = score report.final_confidence = confidence # ── Quality grade ────────────────────────────────────────── abs_score = abs(score) if abs_score >= 50 and confidence >= 75 and report.mtf_agreement: report.signal_quality = "A+" elif abs_score >= 35 and confidence >= 65 and report.mtf_agreement: report.signal_quality = "A" elif abs_score >= 25 and confidence >= 55: report.signal_quality = "B" elif abs_score >= 15 and confidence >= 45: report.signal_quality = "C" elif abs_score >= 10: report.signal_quality = "D" else: report.signal_quality = "F" # ── Action recommendation ────────────────────────────────── if not report.action_recommendation: # not already set by event risk sig_name = report.final_signal.name.replace("_", " ") if report.signal_quality in ("A+", "A"): report.action_recommendation = f"{sig_name} — High-quality setup, consider entry" elif report.signal_quality == "B": report.action_recommendation = f"{sig_name} — Decent setup, wait for confirmation" elif report.signal_quality in ("C", "D"): report.action_recommendation = "WATCH — Signal present but low confidence" else: report.action_recommendation = "NO TRADE — No clear signal" def _generate_trade_setup(self, report: SignalReport, pip_value: float) -> Optional[TradeSetup]: """Generate entry, SL, TP levels for actionable signals.""" if not report.primary_analysis: return None price = report.current_price atr = report.primary_analysis.key_levels.get("atr", 0) if atr == 0: return None is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) rr = self.config.signals.default_rr_ratio # SL = 1.5x ATR from entry (adaptive to volatility) sl_distance = atr * 1.5 tp1_distance = sl_distance * rr tp2_distance = sl_distance * rr * 1.5 if is_buy: entry = price sl = price - sl_distance tp1 = price + tp1_distance tp2 = price + tp2_distance direction = "BUY" else: entry = price sl = price + sl_distance tp1 = price - tp1_distance tp2 = price - tp2_distance direction = "SELL" risk_pips = sl_distance / pip_value reward_pips = tp1_distance / pip_value return TradeSetup( pair=report.symbol, display_name=report.display_name, direction=direction, entry_price=entry, stop_loss=sl, take_profit_1=tp1, take_profit_2=tp2, risk_reward=rr, risk_pips=risk_pips, reward_pips=reward_pips, pip_value=pip_value, ) def _collect_reasons(self, report: SignalReport): """ Collect bullish/bearish reasons from all indicators. Each indicator's description is captured once via the indicators dict. Pattern signals (candlestick) are already included in the 'candlestick' indicator description, so we do NOT re-process pattern_signals to avoid duplicate entries. """ if not report.primary_analysis: return for name, ind in report.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue if ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_BUY else "" report.bullish_reasons.append(f"{prefix}{ind.name}: {ind.description}") elif ind.signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_SELL else "" report.bearish_reasons.append(f"{prefix}{ind.name}: {ind.description}") # Add news context if report.news_sentiment > 0.1: report.bullish_reasons.append(f"News: {report.news_description}") elif report.news_sentiment < -0.1: report.bearish_reasons.append(f"News: {report.news_description}")