""" ForexBot2 — Signal Aggregator ============================== Combines multi-timeframe technical analysis with optional news context into a graded actionable signal. """ import logging import os 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 for live signal analysis.""" 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 pair_info = self.data_provider.get_pair_info(symbol) now_utc = datetime.now(timezone.utc) if pair_info["display"] == "XAU/USD" and self.config.xau_strategy.enabled: xau_cfg = self.config.xau_strategy if xau_cfg.weekend_filter and now_utc.weekday() >= 5: report = SignalReport( symbol=symbol, display_name=pair_info["display"], timestamp=str(now_utc), ) report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.trade_setup = None report.warnings.append("WEEKEND FILTER: XAU/USD market closed — no liquidity") report.action_recommendation = "NO TRADE — Weekend (market closed)" return report if xau_cfg.allowed_weekdays and now_utc.weekday() not in xau_cfg.allowed_weekdays: day_names = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu", 4: "Fri"} allowed = "/".join(day_names.get(d, str(d)) for d in xau_cfg.allowed_weekdays) report = SignalReport( symbol=symbol, display_name=pair_info["display"], timestamp=str(now_utc), ) report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.trade_setup = None report.warnings.append( f"DAY FILTER: {day_names.get(now_utc.weekday(), '?')} not in allowed days ({allowed})" ) report.action_recommendation = f"NO TRADE — Only trading {allowed}" return report 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] = {} stale_tfs = [] for tf in timeframes: try: df = self.data_provider.fetch_ohlcv(symbol, tf) dfs[tf] = df if not self.data_provider._is_data_fresh(df, symbol, tf): stale_tfs.append(tf.value) 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 stale_tfs: report.warnings.append(f"STALE DATA on {', '.join(stale_tfs)} — signal may be unreliable") if self.config.primary_timeframe.value in stale_tfs: report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.trade_setup = None report.warnings.append("Primary timeframe stale — signal suppressed") if bt.regime_filter: self._apply_regime_filter_live(report, dfs) if bt.session_filter: xau_active = ( report.display_name == "XAU/USD" and self.config.xau_strategy.enabled and any("XAU STRATEGY" in r for r in report.bullish_reasons) ) eur_active = ( report.display_name == "EUR/USD" and self.config.eur_strategy.enabled and any("EUR STRATEGY" in r for r in report.bullish_reasons) ) gbp_active = ( report.display_name == "GBP/USD" and self.config.gbp_strategy.enabled and any("GBP STRATEGY" in r for r in report.bullish_reasons) ) if not xau_active and not eur_active and not gbp_active: self._apply_session_filter_live(report) return report def _apply_session_filter_live(self, report: SignalReport) -> None: """Skip signals outside the configured UTC trading window.""" bt = self.config.backtest now_h = datetime.now(timezone.utc).hour if not (bt.session_start_hour <= now_h < bt.session_end_hour): report.warnings.append( f"SESSION FILTER: {now_h:02}:00 UTC outside " f"{bt.session_start_hour:02}:00-{bt.session_end_hour:02}:00 — skipping" ) report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.action_recommendation = ( f"NO TRADE — Outside session " f"({bt.session_start_hour:02}-{bt.session_end_hour:02} UTC; now {now_h:02})" ) report.trade_setup = None else: report.warnings.append( f"Session: {now_h:02}:00 UTC inside trading window " f"{bt.session_start_hour:02}-{bt.session_end_hour:02}" ) 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, adjust=False).mean() plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s) minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).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, adjust=False).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 from primary data (avoids extra API call) ─ try: report.spread_estimate = float( (primary_df["High"] - primary_df["Low"]).median() ) except Exception: report.spread_estimate = 0.0 # ── 6. Final signal ──────────────────────────────────────── self._compute_final_signal(report) # ── 6b. Session filter (works for both live and backtest) ─ if self.config.backtest.session_filter: try: bar_ts = pd.Timestamp(report.timestamp) bar_hour = bar_ts.hour bar_wd = bar_ts.weekday() _is_xau_pair = report.display_name == "XAU/USD" _blocked_hours = (self.config.xau_strategy.blocked_hours_utc if _is_xau_pair else self.config.backtest.blocked_hours_utc) if bar_hour in _blocked_hours: report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.warnings.append(f"SESSION FILTER: {bar_hour:02}:00 UTC blocked") _allowed_wd = (self.config.xau_strategy.allowed_weekdays if _is_xau_pair else self.config.backtest.allowed_weekdays) if _allowed_wd and bar_wd not in _allowed_wd: day_names = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu", 4: "Fri", 5: "Sat", 6: "Sun"} report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.warnings.append( f"DAY FILTER: {day_names.get(bar_wd, '?')} not in allowed days" ) except Exception: pass # ── 6c. XAU/USD oversold bounce strategy (DISABLED — replaced by v2 confluence) ── # if report.display_name == "XAU/USD" and self.config.xau_strategy.enabled: # if self._check_xau_oversold_bounce(report, dfs): # report.final_signal = SignalStrength.STRONG_BUY # report.final_score = 50.0 # report.final_confidence = 85.0 # report.signal_quality = "A+" # report.action_recommendation = ( # "STRONG BUY — XAU oversold bounce " # "(RSI<30 + M15/H1/H4 all above EMA200)" # ) # report.bullish_reasons.insert( # 0, # "XAU STRATEGY: M15 RSI oversold + triple-TF bullish trend confirmation", # ) # report.trade_setup = self._generate_xau_trade_setup(report, dfs) # ── 6d. EUR/USD pullback-buy strategy ───────────────────── if report.display_name == "EUR/USD" and self.config.eur_strategy.enabled: if self._check_eur_pullback_buy(report, dfs): report.final_signal = SignalStrength.BUY report.final_score = 40.0 report.final_confidence = 75.0 report.signal_quality = "A" report.action_recommendation = ( "BUY — EUR pullback entry " "(RSI pullback + not at resistance + M15/H1 above EMA200)" ) report.bullish_reasons.insert( 0, "EUR STRATEGY: M15 RSI pullback + BB mid-band + dual-TF EMA200 confirmation", ) report.trade_setup = self._generate_eur_trade_setup(report, dfs) # ── 6e. GBP/USD pullback-buy strategy ───────────────────── if report.display_name == "GBP/USD" and self.config.gbp_strategy.enabled: if self._check_gbp_pullback_buy(report, dfs): report.final_signal = SignalStrength.BUY report.final_score = 40.0 report.final_confidence = 75.0 report.signal_quality = "A" report.action_recommendation = ( "BUY — GBP pullback entry " "(RSI pullback + not at resistance + M15/H1 above EMA200)" ) report.bullish_reasons.insert( 0, "GBP STRATEGY: M15 RSI pullback + BB mid-band + dual-TF EMA200 confirmation", ) report.trade_setup = self._generate_gbp_trade_setup(report, dfs) # ── 6d. XAU/USD BUY threshold boost (Fix 5) ────────────── if (report.display_name == "XAU/USD" and self.config.xau_strategy.buy_threshold_boost > 0 and report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) and report.signal_quality in ("B", "C")): report.warnings.append( f"XAU BUY FILTER: BUY requires higher confidence (quality {report.signal_quality} insufficient)" ) report.signal_quality = "D" report.trade_setup = None # ── 6f. XAU/USD H4 RSI + H1 MACD directional gates ────── if (report.display_name == "XAU/USD" and self.config.xau_strategy.enabled and report.signal_quality in ("A+", "A", "B", "C") and report.final_signal != SignalStrength.NEUTRAL): xau_cfg = self.config.xau_strategy is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) if xau_cfg.h4_rsi_gate and Timeframe.H4 in dfs: h4_close = dfs[Timeframe.H4]["Close"] delta = h4_close.diff() gain = delta.clip(lower=0).ewm(alpha=1/14, min_periods=14, adjust=False).mean() loss = (-delta.clip(upper=0)).ewm(alpha=1/14, min_periods=14, adjust=False).mean() h4_rsi = 100 - (100 / (1 + gain / loss)) if len(h4_rsi) > 0: h4_rsi_val = float(h4_rsi.iloc[-1]) if is_buy and h4_rsi_val <= 55: report.warnings.append( f"H4 RSI GATE: RSI {h4_rsi_val:.0f} ≤55 — no BUY confirmation" ) report.signal_quality = "D" report.trade_setup = None elif not is_buy and h4_rsi_val >= 45: report.warnings.append( f"H4 RSI GATE: RSI {h4_rsi_val:.0f} ≥45 — no SELL confirmation" ) report.signal_quality = "D" report.trade_setup = None if (xau_cfg.h1_macd_gate and Timeframe.H1 in dfs and report.signal_quality in ("A+", "A", "B", "C")): h1_df = dfs[Timeframe.H1] h1_close = h1_df["Close"] h1_ema_fast = h1_close.ewm(span=12, adjust=False).mean() h1_ema_slow = h1_close.ewm(span=26, adjust=False).mean() h1_macd_line = h1_ema_fast - h1_ema_slow h1_signal_line = h1_macd_line.ewm(span=9, adjust=False).mean() h1_macd_hist = h1_macd_line - h1_signal_line if len(h1_macd_hist) > 0: h1_hist_val = float(h1_macd_hist.iloc[-1]) if is_buy and h1_hist_val <= 0: report.warnings.append( f"H1 MACD GATE: histogram {h1_hist_val:.2f} ≤0 — no BUY confirmation" ) report.signal_quality = "D" report.trade_setup = None elif not is_buy and h1_hist_val >= 0: report.warnings.append( f"H1 MACD GATE: histogram {h1_hist_val:.2f} ≥0 — no SELL confirmation" ) report.signal_quality = "D" report.trade_setup = None # ── RSI zone gate ────────────────────────────────────── if (xau_cfg.rsi_zone and report.signal_quality in ("A+", "A", "B", "C")): m15_analysis = report.primary_analysis if m15_analysis and hasattr(m15_analysis, '_rsi_series'): rsi_val = float(m15_analysis._rsi_series.iloc[-1]) lo, hi = xau_cfg.rsi_zone if not (lo <= rsi_val <= hi): report.warnings.append( f"RSI ZONE GATE: RSI {rsi_val:.0f} outside [{lo:.0f}-{hi:.0f}]" ) report.signal_quality = "D" report.trade_setup = None # ── 6g. XAU/USD SELL quality gate ───────────────────────── xau_is_sell_dir = ( report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL) or (report.final_signal == SignalStrength.NEUTRAL and report.final_score < 0) ) if (report.display_name == "XAU/USD" and self.config.xau_strategy.enabled and xau_is_sell_dir and report.signal_quality not in self.config.xau_strategy.sell_min_quality): report.warnings.append( f"XAU SELL FILTER: quality {report.signal_quality} below minimum for SELL" ) report.signal_quality = "D" report.trade_setup = None # ── 7. Trade setup for actionable signals ────────────────── # Pair-specific generic-trade suppression: when a pair has a dedicated # strategy and that strategy didn't fire, falling through to a generic # swing-based setup tends to chase momentum at unfavorable spots. # 90-day EUR M15 backtest: allow=True → +4.7R, allow=False → +9.5R. suppress_generic = ( (report.display_name == "XAU/USD" and not self.config.xau_strategy.allow_generic_trades) or (report.display_name == "EUR/USD" and self.config.eur_strategy.enabled and not self.config.eur_strategy.allow_generic_trades) or (report.display_name == "GBP/USD" and self.config.gbp_strategy.enabled and not self.config.gbp_strategy.allow_generic_trades) ) if report.trade_setup is None and suppress_generic: pass # generic trade setup suppressed for this pair elif report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"): is_sell_direction = ( report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL) or (report.final_signal == SignalStrength.NEUTRAL and report.final_score < 0) ) if is_sell_direction and self.config.backtest.sell_filter: if report.display_name in self.config.backtest.sell_allowed_pairs: report.trade_setup = self._generate_trade_setup(report, pip_value) elif self._forex_sell_qualifies(report): report.trade_setup = self._generate_trade_setup(report, pip_value) else: report.trade_setup = None else: report.trade_setup = self._generate_trade_setup(report, pip_value) # ── 8. Reasons ───────────────────────────────────────────── self._collect_reasons(report) return report # ── Internals ────────────────────────────────────────────────── def _forex_sell_qualifies(self, report: SignalReport) -> bool: """Always returns False — forex SELL is structurally unprofitable. 0/4 trades won even under best conditions (trending_down + MTF + ADX>=25 + RSI<=50).""" return False def _check_xau_oversold_bounce( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> bool: """Return True when XAU/USD oversold-bounce conditions are met. Conditions: 1. M15 RSI(14) < rsi_entry_threshold 2. Price > EMA(200) on at least min_ema200_tfs of {M15, H1, H4} 3. Current bar hour not in blocked list """ cfg = self.config.xau_strategy try: bar_hour = pd.Timestamp(report.timestamp).hour if bar_hour in cfg.blocked_hours_utc: return False except Exception: return False m15_analysis = None h4_analysis = None for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca elif ca.timeframe == Timeframe.H4.value: h4_analysis = ca h1_analysis = report.primary_analysis if not m15_analysis or not h1_analysis: return False m15_rsi = m15_analysis.indicators.get("rsi") if not m15_rsi or m15_rsi.value >= cfg.rsi_entry_threshold: return False ema200_count = 0 m15_ema200 = m15_analysis.key_levels.get("ema_trend") m15_df = dfs.get(Timeframe.M15) if m15_df is not None and m15_ema200 is not None: m15_price = float(m15_df["Close"].iloc[-1]) if m15_price > m15_ema200: ema200_count += 1 h1_ema200 = h1_analysis.key_levels.get("ema_trend") if h1_ema200 is not None and report.current_price > h1_ema200: ema200_count += 1 if h4_analysis: h4_ema200 = h4_analysis.key_levels.get("ema_trend") h4_df = dfs.get(Timeframe.H4) if h4_df is not None and h4_ema200 is not None: h4_price = float(h4_df["Close"].iloc[-1]) if h4_price > h4_ema200: ema200_count += 1 if ema200_count < cfg.min_ema200_tfs: return False return True def _generate_xau_trade_setup( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> Optional[TradeSetup]: """Generate XAU-specific trade setup with backtested TP/SL.""" cfg = self.config.xau_strategy m15_analysis = None for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca break if not m15_analysis: return None atr = m15_analysis.key_levels.get("atr", 0) if atr <= 0: return None price = report.current_price pip_value = 0.10 tp1 = price + cfg.tp_atr_multiplier * atr sl = price - cfg.sl_atr_multiplier * atr tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr sl_distance = price - sl tp_distance = tp1 - price rr = tp_distance / sl_distance if sl_distance > 0 else 0 return TradeSetup( pair=report.symbol, display_name=report.display_name, direction="BUY", entry_price=price, stop_loss=round(sl, 2), take_profit_1=round(tp1, 2), take_profit_2=round(tp2, 2), risk_reward=round(rr, 2), risk_pips=round(sl_distance / pip_value, 1), reward_pips=round(tp_distance / pip_value, 1), pip_value=pip_value, ) def _generate_xau_atr_trade_setup( self, report: SignalReport, atr: float ) -> Optional[TradeSetup]: """XAU-aware exits for generic signals: wide SL, tight TP via ATR.""" cfg = self.config.xau_strategy price = report.current_price pip_value = 0.10 if report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): is_buy = True elif report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): is_buy = False else: is_buy = report.final_score >= 0 if is_buy: sl = price - cfg.sl_atr_multiplier * atr tp1 = price + cfg.tp_atr_multiplier * atr tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr direction = "BUY" else: sl = price + cfg.sl_atr_multiplier * atr tp1 = price - cfg.tp_atr_multiplier * atr tp2 = price - cfg.tp_atr_multiplier * 2.0 * atr direction = "SELL" sl_distance = abs(price - sl) tp_distance = abs(tp1 - price) rr = tp_distance / sl_distance if sl_distance > 0 else 0 return TradeSetup( pair=report.symbol, display_name=report.display_name, direction=direction, entry_price=price, stop_loss=round(sl, 2), take_profit_1=round(tp1, 2), take_profit_2=round(tp2, 2), risk_reward=round(rr, 2), risk_pips=round(sl_distance / pip_value, 1), reward_pips=round(tp_distance / pip_value, 1), pip_value=pip_value, ) def _check_eur_pullback_buy( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> bool: """Return True when EUR/USD pullback-buy conditions are met. Conditions (all must be true): 0. Bar's weekday is in the universal allowed_weekdays list 1. Current bar hour not in blocked list 2. M15 RSI(14) in 30–55 (pullback, not overbought) 3. M15 BB position < 0.75 (not at resistance) 4. M15 price > EMA200 (uptrend) 5. H1 price > EMA200 (higher-TF confirmation) 6. Price within 1.5×ATR of EMA21 (not chasing) """ cfg = self.config.eur_strategy try: bar_ts = pd.Timestamp(report.timestamp) bar_hour = bar_ts.hour bar_wd = bar_ts.weekday() if bar_hour in cfg.blocked_hours_utc: return False # Universal day filter — must be respected here because the # strategy hook overwrites the section-6b filter's quality reset. allowed_wd = self.config.backtest.allowed_weekdays if allowed_wd and bar_wd not in allowed_wd: return False except Exception: return False m15_analysis = None h1_analysis = None if report.primary_analysis: tf = report.primary_analysis.timeframe if tf == Timeframe.M15.value: m15_analysis = report.primary_analysis elif tf == Timeframe.H1.value: h1_analysis = report.primary_analysis for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca elif ca.timeframe == Timeframe.H1.value: h1_analysis = ca if not m15_analysis or not h1_analysis: return False m15_rsi = m15_analysis.indicators.get("rsi") if not m15_rsi or m15_rsi.value < cfg.rsi_lower or m15_rsi.value > cfg.rsi_upper: return False m15_bb = m15_analysis.indicators.get("bollinger") if m15_bb and m15_bb.value > cfg.bb_max_position: return False m15_ema200 = m15_analysis.key_levels.get("ema_trend") m15_df = dfs.get(Timeframe.M15) if m15_df is None or m15_ema200 is None: return False m15_price = float(m15_df["Close"].iloc[-1]) if m15_price <= m15_ema200: return False h1_ema200 = h1_analysis.key_levels.get("ema_trend") h1_df = dfs.get(Timeframe.H1) if h1_ema200 is None: return False h1_price = float(h1_df["Close"].iloc[-1]) if h1_df is not None else report.current_price if h1_price <= h1_ema200: return False m15_atr = m15_analysis.key_levels.get("atr", 0) m15_ema21 = m15_analysis.key_levels.get("ema_medium") if m15_atr > 0 and m15_ema21: dist = abs(m15_price - m15_ema21) / m15_atr if dist > cfg.chase_max_atr_from_ema21: return False # Minimum indicator confluence — rejects gate-passes with weak # technical agreement. The pullback gate alone fires Quality=A # even when M15 indicators are mostly neutral; this enforces # actual confluence. if cfg.min_indicators_agree > 0: agreeing = sum( 1 for name in ("ema_trend", "rsi", "bollinger", "adx", "candlestick") if (ind := m15_analysis.indicators.get(name)) is not None and ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) ) if agreeing < cfg.min_indicators_agree: return False return True def _generate_eur_trade_setup( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> Optional[TradeSetup]: """Generate EUR/USD-specific trade setup with tight TP / wide SL.""" cfg = self.config.eur_strategy m15_analysis = None if report.primary_analysis and report.primary_analysis.timeframe == Timeframe.M15.value: m15_analysis = report.primary_analysis else: for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca break if not m15_analysis: return None atr = m15_analysis.key_levels.get("atr", 0) if atr <= 0: return None price = report.current_price pip_value = 0.0001 tp1 = price + cfg.tp_atr_multiplier * atr sl = price - cfg.sl_atr_multiplier * atr tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr sl_distance = price - sl tp_distance = tp1 - price rr = tp_distance / sl_distance if sl_distance > 0 else 0 return TradeSetup( pair=report.symbol, display_name=report.display_name, direction="BUY", entry_price=price, stop_loss=round(sl, 5), take_profit_1=round(tp1, 5), take_profit_2=round(tp2 , 5), risk_reward=round(rr, 2), risk_pips=round(sl_distance / pip_value, 1), reward_pips=round(tp_distance / pip_value, 1), pip_value=pip_value, ) def _check_gbp_pullback_buy( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> bool: """GBP/USD pullback-buy — same logic as EUR, different config.""" cfg = self.config.gbp_strategy try: bar_ts = pd.Timestamp(report.timestamp) bar_hour = bar_ts.hour bar_wd = bar_ts.weekday() if bar_hour in cfg.blocked_hours_utc: return False allowed_wd = self.config.backtest.allowed_weekdays if allowed_wd and bar_wd not in allowed_wd: return False except Exception: return False m15_analysis = None h1_analysis = None if report.primary_analysis: tf = report.primary_analysis.timeframe if tf == Timeframe.M15.value: m15_analysis = report.primary_analysis elif tf == Timeframe.H1.value: h1_analysis = report.primary_analysis for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca elif ca.timeframe == Timeframe.H1.value: h1_analysis = ca if not m15_analysis or not h1_analysis: return False m15_rsi = m15_analysis.indicators.get("rsi") if not m15_rsi or m15_rsi.value < cfg.rsi_lower or m15_rsi.value > cfg.rsi_upper: return False m15_bb = m15_analysis.indicators.get("bollinger") if m15_bb and (m15_bb.value < cfg.bb_min_position or m15_bb.value > cfg.bb_max_position): return False m15_ema200 = m15_analysis.key_levels.get("ema_trend") m15_df = dfs.get(Timeframe.M15) if m15_df is None or m15_ema200 is None: return False m15_price = float(m15_df["Close"].iloc[-1]) if m15_price <= m15_ema200: return False h1_ema200 = h1_analysis.key_levels.get("ema_trend") h1_df = dfs.get(Timeframe.H1) if h1_ema200 is None: return False h1_price = float(h1_df["Close"].iloc[-1]) if h1_df is not None else report.current_price if h1_price <= h1_ema200: return False m15_atr = m15_analysis.key_levels.get("atr", 0) m15_ema21 = m15_analysis.key_levels.get("ema_medium") if m15_atr > 0 and m15_ema21: dist = abs(m15_price - m15_ema21) / m15_atr if dist > cfg.chase_max_atr_from_ema21: return False # See EUR equivalent — enforce minimum indicator confluence. if cfg.min_indicators_agree > 0: agreeing = sum( 1 for name in ("ema_trend", "rsi", "bollinger", "adx", "candlestick") if (ind := m15_analysis.indicators.get(name)) is not None and ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) ) if agreeing < cfg.min_indicators_agree: return False return True def _generate_gbp_trade_setup( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> Optional[TradeSetup]: """Generate GBP/USD trade setup with tight TP / balanced SL.""" cfg = self.config.gbp_strategy m15_analysis = None if report.primary_analysis and report.primary_analysis.timeframe == Timeframe.M15.value: m15_analysis = report.primary_analysis else: for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca break if not m15_analysis: return None atr = m15_analysis.key_levels.get("atr", 0) if atr <= 0: return None price = report.current_price pip_value = 0.0001 tp1 = price + cfg.tp_atr_multiplier * atr sl = price - cfg.sl_atr_multiplier * atr tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr sl_distance = price - sl tp_distance = tp1 - price rr = tp_distance / sl_distance if sl_distance > 0 else 0 return TradeSetup( pair=report.symbol, display_name=report.display_name, direction="BUY", entry_price=price, stop_loss=round(sl, 5), take_profit_1=round(tp1, 5), take_profit_2=round(tp2, 5), risk_reward=round(rr, 2), risk_pips=round(sl_distance / pip_value, 1), reward_pips=round(tp_distance / pip_value, 1), pip_value=pip_value, ) 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 if report.mtf_agreement: confidence += 10 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 -= 15 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") if abs(report.news_sentiment) > 0.2: if (report.news_sentiment > 0 and score > 0) or \ (report.news_sentiment < 0 and score < 0): confidence += 5 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 -= 10 if report.event_risk == "danger": report.warnings.append(f"EVENT RISK: {report.event_description}") confidence = min(confidence, 30) report.action_recommendation = f"WAIT — {report.event_description}" elif report.event_risk == "caution": report.warnings.append(f"Upcoming event: {report.event_description}") confidence -= 15 is_counter_trend = ( (report.market_regime == "trending_up" and score < 0) or (report.market_regime == "trending_down" and score > 0) ) if is_counter_trend: confidence -= 15 report.warnings.append("Counter-trend signal — reduced confidence") if report.volatility_regime == "extreme": report.warnings.append("EXTREME volatility — wider stops needed, reduced position size") confidence -= 10 elif report.volatility_regime == "low": report.warnings.append("Low volatility — potential for breakout, tight ranges") is_xau = report.display_name == "XAU/USD" primary = report.primary_analysis if primary is not None: ema200 = primary.key_levels.get("ema_trend") if ema200 and ema200 > 0 and report.current_price > 0: ema_dist_pct = abs(report.current_price - ema200) / ema200 * 100.0 _lo = float(os.environ.get("FB2_EMA_LO", "0.5")) _hi = float(os.environ.get("FB2_EMA_HI", "1.5")) if ema_dist_pct < _lo: report.warnings.append( f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk" ) confidence -= 3 if is_xau else 10 elif ema_dist_pct > (2.5 if is_xau else _hi): report.warnings.append( f"FAR FROM EMA200 ({ema_dist_pct:.2f}%) — over-extended" ) confidence -= 5 if is_xau else 10 atr = primary.key_levels.get("atr", 0) if atr > 0 and score != 0: is_buy_signal = score > 0 if is_buy_signal: swing_lows = sorted( v for k, v in primary.key_levels.items() if k.startswith("swing_low_") ) _exh_mult = 3.5 if is_xau else 2.5 if swing_lows and report.current_price - swing_lows[0] > _exh_mult * atr: confidence -= 5 if is_xau else 15 report.warnings.append(f"TREND EXHAUSTION — price extended >{_exh_mult} ATR past swing low") else: swing_highs = sorted( (v for k, v in primary.key_levels.items() if k.startswith("swing_high_")), reverse=True, ) _exh_mult_s = 3.5 if is_xau else 2.5 if swing_highs and swing_highs[0] - report.current_price > _exh_mult_s * atr: confidence -= 5 if is_xau else 15 report.warnings.append(f"TREND EXHAUSTION — price extended >{_exh_mult_s} ATR past swing high") ema_medium = primary.key_levels.get("ema_medium") if atr > 0 and ema_medium and report.market_regime.startswith("trending"): pullback_dist = abs(report.current_price - ema_medium) / atr _pb_severe = 2.5 if is_xau else 2.0 _pb_mild = 2.0 if is_xau else 1.5 if pullback_dist > _pb_severe: confidence -= 8 if is_xau else 20 report.warnings.append( f"NO PULLBACK — price {pullback_dist:.1f} ATR from EMA21, chasing risk" ) elif pullback_dist > _pb_mild: confidence -= 5 if is_xau else 10 report.warnings.append( f"Weak pullback — price {pullback_dist:.1f} ATR from EMA21" ) if report.market_regime.startswith("trending"): rsi_ind = primary.indicators.get("rsi") bb_ind = primary.indicators.get("bollinger") if rsi_ind: rsi_val = rsi_ind.value _rsi_pen = 5 if is_xau else 15 if report.market_regime in ("trending_up", "trending") and score > 0: if rsi_val > 65: confidence -= _rsi_pen report.warnings.append( f"Extended entry — RSI {rsi_val:.0f} in uptrend (prefer pullback to 40-55)" ) elif 38 <= rsi_val <= 55: confidence += 5 elif report.market_regime == "trending_down" and score < 0: if rsi_val < 35: confidence -= _rsi_pen report.warnings.append( f"Extended entry — RSI {rsi_val:.0f} in downtrend (prefer bounce to 45-62)" ) elif 45 <= rsi_val <= 62: confidence += 5 if bb_ind: bb_pos = bb_ind.value if report.market_regime in ("trending_up", "trending") and score > 0: if bb_pos > 0.85: confidence -= 10 elif bb_pos < 0.4: confidence += 5 elif report.market_regime == "trending_down" and score < 0: if bb_pos < 0.15: confidence -= 10 elif bb_pos > 0.6: confidence += 5 confidence = max(0.0, min(95.0, confidence)) # Final signal if not is_xau and score > 42: confidence -= 10 report.warnings.append("EXHAUSTION: score > 42 — trend overextended") if score >= 35 and confidence >= 55: report.final_signal = SignalStrength.STRONG_BUY elif score >= 25 and confidence >= 50: report.final_signal = SignalStrength.BUY elif score <= -35 and confidence >= 55: report.final_signal = SignalStrength.STRONG_SELL elif score <= -25 and confidence >= 50: report.final_signal = SignalStrength.SELL else: report.final_signal = SignalStrength.NEUTRAL report.final_score = score report.final_confidence = confidence # SELL suppression (pair-aware) sell_suppressed = False if self.config.backtest.sell_filter and report.final_signal in ( SignalStrength.SELL, SignalStrength.STRONG_SELL ): if report.display_name not in self.config.backtest.sell_allowed_pairs: report.final_signal = SignalStrength.NEUTRAL report.warnings.append(f"SELL SUPPRESSED for {report.display_name}") sell_suppressed = True # Quality grade abs_score = abs(score) if sell_suppressed: report.signal_quality = "F" elif abs_score >= 40 and confidence >= 65 and report.mtf_agreement: report.signal_quality = "A+" elif abs_score >= 33 and confidence >= 55 and report.mtf_agreement: report.signal_quality = "A" elif abs_score >= 28 and confidence >= 50: report.signal_quality = "B" elif abs_score >= 25 and confidence >= 45: report.signal_quality = "C" elif abs_score >= 15: 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 # Fix 2: Minimum ATR threshold for XAU/USD if report.display_name == "XAU/USD" and atr < self.config.xau_strategy.min_atr_threshold: report.warnings.append( f"LOW ATR FILTER: ATR ${atr:.2f} < ${self.config.xau_strategy.min_atr_threshold:.2f} minimum" ) return None min_sl_pips = 8.0 if pip_value < 0.01 else 80.0 min_sl_price = min_sl_pips * pip_value if report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): is_buy = True elif report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): is_buy = False else: is_buy = report.final_score >= 0 levels = report.primary_analysis.key_levels min_rr = self.config.signals.min_rr_ratio mature_move = abs(report.final_score) > 30 tp_mult = 0.9 if mature_move else 1.2 tp_swing_cap = 1.5 if mature_move else 2.0 # Fix 3: Wider SL multiplier for XAU/USD (1.8x vs default 1.5x) sl_atr_floor = (self.config.xau_strategy.v2_sl_atr_multiplier if report.display_name == "XAU/USD" else 1.5) if is_buy: swing_lows = sorted( [v for k, v in levels.items() if k.startswith("swing_low_") and v < price], reverse=True, ) if swing_lows: sl_price = swing_lows[0] - atr * 1.0 else: sl_price = price - atr * 2.0 min_sl = max(atr * sl_atr_floor, min_sl_price) if price - sl_price < min_sl: sl_price = price - min_sl sl_distance = price - sl_price swing_highs = sorted( v for k, v in levels.items() if k.startswith("swing_high_") and v > price + atr * 0.3 ) if (swing_highs and swing_highs[0] - price <= atr * tp_swing_cap and swing_highs[0] - price >= sl_distance * min_rr): tp1 = swing_highs[0] else: tp1 = price + max(atr * tp_mult, sl_distance * min_rr) if len(swing_highs) >= 2 and swing_highs[1] - price <= atr * 5.0: tp2 = swing_highs[1] else: tp2 = price + max(atr * (tp_mult + 1.5), sl_distance * min_rr * 1.5) direction = "BUY" entry, sl = price, sl_price else: swing_highs = sorted( v for k, v in levels.items() if k.startswith("swing_high_") and v > price ) if swing_highs: sl_price = swing_highs[0] + atr * 1.0 else: sl_price = price + atr * 2.0 min_sl = max(atr * sl_atr_floor, min_sl_price) if sl_price - price < min_sl: sl_price = price + min_sl sl_distance = sl_price - price swing_lows = sorted( [v for k, v in levels.items() if k.startswith("swing_low_") and v < price - atr * 0.3], reverse=True, ) if (swing_lows and price - swing_lows[0] <= atr * tp_swing_cap and price - swing_lows[0] >= sl_distance * min_rr): tp1 = swing_lows[0] else: tp1 = price - max(atr * tp_mult, sl_distance * min_rr) if len(swing_lows) >= 2 and price - swing_lows[1] <= atr * 5.0: tp2 = swing_lows[1] else: tp2 = price - max(atr * (tp_mult + 1.5), sl_distance * min_rr * 1.5) direction = "SELL" entry, sl = price, sl_price tp1_distance = abs(tp1 - entry) rr = tp1_distance / sl_distance if sl_distance > 0 else 0 if rr < min_rr - 1e-9: return None 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=round(rr, 2), 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}")