diff --git a/engine.py b/engine.py new file mode 100644 index 0000000..5b22069 --- /dev/null +++ b/engine.py @@ -0,0 +1,927 @@ +""" +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 \ No newline at end of file