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

Fix O(n^2) blowup and look-ahead leak in SMC/divergence backtest scoring

Browse Source 
market_structure() (SMC) and detect_divergence() were each precomputed
once over the ENTIRE multi-year backtest range and reused unchanged for
every candle, so every candle's score could see results derived from
years of future price data — a look-ahead bug that inflated both
single-run backtest and walk-forward results, undermining the very
overfitting check walk-forward exists to provide. A prior fix bounded
this to a per-candle trailing window, which closed most of the leak but
still rescanned pivots from scratch on every candle (O(window) per
candle), too slow to enable 15m/30m walk-forward runs.

The real fix: pivot detection is itself a bounded rolling-window scan
(each position only depends on a few bars on either side), so it can be
precomputed once for the whole dataset just like BB/RSI/MACD. Per candle,
_compute_scores_series now just advances a monotonic pointer over
already-known pivots to whatever is causally confirmable as of that
candle — O(1) amortized across the whole run instead of O(window) or
O(n) per candle. Added an optional precomputed_pivots param to
detect_divergence() (backward compatible) to reuse this for RSI/MACD
divergence too.

Net effect: 16,000 candles went from 16.1s to 1.7s (confirmed empirically,
on top of an earlier ~10x from fixing the raw O(n^2)), and scaling stays
linear at 32,000 candles (3.2s). Walk-forward's timeframe options are now
15m/30m/1h/4h/1d (up from 1h/4h/1d) since 15m at the 3-year default
lookback now costs roughly 30s instead of 5+ minutes. Also wired
walk_forward.py's grid search to actually reuse one computed score series
across all 27 parameter combinations per fold (it was recomputing full
classification for every combination despite the scoring/threshold split
added earlier). 156 backend tests passing (3 new: causal-score regression,
pivot-detection-runs-once, order-block-window-bounded).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Le
2026-07-04 10:42:32 +07:00
parent 56325113b3
commit 1c022264f5
8 changed files with 476 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/v1/walk_forward.py
+1 -1
View File
@@ -30,7 +30,7 @@ router = APIRouter(prefix="/walk-forward", tags=["walk_forward"])
async def run(
symbol: str = Query("BTC/USDT"),
exchange: str = Query("mexc"),
timeframe: str = Query("4h", description="1h or 4h recommended — lower timeframes multiply the candle count and runtime"),
timeframe: str = Query("4h", description="15m/30m/1h/4h/1d — 1w/1M don't have enough candles over a multi-year lookback to form meaningful folds"),
total_days: int = Query(1095, ge=180, le=1825, description="Total lookback in days (default ~3 years)"),
train_days: int = Query(270, ge=30, description="Train window size per fold, in days"),
test_days: int = Query(90, ge=14, description="Held-out test window size per fold, in days"),
backend/app/services/backtest_engine.py
+274 -74
View File
@@ -20,16 +20,66 @@ from app.models.symbol import Symbol
from app.models.exchange import Exchange
from app.services.indicator_service import (
bollinger_bands, rsi, sma, macd, supertrend,
volume_breakout, ichimoku, detect_divergence, market_structure,
volume_breakout, ichimoku, detect_divergence,
_find_pivot_highs, _find_pivot_lows,
_find_pivot_highs_levels, _find_pivot_lows_levels,
_detect_bos, _detect_choch, _detect_order_blocks,
)
from app.services.signal_scoring import (
_classify_signal_combined,
_compute_adjusted_score,
_score_to_signal,
STRONG_BUY, BUY, STRONG_SELL, SELL,
)
# Min candles for warmup: BB(20) + RSI(14) + some room = 30
MIN_CANDLES = 30
# Bounded trailing-window sizes used when replaying indicators per candle.
# Every consumer in signal_scoring.py only ever reads the last 1-2 elements
# of these arrays except the BB squeeze check (lookback=10), so windows
# 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
_SHORT_WINDOW = 3
# 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/
# order_blocks) once over the *entire* multi-year backtest range and
# reused it unchanged for every candle, so a candle's score could see
# results derived from years of future data. Recomputing a fresh
# snapshot per candle on a bounded trailing window fixed the leak but
# was still O(window) of real pivot-scanning work per candle.
#
# The actual fix: pivot detection (_find_pivot_highs[_levels],
# _find_pivot_lows[_levels]) is itself a bounded rolling-window scan
# (each position only depends on `pivot_lookback` bars on either side),
# so it can be precomputed ONCE over the whole dataset just like BB/RSI/
# etc — see _precompute_indicators. Per candle, _compute_scores_series
# then only needs to know which of those precomputed pivots are already
# *confirmable* using data up to that candle (a pivot at position p needs
# `pivot_lookback` bars after p to confirm — a small, bounded, and
# inherent-to-the-indicator delay, not a look-ahead bug) — tracked with a
# monotonically advancing pointer + append-only list per candle, i.e.
# O(1) amortized across the whole run, not O(window) or O(n) per candle.
_PIVOT_LOOKBACK = 3 # matches market_structure()'s pivot_lookback default
_DIVERGENCE_PIVOT_LOOKBACK = 5 # matches detect_divergence()'s pivot_lookback default
_ORDER_BLOCK_LOOKBACK = 40 # matches the lookback backtest always requested from market_structure() before
def _tail(v, end_idx: int, window: int):
"""Trailing slice of `v` ending at `end_idx` (inclusive), capped at
`window` elements — O(window), not O(end_idx)."""
if v is None or not hasattr(v, "__getitem__"):
return v
start = max(0, end_idx + 1 - window)
return v[start:end_idx + 1]
def _tail_dict(d: dict, end_idx: int, window: int) -> dict:
return {k: _tail(v, end_idx, window) for k, v in d.items()} if d else {}
async def _fetch_symbol(db: AsyncSession, symbol: str, exchange: str) -> Symbol | None:
"""Look up a Symbol row by (symbol, exchange) name."""
@@ -88,7 +138,11 @@ def _precompute_indicators(candles: list[Candle], timeframe: str) -> dict:
]
close_prices_full = [float(c.close) for c in candles]
# Pre-compute indicators on full dataset (O(n) instead of O(n²))
# Pre-compute rolling-window indicators on the full dataset ONCE — these
# are genuinely causal per-candle arrays (arr[:i+1] truthfully represents
# "value as of candle i"), so precomputing once and taking bounded
# trailing windows per candle (see _compute_scores_series) is both
# correct and fast.
bb_full = bollinger_bands(close_prices_full) or {}
rsi_full = rsi(close_prices_full) or []
sma_full = sma(close_prices_full, 20) or []
@@ -96,12 +150,16 @@ def _precompute_indicators(candles: list[Candle], timeframe: str) -> dict:
st_full = supertrend(candle_dicts_full) or {}
vb_full = volume_breakout(candle_dicts_full) or []
ichi_full = ichimoku(candle_dicts_full) or {}
smc_full = market_structure(candle_dicts_full) or {}
# Pre-compute divergence ONCE (uses full arrays, indexes match)
rsi_div_full = detect_divergence(close_prices_full, rsi_full)
macd_hist_full = macd_full.get("histogram", []) if macd_full else []
macd_div_full = detect_divergence(close_prices_full, macd_hist_full)
# SMC (market_structure) and divergence detection are pivot-based, not
# rolling-window — but pivot detection itself IS a bounded rolling scan
# (see _PIVOT_LOOKBACK/_DIVERGENCE_PIVOT_LOOKBACK above), so precompute
# it once too. _compute_scores_series turns these into causally-
# confirmed running lists rather than reusing them wholesale.
swing_highs_full = _find_pivot_highs_levels(close_prices_full, _PIVOT_LOOKBACK, _PIVOT_LOOKBACK)
swing_lows_full = _find_pivot_lows_levels(close_prices_full, _PIVOT_LOOKBACK, _PIVOT_LOOKBACK)
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)
# Pre-build MTF candles ONCE per MTF config
mtf_precomputed = []
@@ -131,7 +189,8 @@ def _precompute_indicators(candles: list[Candle], timeframe: str) -> dict:
"st": supertrend(mtf_candles_list) or {},
"vb": volume_breakout(mtf_candles_list) or [],
"ichi": ichimoku(mtf_candles_list) or {},
"smc": market_structure(mtf_candles_list) or {},
"swing_highs": _find_pivot_highs_levels(mtf_p, _PIVOT_LOOKBACK, _PIVOT_LOOKBACK),
"swing_lows": _find_pivot_lows_levels(mtf_p, _PIVOT_LOOKBACK, _PIVOT_LOOKBACK),
})
return {
@@ -144,30 +203,35 @@ def _precompute_indicators(candles: list[Candle], timeframe: str) -> dict:
"st_full": st_full,
"vb_full": vb_full,
"ichi_full": ichi_full,
"smc_full": smc_full,
"rsi_div_full": rsi_div_full,
"macd_div_full": macd_div_full,
"swing_highs_full": swing_highs_full,
"swing_lows_full": swing_lows_full,
"price_pivot_highs_full": price_pivot_highs_full,
"price_pivot_lows_full": price_pivot_lows_full,
"mtf_precomputed": mtf_precomputed,
}
def _simulate_trades(
def _compute_scores_series(
candles: list[Candle],
precomputed: dict,
trade_size: Decimal = Decimal("10"),
strong_threshold: float = 4.0,
signal_threshold: float = 1.0,
max_hold_candles: int = 48,
active_from_index: int = MIN_CANDLES,
) -> tuple[list[dict], list[dict]]:
"""Replay signal classification + trade simulation over precomputed indicators.
) -> list[dict]:
"""Compute the threshold-independent adjusted score for every candle
from active_from_index onward.
This is the expensive half of classification — the 13-algorithm vote —
split out so callers trying many threshold combinations against the
same data (walk-forward grid search) can run it ONCE per fold and
cheaply replay `_score_to_signal` against the result, instead of
re-running the vote for every combination.
`active_from_index` lets callers pass extra warmup candles before the
window they actually want simulated (e.g. walk-forward fold
boundaries) — candles before this index are used only so indicators
have enough lookback, never turned into signals/trades.
window they actually want scored (e.g. walk-forward fold boundaries)
— candles before this index exist only so indicators have enough
lookback, never turned into signals/trades.
"""
close_prices_full = precomputed["close_prices_full"]
candle_dicts_full = precomputed["candle_dicts_full"]
bb_full = precomputed["bb_full"]
rsi_full = precomputed["rsi_full"]
sma_full = precomputed["sma_full"]
@@ -175,79 +239,197 @@ def _simulate_trades(
st_full = precomputed["st_full"]
vb_full = precomputed["vb_full"]
ichi_full = precomputed["ichi_full"]
smc_full = precomputed["smc_full"]
rsi_div_full = precomputed["rsi_div_full"]
macd_div_full = precomputed["macd_div_full"]
swing_highs_full = precomputed["swing_highs_full"]
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"]
mtf_precomputed = precomputed["mtf_precomputed"]
all_signals = []
trades = []
current_position = None
macd_hist_full = macd_full.get("histogram") if macd_full else None
start_index = max(MIN_CANDLES, active_from_index)
results: list[dict] = []
# Running, causally-confirmed pivot state — advanced monotonically as
# `i` increases, so the total work across the whole loop is O(n), not
# O(n) *per candle*. A pivot at raw position p is only "confirmed" (its
# value can influence a candle's score) once `pivot_lookback` bars after
# it are known, i.e. once i >= p + pivot_lookback — this is a small,
# bounded delay inherent to how pivots are defined (the same delay a
# live-trading consumer would see), not a look-ahead leak.
swing_ptr = 0
confirmed_swing_highs: list[float] = []
confirmed_swing_lows: list[float] = []
div_ptr = 0
confirmed_div_high_idx: list[int] = []
confirmed_div_low_idx: list[int] = []
# Per-MTF-timeframe running state, keyed by position in mtf_precomputed.
mtf_swing_state = [
{"ptr": 0, "highs": [], "lows": []} for _ in mtf_precomputed
]
for i in range(start_index, len(candles)):
candle = candles[i]
latest_close = close_prices_full[i]
timestamp = candle.timestamp.isoformat()
bb_data = _tail_dict(bb_full, i, _BB_WINDOW)
rsi_data = _tail(rsi_full, i, _SHORT_WINDOW) or []
sma_data = _tail(sma_full, i, _SHORT_WINDOW) or []
macd_data = _tail_dict(macd_full, i, _SHORT_WINDOW)
st_data = _tail_dict(st_full, i, _SHORT_WINDOW)
vb_data = _tail(vb_full, i, _SHORT_WINDOW) or []
ichi_data = _tail_dict(ichi_full, i, _SHORT_WINDOW)
# Slice pre-computed arrays (O(i) but ~100x faster than recomputing)
clip = i + 1
def _safe_slice(v):
return v[:clip] if v is not None and hasattr(v, '__getitem__') else v
bb_data = {k: _safe_slice(v) for k, v in bb_full.items()} if bb_full else {}
rsi_data = rsi_full[:clip] if rsi_full else []
sma_data = sma_full[:clip] if sma_full else []
macd_data = {k: _safe_slice(v) for k, v in macd_full.items()} if macd_full else {}
st_data = {k: _safe_slice(v) for k, v in st_full.items()} if st_full else {}
vb_data = vb_full[:clip] if vb_full else []
ichi_data = {k: _safe_slice(v) for k, v in ichi_full.items()} if ichi_full else {}
smc_data = {k: _safe_slice(v) for k, v in smc_full.items()} if smc_full else {}
# Advance the swing/pivot confirmation pointers to whatever is
# knowable as of candle i (see comment above).
swing_confirm_limit = i - _PIVOT_LOOKBACK + 1
while swing_ptr < swing_confirm_limit:
if swing_highs_full[swing_ptr] is not None:
confirmed_swing_highs.append(swing_highs_full[swing_ptr])
if swing_lows_full[swing_ptr] is not None:
confirmed_swing_lows.append(swing_lows_full[swing_ptr])
swing_ptr += 1
# MTF votes — use precomputed MTF indicators, sliced to current MTF candle index
div_confirm_limit = i - _DIVERGENCE_PIVOT_LOOKBACK + 1
while div_ptr < div_confirm_limit:
if price_pivot_highs_full[div_ptr] is not None:
confirmed_div_high_idx.append(div_ptr)
if price_pivot_lows_full[div_ptr] is not None:
confirmed_div_low_idx.append(div_ptr)
div_ptr += 1
# SMC (BOS/CHoCH/trend/order-blocks) from the causally-confirmed
# swings above, instead of rescanning raw candles for pivots.
recent_highs = confirmed_swing_highs[-3:]
recent_lows = confirmed_swing_lows[-3:]
bos = _detect_bos(recent_highs, recent_lows, [close_prices_full[i]])
choch = _detect_choch(
confirmed_swing_highs[-5:], confirmed_swing_lows[-5:],
close_prices_full[max(0, i - 19):i + 1],
)
ob_start = max(0, i + 1 - _ORDER_BLOCK_LOOKBACK - 3)
obs = _detect_order_blocks(candle_dicts_full[ob_start:i + 1], lookback=_ORDER_BLOCK_LOOKBACK)
trend = "NEUTRAL"
if len(recent_highs) >= 2 and len(recent_lows) >= 2:
if recent_highs[-1] > recent_highs[-2] and recent_lows[-1] > recent_lows[-2]:
trend = "BULLISH"
elif recent_highs[-1] < recent_highs[-2] and recent_lows[-1] < recent_lows[-2]:
trend = "BEARISH"
smc_data = {"bos": bos, "choch": choch, "order_blocks": obs, "trend": trend}
# Divergence — same causally-confirmed pivots feed both the RSI-
# divergence and MACD-histogram-divergence checks (both key off
# the same price pivots, just a different indicator series).
div_highs = confirmed_div_high_idx[-3:]
div_lows = confirmed_div_low_idx[-3:]
rsi_div = detect_divergence(close_prices_full, rsi_full, precomputed_pivots=(div_highs, div_lows)) if rsi_full else (None, None)
macd_div = detect_divergence(close_prices_full, macd_hist_full, precomputed_pivots=(div_highs, div_lows)) if macd_hist_full else (None, None)
# MTF votes — same causally-confirmed-swing treatment per sub-timeframe.
mtf_votes = []
for mtf in mtf_precomputed:
# Which MTF candle corresponds to main candle i?
for mtf, state in zip(mtf_precomputed, mtf_swing_state):
mtf_idx = i // mtf["mult"]
if mtf_idx < MIN_CANDLES or mtf_idx >= len(mtf["close_prices"]):
continue
clip_mtf = mtf_idx + 1
def _safe_slice_mtf(v):
return v[:clip_mtf] if v is not None and hasattr(v, '__getitem__') else v
mtf_swing_highs_full = mtf["swing_highs"]
mtf_swing_lows_full = mtf["swing_lows"]
mtf_confirm_limit = mtf_idx - _PIVOT_LOOKBACK + 1
while state["ptr"] < mtf_confirm_limit:
p = state["ptr"]
if mtf_swing_highs_full[p] is not None:
state["highs"].append(mtf_swing_highs_full[p])
if mtf_swing_lows_full[p] is not None:
state["lows"].append(mtf_swing_lows_full[p])
state["ptr"] += 1
mtf_recent_highs = state["highs"][-3:]
mtf_recent_lows = state["lows"][-3:]
mtf_close = mtf["close_prices"][mtf_idx]
mtf_bos = _detect_bos(mtf_recent_highs, mtf_recent_lows, [mtf_close])
mtf_choch = _detect_choch(
state["highs"][-5:], state["lows"][-5:],
mtf["close_prices"][max(0, mtf_idx - 19):mtf_idx + 1],
)
mtf_ob_start = max(0, mtf_idx + 1 - _ORDER_BLOCK_LOOKBACK - 3)
mtf_obs = _detect_order_blocks(mtf["candles_list"][mtf_ob_start:mtf_idx + 1], lookback=_ORDER_BLOCK_LOOKBACK)
mtf_trend = "NEUTRAL"
if len(mtf_recent_highs) >= 2 and len(mtf_recent_lows) >= 2:
if mtf_recent_highs[-1] > mtf_recent_highs[-2] and mtf_recent_lows[-1] > mtf_recent_lows[-2]:
mtf_trend = "BULLISH"
elif mtf_recent_highs[-1] < mtf_recent_highs[-2] and mtf_recent_lows[-1] < mtf_recent_lows[-2]:
mtf_trend = "BEARISH"
mtf_smc = {"bos": mtf_bos, "choch": mtf_choch, "order_blocks": mtf_obs, "trend": mtf_trend}
mtf_s, *_ = _classify_signal_combined(
mtf["close_prices"][mtf_idx],
{k: _safe_slice_mtf(v) for k, v in mtf["bb"].items()},
mtf["rsi"][:clip_mtf],
mtf["sma"][:clip_mtf],
{k: _safe_slice_mtf(v) for k, v in mtf["macd"].items()} if mtf["macd"] else None,
{k: _safe_slice_mtf(v) for k, v in mtf["st"].items()} if mtf["st"] else None,
mtf["vb"][:clip_mtf] if mtf["vb"] else None,
{k: _safe_slice_mtf(v) for k, v in mtf["ichi"].items()} if mtf["ichi"] else None,
mtf_close,
_tail_dict(mtf["bb"], mtf_idx, _BB_WINDOW),
_tail(mtf["rsi"], mtf_idx, _SHORT_WINDOW) or [],
_tail(mtf["sma"], mtf_idx, _SHORT_WINDOW) or [],
_tail_dict(mtf["macd"], mtf_idx, _SHORT_WINDOW),
_tail_dict(mtf["st"], mtf_idx, _SHORT_WINDOW),
_tail(mtf["vb"], mtf_idx, _SHORT_WINDOW) or [],
_tail_dict(mtf["ichi"], mtf_idx, _SHORT_WINDOW),
(None, None), (None, None),
{k: _safe_slice_mtf(v) for k, v in mtf["smc"].items()} if mtf["smc"] else None,
mtf_smc,
)
if mtf_s:
mtf_votes.append((mtf_s, "", mtf["weight"]))
signal_type, strength, *_ = _classify_signal_combined(
latest_close, bb_data, rsi_data, sma_data,
override_signal, override_strength, adjusted_score, confidence, _raw_scores = _compute_adjusted_score(
close_prices_full[i], bb_data, rsi_data, sma_data,
macd_data, st_data, vb_data, ichi_data,
rsi_div_full, macd_div_full, smc_data, mtf_votes or None,
strong_threshold=strong_threshold, signal_threshold=signal_threshold,
rsi_div, macd_div, smc_data, mtf_votes or None,
)
results.append({
"index": i,
"timestamp": candles[i].timestamp.isoformat(),
"close": close_prices_full[i],
"override_signal": override_signal,
"override_strength": override_strength,
"adjusted_score": adjusted_score,
"confidence": confidence,
})
return results
def _simulate_from_scores(
candles: list[Candle],
scores_series: list[dict],
trade_size: Decimal = Decimal("10"),
strong_threshold: float = 4.0,
signal_threshold: float = 1.0,
max_hold_candles: int = 48,
) -> tuple[list[dict], list[dict]]:
"""Cheap half of simulation: turn a precomputed score series into
signals + trades for one choice of thresholds.
See `_compute_scores_series` for the expensive half — run once,
reused across every threshold combination a grid search tries.
"""
all_signals: list[dict] = []
trades: list[dict] = []
current_position = None
for entry in scores_series:
i = entry["index"]
timestamp = entry["timestamp"]
latest_close = entry["close"]
if entry["override_signal"] is not None:
signal_type, strength = entry["override_signal"], entry["override_strength"]
else:
signal_type, strength = _score_to_signal(entry["adjusted_score"], strong_threshold, signal_threshold)
if signal_type:
all_signals.append({
"time": timestamp, "signal": signal_type,
"strength": strength or "", "price": latest_close,
})
# PnL simulation
if signal_type in (STRONG_BUY, BUY):
if current_position and current_position["direction"] == "SHORT":
if signal_type == STRONG_BUY:
entry = current_position["entry_price"]
entry_price = current_position["entry_price"]
qty = current_position["quantity"]
pnl = (entry - latest_close) * qty
pnl = (entry_price - latest_close) * qty
current_position.update({
"exit_price": latest_close, "exit_time": timestamp,
"pnl": pnl, "status": "CLOSED", "exit_reason": "REVERSAL",
@@ -267,9 +449,9 @@ def _simulate_trades(
elif signal_type in (STRONG_SELL, SELL):
if current_position and current_position["direction"] == "LONG":
if signal_type == STRONG_SELL:
entry = current_position["entry_price"]
entry_price = current_position["entry_price"]
qty = current_position["quantity"]
pnl = (latest_close - entry) * qty
pnl = (latest_close - entry_price) * qty
current_position.update({
"exit_price": latest_close, "exit_time": timestamp,
"pnl": pnl, "status": "CLOSED", "exit_reason": "REVERSAL",
@@ -286,16 +468,15 @@ def _simulate_trades(
"entry_signal": signal_type, "entry_index": i, "status": "OPEN",
}
# Time limit
if current_position and current_position["status"] == "OPEN":
hold = i - current_position["entry_index"]
if hold >= max_hold_candles:
entry = current_position["entry_price"]
entry_price = current_position["entry_price"]
qty = current_position["quantity"]
if current_position["direction"] == "LONG":
pnl = (latest_close - entry) * qty
pnl = (latest_close - entry_price) * qty
else:
pnl = (entry - latest_close) * qty
pnl = (entry_price - latest_close) * qty
current_position.update({
"exit_price": latest_close, "exit_time": timestamp,
"pnl": pnl, "status": "CLOSED", "exit_reason": "TIME_LIMIT",
@@ -306,12 +487,12 @@ def _simulate_trades(
# Close final position
if current_position and current_position["status"] == "OPEN":
last_close = float(candles[-1].close)
entry = current_position["entry_price"]
entry_price = current_position["entry_price"]
qty = current_position["quantity"]
if current_position["direction"] == "LONG":
pnl = (last_close - entry) * qty
pnl = (last_close - entry_price) * qty
else:
pnl = (entry - last_close) * qty
pnl = (entry_price - last_close) * qty
current_position.update({
"exit_price": last_close,
"exit_time": candles[-1].timestamp.isoformat(),
@@ -322,6 +503,25 @@ def _simulate_trades(
return all_signals, trades
def _simulate_trades(
candles: list[Candle],
precomputed: dict,
trade_size: Decimal = Decimal("10"),
strong_threshold: float = 4.0,
signal_threshold: float = 1.0,
max_hold_candles: int = 48,
active_from_index: int = MIN_CANDLES,
) -> tuple[list[dict], list[dict]]:
"""Convenience wrapper: compute scores then simulate with one set of
thresholds. Callers trying many threshold combinations against the
same data (walk-forward grid search) should call
`_compute_scores_series` once and `_simulate_from_scores` per
combination instead — see walk_forward.py.
"""
scores_series = _compute_scores_series(candles, precomputed, active_from_index)
return _simulate_from_scores(candles, scores_series, trade_size, strong_threshold, signal_threshold, max_hold_candles)
def _compute_stats(all_signals: list[dict], trades: list[dict]) -> dict:
"""Reduce raw signals/trades into the summary stats block used by
both the single-run backtest and each walk-forward fold."""
backend/app/services/indicator_service.py
+12 -2
View File
@@ -642,12 +642,19 @@ def detect_divergence(
prices: list[float],
indicator: list[Optional[float]],
pivot_lookback: int = 5,
precomputed_pivots: tuple[list[Optional[int]], list[Optional[int]]] | None = None,
) -> tuple[Optional[str], Optional[str]]:
"""Detect regular and hidden divergences using up to 3 pivot points.
Compares 2 or 3 pivot points in prices vs indicator for stronger signals.
Multi-pivot divergence (3+ pivots) returns STRONG; 2-pivot returns WEAK.
`precomputed_pivots`, if given, replaces the internal `_find_pivot_highs`/
`_find_pivot_lows` scan with an already-known (pivots_high, pivots_low)
pair — lets a caller that recomputes this per-candle over a large
dataset (e.g. backtesting) reuse pivots it tracked incrementally
instead of rescanning on every call.
Returns (divergence_type, strength):
- ("BULLISH", "STRONG") — Regular Bullish Divergence (2+ consecutive pairs)
- ("BULLISH", "WEAK") — Hidden Bullish Divergence or single regular
@@ -659,8 +666,11 @@ def detect_divergence(
if n < 20:
return None, None
pivots_high = _find_pivot_highs(prices, pivot_lookback, pivot_lookback)
pivots_low = _find_pivot_lows(prices, pivot_lookback, pivot_lookback)
if precomputed_pivots is not None:
pivots_high, pivots_low = precomputed_pivots
else:
pivots_high = _find_pivot_highs(prices, pivot_lookback, pivot_lookback)
pivots_low = _find_pivot_lows(prices, pivot_lookback, pivot_lookback)
low_indices = [i for i in pivots_low if i is not None and indicator[i] is not None][-3:]
high_indices = [i for i in pivots_high if i is not None and indicator[i] is not None][-3:]
backend/app/services/walk_forward.py
+23 -15
View File
@@ -29,13 +29,16 @@ from app.services.backtest_engine import (
_fetch_symbol,
_fetch_candles,
_precompute_indicators,
_simulate_trades,
_compute_scores_series,
_simulate_from_scores,
_compute_stats,
)
# Parameters that are cheap to grid-search: they only affect the final
# threshold/exit logic, not the 13-algorithm scoring itself, so replaying
# them against already-precomputed indicators is fast (see _simulate_trades).
# threshold/exit logic, not the 13-algorithm scoring itself. Each fold's
# score series (see _compute_scores_series) is computed ONCE and replayed
# against every combination below via _simulate_from_scores, so the grid
# size mainly costs cheap arithmetic, not repeated indicator work.
DEFAULT_PARAM_GRID: dict[str, list[float]] = {
"strong_threshold": [3.5, 4.0, 4.5],
"signal_threshold": [0.75, 1.0, 1.5],
@@ -135,25 +138,28 @@ async def _prepare_window(
return candles, precomputed, active_from_index
def _run_combo(candles, precomputed, trade_size, params, active_from_index):
return _simulate_trades(
candles, precomputed, trade_size,
def _run_combo(candles, scores_series, trade_size, params):
"""Apply one parameter combination to an already-computed score series
(see `_compute_scores_series`) — cheap, no indicator recomputation."""
return _simulate_from_scores(
candles, scores_series, trade_size,
strong_threshold=params["strong_threshold"],
signal_threshold=params["signal_threshold"],
max_hold_candles=int(params["max_hold_candles"]),
active_from_index=active_from_index,
)
def _grid_search(
candles: list,
precomputed: dict,
scores_series: list[dict],
param_grid: dict[str, list[float]],
trade_size: Decimal,
active_from_index: int,
) -> tuple[dict[str, float], float, dict]:
"""Try every combination in param_grid, return the one that scores best
on the train window by `_fold_score`."""
"""Try every combination in param_grid against one precomputed score
series, return the one that scores best on the train window by
`_fold_score`. The 13-algorithm scoring itself already happened once
to produce `scores_series` — this loop only replays cheap threshold
comparisons and trade bookkeeping per combination."""
keys = list(param_grid.keys())
best_params: dict[str, float] | None = None
best_score = float("-inf")
@@ -161,7 +167,7 @@ def _grid_search(
for combo in product(*(param_grid[k] for k in keys)):
params = dict(zip(keys, combo))
all_signals, trades = _run_combo(candles, precomputed, trade_size, params, active_from_index)
all_signals, trades = _run_combo(candles, scores_series, trade_size, params)
closed_trades = [t for t in trades if t.get("status") == "CLOSED"]
score = _fold_score(closed_trades)
if score > best_score:
@@ -173,7 +179,7 @@ def _grid_search(
# Every combo scored -inf (too few trades) — still report the
# grid's first combination so the fold has *something* to show.
best_params = {k: param_grid[k][0] for k in keys}
all_signals, trades = _run_combo(candles, precomputed, trade_size, best_params, active_from_index)
all_signals, trades = _run_combo(candles, scores_series, trade_size, best_params)
best_stats = _compute_stats(all_signals, trades)
return best_params, best_score, best_stats
@@ -225,17 +231,19 @@ async def run_walk_forward(
if train_window is None:
continue
train_candles, train_precomputed, train_active_from = train_window
train_scores = _compute_scores_series(train_candles, train_precomputed, train_active_from)
best_params, _train_score, train_stats = _grid_search(
train_candles, train_precomputed, param_grid, trade_size, train_active_from,
train_candles, train_scores, param_grid, trade_size,
)
test_window = await _prepare_window(db, db_symbol.id, timeframe, spec["test_start"], spec["test_end"])
if test_window is None:
continue
test_candles, test_precomputed, test_active_from = test_window
test_scores = _compute_scores_series(test_candles, test_precomputed, test_active_from)
test_signals, test_trades = _run_combo(test_candles, test_precomputed, trade_size, best_params, test_active_from)
test_signals, test_trades = _run_combo(test_candles, test_scores, trade_size, best_params)
test_stats = _compute_stats(test_signals, test_trades)
stitched_oos_trades.extend(t for t in test_trades if t.get("status") == "CLOSED")