hanlap/test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Wire ADX/regime detection into shared signal filter, unify the two regime systems

Browse Source 
The market-regime filter (suppress directional signals in choppy/sideways
conditions, downgrade STRONG signals in volatile ones) already existed,
but only as a manual post-classification check inside signal_service.py's
live-trading path. Backtest and walk-forward called signal_scoring.py's
classifier directly, bypassing it entirely — so backtest results
systematically overestimated trade frequency and risk exposure relative
to what live trading actually does.

Factored the filter into a shared _apply_regime_filter() in
signal_scoring.py and added an optional market_regime param to
_classify_signal_combined() (default None preserves existing behavior for
every other caller, e.g. MTF sub-votes). backtest_engine.py now precomputes
adx()/atr() alongside the other rolling-window indicators and computes
detect_market_regime() per candle with the same bounded-window formula
live trading uses (candle_service.py), applying the filter once a
threshold combo resolves a concrete signal type. signal_service.py now
passes its regime into the shared classifier instead of duplicating the
check.

Also found and fixed a second, independent regime system: close_stale_trades()
(SL/TP sizing for open trades) computed its own cruder ATR-percentile-only
regime bucketing, which could disagree with the ADX+BB+Choppiness+
Efficiency-Ratio classification used for entry filtering. It now reads
market_regime from get_indicators() — the same TTL-cached function the
live signal-generation loop already populates — so entry filtering and
exit sizing agree on what "volatile" or "choppy" means for a given symbol.

14 new tests (unit tests for _apply_regime_filter, backtest integration
tests proving the filter suppresses/allows trades by regime, and a
signal_service test confirming close_stale_trades sources its regime from
the shared cache). 170 backend tests passing, frontend build clean.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Le
2026-07-04 11:25:43 +07:00
parent 1c022264f5
commit 0e77c055ba
7 changed files with 381 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/backtest_engine.py
+40 -1
View File
@@ -24,11 +24,13 @@ from app.services.indicator_service import (
_find_pivot_highs, _find_pivot_lows,
_find_pivot_highs_levels, _find_pivot_lows_levels,
_detect_bos, _detect_choch, _detect_order_blocks,
adx, atr, detect_market_regime,
)
from app.services.signal_scoring import (
_classify_signal_combined,
_compute_adjusted_score,
_score_to_signal,
_apply_regime_filter,
STRONG_BUY, BUY, STRONG_SELL, SELL,
)
@@ -41,8 +43,13 @@ MIN_CANDLES = 30
# a little larger than that are enough — the point is each candle's cost
# becomes O(window), not O(candle_index), which is what made the old
# "slice the whole precomputed array up to now" approach O(n^2) overall.
_BB_WINDOW = 20
_BB_WINDOW = 25 # >= detect_market_regime()'s own bb-squeeze lookback (20), with margin
_SHORT_WINDOW = 3
# detect_market_regime() also calls efficiency_ratio()/choppiness_index()
# internally (period=14) on whatever prices/highs/lows it's given — pass a
# bounded tail, not the whole growing history, for the same O(n), not
# O(n^2), reason as everything else in this module.
_REGIME_WINDOW = 30
# SMC (market_structure) and divergence detection are built from pivot
# points (local highs/lows), not rolling windows — the old approach
# precomputed a single current-state snapshot (bos/choch/trend/
@@ -161,6 +168,13 @@ def _precompute_indicators(candles: list[Candle], timeframe: str) -> dict:
price_pivot_highs_full = _find_pivot_highs(close_prices_full, _DIVERGENCE_PIVOT_LOOKBACK, _DIVERGENCE_PIVOT_LOOKBACK)
price_pivot_lows_full = _find_pivot_lows(close_prices_full, _DIVERGENCE_PIVOT_LOOKBACK, _DIVERGENCE_PIVOT_LOOKBACK)
# ADX + ATR feed detect_market_regime() — the same regime filter live
# trading applies (candle_service.py) so backtest results reflect the
# signals live trading would actually have acted on, not the larger
# set every one of the 13 algorithms would produce unfiltered.
adx_full = adx(candle_dicts_full, period=14) or {"adx": [], "plus_di": [], "minus_di": []}
atr_full = atr(candle_dicts_full, period=14) or []
# Pre-build MTF candles ONCE per MTF config
mtf_precomputed = []
for mtf_name, mtf_mult, mtf_w in mtf_config:
@@ -207,6 +221,8 @@ def _precompute_indicators(candles: list[Candle], timeframe: str) -> dict:
"swing_lows_full": swing_lows_full,
"price_pivot_highs_full": price_pivot_highs_full,
"price_pivot_lows_full": price_pivot_lows_full,
"adx_full": adx_full,
"atr_full": atr_full,
"mtf_precomputed": mtf_precomputed,
}
@@ -243,6 +259,8 @@ def _compute_scores_series(
swing_lows_full = precomputed["swing_lows_full"]
price_pivot_highs_full = precomputed["price_pivot_highs_full"]
price_pivot_lows_full = precomputed["price_pivot_lows_full"]
adx_full = precomputed["adx_full"]
atr_full = precomputed["atr_full"]
mtf_precomputed = precomputed["mtf_precomputed"]
macd_hist_full = macd_full.get("histogram") if macd_full else None
@@ -377,6 +395,23 @@ def _compute_scores_series(
rsi_div, macd_div, smc_data, mtf_votes or None,
)
# Market regime — same multi-factor classification live trading
# uses (candle_service.py) to filter/downgrade directional signals
# in choppy/sideways/volatile conditions (see _apply_regime_filter
# in signal_scoring.py, applied in _simulate_from_scores once the
# threshold combo decides a concrete signal_type).
adx_data = _tail_dict(adx_full, i, _SHORT_WINDOW)
atr_tail = _tail(atr_full, i, _SHORT_WINDOW)
last_atr = atr_tail[-1] if atr_tail else None
atr_pct = (last_atr / close_prices_full[i] * 100.0) if last_atr and close_prices_full[i] > 0 else None
regime_start = max(0, i + 1 - _REGIME_WINDOW)
market_regime = detect_market_regime(
adx_data, bb_data, atr_pct, vb_data,
prices=close_prices_full[regime_start:i + 1],
highs=[c["high"] for c in candle_dicts_full[regime_start:i + 1]],
lows=[c["low"] for c in candle_dicts_full[regime_start:i + 1]],
)
results.append({
"index": i,
"timestamp": candles[i].timestamp.isoformat(),
@@ -385,6 +420,7 @@ def _compute_scores_series(
"override_strength": override_strength,
"adjusted_score": adjusted_score,
"confidence": confidence,
"market_regime": market_regime,
})
return results
@@ -417,6 +453,9 @@ def _simulate_from_scores(
signal_type, strength = entry["override_signal"], entry["override_strength"]
else:
signal_type, strength = _score_to_signal(entry["adjusted_score"], strong_threshold, signal_threshold)
signal_type, strength, _confidence = _apply_regime_filter(
signal_type, strength, entry["confidence"], entry.get("market_regime"),
)
if signal_type:
all_signals.append({
backend/app/services/signal_scoring.py
+51
View File
@@ -522,6 +522,50 @@ def _score_to_signal(
return None, None
_DIRECTIONAL_SIGNALS = {STRONG_BUY, BUY, STRONG_SELL, SELL}
def _apply_regime_filter(
signal_type: Optional[str],
strength: Optional[str],
confidence: float,
market_regime: Optional[str],
) -> tuple[Optional[str], Optional[str], float]:
"""Suppress or downgrade directional signals based on market regime.
Choppy/sideways markets (low ADX, high Choppiness Index, low
Efficiency Ratio — see `indicator_service.detect_market_regime`)
generate false trend-following signals, so directional signals are
suppressed entirely. Volatile markets aren't wrong, just risky —
STRONG signals get downgraded and confidence is capped.
This used to live only in signal_service.py's live-trading path,
which meant backtest/walk-forward never applied it and so
systematically overestimated trade frequency and risk exposure
relative to what live trading actually does. Factored out here so
every caller — live and backtest alike — applies the same rule.
Non-directional results (None, or the CAUTION_*/SQUEEZE_ALERT
overrides) pass through unchanged — regime only governs whether/how
strongly to act on a directional BUY/SELL call.
"""
if signal_type not in _DIRECTIONAL_SIGNALS:
return signal_type, strength, confidence
if market_regime in ("choppy", "sideways"):
return None, None, confidence
if market_regime == "volatile":
confidence = min(confidence, 0.3)
strength = "WEAK"
if signal_type == STRONG_BUY:
signal_type = BUY
elif signal_type == STRONG_SELL:
signal_type = SELL
return signal_type, strength, confidence
def _classify_signal_combined(
close_price: float,
bb: dict[str, list[float]],
@@ -544,6 +588,7 @@ def _classify_signal_combined(
enabled_strategies: list[str] | None = None,
strong_threshold: float = 4.0,
signal_threshold: float = 1.0,
market_regime: str | None = None,
) -> tuple[Optional[str], Optional[str], float, dict[str, float]]:
"""Classify market state using 13-algorithm voting with win-rate boosting.
@@ -569,6 +614,11 @@ def _classify_signal_combined(
— left at their defaults for live trading; walk-forward backtesting
overrides them during parameter search.
*market_regime*, if given, suppresses directional signals in choppy/
sideways conditions and downgrades them in volatile ones — see
`_apply_regime_filter`. Left at its default (None = no filtering) for
callers that don't have regime data (e.g. MTF sub-timeframe votes).
Returns (signal_type, strength, confidence, raw_scores) where
confidence is a 0-1 float and raw_scores is a dict of all 9
algorithm scores for ML feature collection.
@@ -582,6 +632,7 @@ def _classify_signal_combined(
return override_signal, override_strength, confidence, raw_scores
signal_type, strength = _score_to_signal(adjusted_score, strong_threshold, signal_threshold)
signal_type, strength, confidence = _apply_regime_filter(signal_type, strength, confidence, market_regime)
return signal_type, strength, confidence, raw_scores
backend/app/services/signal_service.py
+24 -42
View File
@@ -302,43 +302,23 @@ async def _do_analysis(
market_regime: str | None = indicators.get("market_regime")
logger.debug("Market regime for %s: %s", symbol, market_regime)
# Classify signal using combined 13-algorithm voting with win-rate boosting
# Classify signal using combined 13-algorithm voting with win-rate
# boosting. market_regime suppresses choppy/sideways directional
# signals and downgrades volatile ones — see _apply_regime_filter in
# signal_scoring.py. This used to be a manual post-hoc check here,
# only applied to live trading; it's now inside the shared classifier
# so backtest/walk-forward apply the identical rule instead of
# over-estimating trade frequency relative to what live trading does.
signal_type, strength, confidence, algo_scores = _classify_signal_combined(
latest_close, bb, rsi, sma, macd_data, st_data, vol_data, ichi_data,
rsi_div, macd_div, smc_data, mtf_votes, obv_data,
stoch_rsi_data, mfi_data, fvg_data, candlestick_score, rates,
enabled_strategies=enabled_strategies,
market_regime=market_regime,
)
if signal_type is None:
return
# ── Market Regime signal filter ──
directional_signals = {STRONG_BUY, BUY, STRONG_SELL, SELL}
if market_regime == "choppy" and signal_type in directional_signals:
logger.debug(
"Suppressing %s signal for %s — choppy market, avoid trading",
signal_type, symbol,
)
return
elif market_regime == "sideways" and signal_type in directional_signals:
logger.debug(
"Suppressing %s signal for %s — market regime is sideways",
signal_type, symbol,
)
return
elif market_regime == "volatile" and signal_type in directional_signals:
# Downgrade ALL directional signals in volatile conditions
confidence = min(confidence, 0.3)
strength = "WEAK"
if signal_type == STRONG_BUY:
signal_type = BUY
elif signal_type == STRONG_SELL:
signal_type = SELL
logger.debug(
"Downgraded %s → %s for %s — volatile market regime",
signal_type, signal_type if signal_type in (BUY, SELL) else "BUY/SELL", symbol,
)
# Build indicators snapshot
snapshot = {
"close": latest_close,
@@ -1257,20 +1237,22 @@ async def close_stale_trades(
# SL/TP are prices, convert to percentage distance from entry
from app.services.risk_manager import AdaptiveSLTPOptimizer
sltp = AdaptiveSLTPOptimizer()
# Detect regime from ATR percentile (dynamic, not hardcoded)
if atr_pct is not None:
atr_f = float(atr_pct)
if atr_f > 8:
regime = "volatile"
elif atr_f > 5:
regime = "trending"
elif atr_f < 1.5:
regime = "sideways"
elif atr_f < 0.8:
regime = "choppy"
else:
regime = "neutral"
else:
# Regime from the same multi-factor detector live
# signal generation uses (ADX + BB + ATR% + Choppiness
# + Efficiency Ratio — see indicator_service.
# detect_market_regime), not a standalone ATR-percentile
# heuristic. Keeps SL/TP sizing consistent with
# whatever regime allowed/filtered the entry signal in
# the first place, instead of two independent
# definitions of "volatile"/"choppy"/etc that could
# disagree. get_indicators() is TTL-cached per
# (symbol, exchange, timeframe), so this is usually a
# cache hit shared with the live signal-generation
# loop, not a fresh recompute.
try:
cached_indicators = await get_indicators(db, trade.symbol, trade.exchange, trade.timeframe)
regime = cached_indicators.get("market_regime") or "neutral"
except Exception:
regime = "neutral"
result = sltp.compute_sl_tp(
atr=float(latest_atr),