""" ForexBot2 — Signal Aggregator ============================== Combines multi-timeframe technical analysis with optional news context into a graded actionable signal. Same code path is used for live mode and walk-forward backtesting; the only difference is whether the input OHLCV is fetched live or sliced from a pre-loaded historical buffer. """ 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 from data_provider import ForexDataProvider from engine import TechnicalEngine, TechnicalAnalysis from news_engine import NewsEngine, FundamentalContext logger = logging.getLogger(__name__) @dataclass class TradeSetup: pair: str display_name: str direction: str # "BUY" / "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: symbol: str display_name: str timestamp: str primary_signal: SignalStrength = SignalStrength.NEUTRAL primary_score: float = 0.0 primary_confidence: float = 0.0 mtf_agreement: bool = False mtf_signals: Dict[str, SignalStrength] = field(default_factory=dict) mtf_scores: Dict[str, float] = field(default_factory=dict) primary_analysis: Optional[TechnicalAnalysis] = None confirmation_analyses: List[TechnicalAnalysis] = field(default_factory=list) event_risk: str = "clear" event_description: str = "" news_sentiment: float = 0.0 news_description: str = "" 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: Optional[TradeSetup] = None 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: float = 0.0 spread_estimate: float = 0.0 market_regime: str = "" volatility_regime: str = "" class SignalAggregator: """ Master aggregator. Designed so the backtester can drive it bar-by-bar via `analyze_from_dataframes` without re-fetching from the API. """ def __init__(self, config: PlatformConfig): self.config = config self.data_provider = ForexDataProvider(config) self.tech_engine = TechnicalEngine(config.technical) self._news_engine: Optional[NewsEngine] = None # lazy def _get_news_engine(self) -> NewsEngine: if self._news_engine is None: self._news_engine = NewsEngine(self.config.news) return self._news_engine # ── Public: live mode entry point ────────────────────────────── def analyze_pair( self, symbol: Optional[str] = None, context: Optional[FundamentalContext] = None, ) -> SignalReport: """Live-mode entry: fetch all needed timeframes then delegate.""" if symbol is None: symbol = self.config.symbol timeframes = [self.config.primary_timeframe] + list( self.config.confirmation_timeframes ) bt = self.config.backtest if bt.regime_filter and bt.regime_filter_timeframe not in timeframes: timeframes.append(bt.regime_filter_timeframe) dfs: Dict[Timeframe, pd.DataFrame] = {} for tf in timeframes: try: dfs[tf] = self.data_provider.fetch_ohlcv(symbol, tf) except Exception as e: logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}") report = self.analyze_from_dataframes( symbol=symbol, dfs=dfs, context=context, use_news=True, ) if bt.regime_filter: self._apply_regime_filter_live(report, dfs) return report def _apply_regime_filter_live( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> None: """If higher-TF ADX is below threshold, downgrade signal to NEUTRAL.""" bt = self.config.backtest regime_tf = bt.regime_filter_timeframe df = dfs.get(regime_tf) if df is None or df.empty: report.warnings.append( f"Regime filter requested but {regime_tf.value} data unavailable" ) return adx_value = self._latest_adx(df, self.config.technical.adx_period) if adx_value is None: report.warnings.append( f"Regime filter: ADX could not be computed on {regime_tf.value}" ) return if adx_value < bt.regime_filter_min_adx: report.warnings.append( f"REGIME FILTER: {regime_tf.value} ADX={adx_value:.1f} < " f"{bt.regime_filter_min_adx:.0f} — market choppy, skipping signal" ) report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.action_recommendation = ( f"NO TRADE — Regime filter (ADX {adx_value:.1f} on {regime_tf.value})" ) report.trade_setup = None else: report.bullish_reasons.append( f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market" ) if report.final_score > 0 else report.bearish_reasons.append( f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market" ) @staticmethod def _latest_adx(df: pd.DataFrame, period: int = 14) -> Optional[float]: """Wilder ADX on a fresh df; returns the last finite value or None.""" if df is None or len(df) < period * 3: return None high, low, close = df["High"], df["Low"], df["Close"] tr = pd.concat([ high - low, (high - close.shift()).abs(), (low - close.shift()).abs(), ], axis=1).max(axis=1) plus_dm = high.diff() minus_dm = -low.diff() plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0) atr_s = tr.ewm(alpha=1/period, min_periods=period).mean() plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s) minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s) denom = (plus_di + minus_di).replace(0, float("nan")) dx = 100 * (plus_di - minus_di).abs() / denom adx = dx.ewm(alpha=1/period, min_periods=period).mean().dropna() if adx.empty: return None return float(adx.iloc[-1]) # ── Public: bar-replayable entry point ───────────────────────── def analyze_from_dataframes( self, symbol: str, dfs: Dict[Timeframe, pd.DataFrame], context: Optional[FundamentalContext] = None, use_news: bool = True, ) -> SignalReport: """ Run the full analysis pipeline on already-fetched dataframes. `dfs` must include the primary timeframe and ideally the confirmation timeframes. Each df is treated as the visible history at decision time — the backtester is responsible for slicing so no future bars leak in. """ 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 ─────────────────────────────────── primary_df = dfs.get(self.config.primary_timeframe) if primary_df is None or primary_df.empty: report.warnings.append("Primary timeframe data unavailable") report.signal_quality = "F" return report 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.timestamp = str(primary_df.index[-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 # ── 2. Confirmation timeframes ───────────────────────────── for tf in self.config.confirmation_timeframes: cf_df = dfs.get(tf) if cf_df is None or cf_df.empty: continue try: 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. MTF agreement ─────────────────────────────────────── report.mtf_agreement = self._check_mtf_agreement(report) # ── 4. Fundamental context (optional) ────────────────────── if use_news: news = self._get_news_engine() if context is None: context = news.get_fundamental_context() event_risk, event_desc = news.get_event_risk_level(symbol, context) report.event_risk = event_risk report.event_description = event_desc sentiment, sent_desc = news.get_pair_sentiment(symbol, context) report.news_sentiment = sentiment report.news_description = sent_desc else: report.event_risk = "clear" report.event_description = "" report.news_sentiment = 0.0 report.news_description = "" # ── 5. Spread estimate (only meaningful in live mode) ────── if use_news: try: report.spread_estimate = self.data_provider.get_spread_estimate(symbol) except Exception: report.spread_estimate = 0.0 else: # Backtest sets spread externally via BacktestConfig report.spread_estimate = 0.0 # ── 6. Final signal ──────────────────────────────────────── self._compute_final_signal(report) # ── 7. Trade setup for actionable signals ────────────────── if report.signal_quality in ("A+", "A", "B"): report.trade_setup = self._generate_trade_setup(report, pip_value) # ── 8. Reasons ───────────────────────────────────────────── self._collect_reasons(report) return report # ── Internals ────────────────────────────────────────────────── def _check_mtf_agreement(self, report: SignalReport) -> bool: 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): score = report.primary_score confidence = report.primary_confidence # MTF if report.mtf_agreement: 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: 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 if abs(report.news_sentiment) > 0.2: if (report.news_sentiment > 0 and score > 0) or \ (report.news_sentiment < 0 and score < 0): confidence = min(confidence + 5, 95) 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 if report.event_risk == "danger": report.warnings.append(f"EVENT RISK: {report.event_description}") confidence *= 0.3 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 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 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" if not report.action_recommendation: 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]: 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_mult = self.config.signals.atr_sl_multiplier sl_distance = atr * sl_mult tp1_distance = sl_distance * rr tp2_distance = sl_distance * rr * 1.5 if is_buy: entry, sl = price, price - sl_distance tp1, tp2 = price + tp1_distance, price + tp2_distance direction = "BUY" else: entry, sl = price, price + sl_distance tp1, tp2 = price - tp1_distance, 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): 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}") 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}")