""" 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 enum import Enum 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 ───────────────────────────────── self._compute_ema_signals(df, result) self._compute_rsi(df, result) self._compute_macd(df, result) self._compute_bollinger(df, result) self._compute_stochastic(df, result) self._compute_adx(df, result) self._compute_atr(df, result) self._compute_ichimoku(df, result) self._detect_divergences(df, result) self._detect_candlestick_patterns(df, result) self._compute_support_resistance(df, result) self._compute_volume_analysis(df, result) # ── Classify market regime ───────────────────────────────── self._classify_regime(df, 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 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 # Perfect alignment scoring # Bullish: price > fast > medium > slow > trend # Bearish: price < fast < medium < slow < trend 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 # Check for EMA crossover (recent) 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) desc = f"EMA alignment: {alignment_score}/{max_checks}{crossover_desc}" result.indicators["ema_trend"] = IndicatorSignal( name="EMA Trend", value=normalized, signal=signal, confidence=conf, description=desc, weight=1.2, ) 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).mean() avg_loss = loss.ewm(alpha=1/c.rsi_period, min_periods=c.rsi_period).mean() rs = avg_gain / avg_loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) rsi_val = rsi.iloc[-1] if not rsi.empty else 50.0 if np.isnan(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.0, ) # Store RSI series for divergence detection result.indicators["_rsi_series"] = IndicatorSignal( name="_rsi_series", value=0, signal=SignalStrength.NEUTRAL, confidence=0, description="", weight=0, ) result._rsi_series = rsi # attach for divergence use def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis): """MACD with histogram momentum analysis.""" 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 macd_val = macd_line.iloc[-1] signal_val = signal_line.iloc[-1] hist_val = histogram.iloc[-1] # Check for crossover crossover = False crossover_type = "" if len(macd_line) >= 2: prev_macd = macd_line.iloc[-2] prev_signal = signal_line.iloc[-2] if prev_macd <= prev_signal and macd_val > signal_val: crossover = True crossover_type = "bullish" elif prev_macd >= prev_signal and macd_val < signal_val: crossover = True crossover_type = "bearish" # Histogram momentum (is it expanding or contracting?) hist_momentum = "" if len(histogram) >= 3: h1, h2, h3 = histogram.iloc[-3], histogram.iloc[-2], histogram.iloc[-1] if h3 > h2 > h1 and h3 > 0: hist_momentum = "accelerating bullish" elif h3 < h2 < h1 and h3 < 0: hist_momentum = "accelerating bearish" elif abs(h3) < abs(h2) and h3 * h2 > 0: hist_momentum = "decelerating" # Signal if crossover and crossover_type == "bullish": signal = SignalStrength.STRONG_BUY conf = 75 desc = f"MACD bullish crossover | Hist: {hist_val:.6f}" elif crossover and crossover_type == "bearish": signal = SignalStrength.STRONG_SELL conf = 75 desc = f"MACD bearish crossover | Hist: {hist_val:.6f}" elif macd_val > signal_val and hist_val > 0: signal = SignalStrength.BUY conf = 55 desc = f"MACD above signal, positive histogram | {hist_momentum}" elif macd_val < signal_val and hist_val < 0: signal = SignalStrength.SELL conf = 55 desc = f"MACD below signal, negative histogram | {hist_momentum}" else: signal = SignalStrength.NEUTRAL conf = 35 desc = f"MACD mixed | Hist: {hist_val:.6f}" result.indicators["macd"] = IndicatorSignal( name="MACD", value=hist_val, signal=signal, confidence=conf, description=desc, weight=1.1, ) # Store for divergence result._macd_histogram = histogram 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() 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] band_width = (upper_val - lower_val) / sma_val if sma_val != 0 else 0 # Position within bands (0 = lower band, 1 = upper band) bb_position = (latest - lower_val) / (upper_val - lower_val) if (upper_val - lower_val) != 0 else 0.5 # Squeeze detection (band width compared to recent average) if len(std) >= 20: avg_width = ((upper - lower) / sma).rolling(20).mean().iloc[-1] is_squeeze = band_width < avg_width * 0.75 else: is_squeeze = False # 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=0.9, ) result.key_levels["bb_upper"] = upper_val result.key_levels["bb_lower"] = lower_val result.key_levels["bb_mid"] = sma_val def _compute_stochastic(self, df: pd.DataFrame, result: TechnicalAnalysis): """Stochastic oscillator with crossover signals.""" c = self.config high = df["High"] low = df["Low"] close = df["Close"] lowest_low = low.rolling(c.stoch_k).min() highest_high = high.rolling(c.stoch_k).max() denom = highest_high - lowest_low denom = denom.replace(0, np.nan) k_line = 100 * (close - lowest_low) / denom d_line = k_line.rolling(c.stoch_d).mean() k_val = k_line.iloc[-1] if not k_line.empty else 50.0 d_val = d_line.iloc[-1] if not d_line.empty else 50.0 if np.isnan(k_val): k_val = 50.0 if np.isnan(d_val): d_val = 50.0 # Crossover detection crossover = "" if len(k_line) >= 2 and len(d_line) >= 2: prev_k = k_line.iloc[-2] prev_d = d_line.iloc[-2] if not np.isnan(prev_k) and not np.isnan(prev_d): if prev_k <= prev_d and k_val > d_val: crossover = " | Bullish %K/%D cross" elif prev_k >= prev_d and k_val < d_val: crossover = " | Bearish %K/%D cross" if k_val >= c.stoch_overbought and d_val >= c.stoch_overbought: signal = SignalStrength.SELL conf = 60 desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f} — Overbought{crossover}" elif k_val <= c.stoch_oversold and d_val <= c.stoch_oversold: signal = SignalStrength.BUY conf = 60 desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f} — Oversold{crossover}" elif crossover and "Bullish" in crossover: signal = SignalStrength.BUY conf = 55 desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f}{crossover}" elif crossover and "Bearish" in crossover: signal = SignalStrength.SELL conf = 55 desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f}{crossover}" else: signal = SignalStrength.NEUTRAL conf = 30 desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f} — Neutral" result.indicators["stochastic"] = IndicatorSignal( name="Stochastic", value=k_val, signal=signal, confidence=conf, description=desc, weight=0.8, ) 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).mean() plus_di = 100 * (plus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() / atr) minus_di = 100 * (minus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() / atr) dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan) adx = dx.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() adx_val = adx.iloc[-1] if not adx.empty and not np.isnan(adx.iloc[-1]) else 0.0 plus_di_val = plus_di.iloc[-1] if not plus_di.empty else 0.0 minus_di_val = minus_di.iloc[-1] if not minus_di.empty 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" desc = f"ADX {adx_val:.1f} ({trend_strength} {direction}) | +DI={plus_di_val:.1f} -DI={minus_di_val:.1f}" result.indicators["adx"] = IndicatorSignal( name="ADX", value=adx_val, signal=signal, confidence=conf, description=desc, weight=1.0, ) 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).mean() atr_val = atr.iloc[-1] if not atr.empty else 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 doesn't produce buy/sell, it's context ) def _compute_ichimoku(self, df: pd.DataFrame, result: TechnicalAnalysis): """Ichimoku Cloud analysis.""" c = self.config high, low, close = df["High"], df["Low"], df["Close"] if len(df) < c.ichimoku_senkou_b + 26: return # Tenkan-sen (Conversion Line) tenkan = (high.rolling(c.ichimoku_tenkan).max() + low.rolling(c.ichimoku_tenkan).min()) / 2 # Kijun-sen (Base Line) kijun = (high.rolling(c.ichimoku_kijun).max() + low.rolling(c.ichimoku_kijun).min()) / 2 # Senkou Span A (Leading Span A) senkou_a = ((tenkan + kijun) / 2).shift(c.ichimoku_kijun) # Senkou Span B (Leading Span B) senkou_b = ((high.rolling(c.ichimoku_senkou_b).max() + low.rolling(c.ichimoku_senkou_b).min()) / 2).shift(c.ichimoku_kijun) latest = close.iloc[-1] tenkan_val = tenkan.iloc[-1] kijun_val = kijun.iloc[-1] # Use non-shifted values for current cloud sa = ((tenkan.iloc[-1] + kijun.iloc[-1]) / 2) sb_raw = (high.rolling(c.ichimoku_senkou_b).max() + low.rolling(c.ichimoku_senkou_b).min()) / 2 sb = sb_raw.iloc[-1] cloud_top = max(sa, sb) cloud_bottom = min(sa, sb) cloud_is_green = sa > sb # Signal analysis signals = [] score = 0 # Price vs Cloud if latest > cloud_top: signals.append("Price ABOVE cloud") score += 2 elif latest < cloud_bottom: signals.append("Price BELOW cloud") score -= 2 else: signals.append("Price IN cloud (indecision)") # Tenkan vs Kijun if tenkan_val > kijun_val: signals.append("TK cross bullish") score += 1 elif tenkan_val < kijun_val: signals.append("TK cross bearish") score -= 1 # Cloud color if cloud_is_green: signals.append("Green cloud") score += 1 else: signals.append("Red cloud") score -= 1 if score >= 3: signal = SignalStrength.STRONG_BUY conf = 75 elif score >= 1: signal = SignalStrength.BUY conf = 55 elif score <= -3: signal = SignalStrength.STRONG_SELL conf = 75 elif score <= -1: signal = SignalStrength.SELL conf = 55 else: signal = SignalStrength.NEUTRAL conf = 35 result.indicators["ichimoku"] = IndicatorSignal( name="Ichimoku", value=float(score), signal=signal, confidence=conf, description=f"Ichimoku [{score}/4]: {' | '.join(signals)}", weight=1.1, ) result.key_levels["ichimoku_cloud_top"] = cloud_top result.key_levels["ichimoku_cloud_bottom"] = cloud_bottom 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:] # Find local swing lows (for bullish divergence) 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: # Price: lower low, Indicator: higher low 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)") # Find local swing highs (for bearish divergence) 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, # high weight — divergences are strong signals ) result.pattern_signals.extend(divergences) 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 result.indicators["candlestick"] = IndicatorSignal( name="Candlestick", value=float(bullish - bearish), signal=signal, confidence=50, description=" | ".join(patterns), weight=0.7, ) result.pattern_signals.extend(patterns) def _compute_support_resistance(self, df: pd.DataFrame, result: TechnicalAnalysis): """Calculate pivot points and key S/R levels.""" if len(df) < 2: return # Use previous bar for pivot calculation prev_h = df["High"].iloc[-2] prev_l = df["Low"].iloc[-2] prev_c = df["Close"].iloc[-2] pivot = (prev_h + prev_l + prev_c) / 3 r1 = 2 * pivot - prev_l s1 = 2 * pivot - prev_h r2 = pivot + (prev_h - prev_l) s2 = pivot - (prev_h - prev_l) r3 = prev_h + 2 * (pivot - prev_l) s3 = prev_l - 2 * (prev_h - pivot) result.key_levels.update({ "pivot": pivot, "r1": r1, "r2": r2, "r3": r3, "s1": s1, "s2": s2, "s3": s3, }) # Signal based on price position relative to pivot latest = df["Close"].iloc[-1] if latest > r1: desc = f"Price above R1 ({r1:.5f}), next target R2 ({r2:.5f})" elif latest > pivot: desc = f"Price above Pivot ({pivot:.5f}), testing R1 ({r1:.5f})" elif latest > s1: desc = f"Price below Pivot ({pivot:.5f}), support at S1 ({s1:.5f})" else: desc = f"Price below S1 ({s1:.5f}), support at S2 ({s2:.5f})" result.indicators["pivot"] = IndicatorSignal( name="Pivot Points", value=pivot, signal=SignalStrength.BUY if latest > pivot else SignalStrength.SELL, confidence=45, description=desc, weight=0.6, ) def _compute_volume_analysis(self, df: pd.DataFrame, result: TechnicalAnalysis): """Volume analysis (limited on Forex but still useful as a filter).""" c = self.config vol = df.get("Volume") if vol is None or vol.sum() == 0: return vol_ma = vol.rolling(c.volume_ma_period).mean() current_vol = vol.iloc[-1] avg_vol = vol_ma.iloc[-1] if pd.isna(avg_vol) or avg_vol == 0: return vol_ratio = current_vol / avg_vol if vol_ratio > 2.0: desc = f"Volume {vol_ratio:.1f}x average — HIGH ACTIVITY" conf = 55 elif vol_ratio > 1.3: desc = f"Volume {vol_ratio:.1f}x average — above normal" conf = 45 elif vol_ratio < 0.5: desc = f"Volume {vol_ratio:.1f}x average — LOW (signals less reliable)" conf = 25 else: desc = f"Volume {vol_ratio:.1f}x average — normal" conf = 35 result.indicators["volume"] = IndicatorSignal( name="Volume", value=vol_ratio, signal=SignalStrength.NEUTRAL, confidence=conf, description=desc, weight=0.0, ) # ══════════════════════════════════════════════════════════════ # MARKET REGIME & COMPOSITE SCORING # ══════════════════════════════════════════════════════════════ def _classify_regime(self, df: pd.DataFrame, result: TechnicalAnalysis): """Classify the market regime (trending vs ranging).""" adx_sig = result.indicators.get("adx") ema_sig = result.indicators.get("ema_trend") 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" 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) - Divergence boost (divergences override in ranging markets) - Confidence scaling """ total_score = 0.0 total_weight = 0.0 total_confidence = 0.0 regime_weights = { "trending_up": {"ema_trend": 1.5, "adx": 1.3, "macd": 1.2, "rsi": 0.7, "bollinger": 0.6}, "trending_down": {"ema_trend": 1.5, "adx": 1.3, "macd": 1.2, "rsi": 0.7, "bollinger": 0.6}, "ranging": {"rsi": 1.4, "stochastic": 1.3, "bollinger": 1.4, "ema_trend": 0.6, "adx": 0.5}, } active_regime_weights = regime_weights.get(result.market_regime, {}) for name, ind in result.indicators.items(): if name.startswith("_") or ind.weight == 0: continue # Base weight from indicator weight = ind.weight # Regime adjustment regime_mult = active_regime_weights.get(name, 1.0) weight *= regime_mult # Signal to score (-2 to +2) signal_score = ind.signal.value # Weight by confidence 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 # scale to -100..+100 avg_confidence = total_confidence / total_weight else: normalized_score = 0 avg_confidence = 0 # Clamp normalized_score = max(-100, min(100, normalized_score)) # Determine composite signal 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 if mode == "low": if all(vals[i] <= vals[i-j] for j in range(1, window+1) if not np.isnan(vals[i-j])) and \ all(vals[i] <= vals[i+j] for j in range(1, window+1) if i+j < len(vals) and not np.isnan(vals[i+j])): points.append((i, vals[i])) else: if all(vals[i] >= vals[i-j] for j in range(1, window+1) if not np.isnan(vals[i-j])) and \ all(vals[i] >= vals[i+j] for j in range(1, window+1) if i+j < len(vals) and not np.isnan(vals[i+j])): points.append((i, vals[i])) return points