""" 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 chart_patterns import detect_chart_pattern from config import PlatformConfig, SignalStrength, Timeframe from providers import BaseDataProvider, make_data_provider 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" # Trailing exit params stamped from the PRODUCING config, so per-TF # factory overrides (e.g. M15 arm 0.70, M5 dist 0.85) reach every # consumer. None = legacy/unstamped setup — formatters fall back to # platform defaults; 0.0 means trailing is genuinely disabled. trail_arm_r: Optional[float] = None trail_dist_r: Optional[float] = None @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] = {} # Same, for the chart-pattern route (XAU_CP). cp_funnel: Dict[str, int] = {} def __init__(self, config: PlatformConfig): self.config = config self.data_provider = make_data_provider(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) # 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=BaseDataProvider.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: # Fail-open by design (an unparseable timestamp shouldn't # kill the scan) but never silently: this filter guards # real money hours. logger.exception( f"{symbol}: live session filter errored — filter SKIPPED this scan" ) 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: _reject("spread_gate") 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", trail_arm_r=self.config.signals.pb_trail_arm_r, trail_dist_r=self.config.signals.pb_trail_dist_r, ) _reject("accepted") # funnel tick, mirrors the CP path 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) # ── Chart-pattern entry (XAU_CP) ──────────────────────────────── def _try_chart_pattern( self, report: SignalReport, primary_df: pd.DataFrame, pip_value: float ) -> None: """Build an XAU_CP trade setup if a completed chart pattern passes the pair-level protections. Unlike the composite paths there is no minimum-score gate: a reversal pattern completes exactly where the composite is still pointing the old way. Protection comes from the pattern's own structural gates plus the validated directional filters below. """ cfg = self.config.xau_strategy funnel = SignalAggregator.cp_funnel def _reject(reason: str) -> None: funnel[reason] = funnel.get(reason, 0) + 1 pa = report.primary_analysis if pa is None: return atr = pa.key_levels.get("atr", 0.0) # Same $-ATR floor the swing generator applies (dead tape). if atr < cfg.min_atr_threshold: return pat = detect_chart_pattern(primary_df, atr, cfg) if pat is None: return _reject("detected") direction = pat["direction"] is_buy = direction == "BUY" # 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 # Directional protections — gold's macro-uptrend asymmetry is # structural, so the validated SELL gates apply to pattern # shorts too (quiet-vol shorts fade, blocked SELL hours lose). if not is_buy: if (report.volatility_regime and report.volatility_regime not in cfg.sell_vol_regimes): _reject("sell_vol") return if cfg.sell_blocked_hours_utc: try: close_h = ( pd.Timestamp(report.timestamp) + pd.Timedelta( seconds=BaseDataProvider.CANDLE_SECONDS[ self.config.primary_timeframe ] ) ).hour except Exception: close_h = None logger.warning( f"{report.symbol}: bar-close hour underivable — " f"SELL hour gate skipped" ) if close_h in cfg.sell_blocked_hours_utc: _reject("sell_hour") return if self.config.primary_timeframe == Timeframe.M5: rsi_ind = pa.indicators.get("rsi") if rsi_ind is not None and rsi_ind.value < cfg.m5_sell_min_rsi: _reject("m5_sell_rsi") return # M5 chop-transition volatility block (validated, both ways). if (self.config.primary_timeframe == Timeframe.M5 and report.volatility_regime in cfg.m5_blocked_vol_regimes): _reject("vol_regime") return entry = report.current_price min_sl_price = self.config.signals.min_sl_pips * pip_value height = pat["height"] if is_buy: sl_dist = max(entry - (pat["anchor"] - 0.5 * atr), min_sl_price) sl = entry - sl_dist tp1 = entry + height tp2 = entry + 1.5 * height else: sl_dist = max((pat["anchor"] + 0.5 * atr) - entry, min_sl_price) sl = entry + sl_dist tp1 = entry - height tp2 = entry - 1.5 * height # The measured move IS the pattern target — if it can't clear # the RR floor or the spread gate, the pattern is untradeable. rr = height / sl_dist if sl_dist > 0 else 0.0 if rr < self.config.signals.min_rr_ratio - 1e-9: _reject("rr") return min_tp = (self.config.signals.round_trip_cost * self.config.signals.min_tp_cost_multiple) if height < min_tp: _reject("spread_gate") 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=round(rr, 2), risk_pips=sl_dist / pip_value, reward_pips=height / pip_value, pip_value=pip_value, strategy="XAU_CP", trail_arm_r=self.config.signals.trail_arm_r, trail_dist_r=self.config.signals.trail_dist_r, ) 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} — {pat['pattern']} pattern completion" ) marker = (f"XAU CHART PATTERN: {pat['pattern']} " f"({direction}) break of {pat['trigger']:.2f}, " f"height {height / atr:.1f} ATR") target = report.bullish_reasons if is_buy else report.bearish_reasons target.insert(0, marker) _reject("accepted") # ── 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: # Fail-open by design, but never silently (see live wrapper). logger.exception( f"{symbol}: session/day filter errored — filter SKIPPED" ) # ── 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=BaseDataProvider.CANDLE_SECONDS[ self.config.primary_timeframe ] ) ).hour except Exception: close_h = None logger.warning( f"{report.symbol}: bar-close hour underivable — " f"SELL hour gate skipped" ) 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 # ── 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) # ── 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) # ── 10. Chart-pattern entry (XAU_CP) ──────────────────────── # Third archetype: completion of a classical chart pattern # (double top/bottom, H&S, flag — see chart_patterns.py). Only # fills empty slots (core + pullback setups take precedence). # Unlike XAU_PB this must NOT require quality != "F": a weak # composite (grade F on |score| < 15) is the normal state at a # reversal-pattern completion. Session/day blocks still bind — # they are detected by their warnings (and the live wrapper + # backtest replay apply the bar-close session predicate again # on top, so a pattern setup can never trade a blocked hour). if (report.display_name == "XAU/USD" and self.config.xau_strategy.cp_enabled and report.trade_setup is None and not any(w.startswith(("SESSION FILTER", "DAY FILTER", "WEEKEND FILTER")) for w in report.warnings)): self._try_chart_pattern(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, trail_arm_r=self.config.signals.trail_arm_r, trail_dist_r=self.config.signals.trail_dist_r, ) 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 self.config.news.sentiment_affects_signals and 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") if self.config.news.sentiment_affects_signals: 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: # Distance is measured from the FARTHEST of the 3 tracked # swings (sorted()[0] = deepest low / highest high), so the # penalty fires earlier than a nearest-swing rule would. # Calibrated as-is in every validated book — changing the # anchor is a signal-logic change requiring re-validation. 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 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). 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, trail_arm_r=self.config.signals.trail_arm_r, trail_dist_r=self.config.signals.trail_dist_r, ) 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}")