harwinder
e6e09c587e
Bug fixes:
- Swing point detection: false positives when all neighbors are NaN (all([]) == True)
- ROC indicator: division by zero when historical close price is zero
- News currency extraction: substring matching ("usd" in "usda") replaced with word-boundary regex
- ADX DX calculation: use epsilon instead of NaN to avoid cascade
- Signal aggregator: refactor ternary append, add pip_value<=0 guard
New tests (77) cover: swing point edge cases, ROC/ADX/RSI/MACD/BB boundaries,
regime classification, composite scoring, quality grade thresholds, EUR filter
pipeline, trade setup edge cases, MTF agreement, weekend filter, and integration.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1130 lines
46 KiB
Python
Executable File
1130 lines
46 KiB
Python
Executable File
"""
|
|
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
|