From ce4a12877d96930df2c16163cf2d2f6a41723969 Mon Sep 17 00:00:00 2001 From: harry Date: Mon, 4 May 2026 04:10:01 +0000 Subject: [PATCH] Add signal_aggregator.py --- signal_aggregator.py | 408 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 408 insertions(+) create mode 100644 signal_aggregator.py diff --git a/signal_aggregator.py b/signal_aggregator.py new file mode 100644 index 0000000..8a474e6 --- /dev/null +++ b/signal_aggregator.py @@ -0,0 +1,408 @@ +""" +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, Tuple + +import pandas as pd + +from config import ( + PlatformConfig, SignalStrength, Timeframe, + SYMBOL_DISPLAY, MAJOR_PAIRS, CROSS_PAIRS, +) +from data_provider import ForexDataProvider +from technical_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() + + 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 + confidence = min(confidence + 15, 95) + # Average the MTF scores for more stable signal + 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()) + # Check for conflicting directions + 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 ────────────────────────────── + sent_weight = self.config.signals.weights.get("sentiment", 0.1) + 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 + 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 = f"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.""" + 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 pattern signals + for pattern in report.primary_analysis.pattern_signals: + if any(kw in pattern.lower() for kw in ["bullish", "hammer", "buy"]): + report.bullish_reasons.append(f"Pattern: {pattern}") + elif any(kw in pattern.lower() for kw in ["bearish", "shooting", "sell"]): + report.bearish_reasons.append(f"Pattern: {pattern}") + + # 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}") \ No newline at end of file