fix: order block displacement/volume confirmation, Tweezer Bottom epsilon inconsistency

- indicator_service.py: _detect_order_blocks() now requires the
  confirming candles to actually displace price past the OB candle's own
  high/low (the "imbalance" a real SMC order block is supposed to leave
  behind) instead of accepting any 2-candle same-direction sequence, even
  one that stays entirely inside the OB candle's range. strength (1-3)
  now reflects real displacement/volume confirmation instead of always 1.
- indicator_service.py: removed a stray "+ 0.001" from Tweezer Bottom's
  zero-division guard that Tweezer Top's otherwise-identical formula
  didn't have -- both already guard via an explicit > 0 check, so the
  epsilon just made the two directions inconsistent for no reason.

237 backend tests pass (+5). This closes out the full 13-algorithm audit
started in v10 -- only (tt) parameter re-optimization (needs empirical
walk-forward runs, not a code change) and (yy) multi-position portfolio
backtesting (needs a new architecture) remain, both flagged as separate
larger investment decisions.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Le
2026-07-04 21:39:17 +07:00
parent 842631744c
commit 0a93af10a9
3 changed files with 203 additions and 9 deletions
+41 -9
View File
@@ -847,40 +847,67 @@ def _detect_order_blocks(
Returns list of order blocks, each with:
- type: \"BULLISH\" (support) or \"BEARISH\" (resistance)
- price_high, price_low
- strength: int (number of consecutive strong moves)
- strength: int (1-3; see below)
(fix vv) Previously ANY 2-candle same-direction sequence after the OB
candle counted, even if price never actually moved away from that
candle's own range — the defining feature of a real SMC order block
(the "imbalance" it leaves behind) wasn't checked at all, and
`strength` was hardcoded to 1 regardless of conviction. Now requires
the confirming move to displace price past the OB candle's high/low,
and strength reflects (a) that displacement exceeding the OB candle's
own range and (b) above-(local-)average volume on either confirming
candle. Volume only ever *boosts* strength — it's not a hard gate, so
a genuine displacement isn't rejected just because volume data happens
to look flat (e.g. synthetic/backtest data with uniform volume).
"""
n = len(candles)
if n < lookback + 3:
return []
window_start = max(0, n - lookback - 3)
volumes = [float(c.get("volume", 0.0)) for c in candles[window_start:n]]
avg_volume = sum(volumes) / len(volumes) if volumes else 0.0
blocks: list[dict] = []
for i in range(max(1, n - lookback), n - 2):
prev = candles[i - 1]
curr = candles[i]
nxt1 = candles[i + 1]
nxt2 = candles[i + 2]
curr_high = float(curr["high"])
curr_low = float(curr["low"])
curr_range = curr_high - curr_low
nxt1_close = float(nxt1["close"])
nxt2_close = float(nxt2["close"])
nxt1_vol = float(nxt1.get("volume", 0.0))
nxt2_vol = float(nxt2.get("volume", 0.0))
volume_confirmed = avg_volume > 0 and max(nxt1_vol, nxt2_vol) >= avg_volume * 1.2
# Bullish OB: bearish candle followed by 2 consecutive bullish candles
if float(curr["close"]) < float(curr["open"]) and nxt1_close > float(nxt1["open"]) and nxt2_close > float(nxt2["open"]):
# Bullish OB: bearish candle followed by 2 consecutive bullish
# candles that actually clear above the OB candle's own high.
displacement_up = nxt2_close - curr_high
if (float(curr["close"]) < float(curr["open"]) and nxt1_close > float(nxt1["open"])
and nxt2_close > float(nxt2["open"]) and curr_range > 0 and displacement_up > 0):
strength = 1 + int(volume_confirmed) + int(displacement_up > curr_range)
blocks.append({
"type": "BULLISH",
"price_high": curr_high,
"price_low": curr_low,
"strength": 1,
"strength": strength,
})
# Bearish OB: bullish candle followed by 2 consecutive bearish candles
if float(curr["close"]) > float(curr["open"]) and nxt1_close < float(nxt1["open"]) and nxt2_close < float(nxt2["open"]):
# Bearish OB: bullish candle followed by 2 consecutive bearish
# candles that actually clear below the OB candle's own low.
displacement_down = curr_low - nxt2_close
if (float(curr["close"]) > float(curr["open"]) and nxt1_close < float(nxt1["open"])
and nxt2_close < float(nxt2["open"]) and curr_range > 0 and displacement_down > 0):
strength = 1 + int(volume_confirmed) + int(displacement_down > curr_range)
blocks.append({
"type": "BEARISH",
"price_high": curr_high,
"price_low": curr_low,
"strength": 1,
"strength": strength,
})
# Keep only the most recent blocks
@@ -1571,8 +1598,13 @@ def detect_candlestick_patterns(candles: list[dict]) -> float:
if bull1 and bear0 and o0 > h1 and c0c < (o1 + c1c) / 2: _up(-1.5)
# ── Tweezer ──
# (fix ww) Tweezer Bottom's denominator had a stray `+ 0.001` not
# present in Tweezer Top's otherwise-identical formula just above —
# both already guard `l1 > 0 and l0 > 0` before dividing, so the
# epsilon was never needed to prevent a zero-division and just
# asymmetrically skewed the bottom's threshold vs. the top's.
if h1 > 0 and h0 > 0 and abs(h1 - h0) / max(h1, h0) * 100 < 0.1 and bull1 and bear0: _up(-1.5)
if l1 > 0 and l0 > 0 and abs(l1 - l0) / max(l1 + 0.001, l0) * 100 < 0.1 and bear1 and bull0: _up(1.5)
if l1 > 0 and l0 > 0 and abs(l1 - l0) / max(l1, l0) * 100 < 0.1 and bear1 and bull0: _up(1.5)
# ── Morning / Evening Star ──
if bear2 and bull0:
+82
View File
@@ -34,6 +34,8 @@ from app.services.indicator_service import (
_BOS_ATR_BUFFER_MULT,
detect_fvg,
ichimoku,
detect_candlestick_patterns,
_detect_order_blocks,
)
@@ -41,6 +43,10 @@ def candle(high, low, close, volume=100.0):
return {"high": high, "low": low, "close": close, "volume": volume}
def ohlc(open, high, low, close, volume=100.0):
return {"open": open, "high": high, "low": low, "close": close, "volume": volume}
class TestSma:
def test_leading_entries_are_none(self):
result = sma([1, 2, 3, 4, 5], period=3)
@@ -600,3 +606,79 @@ class TestIchimokuConfigurablePeriods:
i = 70
assert custom_result["tenkan"][i] != pytest.approx(default_result["tenkan"][i])
assert custom_result["kijun"][i] != pytest.approx(default_result["kijun"][i])
class TestCandlestickTweezerSymmetry:
"""Regression test for fix (ww): Tweezer Bottom's zero-division guard
had a stray `+ 0.001` epsilon that Tweezer Top's otherwise-identical
formula didn't have — both already guard against a zero denominator
via an explicit `> 0` check, so the epsilon was dead weight that made
the two directions inconsistent for no reason. Locks in that a small
(within-tolerance) gap is detected symmetrically in both directions.
"""
def test_tweezer_bottom_detects_small_matching_low_gap(self):
c2 = ohlc(100, 101, 99, 100)
c1 = ohlc(105, 106, 95.0, 96) # bearish, low=95.0
c0 = ohlc(96, 97, 95.02, 96.5) # bullish, low=95.02 -> gap ~0.021%
assert detect_candlestick_patterns([c2, c1, c0]) == pytest.approx(1.5)
def test_tweezer_top_detects_small_matching_high_gap(self):
c2 = ohlc(100, 101, 99, 100)
c1 = ohlc(95, 106.0, 94, 105) # bullish, high=106.0
c0 = ohlc(105, 105.98, 94, 95) # bearish, high=105.98 -> gap ~0.019%
assert detect_candlestick_patterns([c2, c1, c0]) == pytest.approx(-1.5)
class TestOrderBlockDisplacementAndVolume:
"""Regression tests for fix (vv): any 2-candle same-direction sequence
after the OB candle used to count as a valid order block, even if
price never actually cleared that candle's own high/low — the
"imbalance" a real SMC order block is supposed to leave behind wasn't
checked at all, and `strength` was hardcoded to 1. Now requires actual
displacement past the OB candle's range, and boosts (never gates on)
strength when volume and/or displacement are unusually large.
"""
def _candles_with_ob(self, confirm_high=None, confirm_low=None, volume_spike=False):
# Bullish OB candidate: bearish OB candle at some index, followed
# by 2 bullish confirming candles.
base = [ohlc(100, 101, 99, 100, volume=100) for _ in range(20)]
ob_candle = candle(105, 95, 96, volume=100) # bearish (open > close conceptually via high/low only used)
# candle() here only carries high/low/close/volume (no open) since
# _detect_order_blocks reads curr["open"]/["close"] — build OHLC
# explicitly instead for the 3 candles that matter.
ob = ohlc(104, 105, 95, 96, volume=100) # bearish: close(96) < open(104)
nxt1_close = confirm_high - 1 if confirm_high else 106
nxt1 = ohlc(97, nxt1_close + 1, 96, nxt1_close, volume=200 if volume_spike else 100)
nxt2_close = confirm_high if confirm_high else 107
nxt2 = ohlc(nxt1_close, nxt2_close + 1, nxt1_close - 1, nxt2_close, volume=200 if volume_spike else 100)
return base + [ob, nxt1, nxt2]
def test_bullish_ob_requires_displacement_past_ob_high(self):
# Confirming candles close ABOVE the OB candle's own high (105) -> valid.
candles = self._candles_with_ob(confirm_high=110)
blocks = _detect_order_blocks(candles, lookback=20)
assert any(b["type"] == "BULLISH" for b in blocks)
def test_bullish_ob_rejected_when_confirmation_stays_inside_ob_range(self):
# Confirming candles close ABOVE their own open (still "bullish"
# candles) but never clear the OB candle's high (105) — the old
# code accepted this; the fix requires real displacement.
base = [ohlc(100, 101, 99, 100, volume=100) for _ in range(20)]
ob = ohlc(104, 105, 95, 96, volume=100)
nxt1 = ohlc(96.2, 96.8, 96.0, 96.6, volume=100) # bullish but tiny, inside OB range
nxt2 = ohlc(96.6, 97.5, 96.5, 97.0, volume=100) # bullish but still under OB high (105)
candles = base + [ob, nxt1, nxt2]
blocks = _detect_order_blocks(candles, lookback=20)
assert not any(b["type"] == "BULLISH" for b in blocks)
def test_strength_increases_with_volume_confirmation(self):
candles_no_spike = self._candles_with_ob(confirm_high=110, volume_spike=False)
candles_with_spike = self._candles_with_ob(confirm_high=110, volume_spike=True)
blocks_no_spike = [b for b in _detect_order_blocks(candles_no_spike, lookback=20) if b["type"] == "BULLISH"]
blocks_with_spike = [b for b in _detect_order_blocks(candles_with_spike, lookback=20) if b["type"] == "BULLISH"]
assert blocks_no_spike and blocks_with_spike
assert blocks_with_spike[-1]["strength"] > blocks_no_spike[-1]["strength"]