""" Forex Signal Intelligence Platform - Technical Analysis Engine =============================================================== Computes indicators, detects patterns, and generates technical signals. Each indicator returns both a value and a signal interpretation. Design philosophy: - No single indicator is trusted alone - Confluence of multiple signals = higher confidence - Divergences are weighted heavily (they predict reversals) - Market regime (trending vs ranging) changes which signals matter """ import logging from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd from config import TechnicalConfig, SignalStrength logger = logging.getLogger(__name__) @dataclass class IndicatorSignal: """Individual indicator's signal output.""" name: str value: float signal: SignalStrength confidence: float # 0-100 description: str weight: float = 1.0 @dataclass class TechnicalAnalysis: """Complete technical analysis result for a pair/timeframe.""" symbol: str timeframe: str timestamp: str indicators: Dict[str, IndicatorSignal] = field(default_factory=dict) composite_signal: SignalStrength = SignalStrength.NEUTRAL composite_score: float = 0.0 # -100 to +100 confidence: float = 0.0 # 0 to 100 market_regime: str = "unknown" # trending_up, trending_down, ranging volatility_regime: str = "normal" # low, normal, high, extreme key_levels: Dict[str, float] = field(default_factory=dict) pattern_signals: List[str] = field(default_factory=list) class TechnicalEngine: """ Core technical analysis engine. Computes all indicators on a given OHLCV DataFrame and produces a composite signal with confidence scoring. """ def __init__(self, config: TechnicalConfig): self.config = config def analyze(self, df: pd.DataFrame, symbol: str, timeframe: str) -> TechnicalAnalysis: """ Run full technical analysis on OHLCV data. Returns TechnicalAnalysis with all indicators, composite signal, and market regime classification. """ if df is None or len(df) < 30: return TechnicalAnalysis( symbol=symbol, timeframe=timeframe, timestamp=str(pd.Timestamp.utcnow()), confidence=0.0, ) result = TechnicalAnalysis( symbol=symbol, timeframe=timeframe, timestamp=str(df.index[-1]), ) # ── Compute all indicators ───────────────────────────────── disabled = set(self.config.disabled_indicators or []) if "ema_trend" not in disabled: self._compute_ema_signals(df, result) if "rsi" not in disabled: self._compute_rsi(df, result) if "bollinger" not in disabled: self._compute_bollinger(df, result) if "adx" not in disabled: self._compute_adx(df, result) if "atr" not in disabled: self._compute_atr(df, result) self._compute_macd(df, result) if "roc" not in disabled: self._compute_roc(df, result) if "keltner" not in disabled: self._compute_keltner(df, result) if "divergence" not in disabled: self._detect_divergences(df, result) if "structure" not in disabled: self._compute_structure_sr(df, result) if "candlestick" not in disabled: self._detect_candlestick_patterns(df, result) # ── Classify market regime ───────────────────────────────── self._classify_regime(df, result) # ── Regime-aware RSI reinterpretation ────────────────────── if self.config.rsi_mode == "regime_aware" and "rsi" not in disabled: self._reinterpret_rsi_for_regime(result) # ── Compute composite signal ─────────────────────────────── self._compute_composite(result) return result # ══════════════════════════════════════════════════════════════ # INDICATOR COMPUTATIONS # ══════════════════════════════════════════════════════════════ def _compute_ema_signals(self, df: pd.DataFrame, result: TechnicalAnalysis): """EMA trend alignment — supports 4-EMA and 2-EMA modes.""" if self.config.ema_mode == "2ema": self._compute_ema_2(df, result) else: self._compute_ema_4(df, result) def _compute_ema_2(self, df: pd.DataFrame, result: TechnicalAnalysis): """EMA 50/200 trend alignment — cleaner, less noise on H1.""" c = self.config close = df["Close"] ema50 = close.ewm(span=c.ema_slow, adjust=False).mean() ema200 = close.ewm(span=c.ema_trend, adjust=False).mean() if len(close) >= c.ema_trend else None ema21 = close.ewm(span=c.ema_medium, adjust=False).mean() latest = close.iloc[-1] s = ema50.iloc[-1] m = ema21.iloc[-1] t = ema200.iloc[-1] if ema200 is not None and len(ema200) >= c.ema_trend else None if t is None: result.indicators["ema_trend"] = IndicatorSignal( name="EMA Trend", value=0, signal=SignalStrength.NEUTRAL, confidence=30, description="EMA200 data insufficient", weight=1.0, ) result.key_levels["ema_medium"] = float(m) result.key_levels["ema_slow"] = float(s) return # 3-level alignment: price vs EMA50, EMA50 vs EMA200 if latest > s > t: signal = SignalStrength.STRONG_BUY conf = 85 desc = "Full bullish: price > EMA50 > EMA200" elif latest > s and s < t: signal = SignalStrength.BUY conf = 60 desc = "Emerging uptrend: price > EMA50, EMA50 < EMA200" elif latest < s < t: signal = SignalStrength.STRONG_SELL conf = 85 desc = "Full bearish: price < EMA50 < EMA200" elif latest < s and s > t: signal = SignalStrength.SELL conf = 60 desc = "Emerging downtrend: price < EMA50, EMA50 > EMA200" else: signal = SignalStrength.NEUTRAL conf = 40 desc = "EMA alignment mixed" # EMA50/200 crossover detection (recent 10 bars) if len(ema50) >= 10 and ema200 is not None: for lookback in range(2, min(11, len(ema50))): prev_s = ema50.iloc[-lookback] prev_t = ema200.iloc[-lookback] if prev_s <= prev_t and s > t: desc += " | Golden cross (50/200)" conf = min(conf + 10, 95) break elif prev_s >= prev_t and s < t: desc += " | Death cross (50/200)" conf = min(conf + 10, 95) break normalized = 1.0 if signal == SignalStrength.STRONG_BUY else ( 0.5 if signal == SignalStrength.BUY else ( -1.0 if signal == SignalStrength.STRONG_SELL else ( -0.5 if signal == SignalStrength.SELL else 0.0))) result.indicators["ema_trend"] = IndicatorSignal( name="EMA Trend", value=normalized, signal=signal, confidence=conf, description=desc, weight=1.0, ) result.key_levels["ema_trend"] = float(t) result.key_levels["ema_medium"] = float(m) result.key_levels["ema_slow"] = float(s) def _compute_ema_4(self, df: pd.DataFrame, result: TechnicalAnalysis): """Original 4-EMA (9/21/50/200) trend alignment analysis.""" c = self.config close = df["Close"] ema_fast = close.ewm(span=c.ema_fast, adjust=False).mean() ema_medium = close.ewm(span=c.ema_medium, adjust=False).mean() ema_slow = close.ewm(span=c.ema_slow, adjust=False).mean() ema_trend = close.ewm(span=c.ema_trend, adjust=False).mean() if len(close) >= c.ema_trend else None latest = close.iloc[-1] f, m, s = ema_fast.iloc[-1], ema_medium.iloc[-1], ema_slow.iloc[-1] t = ema_trend.iloc[-1] if ema_trend is not None and len(ema_trend) >= c.ema_trend else None alignment_score = 0 checks = [(latest, f), (f, m), (m, s)] if t is not None: checks.append((s, t)) for above, below in checks: if above > below: alignment_score += 1 elif above < below: alignment_score -= 1 max_checks = len(checks) normalized = alignment_score / max_checks if max_checks > 0 else 0 if normalized > 0.6: signal = SignalStrength.STRONG_BUY conf = min(90, 60 + normalized * 30) elif normalized > 0.2: signal = SignalStrength.BUY conf = 55 + normalized * 20 elif normalized < -0.6: signal = SignalStrength.STRONG_SELL conf = min(90, 60 + abs(normalized) * 30) elif normalized < -0.2: signal = SignalStrength.SELL conf = 55 + abs(normalized) * 20 else: signal = SignalStrength.NEUTRAL conf = 40 crossover_desc = "" if len(ema_fast) >= 3: prev_f, prev_m = ema_fast.iloc[-2], ema_medium.iloc[-2] if prev_f <= prev_m and f > m: crossover_desc = " | Golden cross (fast/medium)" conf = min(conf + 10, 95) elif prev_f >= prev_m and f < m: crossover_desc = " | Death cross (fast/medium)" conf = min(conf + 10, 95) # EMA ribbon compression/expansion detection ribbon_desc = "" atr_val = result.key_levels.get("atr", 0) ema_values = [f, m, s] if atr_val > 0: ribbon_width = (max(ema_values) - min(ema_values)) / atr_val if ribbon_width < 1.0: ribbon_desc = " | Ribbon COMPRESSED (trend change imminent)" if len(ema_fast) >= 6: prev_vals = [ema_fast.iloc[-5], ema_medium.iloc[-5], ema_slow.iloc[-5]] prev_width = (max(prev_vals) - min(prev_vals)) / atr_val if atr_val > 0 else 0 if prev_width < 1.0 and ribbon_width >= 1.0: ribbon_desc = " | Ribbon EXPANDING (new trend starting)" conf = min(conf + 10, 95) desc = f"EMA alignment: {alignment_score}/{max_checks}{crossover_desc}{ribbon_desc}" result.indicators["ema_trend"] = IndicatorSignal( name="EMA Trend", value=normalized, signal=signal, confidence=conf, description=desc, weight=1.0, ) if t is not None: result.key_levels["ema_trend"] = float(t) result.key_levels["ema_fast"] = float(f) result.key_levels["ema_medium"] = float(m) result.key_levels["ema_slow"] = float(s) def _compute_rsi(self, df: pd.DataFrame, result: TechnicalAnalysis): """RSI with trend-adjusted interpretation.""" c = self.config close = df["Close"] delta = close.diff() gain = delta.where(delta > 0, 0.0) loss = (-delta).where(delta < 0, 0.0) avg_gain = gain.ewm(alpha=1/c.rsi_period, min_periods=c.rsi_period, adjust=False).mean() avg_loss = loss.ewm(alpha=1/c.rsi_period, min_periods=c.rsi_period, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, 1e-10) rsi = 100 - (100 / (1 + rs)) rsi_val = rsi.iloc[-1] if not rsi.empty else 50.0 if pd.isna(rsi_val): rsi_val = 50.0 # Signal interpretation (context-aware) if rsi_val >= c.rsi_overbought: signal = SignalStrength.SELL conf = min(80, 50 + (rsi_val - c.rsi_overbought)) desc = f"RSI {rsi_val:.1f} — Overbought" elif rsi_val <= c.rsi_oversold: signal = SignalStrength.BUY conf = min(80, 50 + (c.rsi_oversold - rsi_val)) desc = f"RSI {rsi_val:.1f} — Oversold" elif rsi_val >= c.rsi_neutral_upper: signal = SignalStrength.BUY # bullish momentum conf = 45 desc = f"RSI {rsi_val:.1f} — Bullish momentum zone" elif rsi_val <= c.rsi_neutral_lower: signal = SignalStrength.SELL conf = 45 desc = f"RSI {rsi_val:.1f} — Bearish momentum zone" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"RSI {rsi_val:.1f} — Neutral" result.indicators["rsi"] = IndicatorSignal( name="RSI", value=rsi_val, signal=signal, confidence=conf, description=desc, weight=1.5, ) # Attach RSI series for divergence detection (dynamic attribute, filtered by _prefix) result._rsi_series = rsi # type: ignore[attr-defined] def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis): """MACD computation — always runs for divergence; optionally a signal pillar.""" c = self.config close = df["Close"] ema_fast = close.ewm(span=c.macd_fast, adjust=False).mean() ema_slow = close.ewm(span=c.macd_slow, adjust=False).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=c.macd_signal, adjust=False).mean() histogram = macd_line - signal_line result._macd_histogram = histogram # type: ignore[attr-defined] if not c.macd_as_pillar: return if len(histogram) < 3: return hist_now = histogram.iloc[-1] hist_prev = histogram.iloc[-2] hist_prev2 = histogram.iloc[-3] if pd.isna(hist_now) or pd.isna(hist_prev): return # Histogram zero-line crossover crossed_up = hist_prev <= 0 < hist_now crossed_down = hist_prev >= 0 > hist_now # Momentum acceleration accel_up = hist_now > 0 and hist_now > hist_prev > 0 accel_down = hist_now < 0 and hist_now < hist_prev < 0 # Momentum fading fading_up = hist_now > 0 and hist_now < hist_prev fading_down = hist_now < 0 and hist_now > hist_prev # Normalize histogram relative to recent range for confidence hist_abs = histogram.abs() if len(hist_abs) >= 20: hist_pct = (abs(hist_now) / hist_abs.iloc[-20:].mean()) if hist_abs.iloc[-20:].mean() > 0 else 1.0 else: hist_pct = 1.0 if crossed_up or accel_up: signal = SignalStrength.STRONG_BUY if accel_up else SignalStrength.BUY conf = min(80, 55 + hist_pct * 10) desc = "MACD histogram" + (" accelerating ↑" if accel_up else " crossed above zero") elif crossed_down or accel_down: signal = SignalStrength.STRONG_SELL if accel_down else SignalStrength.SELL conf = min(80, 55 + hist_pct * 10) desc = "MACD histogram" + (" accelerating ↓" if accel_down else " crossed below zero") elif fading_up: signal = SignalStrength.NEUTRAL conf = 35 desc = f"MACD histogram positive but fading" elif fading_down: signal = SignalStrength.NEUTRAL conf = 35 desc = f"MACD histogram negative but fading" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"MACD histogram flat" result.indicators["macd"] = IndicatorSignal( name="MACD", value=float(hist_now), signal=signal, confidence=conf, description=desc, weight=1.5, ) def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis): """Bollinger Bands position and squeeze detection.""" c = self.config close = df["Close"] sma = close.rolling(c.bb_period).mean() std = close.rolling(c.bb_period).std(ddof=0) upper = sma + c.bb_std * std lower = sma - c.bb_std * std latest = close.iloc[-1] upper_val = upper.iloc[-1] lower_val = lower.iloc[-1] sma_val = sma.iloc[-1] # Guard against NaN (insufficient data) if pd.isna(upper_val) or pd.isna(lower_val) or pd.isna(sma_val) or sma_val == 0: return band_width = (upper_val - lower_val) / sma_val # Position within bands (0 = lower band, 1 = upper band) bb_range = upper_val - lower_val bb_position = (latest - lower_val) / bb_range if bb_range != 0 else 0.5 # Squeeze detection (band width compared to recent average) is_squeeze = False if len(std) >= 20: avg_width_series = ((upper - lower) / sma).rolling(20).mean() avg_width = avg_width_series.iloc[-1] if not pd.isna(avg_width) and avg_width > 0: is_squeeze = band_width < avg_width * 0.75 # Signal squeeze_note = " | SQUEEZE detected (expect breakout)" if is_squeeze else "" if bb_position > 1.0: signal = SignalStrength.SELL conf = 65 desc = f"Price above upper BB ({bb_position:.2f}){squeeze_note}" elif bb_position < 0.0: signal = SignalStrength.BUY conf = 65 desc = f"Price below lower BB ({bb_position:.2f}){squeeze_note}" elif bb_position > 0.8: signal = SignalStrength.SELL conf = 45 desc = f"Price near upper BB ({bb_position:.2f}){squeeze_note}" elif bb_position < 0.2: signal = SignalStrength.BUY conf = 45 desc = f"Price near lower BB ({bb_position:.2f}){squeeze_note}" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"Price mid-band ({bb_position:.2f}){squeeze_note}" if is_squeeze: conf = min(conf + 10, 85) result.indicators["bollinger"] = IndicatorSignal( name="Bollinger Bands", value=bb_position, signal=signal, confidence=conf, description=desc, weight=1.5, ) result.key_levels["bb_upper"] = upper_val result.key_levels["bb_lower"] = lower_val result.key_levels["bb_mid"] = sma_val def _compute_adx(self, df: pd.DataFrame, result: TechnicalAnalysis): """ADX trend strength measurement.""" c = self.config high = df["High"] low = df["Low"] close = df["Close"] # True Range tr1 = high - low tr2 = (high - close.shift()).abs() tr3 = (low - close.shift()).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) # Directional Movement 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 = tr.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() plus_di = 100 * (plus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() / atr) minus_di = 100 * (minus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() / atr) dx = 100 * (plus_di - minus_di).abs() / np.maximum(plus_di + minus_di, 1e-10) adx = dx.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() adx_val = adx.iloc[-1] if not adx.empty and not pd.isna(adx.iloc[-1]) else 0.0 plus_di_last = plus_di.iloc[-1] if not plus_di.empty else np.nan minus_di_last = minus_di.iloc[-1] if not minus_di.empty else np.nan # Guard NaN — treat as equal (no directional edge) if data is missing plus_di_val = float(plus_di_last) if not pd.isna(plus_di_last) else 0.0 minus_di_val = float(minus_di_last) if not pd.isna(minus_di_last) else 0.0 # Signal (ADX tells strength, DI tells direction) if adx_val >= c.adx_very_strong: trend_strength = "Very strong" conf = 80 elif adx_val >= c.adx_strong_trend: trend_strength = "Strong" conf = 65 else: trend_strength = "Weak/Ranging" conf = 40 if plus_di_val > minus_di_val: signal = SignalStrength.BUY if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL direction = "bullish" else: signal = SignalStrength.SELL if adx_val >= c.adx_strong_trend else SignalStrength.NEUTRAL direction = "bearish" # ADX slope — trend acceleration detection slope_lookback = self.config.adx_slope_lookback adx_slope = 0.0 if len(adx) > slope_lookback: prev_adx = adx.iloc[-slope_lookback - 1] if not pd.isna(prev_adx): adx_slope = adx_val - prev_adx slope_desc = "" emerging_threshold = self.config.adx_slope_emerging_threshold if adx_val < self.config.adx_strong_trend and adx_slope > emerging_threshold: slope_desc = " | TREND EMERGING (ADX rising fast)" conf = min(conf + 15, 85) elif adx_val >= self.config.adx_strong_trend and adx_slope > 0: slope_desc = f" | Trend strengthening (slope +{adx_slope:.1f})" conf = min(conf + 5, 85) elif adx_val >= self.config.adx_strong_trend and adx_slope < -emerging_threshold: slope_desc = f" | Trend weakening (slope {adx_slope:.1f})" conf = max(conf - 10, 30) desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}{slope_desc}" result.indicators["adx"] = IndicatorSignal( name="ADX", value=adx_val, signal=signal, confidence=conf, description=desc, weight=0.5, ) result._adx_slope = adx_slope # type: ignore[attr-defined] def _compute_atr(self, df: pd.DataFrame, result: TechnicalAnalysis): """ATR volatility measurement and regime classification.""" c = self.config high, low, close = df["High"], df["Low"], df["Close"] tr1 = high - low tr2 = (high - close.shift()).abs() tr3 = (low - close.shift()).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) atr = tr.ewm(alpha=1/c.atr_period, min_periods=c.atr_period, adjust=False).mean() atr_val = atr.iloc[-1] if not atr.empty else 0.0 if pd.isna(atr_val): atr_val = 0.0 # ATR as percentage of price atr_pct = (atr_val / close.iloc[-1]) * 100 if close.iloc[-1] != 0 else 0 # Compare to historical ATR if len(atr) >= 50: atr_percentile = (atr.iloc[-1] > atr.iloc[-50:]).mean() * 100 else: atr_percentile = 50.0 if atr_percentile > 80: vol_regime = "extreme" elif atr_percentile > 60: vol_regime = "high" elif atr_percentile > 30: vol_regime = "normal" else: vol_regime = "low" result.volatility_regime = vol_regime result.key_levels["atr"] = atr_val result.key_levels["atr_pct"] = atr_pct result.indicators["atr"] = IndicatorSignal( name="ATR", value=atr_val, signal=SignalStrength.NEUTRAL, confidence=50, description=f"ATR {atr_val:.5f} ({atr_pct:.3f}%) | Volatility: {vol_regime} (P{atr_percentile:.0f})", weight=0.0, # ATR is context, not a directional signal ) def _compute_roc(self, df: pd.DataFrame, result: TechnicalAnalysis): """Rate of Change — momentum acceleration indicator.""" c = self.config close = df["Close"] n = c.roc_period if len(close) < n + 5: return denom1 = close.iloc[-n - 1] denom2 = close.iloc[-n - 2] if denom1 == 0 or denom2 == 0: return roc_val = ((close.iloc[-1] - denom1) / denom1) * 100 roc_prev = ((close.iloc[-2] - denom2) / denom2) * 100 if pd.isna(roc_val) or pd.isna(roc_prev): return roc_rising = roc_val > roc_prev roc_abs = abs(roc_val) threshold = c.roc_threshold if roc_val > 0 and roc_rising: if roc_abs > threshold * 2: signal = SignalStrength.STRONG_BUY conf = min(80, 60 + roc_abs * 5) desc = f"ROC +{roc_val:.2f}% — Strong momentum acceleration" else: signal = SignalStrength.BUY conf = min(70, 50 + roc_abs * 10) desc = f"ROC +{roc_val:.2f}% — Rising momentum" elif roc_val < 0 and not roc_rising: if roc_abs > threshold * 2: signal = SignalStrength.STRONG_SELL conf = min(80, 60 + roc_abs * 5) desc = f"ROC {roc_val:.2f}% — Strong bearish momentum" else: signal = SignalStrength.SELL conf = min(70, 50 + roc_abs * 10) desc = f"ROC {roc_val:.2f}% — Falling momentum" elif roc_prev <= 0 < roc_val: signal = SignalStrength.BUY conf = 65 desc = f"ROC crossed above zero (+{roc_val:.2f}%)" elif roc_prev >= 0 > roc_val: signal = SignalStrength.SELL conf = 65 desc = f"ROC crossed below zero ({roc_val:.2f}%)" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"ROC {roc_val:+.2f}% — No clear momentum" result.indicators["roc"] = IndicatorSignal( name="ROC", value=roc_val, signal=signal, confidence=conf, description=desc, weight=1.5, ) def _compute_keltner(self, df: pd.DataFrame, result: TechnicalAnalysis): """Keltner Channels + Bollinger-Keltner squeeze detection (TTM Squeeze).""" c = self.config close = df["Close"] high, low = df["High"], df["Low"] if len(close) < max(c.kc_period, c.kc_atr_period) + 5: return kc_mid = close.ewm(span=c.kc_period, adjust=False).mean() tr1 = high - low tr2 = (high - close.shift()).abs() tr3 = (low - close.shift()).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) kc_atr = tr.ewm(alpha=1/c.kc_atr_period, min_periods=c.kc_atr_period, adjust=False).mean() kc_upper = kc_mid + c.kc_multiplier * kc_atr kc_lower = kc_mid - c.kc_multiplier * kc_atr latest = close.iloc[-1] kc_upper_val = kc_upper.iloc[-1] kc_lower_val = kc_lower.iloc[-1] kc_mid_val = kc_mid.iloc[-1] if pd.isna(kc_upper_val) or pd.isna(kc_lower_val): return bb_upper = result.key_levels.get("bb_upper") bb_lower = result.key_levels.get("bb_lower") is_squeeze = False squeeze_released = False if bb_upper is not None and bb_lower is not None: is_squeeze = bb_lower > kc_lower_val and bb_upper < kc_upper_val if len(close) >= 3: bb_sma = close.rolling(c.bb_period).mean() bb_std = close.rolling(c.bb_period).std(ddof=0) if len(bb_sma) >= 2 and not pd.isna(bb_sma.iloc[-2]): prev_bb_upper = bb_sma.iloc[-2] + c.bb_std * bb_std.iloc[-2] prev_bb_lower = bb_sma.iloc[-2] - c.bb_std * bb_std.iloc[-2] prev_kc_upper = kc_upper.iloc[-2] prev_kc_lower = kc_lower.iloc[-2] was_squeeze = prev_bb_lower > prev_kc_lower and prev_bb_upper < prev_kc_upper squeeze_released = was_squeeze and not is_squeeze macd_hist = getattr(result, "_macd_histogram", None) macd_direction = 0 if macd_hist is not None and len(macd_hist) >= 1 and not pd.isna(macd_hist.iloc[-1]): macd_direction = 1 if macd_hist.iloc[-1] > 0 else -1 if squeeze_released: if macd_direction > 0: signal = SignalStrength.STRONG_BUY desc = "Keltner SQUEEZE RELEASED — bullish breakout" elif macd_direction < 0: signal = SignalStrength.STRONG_SELL desc = "Keltner SQUEEZE RELEASED — bearish breakout" else: signal = SignalStrength.NEUTRAL desc = "Keltner squeeze released — direction unclear" conf = 80 elif is_squeeze: signal = SignalStrength.NEUTRAL conf = 60 desc = "Keltner SQUEEZE active — breakout imminent" elif latest > kc_upper_val: signal = SignalStrength.BUY conf = 65 desc = f"Price above Keltner upper — strong uptrend continuation" elif latest < kc_lower_val: signal = SignalStrength.SELL conf = 65 desc = f"Price below Keltner lower — strong downtrend continuation" else: signal = SignalStrength.NEUTRAL conf = 30 desc = "Price within Keltner Channels" result.indicators["keltner"] = IndicatorSignal( name="Keltner", value=float(latest - kc_mid_val), signal=signal, confidence=conf, description=desc, weight=1.5, ) result.key_levels["kc_upper"] = float(kc_upper_val) result.key_levels["kc_lower"] = float(kc_lower_val) result.key_levels["kc_mid"] = float(kc_mid_val) result._kc_squeeze = is_squeeze # type: ignore[attr-defined] result._kc_squeeze_released = squeeze_released # type: ignore[attr-defined] def _detect_divergences(self, df: pd.DataFrame, result: TechnicalAnalysis): """ Detect RSI and MACD divergences (powerful reversal signals). Bullish divergence: Price makes lower low, indicator makes higher low Bearish divergence: Price makes higher high, indicator makes lower high """ c = self.config close = df["Close"] lookback = min(c.divergence_lookback, len(close) - 1) if lookback < 5: return rsi_series = getattr(result, "_rsi_series", None) macd_hist = getattr(result, "_macd_histogram", None) divergences = [] for name, indicator_series in [("RSI", rsi_series), ("MACD", macd_hist)]: if indicator_series is None or len(indicator_series) < lookback: continue price_window = close.iloc[-lookback:] ind_window = indicator_series.iloc[-lookback:] # Bullish divergence: lower price low but higher indicator low price_lows = self._find_swing_points(price_window, mode="low") ind_lows = self._find_swing_points(ind_window, mode="low") if len(price_lows) >= 2 and len(ind_lows) >= 2: if price_lows[-1][1] < price_lows[-2][1] and ind_lows[-1][1] > ind_lows[-2][1]: divergences.append(f"Bullish {name} divergence (reversal warning)") # Bearish divergence: higher price high but lower indicator high price_highs = self._find_swing_points(price_window, mode="high") ind_highs = self._find_swing_points(ind_window, mode="high") if len(price_highs) >= 2 and len(ind_highs) >= 2: if price_highs[-1][1] > price_highs[-2][1] and ind_highs[-1][1] < ind_highs[-2][1]: divergences.append(f"Bearish {name} divergence (reversal warning)") if divergences: is_bullish = any("Bullish" in d for d in divergences) is_bearish = any("Bearish" in d for d in divergences) if is_bullish and not is_bearish: signal = SignalStrength.BUY elif is_bearish and not is_bullish: signal = SignalStrength.SELL else: signal = SignalStrength.NEUTRAL result.indicators["divergence"] = IndicatorSignal( name="Divergence", value=len(divergences), signal=signal, confidence=70, description=" | ".join(divergences), weight=1.5, ) # Note: divergences are NOT added to pattern_signals to avoid # duplicate reason collection (the indicator entry covers them). def _compute_structure_sr(self, df: pd.DataFrame, result: TechnicalAnalysis): """Detect swing high/low structure levels for S/R and trade setup.""" lookback = min(self.config.swing_sr_lookback, len(df) - 1) if lookback < 10: return price = df["Close"].iloc[-1] high_window = df["High"].iloc[-lookback:] low_window = df["Low"].iloc[-lookback:] swing_highs = self._find_swing_points(high_window, mode="high", window=3) swing_lows = self._find_swing_points(low_window, mode="low", window=3) highs_above = sorted([v for _, v in swing_highs if v > price]) highs_below = sorted([v for _, v in swing_highs if v <= price], reverse=True) lows_below = sorted([v for _, v in swing_lows if v < price], reverse=True) lows_above = sorted([v for _, v in swing_lows if v >= price]) for i, val in enumerate(highs_above[:3], 1): result.key_levels[f"swing_high_{i}"] = float(val) for i, val in enumerate(lows_below[:3], 1): result.key_levels[f"swing_low_{i}"] = float(val) if not highs_above and highs_below: result.key_levels["swing_high_1"] = float(highs_below[0]) if not lows_below and lows_above: result.key_levels["swing_low_1"] = float(lows_above[0]) def _detect_candlestick_patterns(self, df: pd.DataFrame, result: TechnicalAnalysis): """Detect significant candlestick patterns on recent bars.""" if len(df) < 5: return patterns = [] o, h, l, c = df["Open"].iloc[-1], df["High"].iloc[-1], df["Low"].iloc[-1], df["Close"].iloc[-1] prev_o, prev_c = df["Open"].iloc[-2], df["Close"].iloc[-2] body = abs(c - o) upper_wick = h - max(o, c) lower_wick = min(o, c) - l full_range = h - l if full_range == 0: return body_ratio = body / full_range # Doji (indecision) if body_ratio < 0.1: patterns.append("Doji (indecision)") # Hammer / Hanging Man if lower_wick > 2 * body and upper_wick < body * 0.5 and body_ratio > 0.1: if c > o: patterns.append("Hammer (bullish reversal)") else: patterns.append("Hanging Man (bearish warning)") # Shooting Star / Inverted Hammer if upper_wick > 2 * body and lower_wick < body * 0.5 and body_ratio > 0.1: if c < o: patterns.append("Shooting Star (bearish reversal)") else: patterns.append("Inverted Hammer (bullish)") # Engulfing if len(df) >= 3: if prev_c < prev_o and c > o and c > prev_o and o < prev_c: patterns.append("Bullish Engulfing") elif prev_c > prev_o and c < o and c < prev_o and o > prev_c: patterns.append("Bearish Engulfing") # Marubozu (strong conviction candle) if body_ratio > 0.85: if c > o: patterns.append("Bullish Marubozu (strong buyers)") else: patterns.append("Bearish Marubozu (strong sellers)") if patterns: bullish = sum(1 for p in patterns if "bullish" in p.lower() or "Hammer" in p) bearish = sum(1 for p in patterns if "bearish" in p.lower() or "Shooting" in p) if bullish > bearish: signal = SignalStrength.BUY elif bearish > bullish: signal = SignalStrength.SELL else: signal = SignalStrength.NEUTRAL conf = 50 atr_val = result.key_levels.get("atr", 0) latest = c proximity = atr_val * 0.5 if atr_val > 0 else 0 if proximity > 0: swing_lows = sorted( [v for k, v in result.key_levels.items() if k.startswith("swing_low_")] ) swing_highs = sorted( [v for k, v in result.key_levels.items() if k.startswith("swing_high_")] ) near_support = any(abs(latest - sl) <= proximity for sl in swing_lows) near_resistance = any(abs(latest - sh) <= proximity for sh in swing_highs) if bullish > bearish and near_support: conf += 20 patterns.append("(at support)") elif bearish > bullish and near_resistance: conf += 20 patterns.append("(at resistance)") result.indicators["candlestick"] = IndicatorSignal( name="Candlestick", value=float(bullish - bearish), signal=signal, confidence=conf, description=" | ".join(patterns), weight=1.0, ) result.pattern_signals.extend(patterns) # ══════════════════════════════════════════════════════════════ # MARKET REGIME & COMPOSITE SCORING # ══════════════════════════════════════════════════════════════ def _reinterpret_rsi_for_regime(self, result: TechnicalAnalysis): """Reinterpret RSI signal based on market regime — called after regime classification.""" rsi_ind = result.indicators.get("rsi") if not rsi_ind: return rsi_val = rsi_ind.value regime = result.market_regime if regime in ("trending_up", "trending"): if 38 <= rsi_val <= 55: signal = SignalStrength.STRONG_BUY conf = 75 desc = f"RSI {rsi_val:.1f} — Pullback zone in uptrend (ideal entry)" elif 30 <= rsi_val < 38: signal = SignalStrength.BUY conf = 60 desc = f"RSI {rsi_val:.1f} — Deep pullback in uptrend" elif 55 < rsi_val <= 65: signal = SignalStrength.BUY conf = 50 desc = f"RSI {rsi_val:.1f} — Momentum ok in uptrend" elif rsi_val > 65: signal = SignalStrength.NEUTRAL conf = 30 desc = f"RSI {rsi_val:.1f} — Extended in uptrend, wait for pullback" elif rsi_val < 30: signal = SignalStrength.NEUTRAL conf = 25 desc = f"RSI {rsi_val:.1f} — Oversold in uptrend, possible reversal" else: return elif regime == "trending_down": if 45 <= rsi_val <= 62: signal = SignalStrength.STRONG_SELL conf = 75 desc = f"RSI {rsi_val:.1f} — Bounce zone in downtrend (ideal short)" elif 62 < rsi_val <= 70: signal = SignalStrength.SELL conf = 60 desc = f"RSI {rsi_val:.1f} — Deep bounce in downtrend" elif 35 <= rsi_val < 45: signal = SignalStrength.SELL conf = 50 desc = f"RSI {rsi_val:.1f} — Momentum ok in downtrend" elif rsi_val < 35: signal = SignalStrength.NEUTRAL conf = 30 desc = f"RSI {rsi_val:.1f} — Extended in downtrend, wait for bounce" elif rsi_val > 70: signal = SignalStrength.NEUTRAL conf = 25 desc = f"RSI {rsi_val:.1f} — Overbought in downtrend, possible reversal" else: return else: # ranging if rsi_val >= 70: signal = SignalStrength.SELL conf = 70 desc = f"RSI {rsi_val:.1f} — Overbought in range" elif rsi_val <= 30: signal = SignalStrength.BUY conf = 70 desc = f"RSI {rsi_val:.1f} — Oversold in range" elif rsi_val >= 60: signal = SignalStrength.SELL conf = 45 desc = f"RSI {rsi_val:.1f} — Upper range zone" elif rsi_val <= 40: signal = SignalStrength.BUY conf = 45 desc = f"RSI {rsi_val:.1f} — Lower range zone" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"RSI {rsi_val:.1f} — Mid-range, no edge" result.indicators["rsi"] = IndicatorSignal( name="RSI", value=rsi_val, signal=signal, confidence=conf, description=desc, weight=rsi_ind.weight, ) def _classify_regime(self, df: pd.DataFrame, result: TechnicalAnalysis): """Classify the market regime (trending vs ranging vs transitioning).""" adx_sig = result.indicators.get("adx") ema_sig = result.indicators.get("ema_trend") adx_slope = getattr(result, "_adx_slope", 0.0) if adx_sig and adx_sig.value >= self.config.adx_strong_trend: if ema_sig and ema_sig.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): result.market_regime = "trending_up" elif ema_sig and ema_sig.signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL): result.market_regime = "trending_down" else: result.market_regime = "trending" elif adx_sig and adx_slope > self.config.adx_slope_emerging_threshold: result.market_regime = "transitioning" else: result.market_regime = "ranging" def _compute_composite(self, result: TechnicalAnalysis): """ Compute the final composite signal score. Weighted average of all indicator signals, adjusted for: - Market regime (trend indicators weighted more in trends) - Confidence scaling """ total_score = 0.0 total_weight = 0.0 total_confidence = 0.0 # Use override weights if provided, else gold-optimized defaults if self.config.regime_weights_override: regime_weights = self.config.regime_weights_override else: _trending = { "ema_trend": 2.0, "rsi": 1.0, "macd": 2.0, "bollinger": 0.5, "keltner": 1.5, "roc": 1.5, "candlestick": 0.8, "divergence": 2.0, "adx": 0.8, } regime_weights = { "trending_up": _trending, "trending_down": _trending, "trending": _trending, "ranging": { "ema_trend": 0.3, "rsi": 1.5, "macd": 1.5, "bollinger": 1.5, "keltner": 2.0, "roc": 0.5, "candlestick": 1.0, "divergence": 2.5, "adx": 0.5, }, "transitioning": { "ema_trend": 1.5, "rsi": 1.0, "macd": 2.5, "bollinger": 1.0, "keltner": 2.5, "roc": 2.0, "candlestick": 1.0, "divergence": 2.0, "adx": 1.0, }, } active_regime_weights = regime_weights.get(result.market_regime, {}) for name, ind in result.indicators.items(): if name.startswith("_") or ind.weight == 0: continue weight = ind.weight regime_mult = active_regime_weights.get(name, 1.0) weight *= regime_mult signal_score = ind.signal.value weighted_score = signal_score * weight * (ind.confidence / 100.0) total_score += weighted_score total_weight += weight total_confidence += ind.confidence * weight if total_weight > 0: normalized_score = (total_score / total_weight) * 50 avg_confidence = total_confidence / total_weight else: normalized_score = 0 avg_confidence = 0 normalized_score = max(-100, min(100, normalized_score)) if normalized_score >= 40: result.composite_signal = SignalStrength.STRONG_BUY elif normalized_score >= 15: result.composite_signal = SignalStrength.BUY elif normalized_score <= -40: result.composite_signal = SignalStrength.STRONG_SELL elif normalized_score <= -15: result.composite_signal = SignalStrength.SELL else: result.composite_signal = SignalStrength.NEUTRAL result.composite_score = normalized_score result.confidence = avg_confidence # ══════════════════════════════════════════════════════════════ # HELPER METHODS # ══════════════════════════════════════════════════════════════ @staticmethod def _find_swing_points( series: pd.Series, mode: str = "low", window: int = 3, ) -> List[Tuple[int, float]]: """Find local swing highs or lows in a series.""" points = [] vals = series.values for i in range(window, len(vals) - window): if np.isnan(vals[i]): continue left_vals = [vals[i-j] for j in range(1, window+1) if not np.isnan(vals[i-j])] right_vals = [vals[i+j] for j in range(1, window+1) if i+j < len(vals) and not np.isnan(vals[i+j])] if not left_vals or not right_vals: continue if mode == "low": if all(vals[i] <= v for v in left_vals) and all(vals[i] <= v for v in right_vals): points.append((i, vals[i])) else: if all(vals[i] >= v for v in left_vals) and all(vals[i] >= v for v in right_vals): points.append((i, vals[i])) return points