""" ForexBot2 — Signal Aggregator ============================== Combines multi-timeframe technical analysis with optional news context into a graded actionable signal. """ import logging import os from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Dict, List, Optional import pandas as pd from config import PlatformConfig, SignalStrength, Timeframe from data_provider import ForexDataProvider from engine import TechnicalEngine, TechnicalAnalysis from news_engine import NewsEngine, FundamentalContext logger = logging.getLogger(__name__) @dataclass class TradeSetup: pair: str display_name: str direction: str # "BUY" / "SELL" entry_price: float stop_loss: float take_profit_1: float take_profit_2: float risk_reward: float risk_pips: float reward_pips: float pip_value: float @dataclass class SignalReport: symbol: str display_name: str timestamp: str primary_signal: SignalStrength = SignalStrength.NEUTRAL primary_score: float = 0.0 primary_confidence: float = 0.0 mtf_agreement: bool = False mtf_signals: Dict[str, SignalStrength] = field(default_factory=dict) mtf_scores: Dict[str, float] = field(default_factory=dict) primary_analysis: Optional[TechnicalAnalysis] = None confirmation_analyses: List[TechnicalAnalysis] = field(default_factory=list) event_risk: str = "clear" event_description: str = "" news_sentiment: float = 0.0 news_description: str = "" final_signal: SignalStrength = SignalStrength.NEUTRAL final_score: float = 0.0 final_confidence: float = 0.0 action_recommendation: str = "" signal_quality: str = "" # A+, A, B, C, D, F trade_setup: Optional[TradeSetup] = None bullish_reasons: List[str] = field(default_factory=list) bearish_reasons: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) current_price: float = 0.0 spread_estimate: float = 0.0 market_regime: str = "" volatility_regime: str = "" class SignalAggregator: """Master aggregator for live signal analysis.""" def __init__(self, config: PlatformConfig): self.config = config self.data_provider = ForexDataProvider(config) self.tech_engine = TechnicalEngine(config.technical) self._news_engine: Optional[NewsEngine] = None # lazy def _get_news_engine(self) -> NewsEngine: if self._news_engine is None: self._news_engine = NewsEngine(self.config.news) return self._news_engine # ── Public: live mode entry point ────────────────────────────── def analyze_pair( self, symbol: Optional[str] = None, context: Optional[FundamentalContext] = None, ) -> SignalReport: """Live-mode entry: fetch all needed timeframes then delegate.""" if symbol is None: symbol = self.config.symbol timeframes = [self.config.primary_timeframe] + list( self.config.confirmation_timeframes ) bt = self.config.backtest if bt.regime_filter and bt.regime_filter_timeframe not in timeframes: timeframes.append(bt.regime_filter_timeframe) dfs: Dict[Timeframe, pd.DataFrame] = {} stale_tfs = [] for tf in timeframes: try: df = self.data_provider.fetch_ohlcv(symbol, tf) dfs[tf] = df if not self.data_provider._is_data_fresh(df, symbol, tf): stale_tfs.append(tf.value) except Exception as e: logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}") report = self.analyze_from_dataframes( symbol=symbol, dfs=dfs, context=context, use_news=True, ) if stale_tfs: report.warnings.append(f"STALE DATA on {', '.join(stale_tfs)} — signal may be unreliable") if self.config.primary_timeframe.value in stale_tfs: report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.trade_setup = None report.warnings.append("Primary timeframe stale — signal suppressed") if bt.regime_filter: self._apply_regime_filter_live(report, dfs) if bt.session_filter: xau_active = ( report.display_name == "XAU/USD" and self.config.xau_strategy.enabled and any("XAU STRATEGY" in r for r in report.bullish_reasons) ) if not xau_active: self._apply_session_filter_live(report) return report def _apply_session_filter_live(self, report: SignalReport) -> None: """Skip signals outside the configured UTC trading window.""" bt = self.config.backtest now_h = datetime.now(timezone.utc).hour if not (bt.session_start_hour <= now_h < bt.session_end_hour): report.warnings.append( f"SESSION FILTER: {now_h:02}:00 UTC outside " f"{bt.session_start_hour:02}:00-{bt.session_end_hour:02}:00 — skipping" ) report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.action_recommendation = ( f"NO TRADE — Outside session " f"({bt.session_start_hour:02}-{bt.session_end_hour:02} UTC; now {now_h:02})" ) report.trade_setup = None else: report.warnings.append( f"Session: {now_h:02}:00 UTC inside trading window " f"{bt.session_start_hour:02}-{bt.session_end_hour:02}" ) def _apply_regime_filter_live( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> None: """If higher-TF ADX is below threshold, downgrade signal to NEUTRAL.""" bt = self.config.backtest regime_tf = bt.regime_filter_timeframe df = dfs.get(regime_tf) if df is None or df.empty: report.warnings.append( f"Regime filter requested but {regime_tf.value} data unavailable" ) return adx_value = self._latest_adx(df, self.config.technical.adx_period) if adx_value is None: report.warnings.append( f"Regime filter: ADX could not be computed on {regime_tf.value}" ) return if adx_value < bt.regime_filter_min_adx: report.warnings.append( f"REGIME FILTER: {regime_tf.value} ADX={adx_value:.1f} < " f"{bt.regime_filter_min_adx:.0f} — market choppy, skipping signal" ) report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.action_recommendation = ( f"NO TRADE — Regime filter (ADX {adx_value:.1f} on {regime_tf.value})" ) report.trade_setup = None else: report.bullish_reasons.append( f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market" ) if report.final_score > 0 else report.bearish_reasons.append( f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market" ) @staticmethod def _latest_adx(df: pd.DataFrame, period: int = 14) -> Optional[float]: """Wilder ADX on a fresh df; returns the last finite value or None.""" if df is None or len(df) < period * 3: return None high, low, close = df["High"], df["Low"], df["Close"] tr = pd.concat([ high - low, (high - close.shift()).abs(), (low - close.shift()).abs(), ], axis=1).max(axis=1) plus_dm = high.diff() minus_dm = -low.diff() plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0) atr_s = tr.ewm(alpha=1/period, min_periods=period, adjust=False).mean() plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s) minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s) denom = (plus_di + minus_di).replace(0, float("nan")) dx = 100 * (plus_di - minus_di).abs() / denom adx = dx.ewm(alpha=1/period, min_periods=period, adjust=False).mean().dropna() if adx.empty: return None return float(adx.iloc[-1]) # ── Public: bar-replayable entry point ───────────────────────── def analyze_from_dataframes( self, symbol: str, dfs: Dict[Timeframe, pd.DataFrame], context: Optional[FundamentalContext] = None, use_news: bool = True, ) -> SignalReport: """ Run the full analysis pipeline on already-fetched dataframes. `dfs` must include the primary timeframe and ideally the confirmation timeframes. Each df is treated as the visible history at decision time — the backtester is responsible for slicing so no future bars leak in. """ pair_info = self.data_provider.get_pair_info(symbol) display = pair_info["display"] pip_value = pair_info["pip_value"] report = SignalReport( symbol=symbol, display_name=display, timestamp=str(datetime.now(timezone.utc)), ) # ── 1. Primary timeframe ─────────────────────────────────── primary_df = dfs.get(self.config.primary_timeframe) if primary_df is None or primary_df.empty: report.warnings.append("Primary timeframe data unavailable") report.signal_quality = "F" return report primary_analysis = self.tech_engine.analyze( primary_df, symbol, self.config.primary_timeframe.value ) report.primary_analysis = primary_analysis report.primary_signal = primary_analysis.composite_signal report.primary_score = primary_analysis.composite_score report.primary_confidence = primary_analysis.confidence report.market_regime = primary_analysis.market_regime report.volatility_regime = primary_analysis.volatility_regime report.current_price = float(primary_df["Close"].iloc[-1]) report.timestamp = str(primary_df.index[-1]) report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score # ── 2. Confirmation timeframes ───────────────────────────── for tf in self.config.confirmation_timeframes: cf_df = dfs.get(tf) if cf_df is None or cf_df.empty: continue try: cf_analysis = self.tech_engine.analyze(cf_df, symbol, tf.value) report.confirmation_analyses.append(cf_analysis) report.mtf_signals[tf.value] = cf_analysis.composite_signal report.mtf_scores[tf.value] = cf_analysis.composite_score except Exception as e: logger.warning(f"Confirmation TF {tf.value} failed for {symbol}: {e}") # ── 3. MTF agreement ─────────────────────────────────────── report.mtf_agreement = self._check_mtf_agreement(report) # ── 4. Fundamental context (optional) ────────────────────── if use_news: news = self._get_news_engine() if context is None: context = news.get_fundamental_context() event_risk, event_desc = news.get_event_risk_level(symbol, context) report.event_risk = event_risk report.event_description = event_desc sentiment, sent_desc = news.get_pair_sentiment(symbol, context) report.news_sentiment = sentiment report.news_description = sent_desc else: report.event_risk = "clear" report.event_description = "" report.news_sentiment = 0.0 report.news_description = "" # ── 5. Spread estimate from primary data (avoids extra API call) ─ try: report.spread_estimate = float( (primary_df["High"] - primary_df["Low"]).median() ) except Exception: report.spread_estimate = 0.0 # ── 6. Final signal ──────────────────────────────────────── self._compute_final_signal(report) # ── 6b. Session filter (works for both live and backtest) ─ if self.config.backtest.session_filter: try: bar_hour = pd.Timestamp(report.timestamp).hour if bar_hour in self.config.backtest.blocked_hours_utc: report.final_signal = SignalStrength.NEUTRAL report.signal_quality = "F" report.warnings.append(f"SESSION FILTER: {bar_hour:02}:00 UTC blocked") except Exception: pass # ── 6c. XAU/USD oversold bounce strategy ───────────────── if report.display_name == "XAU/USD" and self.config.xau_strategy.enabled: if self._check_xau_oversold_bounce(report, dfs): report.final_signal = SignalStrength.STRONG_BUY report.final_score = 50.0 report.final_confidence = 85.0 report.signal_quality = "A+" report.action_recommendation = ( "STRONG BUY — XAU oversold bounce " "(RSI<30 + M15/H1/H4 all above EMA200)" ) report.bullish_reasons.insert( 0, "XAU STRATEGY: M15 RSI oversold + triple-TF bullish trend confirmation", ) report.trade_setup = self._generate_xau_trade_setup(report, dfs) # ── 7. Trade setup for actionable signals ────────────────── if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"): is_sell_direction = ( report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL) or (report.final_signal == SignalStrength.NEUTRAL and report.final_score < 0) ) if is_sell_direction and self.config.backtest.sell_filter: if report.display_name in self.config.backtest.sell_allowed_pairs: report.trade_setup = self._generate_trade_setup(report, pip_value) elif self._forex_sell_qualifies(report): report.trade_setup = self._generate_trade_setup(report, pip_value) else: report.trade_setup = None else: report.trade_setup = self._generate_trade_setup(report, pip_value) # ── 8. Reasons ───────────────────────────────────────────── self._collect_reasons(report) return report # ── Internals ────────────────────────────────────────────────── def _forex_sell_qualifies(self, report: SignalReport) -> bool: """Always returns False — forex SELL is structurally unprofitable. 0/4 trades won even under best conditions (trending_down + MTF + ADX>=25 + RSI<=50).""" return False def _check_xau_oversold_bounce( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> bool: """Return True when XAU/USD oversold-bounce conditions are met. Conditions (all must be true): 1. M15 RSI(14) < 30 2. M15 price > M15 EMA(200) 3. H1 price > H1 EMA(200) 4. H4 price > H4 EMA(200) 5. Current bar hour not in blocked list """ cfg = self.config.xau_strategy try: bar_hour = pd.Timestamp(report.timestamp).hour if bar_hour in cfg.blocked_hours_utc: return False except Exception: return False m15_analysis = None h4_analysis = None for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca elif ca.timeframe == Timeframe.H4.value: h4_analysis = ca h1_analysis = report.primary_analysis if not m15_analysis or not h1_analysis or not h4_analysis: return False m15_rsi = m15_analysis.indicators.get("rsi") if not m15_rsi or m15_rsi.value >= cfg.rsi_entry_threshold: return False m15_ema200 = m15_analysis.key_levels.get("ema_trend") m15_df = dfs.get(Timeframe.M15) if m15_df is None or m15_ema200 is None: return False m15_price = float(m15_df["Close"].iloc[-1]) if m15_price <= m15_ema200: return False h1_ema200 = h1_analysis.key_levels.get("ema_trend") if h1_ema200 is None or report.current_price <= h1_ema200: return False h4_ema200 = h4_analysis.key_levels.get("ema_trend") h4_df = dfs.get(Timeframe.H4) if h4_df is None or h4_ema200 is None: return False h4_price = float(h4_df["Close"].iloc[-1]) if h4_price <= h4_ema200: return False return True def _generate_xau_trade_setup( self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame] ) -> Optional[TradeSetup]: """Generate XAU-specific trade setup with backtested TP/SL.""" cfg = self.config.xau_strategy m15_analysis = None for ca in report.confirmation_analyses: if ca.timeframe == Timeframe.M15.value: m15_analysis = ca break if not m15_analysis: return None atr = m15_analysis.key_levels.get("atr", 0) if atr <= 0: return None price = report.current_price pip_value = 0.10 tp1 = price + cfg.tp_atr_multiplier * atr sl = price - cfg.sl_atr_multiplier * atr tp2 = price + cfg.tp_atr_multiplier * 2.0 * atr sl_distance = price - sl tp_distance = tp1 - price rr = tp_distance / sl_distance if sl_distance > 0 else 0 return TradeSetup( pair=report.symbol, display_name=report.display_name, direction="BUY", entry_price=price, stop_loss=round(sl, 2), take_profit_1=round(tp1, 2), take_profit_2=round(tp2, 2), risk_reward=round(rr, 2), risk_pips=round(sl_distance / pip_value, 1), reward_pips=round(tp_distance / pip_value, 1), pip_value=pip_value, ) def _check_mtf_agreement(self, report: SignalReport) -> bool: signals = list(report.mtf_signals.values()) if len(signals) < 2: return False bullish = sum(1 for s in signals if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY)) bearish = sum(1 for s in signals if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL)) threshold = self.config.signals.mtf_agreement_threshold return bullish >= threshold or bearish >= threshold def _compute_final_signal(self, report: SignalReport): score = report.primary_score confidence = report.primary_confidence if report.mtf_agreement: confidence += 10 all_scores = list(report.mtf_scores.values()) if all_scores: mtf_avg = sum(all_scores) / len(all_scores) score = score * 0.6 + mtf_avg * 0.4 else: confidence -= 15 if len(report.mtf_scores) >= 2: scores = list(report.mtf_scores.values()) has_bull = any(s > 10 for s in scores) has_bear = any(s < -10 for s in scores) if has_bull and has_bear: report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes") if abs(report.news_sentiment) > 0.2: if (report.news_sentiment > 0 and score > 0) or \ (report.news_sentiment < 0 and score < 0): confidence += 5 elif (report.news_sentiment > 0 and score < -10) or \ (report.news_sentiment < 0 and score > 10): report.warnings.append("News sentiment CONFLICTS with technical signal") confidence -= 10 if report.event_risk == "danger": report.warnings.append(f"EVENT RISK: {report.event_description}") confidence = min(confidence, 30) report.action_recommendation = f"WAIT — {report.event_description}" elif report.event_risk == "caution": report.warnings.append(f"Upcoming event: {report.event_description}") confidence -= 15 is_counter_trend = ( (report.market_regime == "trending_up" and score < 0) or (report.market_regime == "trending_down" and score > 0) ) if is_counter_trend: confidence -= 15 report.warnings.append("Counter-trend signal — reduced confidence") if report.volatility_regime == "extreme": report.warnings.append("EXTREME volatility — wider stops needed, reduced position size") confidence -= 10 elif report.volatility_regime == "low": report.warnings.append("Low volatility — potential for breakout, tight ranges") primary = report.primary_analysis if primary is not None: ema200 = primary.key_levels.get("ema_trend") if ema200 and ema200 > 0 and report.current_price > 0: ema_dist_pct = abs(report.current_price - ema200) / ema200 * 100.0 _lo = float(os.environ.get("FB2_EMA_LO", "0.5")) _hi = float(os.environ.get("FB2_EMA_HI", "1.5")) if ema_dist_pct < _lo: report.warnings.append( f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk" ) confidence -= 10 elif ema_dist_pct > _hi: report.warnings.append( f"FAR FROM EMA200 ({ema_dist_pct:.2f}%) — over-extended" ) confidence -= 10 atr = primary.key_levels.get("atr", 0) if atr > 0 and score != 0: is_buy_signal = score > 0 if is_buy_signal: swing_lows = sorted( v for k, v in primary.key_levels.items() if k.startswith("swing_low_") ) if swing_lows and report.current_price - swing_lows[0] > 2.5 * atr: confidence -= 15 report.warnings.append("TREND EXHAUSTION — price extended >2.5 ATR past swing low") 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 > 2.5 * atr: confidence -= 15 report.warnings.append("TREND EXHAUSTION — price extended >2.5 ATR past swing high") ema_medium = primary.key_levels.get("ema_medium") if atr > 0 and ema_medium and report.market_regime.startswith("trending"): pullback_dist = abs(report.current_price - ema_medium) / atr if pullback_dist > 2.0: confidence -= 20 report.warnings.append( f"NO PULLBACK — price {pullback_dist:.1f} ATR from EMA21, chasing risk" ) elif pullback_dist > 1.5: confidence -= 10 report.warnings.append( f"Weak pullback — price {pullback_dist:.1f} ATR from EMA21" ) if report.market_regime.startswith("trending"): rsi_ind = primary.indicators.get("rsi") bb_ind = primary.indicators.get("bollinger") if rsi_ind: rsi_val = rsi_ind.value if report.market_regime in ("trending_up", "trending") and score > 0: if rsi_val > 65: confidence -= 15 report.warnings.append( f"Extended entry — RSI {rsi_val:.0f} in uptrend (prefer pullback to 40-55)" ) elif 38 <= rsi_val <= 55: confidence += 5 elif report.market_regime == "trending_down" and score < 0: if rsi_val < 35: confidence -= 15 report.warnings.append( f"Extended entry — RSI {rsi_val:.0f} in downtrend (prefer bounce to 45-62)" ) elif 45 <= rsi_val <= 62: confidence += 5 if bb_ind: bb_pos = bb_ind.value if report.market_regime in ("trending_up", "trending") and score > 0: if bb_pos > 0.85: confidence -= 10 elif bb_pos < 0.4: confidence += 5 elif report.market_regime == "trending_down" and score < 0: if bb_pos < 0.15: confidence -= 10 elif bb_pos > 0.6: confidence += 5 confidence = max(0.0, min(95.0, confidence)) # Final signal if score > 38: confidence -= 20 report.warnings.append("EXHAUSTION: score > 38 — trend overextended") if score >= 40 and confidence >= 60: report.final_signal = SignalStrength.STRONG_BUY elif score >= 30 and confidence >= 55: report.final_signal = SignalStrength.BUY elif score <= -40 and confidence >= 60: report.final_signal = SignalStrength.STRONG_SELL elif score <= -30 and confidence >= 55: report.final_signal = SignalStrength.SELL else: report.final_signal = SignalStrength.NEUTRAL report.final_score = score report.final_confidence = confidence # SELL suppression (pair-aware) sell_suppressed = False if self.config.backtest.sell_filter and report.final_signal in ( SignalStrength.SELL, SignalStrength.STRONG_SELL ): if report.display_name not in self.config.backtest.sell_allowed_pairs: report.final_signal = SignalStrength.NEUTRAL report.warnings.append(f"SELL SUPPRESSED for {report.display_name}") sell_suppressed = True # Quality grade abs_score = abs(score) if sell_suppressed: report.signal_quality = "F" elif abs_score >= 45 and confidence >= 70 and report.mtf_agreement: report.signal_quality = "A+" elif abs_score >= 38 and confidence >= 60 and report.mtf_agreement: report.signal_quality = "A" elif abs_score >= 33 and confidence >= 55: report.signal_quality = "B" elif abs_score >= 30 and confidence >= 50: 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 min_sl_pips = 8.0 if pip_value < 0.01 else 80.0 min_sl_price = min_sl_pips * pip_value if report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): is_buy = True elif report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): is_buy = False else: is_buy = report.final_score >= 0 levels = report.primary_analysis.key_levels min_rr = self.config.signals.min_rr_ratio mature_move = abs(report.final_score) > 30 tp_mult = 1.5 if mature_move else 2.0 tp_swing_cap = 2.5 if mature_move else 3.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 * 1.0 else: sl_price = price - atr * 2.0 min_sl = max(atr * 1.5, min_sl_price) if price - sl_price < min_sl: sl_price = price - min_sl sl_distance = price - sl_price swing_highs = sorted( v for k, v in levels.items() if k.startswith("swing_high_") and v > price + atr * 0.3 ) if (swing_highs and swing_highs[0] - price <= atr * tp_swing_cap and swing_highs[0] - price >= sl_distance * min_rr): tp1 = swing_highs[0] else: tp1 = price + max(atr * tp_mult, sl_distance * min_rr) if len(swing_highs) >= 2 and swing_highs[1] - price <= atr * 5.0: tp2 = swing_highs[1] else: tp2 = price + max(atr * (tp_mult + 1.5), sl_distance * min_rr * 1.5) direction = "BUY" entry, sl = price, sl_price else: swing_highs = sorted( v for k, v in levels.items() if k.startswith("swing_high_") and v > price ) if swing_highs: sl_price = swing_highs[0] + atr * 1.0 else: sl_price = price + atr * 2.0 min_sl = max(atr * 1.5, min_sl_price) if sl_price - price < min_sl: sl_price = price + min_sl sl_distance = sl_price - price swing_lows = sorted( [v for k, v in levels.items() if k.startswith("swing_low_") and v < price - atr * 0.3], reverse=True, ) if (swing_lows and price - swing_lows[0] <= atr * tp_swing_cap and price - swing_lows[0] >= sl_distance * min_rr): tp1 = swing_lows[0] else: tp1 = price - max(atr * tp_mult, sl_distance * min_rr) if len(swing_lows) >= 2 and price - swing_lows[1] <= atr * 5.0: tp2 = swing_lows[1] else: tp2 = price - max(atr * (tp_mult + 1.5), sl_distance * min_rr * 1.5) direction = "SELL" entry, sl = price, sl_price tp1_distance = abs(tp1 - entry) rr = tp1_distance / sl_distance if sl_distance > 0 else 0 if rr < min_rr - 1e-9: return None risk_pips = sl_distance / pip_value reward_pips = tp1_distance / pip_value return TradeSetup( pair=report.symbol, display_name=report.display_name, direction=direction, entry_price=entry, stop_loss=sl, take_profit_1=tp1, take_profit_2=tp2, risk_reward=round(rr, 2), risk_pips=risk_pips, reward_pips=reward_pips, pip_value=pip_value, ) def _collect_reasons(self, report: SignalReport): if not report.primary_analysis: return for name, ind in report.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue if ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_BUY else "" report.bullish_reasons.append(f"{prefix}{ind.name}: {ind.description}") elif ind.signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_SELL else "" report.bearish_reasons.append(f"{prefix}{ind.name}: {ind.description}") if report.news_sentiment > 0.1: report.bullish_reasons.append(f"News: {report.news_description}") elif report.news_sentiment < -0.1: report.bearish_reasons.append(f"News: {report.news_description}")