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

Fixed code

Browse Source
This commit is contained in:
harry
2026-05-04 14:29:32 +00:00
parent 1637b93f1c
commit e4138ee60a
5 changed files with 908 additions and 889 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engine.py
+47 -54
View File
@@ -13,7 +13,6 @@ Design philosophy:
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Tuple
import numpy as np
@@ -54,7 +53,7 @@ class TechnicalAnalysis:
class TechnicalEngine:
"""
Core technical analysis engine.
Computes all indicators on a given OHLCV DataFrame and produces
a composite signal with confidence scoring.
"""
@@ -65,7 +64,7 @@ class TechnicalEngine:
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.
"""
@@ -123,7 +122,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
# Perfect alignment scoring
# Bullish: price > fast > medium > slow > trend
# Bearish: price < fast < medium < slow < trend
alignment_score = 0
@@ -190,7 +188,7 @@ class TechnicalEngine:
rsi = 100 - (100 / (1 + rs))
rsi_val = rsi.iloc[-1] if not rsi.empty else 50.0
if np.isnan(rsi_val):
if pd.isna(rsi_val):
rsi_val = 50.0
# Signal interpretation (context-aware)
@@ -220,12 +218,8 @@ class TechnicalEngine:
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
# 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 with histogram momentum analysis."""
@@ -293,8 +287,8 @@ class TechnicalEngine:
confidence=conf, description=desc, weight=1.1,
)
# Store for divergence
result._macd_histogram = histogram
# Attach histogram for divergence detection
result._macd_histogram = histogram # type: ignore[attr-defined]
def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Bollinger Bands position and squeeze detection."""
@@ -310,17 +304,24 @@ class TechnicalEngine:
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
# 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_position = (latest - lower_val) / (upper_val - lower_val) if (upper_val - lower_val) != 0 else 0.5
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 = ((upper - lower) / sma).rolling(20).mean().iloc[-1]
is_squeeze = band_width < avg_width * 0.75
else:
is_squeeze = False
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 ""
@@ -376,9 +377,9 @@ class TechnicalEngine:
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):
if pd.isna(k_val):
k_val = 50.0
if np.isnan(d_val):
if pd.isna(d_val):
d_val = 50.0
# Crossover detection
@@ -386,7 +387,7 @@ class TechnicalEngine:
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 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:
@@ -445,9 +446,13 @@ class TechnicalEngine:
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
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:
@@ -486,6 +491,8 @@ class TechnicalEngine:
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
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
@@ -513,7 +520,7 @@ class TechnicalEngine:
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
weight=0.0, # ATR is context, not a directional signal
)
def _compute_ichimoku(self, df: pd.DataFrame, result: TechnicalAnalysis):
@@ -532,32 +539,26 @@ class TechnicalEngine:
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)
# 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
# Signal analysis
signals = []
score = 0
# Price vs Cloud
if latest > cloud_top:
signals.append("Price ABOVE cloud")
score += 2
@@ -567,7 +568,6 @@ class TechnicalEngine:
else:
signals.append("Price IN cloud (indecision)")
# Tenkan vs Kijun
if tenkan_val > kijun_val:
signals.append("TK cross bullish")
score += 1
@@ -575,7 +575,6 @@ class TechnicalEngine:
signals.append("TK cross bearish")
score -= 1
# Cloud color
if cloud_is_green:
signals.append("Green cloud")
score += 1
@@ -612,7 +611,7 @@ class TechnicalEngine:
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
"""
@@ -635,16 +634,15 @@ class TechnicalEngine:
price_window = close.iloc[-lookback:]
ind_window = indicator_series.iloc[-lookback:]
# Find local swing lows (for bullish divergence)
# 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:
# 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)
# 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")
@@ -667,9 +665,10 @@ class TechnicalEngine:
name="Divergence", value=len(divergences),
signal=signal, confidence=70,
description=" | ".join(divergences),
weight=1.5, # high weight — divergences are strong signals
weight=1.5, # divergences are strong reversal signals
)
result.pattern_signals.extend(divergences)
# Note: divergences are NOT added to pattern_signals to avoid
# duplicate reason collection (the indicator entry covers them).
def _detect_candlestick_patterns(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Detect significant candlestick patterns on recent bars."""
@@ -838,7 +837,7 @@ class TechnicalEngine:
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)
@@ -860,17 +859,11 @@ class TechnicalEngine:
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
signal_score = ind.signal.value # -2 to +2
weighted_score = signal_score * weight * (ind.confidence / 100.0)
total_score += weighted_score
@@ -924,4 +917,4 @@ class TechnicalEngine:
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
return points