""" 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__) def _is_forex_weekend(now_utc: datetime) -> bool: """Forex market closed: Friday 22:00 UTC → Sunday 22:00 UTC (Sydney open).""" wd = now_utc.weekday() hour = now_utc.hour if wd == 4 and hour >= 22: # Friday after 22:00 return True if wd == 5: # All Saturday return True if wd == 6 and hour < 22: # Sunday before 22:00 return True return False @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 = "" exhaustion_count: int = 0 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 _is_forex_weekend(now_utc): 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", 5: "Sat", 6: "Sun"} 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 = ( self.config.xau_strategy.enabled and any("XAU STRATEGY" in r for r in report.bullish_reasons) ) if not xau_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. Transitioning regime quality boost ────────────────── if report.market_regime == "transitioning" and report.signal_quality in ("B", "C", "D"): abs_score = abs(report.final_score) if abs_score >= 30 and report.final_confidence >= 50 and report.mtf_agreement: report.signal_quality = "A" report.action_recommendation = f"{report.final_signal.name.replace('_', ' ')} — Trend emerging, early entry" elif abs_score >= 25 and report.final_confidence >= 45: report.signal_quality = "B" report.action_recommendation = f"{report.final_signal.name.replace('_', ' ')} — Trend forming, watch closely" # ── 6d. 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 (self.config.xau_strategy.enabled and report.display_name == "XAU/USD" and xau_is_sell_dir and report.signal_quality not in self.config.xau_strategy.sell_min_quality): report.warnings.append( f"SELL FILTER: quality {report.signal_quality} below minimum for SELL" ) report.signal_quality = "D" report.trade_setup = None # ── 7. Trade setup for actionable signals ────────────────── if report.trade_setup is None and report.signal_quality in ("A+", "A") and report.display_name == "XAU/USD": atr = (report.primary_analysis.key_levels.get("atr", 0) if report.primary_analysis else 0) if atr > 0: report.trade_setup = self._generate_xau_atr_trade_setup(report, atr) if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"): report.trade_setup = self._generate_trade_setup(report, pip_value) # ── 8. Reasons ───────────────────────────────────────────── self._collect_reasons(report) return report # ── Internals ────────────────────────────────────────────────── def _check_xau_oversold_bounce( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> bool: """Return True when XAU/USD oversold-bounce conditions are met.""" 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 quality_scale = self._TP_QUALITY_SCALE.get(report.signal_quality, 0.75) tp_mult = cfg.tp_atr_multiplier * quality_scale tp1 = price + tp_mult * atr sl = price - cfg.sl_atr_multiplier * atr tp2 = price + tp_mult * 1.5 * 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, ) _TP_QUALITY_SCALE = {"A+": 2.0, "A": 1.5, "B": 1.0, "C": 1.0} def _generate_xau_atr_trade_setup( self, report: SignalReport, atr: float ) -> Optional[TradeSetup]: """XAU ATR-based trade setup with quality-tiered TP scaling.""" 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 quality_scale = self._TP_QUALITY_SCALE.get(report.signal_quality, 0.75) if report.signal_quality == "A" and report.exhaustion_count >= 2: quality_scale = 1.0 tp_mult = cfg.tp_atr_multiplier * quality_scale if is_buy: sl = price - cfg.sl_atr_multiplier * atr tp1 = price + tp_mult * atr tp2 = price + tp_mult * 1.5 * atr direction = "BUY" else: sl = price + cfg.sl_atr_multiplier * atr tp1 = price - tp_mult * atr tp2 = price - tp_mult * 1.5 * atr direction = "SELL" sl_distance = abs(price - sl) tp_distance = abs(tp1 - price) rr = tp_distance / sl_distance if sl_distance > 0 else 0 min_rr = self.config.signals.min_rr_ratio if rr < min_rr - 1e-9: return None 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_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 # ── Bonuses (additive, applied first) ───────────────────── 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 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 primary = report.primary_analysis if primary and 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 if report.market_regime in ("trending_up", "trending") and score > 0 and 38 <= rsi_val <= 55: confidence += 5 elif report.market_regime == "trending_down" and score < 0 and 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 and bb_pos < 0.4: confidence += 5 elif report.market_regime == "trending_down" and score < 0 and bb_pos > 0.6: confidence += 5 # ── Event risk (hard cap, separate from penalty system) ─── 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}" # ── Penalties (diminishing cap system) ──────────────────── # Collect all applicable penalties, then apply with exponential # decay so stacking never guts a signal beyond 30%. penalties = [] if not report.mtf_agreement: penalties.append(("mtf_disagreement", 15)) if len(report.mtf_scores) >= 2: scores_list = list(report.mtf_scores.values()) if any(s > 10 for s in scores_list) and any(s < -10 for s in scores_list): report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes") if abs(report.news_sentiment) > 0.2: if (report.news_sentiment > 0 and score < -10) or \ (report.news_sentiment < 0 and score > 10): report.warnings.append("News sentiment CONFLICTS with technical signal") penalties.append(("news_conflict", 10)) if report.event_risk == "caution": report.warnings.append(f"Upcoming event: {report.event_description}") penalties.append(("event_caution", 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: report.warnings.append("Counter-trend signal — reduced confidence") penalties.append(("counter_trend", 15)) if report.volatility_regime == "extreme": report.warnings.append("EXTREME volatility — wider stops needed") penalties.append(("extreme_vol", 10)) elif report.volatility_regime == "low": report.warnings.append("Low volatility — potential for breakout") 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")) if ema_dist_pct < _lo: report.warnings.append(f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk") penalties.append(("ema200_near", 5)) elif ema_dist_pct > 2.5: report.warnings.append(f"FAR FROM EMA200 ({ema_dist_pct:.2f}%) — over-extended") penalties.append(("ema200_far", 5)) atr = primary.key_levels.get("atr", 0) if atr > 0 and score != 0: if score > 0: swing_lows = sorted( v for k, v in primary.key_levels.items() if k.startswith("swing_low_") ) if swing_lows and report.current_price - swing_lows[0] > 3.5 * atr: report.warnings.append("TREND EXHAUSTION — price extended >3.5 ATR past swing low") penalties.append(("exhaustion", 8)) else: swing_highs = sorted( (v for k, v in primary.key_levels.items() if k.startswith("swing_high_")), reverse=True, ) if swing_highs and swing_highs[0] - report.current_price > 3.5 * atr: report.warnings.append("TREND EXHAUSTION — price extended >3.5 ATR past swing high") penalties.append(("exhaustion", 8)) 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 if pullback_dist > 2.5: report.warnings.append(f"NO PULLBACK — price {pullback_dist:.1f} ATR from EMA21") penalties.append(("no_pullback", 8)) elif pullback_dist > 2.0: report.warnings.append(f"Weak pullback — price {pullback_dist:.1f} ATR from EMA21") penalties.append(("weak_pullback", 5)) 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 if report.market_regime in ("trending_up", "trending") and score > 0 and rsi_val > 65: report.warnings.append(f"Extended — RSI {rsi_val:.0f} in uptrend") penalties.append(("rsi_extended", 5)) elif report.market_regime == "trending_down" and score < 0 and rsi_val < 35: report.warnings.append(f"Extended — RSI {rsi_val:.0f} in downtrend") penalties.append(("rsi_extended", 5)) if bb_ind: bb_pos = bb_ind.value if report.market_regime in ("trending_up", "trending") and score > 0 and bb_pos > 0.85: penalties.append(("bb_overextended", 8)) elif report.market_regime == "trending_down" and score < 0 and bb_pos < 0.15: penalties.append(("bb_overextended", 8)) # Hard block: static exhaustion + fading momentum → kill the trade # Static signals alone are normal in gold trends. Only block when # price is extended AND momentum indicators confirm deceleration. _EXHAUSTION_TAGS = {"exhaustion", "no_pullback", "rsi_extended", "bb_overextended"} exhaustion_count = sum(1 for name, _ in penalties if name in _EXHAUSTION_TAGS) report.exhaustion_count = exhaustion_count momentum_fading = False if primary is not None and exhaustion_count >= 2: roc_ind = primary.indicators.get("roc") macd_ind = primary.indicators.get("macd") roc_fading = False macd_fading = False if roc_ind: if score > 0 and roc_ind.value < 0: roc_fading = True elif score < 0 and roc_ind.value > 0: roc_fading = True if macd_ind: if score > 0 and macd_ind.signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): macd_fading = True elif score < 0 and macd_ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): macd_fading = True momentum_fading = roc_fading or macd_fading exhaustion_blocked = exhaustion_count >= 2 and momentum_fading # Apply penalties with diminishing returns (cap at 30%) MAX_PENALTY = 30.0 penalties.sort(key=lambda x: x[1], reverse=True) total_penalty = 0.0 decay = 1.0 for _name, amount in penalties: applied = amount * decay if total_penalty + applied > MAX_PENALTY: applied = MAX_PENALTY - total_penalty total_penalty += applied decay *= 0.5 if total_penalty >= MAX_PENALTY: break confidence -= total_penalty confidence = max(0.0, min(95.0, confidence)) if exhaustion_blocked: report.warnings.append( f"EXHAUSTION BLOCK — {exhaustion_count} exhaustion signals fired, trade rejected" ) confidence = 0.0 # Final signal 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 # Quality grade abs_score = abs(score) if 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 _is_eur = report.display_name == "EUR/USD" and self.config.eur_strategy.enabled if _is_eur: eur = self.config.eur_strategy q_scale = eur.tp_quality_scale.get(report.signal_quality, 1.0) min_rr = eur.min_rr_ratio * q_scale mature_move = abs(report.final_score) > eur.mature_move_threshold base_tp = eur.tp_atr_multiplier if mature_move else eur.tp_atr_multiplier * 1.5 tp_mult = base_tp * q_scale tp_swing_cap = (2.0 if mature_move else 2.5) * q_scale sl_atr_floor = eur.sl_atr_floor else: min_rr = self.config.signals.min_rr_ratio mature_move = abs(report.final_score) > 30 tp_mult = 1.0 if mature_move else 1.5 tp_swing_cap = 2.0 if mature_move else 2.5 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}")