harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backtest tuning, data freshness guard, Telegram improvements

Browse Source 
- Drop USD/JPY from scan pairs (zero signals across all timeframes)
- Disable divergence indicator (zero impact in 14-config lab test)
- Parallelize evaluate.py with ProcessPoolExecutor (16 cores)
- Add --pair CLI filter to evaluate.py for single-pair backtests
- Add data freshness guard: stale cache auto-invalidated, primary TF
  stale data suppresses signals to prevent trading on old data
- Add /cache command: shows fetch age and last bar per pair/TF
- Add /clearcache command: wipes in-memory cache for fresh API fetch
- Remove BB from WebSocket breach alerts (keep structural levels only)
- Show live prices only for WS-monitored pairs (EUR/USD, XAU/USD)
- Replace arrow emojis with text tags in indicator lines
This commit is contained in:
harwinder
2026-05-12 22:56:55 -04:00
parent 59a18b8bc6
commit 89030c96b6
6 changed files with 1148 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engine.py
+261 -262
View File
@@ -91,7 +91,7 @@ class TechnicalEngine:
if "adx" not in disabled: self._compute_adx(df, result)
if "atr" not in disabled: self._compute_atr(df, result)
self._compute_macd_internal(df, result)
self._compute_macd(df, result)
if "divergence" not in disabled: self._detect_divergences(df, result)
if "structure" not in disabled: self._compute_structure_sr(df, result)
@@ -100,6 +100,10 @@ class TechnicalEngine:
# ── 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)
@@ -110,7 +114,87 @@ class TechnicalEngine:
# ══════════════════════════════════════════════════════════════
def _compute_ema_signals(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""EMA trend alignment analysis."""
"""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"]
@@ -123,8 +207,6 @@ class TechnicalEngine:
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
# 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:
@@ -155,7 +237,6 @@ class TechnicalEngine:
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]
@@ -187,10 +268,10 @@ class TechnicalEngine:
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()
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, np.nan)
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
@@ -227,8 +308,8 @@ class TechnicalEngine:
# Attach RSI series for divergence detection (dynamic attribute, filtered by _prefix)
result._rsi_series = rsi # type: ignore[attr-defined]
def _compute_macd_internal(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Compute MACD histogram internally for divergence detection (not a signal pillar)."""
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"]
@@ -240,13 +321,70 @@ class TechnicalEngine:
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()
std = close.rolling(c.bb_period).std(ddof=0)
upper = sma + c.bb_std * std
lower = sma - c.bb_std * std
@@ -308,67 +446,6 @@ class TechnicalEngine:
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 pd.isna(k_val):
k_val = 50.0
if pd.isna(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 pd.isna(prev_k) and not pd.isna(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=2.5,
)
def _compute_adx(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""ADX trend strength measurement."""
c = self.config
@@ -389,12 +466,12 @@ class TechnicalEngine:
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)
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() / (plus_di + minus_di).replace(0, np.nan)
adx = dx.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean()
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
@@ -439,7 +516,7 @@ class TechnicalEngine:
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 = 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
@@ -473,91 +550,6 @@ class TechnicalEngine:
weight=0.0, # ATR is context, not a directional signal
)
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
latest = close.iloc[-1]
tenkan_val = tenkan.iloc[-1]
kijun_val = kijun.iloc[-1]
# Current (unshifted) 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]
if pd.isna(sa) or pd.isna(sb) or pd.isna(tenkan_val) or pd.isna(kijun_val):
return
cloud_top = max(sa, sb)
cloud_bottom = min(sa, sb)
cloud_is_green = sa > sb
signals = []
score = 0
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)")
if tenkan_val > kijun_val:
signals.append("TK cross bullish")
score += 1
elif tenkan_val < kijun_val:
signals.append("TK cross bearish")
score -= 1
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=0.3,
)
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).
@@ -615,7 +607,7 @@ class TechnicalEngine:
name="Divergence", value=len(divergences),
signal=signal, confidence=70,
description=" | ".join(divergences),
weight=1.0,
weight=1.5,
)
# Note: divergences are NOT added to pattern_signals to avoid
# duplicate reason collection (the indicator entry covers them).
@@ -735,90 +727,96 @@ class TechnicalEngine:
name="Candlestick", value=float(bullish - bearish),
signal=signal, confidence=conf,
description=" | ".join(patterns),
weight=1.5,
weight=1.0,
)
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.8,
)
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 _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)."""
adx_sig = result.indicators.get("adx")
@@ -840,26 +838,29 @@ class TechnicalEngine:
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
_trending = {
"ema_trend": 2.0, "rsi": 1.5, "bollinger": 0.8,
"candlestick": 1.5, "divergence": 1.0,
}
regime_weights = {
"trending_up": _trending,
"trending_down": _trending,
"trending": _trending,
"ranging": {
"ema_trend": 0.5, "rsi": 2.0, "bollinger": 2.0,
"candlestick": 1.5, "divergence": 1.5,
},
}
# Use override weights if provided, else defaults
if self.config.regime_weights_override:
regime_weights = self.config.regime_weights_override
else:
_trending = {
"ema_trend": 2.0, "rsi": 1.5, "bollinger": 0.8,
"candlestick": 1.0, "divergence": 1.5,
}
regime_weights = {
"trending_up": _trending,
"trending_down": _trending,
"trending": _trending,
"ranging": {
"ema_trend": 0.5, "rsi": 2.0, "bollinger": 2.0,
"candlestick": 1.0, "divergence": 2.0,
},
}
active_regime_weights = regime_weights.get(result.market_regime, {})
@@ -871,7 +872,7 @@ class TechnicalEngine:
regime_mult = active_regime_weights.get(name, 1.0)
weight *= regime_mult
signal_score = ind.signal.value # -2 to +2
signal_score = ind.signal.value
weighted_score = signal_score * weight * (ind.confidence / 100.0)
total_score += weighted_score
@@ -879,23 +880,21 @@ class TechnicalEngine:
total_confidence += ind.confidence * weight
if total_weight > 0:
normalized_score = (total_score / total_weight) * 50 # scale to -100..+100
normalized_score = (total_score / total_weight) * 50
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:
if normalized_score >= 45:
result.composite_signal = SignalStrength.STRONG_BUY
elif normalized_score >= 15:
elif normalized_score >= 20:
result.composite_signal = SignalStrength.BUY
elif normalized_score <= -40:
elif normalized_score <= -45:
result.composite_signal = SignalStrength.STRONG_SELL
elif normalized_score <= -15:
elif normalized_score <= -20:
result.composite_signal = SignalStrength.SELL
else:
result.composite_signal = SignalStrength.NEUTRAL