Fix 3 engine bugs: swing-point NaN false positives, ROC div-by-zero, ADX epsilon
Ported from feature/eurusd-support (e6e09c5), pair-neutral fixes:
- _find_swing_points: all(<empty generator>) is vacuously True, so a value
surrounded by NaNs was reported as a swing point; now requires at least
one valid neighbor on each side
- ROC: guard against zero close at the lookback positions (inf scores)
- ADX: epsilon denominator instead of NaN-replace to avoid NaN cascade
Adds 8 regression tests (Module 9). 150/150 pass.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -492,7 +492,7 @@ class TechnicalEngine:
|
|||||||
plus_di = 100 * (plus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() / atr)
|
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)
|
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)
|
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 = 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
|
adx_val = adx.iloc[-1] if not adx.empty and not pd.isna(adx.iloc[-1]) else 0.0
|
||||||
@@ -602,8 +602,12 @@ class TechnicalEngine:
|
|||||||
if len(close) < n + 5:
|
if len(close) < n + 5:
|
||||||
return
|
return
|
||||||
|
|
||||||
roc_val = ((close.iloc[-1] - close.iloc[-n - 1]) / close.iloc[-n - 1]) * 100
|
denom1 = close.iloc[-n - 1]
|
||||||
roc_prev = ((close.iloc[-2] - close.iloc[-n - 2]) / close.iloc[-n - 2]) * 100
|
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):
|
if pd.isna(roc_val) or pd.isna(roc_prev):
|
||||||
return
|
return
|
||||||
@@ -1154,12 +1158,14 @@ class TechnicalEngine:
|
|||||||
for i in range(window, len(vals) - window):
|
for i in range(window, len(vals) - window):
|
||||||
if np.isnan(vals[i]):
|
if np.isnan(vals[i]):
|
||||||
continue
|
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 mode == "low":
|
||||||
if all(vals[i] <= vals[i-j] for j in range(1, window+1) if not np.isnan(vals[i-j])) and \
|
if all(vals[i] <= v for v in left_vals) and all(vals[i] <= v for v in right_vals):
|
||||||
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]))
|
points.append((i, vals[i]))
|
||||||
else:
|
else:
|
||||||
if all(vals[i] >= vals[i-j] for j in range(1, window+1) if not np.isnan(vals[i-j])) and \
|
if all(vals[i] >= v for v in left_vals) and all(vals[i] >= v for v in right_vals):
|
||||||
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]))
|
points.append((i, vals[i]))
|
||||||
return points
|
return points
|
||||||
|
|||||||
+79
@@ -1761,6 +1761,85 @@ def _():
|
|||||||
assert r1.market_regime == r2.market_regime
|
assert r1.market_regime == r2.market_regime
|
||||||
|
|
||||||
|
|
||||||
|
print("\n═══ Module 9: Engine Bug Fix Regressions ═══")
|
||||||
|
|
||||||
|
# ── 9a. Swing point NaN fix ──────────────────────────────────────────
|
||||||
|
|
||||||
|
@test("Swing points: all-NaN neighbors produces no false positives")
|
||||||
|
def _():
|
||||||
|
series = pd.Series([np.nan, np.nan, np.nan, 5.0, np.nan, np.nan, np.nan])
|
||||||
|
lows = TechnicalEngine._find_swing_points(series, mode="low", window=3)
|
||||||
|
highs = TechnicalEngine._find_swing_points(series, mode="high", window=3)
|
||||||
|
assert lows == [], f"Expected no lows with all-NaN neighbors, got {lows}"
|
||||||
|
assert highs == [], f"Expected no highs with all-NaN neighbors, got {highs}"
|
||||||
|
|
||||||
|
|
||||||
|
@test("Swing points: partial NaN neighbors still detect valid swings")
|
||||||
|
def _():
|
||||||
|
series = pd.Series([10.0, 8.0, np.nan, 4.0, np.nan, 8.0, 10.0])
|
||||||
|
lows = TechnicalEngine._find_swing_points(series, mode="low", window=2)
|
||||||
|
low_vals = [v for _, v in lows]
|
||||||
|
assert 4.0 in low_vals, f"Expected 4.0 as swing low, got {low_vals}"
|
||||||
|
|
||||||
|
|
||||||
|
@test("Swing points: empty series returns empty list")
|
||||||
|
def _():
|
||||||
|
series = pd.Series([], dtype=float)
|
||||||
|
assert TechnicalEngine._find_swing_points(series, mode="low") == []
|
||||||
|
assert TechnicalEngine._find_swing_points(series, mode="high") == []
|
||||||
|
|
||||||
|
|
||||||
|
@test("Swing points: monotonic increasing has no interior lows")
|
||||||
|
def _():
|
||||||
|
series = pd.Series([float(i) for i in range(20)])
|
||||||
|
lows = TechnicalEngine._find_swing_points(series, mode="low", window=2)
|
||||||
|
assert len(lows) == 0, f"Monotonic increase should have no interior lows, got {lows}"
|
||||||
|
|
||||||
|
|
||||||
|
@test("Swing points: V-shape detects low at bottom")
|
||||||
|
def _():
|
||||||
|
series = pd.Series([10.0, 8.0, 6.0, 4.0, 6.0, 8.0, 10.0])
|
||||||
|
lows = TechnicalEngine._find_swing_points(series, mode="low", window=2)
|
||||||
|
low_vals = [v for _, v in lows]
|
||||||
|
assert 4.0 in low_vals, f"V-shape should detect 4.0 as low, got {low_vals}"
|
||||||
|
|
||||||
|
|
||||||
|
@test("Swing points: inverted V-shape detects high at top")
|
||||||
|
def _():
|
||||||
|
series = pd.Series([4.0, 6.0, 8.0, 10.0, 8.0, 6.0, 4.0])
|
||||||
|
highs = TechnicalEngine._find_swing_points(series, mode="high", window=2)
|
||||||
|
high_vals = [v for _, v in highs]
|
||||||
|
assert 10.0 in high_vals, f"Inverted V should detect 10.0 as high, got {high_vals}"
|
||||||
|
|
||||||
|
|
||||||
|
# ── 9b. ROC div-by-zero fix ──────────────────────────────────────────
|
||||||
|
|
||||||
|
@test("ROC: zero close at lookback position does not produce inf")
|
||||||
|
def _():
|
||||||
|
df = make_ohlcv(n=100, seed=42)
|
||||||
|
df.iloc[-11, df.columns.get_loc("Close")] = 0.0
|
||||||
|
engine = TechnicalEngine(PlatformConfig().technical)
|
||||||
|
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||||||
|
if "roc" in result.indicators:
|
||||||
|
assert np.isfinite(result.indicators["roc"].value), "ROC should not be inf"
|
||||||
|
|
||||||
|
|
||||||
|
# ── 9c. ADX epsilon fix ──────────────────────────────────────────────
|
||||||
|
|
||||||
|
@test("ADX: flat data produces finite ADX value")
|
||||||
|
def _():
|
||||||
|
n = 100
|
||||||
|
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||||||
|
df = pd.DataFrame({
|
||||||
|
"Open": [100.0]*n, "High": [100.01]*n, "Low": [99.99]*n,
|
||||||
|
"Close": [100.0]*n, "Volume": [1000.0]*n,
|
||||||
|
}, index=dates)
|
||||||
|
engine = TechnicalEngine(PlatformConfig().technical)
|
||||||
|
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||||||
|
if "adx" in result.indicators:
|
||||||
|
assert np.isfinite(result.indicators["adx"].value), "ADX should be finite on flat data"
|
||||||
|
|
||||||
|
|
||||||
# ══════════════════════════════════════════════════════════════════════
|
# ══════════════════════════════════════════════════════════════════════
|
||||||
# SUMMARY
|
# SUMMARY
|
||||||
# ══════════════════════════════════════════════════════════════════════
|
# ══════════════════════════════════════════════════════════════════════
|
||||||
|
|||||||
Reference in New Issue
Block a user