""" 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 # Entry archetype ("GENERIC", "XAU_PB") — selects trail params. strategy: str = "GENERIC" @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.""" # Diagnostic funnel for the M5 pullback route (shared across # instances; read by backtest tooling, harmless in live use). pb_funnel: Dict[str, int] = {} 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 if pair_info["display"] == "EUR/USD" and self.config.eur_strategy.enabled: eur_cfg = self.config.eur_strategy if eur_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: EUR/USD market closed — no liquidity") report.action_recommendation = "NO TRADE — Weekend (market closed)" return report if eur_cfg.allowed_weekdays and now_utc.weekday() not in eur_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 eur_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) # Session window + blocked hours on the signal bar's CLOSE time — # the exact predicate the backtester's replay loop applies before # each bar, so live and backtest cannot diverge. Bar-close time # (not wall clock) makes manual mid-candle scans behave like # boundary auto-scans. This replaces an older wall-clock filter # that was guarded by a marker string no code wrote. if bt.session_filter and report.primary_analysis is not None: try: bar_close = ( pd.Timestamp(report.timestamp) + pd.Timedelta( seconds=ForexDataProvider.CANDLE_SECONDS[ self.config.primary_timeframe ] ) ) close_h = bar_close.hour _is_xau = pair_info["display"] == "XAU/USD" _blocked = (self.config.xau_strategy.blocked_hours_utc if _is_xau else bt.blocked_hours_utc) outside = not (bt.session_start_hour <= close_h < bt.session_end_hour) if outside or close_h in _blocked: report.warnings.append( f"SESSION FILTER: bar closes {close_h:02}:00 UTC — " + ("outside window" if outside else "blocked hour") ) report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.action_recommendation = "NO TRADE — Outside tradeable session" report.trade_setup = None except Exception: pass 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_raw = high.diff() minus_dm_raw = -low.diff() plus_dm = plus_dm_raw.where((plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0), 0.0) minus_dm = minus_dm_raw.where((minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 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]) def _apply_m5_filters(self, report: SignalReport) -> None: """Extra selectivity for M5 — noisy TF needs stronger confirmation.""" if report.signal_quality not in ("A+", "A", "B", "C"): return xau_cfg = self.config.xau_strategy abs_score = abs(report.final_score) # Gate 1: minimum absolute score (M5 composite scores near the # threshold are noise — require a wider margin). if abs_score < xau_cfg.m5_min_abs_score: report.warnings.append( f"M5 FILTER: |score| {abs_score:.1f} < {xau_cfg.m5_min_abs_score:.0f} minimum" ) report.signal_quality = "D" report.trade_setup = None return # Gate 2: require MTF agreement (M5 alone is too noisy). if xau_cfg.m5_require_mtf and not report.mtf_agreement: report.warnings.append( "M5 FILTER: no MTF agreement — M5 signals require confirmation" ) report.signal_quality = "D" report.trade_setup = None return # Gate 3: at least one confirmation TF must have a meaningful # score in the same direction — NEUTRAL confirmations don't # count for M5 since they indicate the move isn't visible on # higher timeframes yet. min_conf_score = xau_cfg.m5_min_confirmation_score is_buy = report.final_score > 0 confirmed = False primary_tf = self.config.primary_timeframe.value for tf_val, tf_score in report.mtf_scores.items(): if tf_val == primary_tf: continue if is_buy and tf_score >= min_conf_score: confirmed = True break if not is_buy and tf_score <= -min_conf_score: confirmed = True break if not confirmed: report.warnings.append( "M5 FILTER: no confirmation TF shows directional conviction" ) report.signal_quality = "D" report.trade_setup = None return # Gate 4: cap SELL quality on M5 — A-grade SELLs enter at the # extreme of temporary dips in gold's macro uptrend and lose # money. B-grade SELLs (weaker conviction, shallower entries) # are marginally profitable. is_sell = report.final_score < 0 if is_sell and report.signal_quality not in xau_cfg.m5_sell_max_quality: report.warnings.append( f"M5 FILTER: SELL quality {report.signal_quality} capped — " f"high-conviction M5 shorts enter at extremes" ) report.signal_quality = "D" report.trade_setup = None return # Gate 5: block chop-transition volatility. 'high' vol on M5 is # elevated noise without the directional expansion of 'extreme' # — stops get swept before targets (61% WR, negative total in # every backtest window; low/normal/extreme all positive). if report.volatility_regime in xau_cfg.m5_blocked_vol_regimes: report.warnings.append( f"M5 FILTER: '{report.volatility_regime}' volatility — " f"chop-transition regime, stops get swept" ) report.signal_quality = "D" report.trade_setup = None return # Gate 6: SELL RSI floor — shorting with RSI already < 25 means # entering at the extreme of the dip (same pathology as Gate 4, # visible through RSI instead of quality grade). if is_sell: rsi_ind = report.primary_analysis.indicators.get("rsi") if rsi_ind is not None and rsi_ind.value < xau_cfg.m5_sell_min_rsi: report.warnings.append( f"M5 FILTER: SELL blocked — RSI {rsi_ind.value:.0f} < " f"{xau_cfg.m5_sell_min_rsi:.0f}, dip already extended" ) report.signal_quality = "D" report.trade_setup = None return # ── M5 pullback entry (XAU_PB) ────────────────────────────────── def _detect_m5_pullback( self, report: SignalReport, primary_df: pd.DataFrame ) -> Optional[Dict]: """Detect a pullback-continuation candidate on the primary TF. Trend intact (EMA alignment >= 3/4, ADX floor) + price retraced to EMA21/KC-mid within the last few bars + resumption bar closed back in trend direction + RSI in the pullback zone (not the extended zone breakout entries fire in). """ cfg = self.config.xau_strategy pa = report.primary_analysis if pa is None or len(primary_df) < 60: return None levels = pa.key_levels atr = levels.get("atr", 0.0) ema_f = levels.get("ema_fast") ema_m = levels.get("ema_medium") ema_s = levels.get("ema_slow") ema_t = levels.get("ema_trend") if not atr or None in (ema_f, ema_m, ema_s, ema_t): return None adx_ind = pa.indicators.get("adx") rsi_ind = pa.indicators.get("rsi") if adx_ind is None or rsi_ind is None: return None if adx_ind.value < cfg.m5_pullback_adx_min: return None close = float(primary_df["Close"].iloc[-1]) open_ = float(primary_df["Open"].iloc[-1]) n_touch = cfg.m5_pullback_touch_bars recent_low = float(primary_df["Low"].iloc[-n_touch:].min()) recent_high = float(primary_df["High"].iloc[-n_touch:].max()) # Pullback swing anchor for SL: extreme of the last 8 bars. pb_low = float(primary_df["Low"].iloc[-8:].min()) pb_high = float(primary_df["High"].iloc[-8:].max()) kc_mid = levels.get("kc_mid", ema_m) lo_atr, hi_atr = cfg.m5_pullback_retrace_atr rsi = rsi_ind.value # BUY: uptrend structure, dip to the band, bullish resumption. up_align = sum([ema_f > ema_m, ema_m > ema_s, ema_s > ema_t, close > ema_t]) if up_align >= 3: touched = recent_low <= max(ema_m, kc_mid) swing_high = float(primary_df["High"].iloc[-40:].max()) depth = (swing_high - pb_low) / atr resumed = close > open_ and close > ema_f lo_z, hi_z = cfg.m5_pullback_rsi_buy if (touched and resumed and lo_atr <= depth <= hi_atr and lo_z <= rsi <= hi_z): return {"direction": "BUY", "anchor": pb_low, "retrace_atr": depth} # SELL: mirrored. dn_align = sum([ema_f < ema_m, ema_m < ema_s, ema_s < ema_t, close < ema_t]) if dn_align >= 3: touched = recent_high >= min(ema_m, kc_mid) swing_low = float(primary_df["Low"].iloc[-40:].min()) depth = (pb_high - swing_low) / atr resumed = close < open_ and close < ema_f lo_z, hi_z = cfg.m5_pullback_rsi_sell if (touched and resumed and lo_atr <= depth <= hi_atr and lo_z <= rsi <= hi_z): return {"direction": "SELL", "anchor": pb_high, "retrace_atr": depth} return None def _try_m5_pullback( self, report: SignalReport, primary_df: pd.DataFrame, pip_value: float ) -> None: """Build an XAU_PB trade setup if a pullback candidate passes gates.""" cfg = self.config.xau_strategy # Rejection funnel counters (class-level diagnostic, cheap). funnel = SignalAggregator.pb_funnel def _reject(reason: str) -> None: funnel[reason] = funnel.get(reason, 0) + 1 pb = self._detect_m5_pullback(report, primary_df) if pb is None: return _reject("detected") direction = pb["direction"] is_buy = direction == "BUY" # Directional composite floor (lower than Gate 1 by design — # price mid-pullback scores weaker on the momentum indicators). signed = report.final_score if is_buy else -report.final_score if signed < cfg.m5_pullback_min_score: _reject("score") return # Confirmation TF must agree (reuse Gate-3 threshold). min_conf = cfg.m5_min_confirmation_score confirmed = False primary_tf = self.config.primary_timeframe.value for tf_val, tf_score in report.mtf_scores.items(): if tf_val == primary_tf: continue if (is_buy and tf_score >= min_conf) or \ (not is_buy and tf_score <= -min_conf): confirmed = True break if not confirmed: _reject("confirmation_tf") return # Never against multi-TF consensus (mirror of gate 6f). sigs = list(report.mtf_signals.values()) bull_n = sum(1 for s in sigs if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY)) bear_n = sum(1 for s in sigs if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL)) thr = self.config.signals.mtf_agreement_threshold if (is_buy and bear_n >= thr) or (not is_buy and bull_n >= thr): _reject("mtf_opposition") return # Volatility gates match the breakout path. if report.volatility_regime in cfg.m5_blocked_vol_regimes: _reject("vol_regime") return if (not is_buy and report.volatility_regime and report.volatility_regime not in cfg.sell_vol_regimes): _reject("sell_vol") return atr = report.primary_analysis.key_levels.get("atr", 0.0) entry = report.current_price min_sl_price = self.config.signals.min_sl_pips * pip_value if is_buy: sl_dist = max(entry - (pb["anchor"] - 0.5 * atr), min_sl_price) sl = entry - sl_dist tp1 = entry + 1.5 * sl_dist tp2 = entry + 2.5 * sl_dist else: sl_dist = max((pb["anchor"] + 0.5 * atr) - entry, min_sl_price) sl = entry + sl_dist tp1 = entry - 1.5 * sl_dist tp2 = entry - 2.5 * sl_dist # Spread gate — same economics as the breakout generators. min_tp = (self.config.signals.round_trip_cost * self.config.signals.min_tp_cost_multiple) if 1.5 * sl_dist < min_tp: return report.trade_setup = 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=1.5, risk_pips=sl_dist / pip_value, reward_pips=1.5 * sl_dist / pip_value, pip_value=pip_value, strategy="XAU_PB", ) if report.signal_quality not in ("A+", "A", "B"): report.signal_quality = "B" report.final_signal = (SignalStrength.BUY if is_buy else SignalStrength.SELL) report.action_recommendation = ( f"{direction} — pullback continuation entry" ) marker = (f"XAU PULLBACK STRATEGY: retrace {pb['retrace_atr']:.1f} " f"ATR to EMA21/KC-mid, trend resumed") target = report.bullish_reasons if is_buy else report.bearish_reasons target.insert(0, marker) # ── 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. Exhausted trend quality cap ─────────────────────── if (report.market_regime.startswith("trending") and report.exhaustion_count >= 2 and report.signal_quality in ("C", "D")): report.warnings.append( "EXHAUSTED TREND FILTER: C-grade blocked in exhausted trend" ) report.signal_quality = "D" report.trade_setup = None # ── 6e. 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) ) # Actionable grades only: without this guard the gate would "upgrade" # a session-blocked F report to D, which re-opens the M5 pullback # path (its entry check is quality != "F"). if (self.config.xau_strategy.enabled and report.display_name == "XAU/USD" and xau_is_sell_dir and report.signal_quality in ("A+", "A", "B", "C") 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 # ── 6f. MTF opposition gate ─────────────────────────────── # Never trade against a multi-timeframe consensus: if the # threshold number of TFs point opposite to the trade direction, # block the setup (counter-trend entries in gold bleed). if report.signal_quality in ("A+", "A", "B", "C") and report.mtf_signals: sigs = list(report.mtf_signals.values()) bull_n = sum(1 for s in sigs if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY)) bear_n = sum(1 for s in sigs if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL)) thr = self.config.signals.mtf_agreement_threshold opposed = (bear_n >= thr and report.final_score > 0) or \ (bull_n >= thr and report.final_score < 0) if opposed: report.warnings.append( "MTF OPPOSITION: majority of timeframes point against signal — no trade" ) report.signal_quality = "D" report.trade_setup = None # ── 6g. SELL volatility gate ────────────────────────────── # Shorts in quiet markets fade against gold's macro uptrend; # only take sells when volatility regime confirms real selling. if (self.config.xau_strategy.enabled and report.display_name == "XAU/USD" and xau_is_sell_dir and report.signal_quality in ("A+", "A", "B", "C") and report.volatility_regime and report.volatility_regime not in self.config.xau_strategy.sell_vol_regimes): report.warnings.append( f"SELL FILTER: volatility '{report.volatility_regime}' too quiet for shorts" ) report.signal_quality = "D" report.trade_setup = None # ── 6g2. XAU regime gates (opt-in via per-TF factory overrides) ── if (self.config.xau_strategy.enabled and report.display_name == "XAU/USD" and report.signal_quality in ("A+", "A", "B", "C")): if report.market_regime in self.config.xau_strategy.blocked_regimes: report.warnings.append( f"REGIME FILTER: '{report.market_regime}' blocked for XAU" ) report.signal_quality = "D" report.trade_setup = None elif (xau_is_sell_dir and report.market_regime in self.config.xau_strategy.sell_blocked_regimes): report.warnings.append( f"SELL FILTER: regime '{report.market_regime}' blocked for shorts" ) report.signal_quality = "D" report.trade_setup = None elif xau_is_sell_dir and self.config.xau_strategy.sell_blocked_hours_utc: # ── 6g3. XAU SELL hour gate (opt-in via per-TF factories) ── # Evaluated on the bar's CLOSE time (= entry time under # next-open fills), the same clock the session filter and # backtest replay use — NOT the bar label hour. try: close_h = ( pd.Timestamp(report.timestamp) + pd.Timedelta( seconds=ForexDataProvider.CANDLE_SECONDS[ self.config.primary_timeframe ] ) ).hour except Exception: close_h = None if close_h in self.config.xau_strategy.sell_blocked_hours_utc: report.warnings.append( f"SELL FILTER: {close_h:02}:00 UTC entry blocked for shorts" ) report.signal_quality = "D" report.trade_setup = None # ── 6h. EUR/USD regime + quality + risk gate ────────────── if report.display_name == "EUR/USD" and self.config.eur_strategy.enabled: eur = self.config.eur_strategy if eur.blocked_regimes and report.market_regime in eur.blocked_regimes: report.warnings.append( f"EUR REGIME FILTER: {report.market_regime} blocked" ) report.signal_quality = "F" if (eur.min_quality_grades and report.signal_quality not in eur.min_quality_grades and report.signal_quality != "F"): report.warnings.append( f"EUR QUALITY FILTER: {report.signal_quality} below minimum" ) report.signal_quality = "F" if (report.signal_quality != "F" and getattr(report, 'exhaustion_count', 0) > eur.max_exhaustion_tags): report.warnings.append( f"EUR EXHAUSTION FILTER: {report.exhaustion_count} tags > max {eur.max_exhaustion_tags}" ) report.signal_quality = "F" if (report.signal_quality != "F" and eur.block_near_ema200 and any(w.startswith("NEAR EMA200") for w in report.warnings)): report.warnings.append("EUR EMA200 FILTER: too close to EMA200") report.signal_quality = "F" if eur.session_hours and report.signal_quality != "F": try: bar_hour = pd.Timestamp(report.timestamp).hour if bar_hour not in eur.session_hours: report.warnings.append( f"EUR SESSION FILTER: {bar_hour:02}:00 UTC outside allowed hours" ) report.signal_quality = "F" except Exception: pass if report.signal_quality != "F": abs_s = abs(report.final_score) if abs_s < eur.min_entry_score: report.warnings.append( f"EUR SCORE FILTER: |{abs_s:.1f}| < {eur.min_entry_score}" ) report.signal_quality = "F" elif report.final_confidence < eur.min_entry_confidence: report.warnings.append( f"EUR CONF FILTER: {report.final_confidence} < {eur.min_entry_confidence}" ) report.signal_quality = "F" if (report.signal_quality != "F" and eur.blocked_vol_regimes and report.volatility_regime): if report.volatility_regime in eur.blocked_vol_regimes: report.warnings.append( f"EUR VOL FILTER: {report.volatility_regime} volatility blocked" ) report.signal_quality = "F" if (report.signal_quality != "F" and eur.min_indicator_agreement > 0 and report.primary_analysis): agree_count = sum( 1 for ind in report.primary_analysis.indicators.values() if ind.signal.value != 0 and (ind.signal.value > 0) == (report.final_score > 0) ) if agree_count < eur.min_indicator_agreement: report.warnings.append( f"EUR AGREE FILTER: {agree_count} < {eur.min_indicator_agreement}" ) report.signal_quality = "F" if (report.signal_quality != "F" and eur.block_rsi_extended and report.primary_analysis): rsi_ind = report.primary_analysis.indicators.get("rsi") if rsi_ind: rsi_v = rsi_ind.value is_buy = report.final_score > 0 if is_buy and rsi_v > 70: report.warnings.append( f"EUR RSI FILTER: RSI {rsi_v:.0f} overbought for BUY" ) report.signal_quality = "F" elif not is_buy and rsi_v < 30: report.warnings.append( f"EUR RSI FILTER: RSI {rsi_v:.0f} oversold for SELL" ) report.signal_quality = "F" # ── 6i. M5 noise gate (gold-tuned thresholds — XAU only) ── if (self.config.primary_timeframe == Timeframe.M5 and report.display_name == "XAU/USD"): self._apply_m5_filters(report) # ── 7. Trade setup for actionable signals ────────────────── # ATR generator is XAU-only: it hardcodes gold pip value and # 2-decimal price rounding. 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) # ── 8b. EUR strategy marker ──────────────────────────────── # The backtester tags trades from reasons[0]; mark EUR setups so # they report under the EUR strategy instead of GENERIC. if (report.display_name == "EUR/USD" and self.config.eur_strategy.enabled and report.trade_setup is not None): marker = (f"EUR STRATEGY: {report.market_regime} regime, " f"score {report.final_score:+.1f}, quality {report.signal_quality}") target = (report.bullish_reasons if report.trade_setup.direction == "BUY" else report.bearish_reasons) target.insert(0, marker) # ── 9. M5 pullback entry (XAU_PB) ────────────────────────── # Second archetype: when no breakout setup fired, look for a # trend-continuation pullback (retrace to EMA21/KC-mid). Runs # after reasons so its marker lands at reasons[0] for tagging. if (self.config.primary_timeframe == Timeframe.M5 and report.display_name == "XAU/USD" and self.config.xau_strategy.m5_pullback_enabled and report.trade_setup is None and report.signal_quality != "F"): self._try_m5_pullback(report, primary_df, pip_value) return report # ── Internals ────────────────────────────────────────────────── _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" tf = self.config.signals.tp_tightening_factor tp1 = price + (tp1 - price) * tf tp2 = price + (tp2 - price) * tf 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 # Spread gate (same rule as _generate_trade_setup). min_tp = self.config.signals.round_trip_cost * self.config.signals.min_tp_cost_multiple if tp_distance < min_tp: report.warnings.append( f"SPREAD GATE: target ${tp_distance:.2f} < ${min_tp:.2f} minimum — " f"costs would eat the edge" ) 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 # Agreement must point the same way as the primary timeframe — # two bullish confirmation TFs must not "confirm" a bearish primary. if report.primary_score > 0: return bullish >= threshold if report.primary_score < 0: return bearish >= 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") # Pullback bonuses require a KNOWN trend direction — the mixed # "trending" regime (EMAs unaligned) gets no directional bonus. if rsi_ind: rsi_val = rsi_ind.value if report.market_regime == "trending_up" 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 == "trending_up" 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) ─── # Warning always shown; the confidence cap is config-gated # (news.event_risk_caps_signals) — see NewsConfig for the audit # showing the cap costs R on XAU. if report.event_risk == "danger": report.warnings.append(f"EVENT RISK: {report.event_description}") if self.config.news.event_risk_caps_signals: 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}") if self.config.news.event_risk_caps_signals: 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") # Extension penalties apply in ANY strong-ADX regime; the # mixed "trending" regime penalizes BOTH directions (it # previously penalized only BUYs — the SELL side of the # same bullish-bias bug fixed in the RSI reinterpretation). 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 in ("trending_down", "trending") 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 in ("trending_down", "trending") 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 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 if pip_value <= 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 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 geometry reads only EurUsdStrategyConfig — nothing # gold-calibrated from SignalConfig applies here. eur = self.config.eur_strategy min_sl_price = eur.min_sl_pips * pip_value min_rr = eur.min_rr_ratio # No TP-tightening interplay for EUR: the RR floor used for # target selection IS min_rr (tp_tightening_factor is 1.0; # revisit if EUR ever tunes it). pre_tight_min_rr = min_rr mature_move = abs(report.final_score) > eur.mature_move_threshold quality_scale = eur.tp_quality_scale.get(report.signal_quality, 1.0) tp_mult = (eur.tp_atr_multiplier if mature_move else eur.tp_atr_multiplier * 1.5) * quality_scale tp_swing_cap = (2.0 if mature_move else 2.5) * quality_scale sl_atr_floor = eur.sl_atr_floor sl_swing_buffer = eur.sl_atr_multiplier else: min_sl_price = self.config.signals.min_sl_pips * pip_value min_rr = self.config.signals.min_rr_ratio tp_floor = max(min_rr, self.config.signals.tp_floor_rr) tf = self.config.signals.tp_tightening_factor pre_tight_min_rr = tp_floor / tf if tf > 0 else tp_floor tp_mult = 1.5 tp_swing_cap = 2.5 sl_atr_floor = (self.config.xau_strategy.v2_sl_atr_multiplier if report.display_name == "XAU/USD" else 1.5) sl_swing_buffer = 1.0 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 * sl_swing_buffer else: sl_price = price - atr * (1.0 + sl_swing_buffer) 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 * pre_tight_min_rr): tp1 = swing_highs[0] else: tp1 = price + max(atr * tp_mult, sl_distance * pre_tight_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 * pre_tight_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 * sl_swing_buffer else: sl_price = price + atr * (1.0 + sl_swing_buffer) 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 * pre_tight_min_rr): tp1 = swing_lows[0] else: tp1 = price - max(atr * tp_mult, sl_distance * pre_tight_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 * pre_tight_min_rr * 1.5) direction = "SELL" entry, sl = price, sl_price tf = self.config.signals.tp_tightening_factor tp1 = entry + (tp1 - entry) * tf tp2 = entry + (tp2 - entry) * tf # TP2 must sit beyond TP1 — swing picks can invert when TP1 falls # back to the ATR target while TP2 takes a nearer second swing. if is_buy: tp2 = max(tp2, tp1 + atr) else: tp2 = min(tp2, tp1 - atr) tp1_distance = abs(tp1 - entry) rr = tp1_distance / sl_distance if sl_distance > 0 else 0 if rr < min_rr - 1e-9: return None # Spread gate: a target too close to entry hands most of the win # to transaction costs (validated on M5: sub-$10 targets lose). if _is_eur: min_tp = eur.round_trip_cost * eur.min_tp_cost_multiple if tp1_distance < min_tp: report.warnings.append( f"SPREAD GATE: target {tp1_distance / pip_value:.1f} pips " f"< {min_tp / pip_value:.1f} pip minimum — costs would eat the edge" ) return None else: min_tp = self.config.signals.round_trip_cost * self.config.signals.min_tp_cost_multiple if tp1_distance < min_tp: report.warnings.append( f"SPREAD GATE: target ${tp1_distance:.2f} < ${min_tp:.2f} minimum — " f"costs would eat the edge" ) 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}")