dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Test clean S/R as production rank overlay

Browse Source
This commit is contained in:
dennisthiessen
2026-07-13 10:33:42 +02:00
parent 2e9afeec0f
commit 01e6f7e2c3
6 changed files with 327 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/backtest_service.py
+173 -13
View File
@@ -94,6 +94,12 @@ HORIZON = 30 # trading days to resolve an outcome (matches the evaluat
ATR_MULTIPLIER = 1.5
RANGE_FACTOR_LOOKBACK = 504
RANGE_FACTOR_MIN_LOG = 1.0 # approximately a 2.7x high/low span
STRUCTURAL_OVERLAY_VARIANT = "production_structural_overlay"
STRUCTURAL_OVERLAY_SOURCE_VARIANT = (
"rewrite_range504_structural_legacy_primary"
)
STRUCTURAL_OVERLAY_WEIGHT = 0.05
STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score"
RANGE_RESIDUAL_VARIANTS = {
"rewrite_range504_structural_legacy_primary",
"rewrite_range504_structural_primary2",
@@ -157,6 +163,7 @@ SR_RESEARCH_VARIANTS = {
"legacy_traffic_grid_only",
"legacy_range_grid_touch",
"legacy_range_grid_neutral",
STRUCTURAL_OVERLAY_VARIANT,
*RANGE_FACTOR_VARIANTS,
}
@@ -172,7 +179,7 @@ def _sr_research_variant() -> str:
def _sr_detector_variant(sr_variant: str) -> str:
"""Map factor-gated research arms to the detector they hold fixed."""
if sr_variant == "production_range504":
if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}:
return "production_control"
if (
sr_variant == "rewrite_range504_legacy_primary"
@@ -206,7 +213,11 @@ def _range_factor_allows(sr_variant: str, range_504_log: float) -> bool:
def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float:
"""Preserve the deployed selector except in explicit primary-2 research."""
if (
sr_variant in {"production_control", "production_range504"}
sr_variant in {
"production_control",
"production_range504",
STRUCTURAL_OVERLAY_VARIANT,
}
or sr_variant.endswith("_legacy_primary")
or sr_variant.startswith("legacy_")
):
@@ -317,6 +328,8 @@ def _window_setups(
window_records: list,
config: dict,
activation: dict,
*,
sr_variant: str | None = None,
) -> list[dict]:
"""Rebuild the setup(s) at the last bar of ``window_records`` (the as-of date),
using only those bars. Returns one dict per tradeable direction."""
@@ -343,7 +356,7 @@ def _window_setups(
if atr <= 0:
return []
sr_variant = _sr_research_variant()
sr_variant = sr_variant or _sr_research_variant()
detector_variant = _sr_detector_variant(sr_variant)
range_504_log = _range_504_log(highs, lows)
if sr_variant == "legacy_geometry_neutral":
@@ -508,6 +521,55 @@ def _window_setups(
return out
def _structural_overlay_window_setups(
window_records: list,
config: dict,
activation: dict,
) -> list[dict]:
"""Production setups tagged by the clean structural/range candidate.
The returned population and setup geometry remain production-identical.
The clean detector is only a point-in-time feature, so this arm can test a
ranking overlay without silently changing admission breadth or targets.
"""
production = _window_setups(
window_records,
config,
activation,
sr_variant="production_control",
)
if not production:
return []
structural = _window_setups(
window_records,
config,
activation,
sr_variant=STRUCTURAL_OVERLAY_SOURCE_VARIANT,
)
structural_by_direction = {row["direction"]: row for row in structural}
tagged: list[dict] = []
for production_row in production:
row = dict(production_row)
structural_row = structural_by_direction.get(row["direction"])
row["sr_variant"] = STRUCTURAL_OVERLAY_VARIANT
row["structural_overlay_pass"] = bool(
structural_row and structural_row.get("meets_core")
)
row["structural_overlay_rr"] = (
float(structural_row["rr"]) if structural_row is not None else None
)
row["structural_overlay_sources"] = (
list(structural_row.get("primary_sources") or [])
if structural_row is not None else []
)
row["structural_overlay_gate_level_count"] = (
int(structural_row.get("gate_level_count", 0) or 0)
if structural_row is not None else 0
)
tagged.append(row)
return tagged
def _stop_fill_r(direction: str, entry: float, stop: float, bar) -> float:
"""Realized R when the stop is hit on ``bar``: filled at the stop, or at the
bar's open when price gapped through it — so a gap can lose more than −1R,
@@ -595,6 +657,7 @@ def _replay_ticker(
return candidates
entry_start, entry_end = _backtest_entry_bounds()
sr_variant = _sr_research_variant()
for i in range(MIN_LOOKBACK - 1, n - HORIZON, STEP_DAYS):
as_of = records[i].date
if entry_start is not None and as_of < entry_start:
@@ -611,7 +674,17 @@ def _replay_ticker(
)
vol_6m = _realized_vol_6m(closes, len(window) - 1)
for s in _window_setups(window, config, activation):
setups = (
_structural_overlay_window_setups(window, config, activation)
if sr_variant == STRUCTURAL_OVERLAY_VARIANT
else _window_setups(
window,
config,
activation,
sr_variant=sr_variant,
)
)
for s in setups:
outcome, outcome_date = evaluate_setup_against_bars(
s["direction"], s["stop"], s["target"], forward_bars, HORIZON
)
@@ -670,6 +743,12 @@ def _replay_ticker(
"range_504_log": s["range_504_log"],
"range_504_ratio": s["range_504_ratio"],
"range_factor_pass": s["range_factor_pass"],
"structural_overlay_pass": s.get("structural_overlay_pass"),
"structural_overlay_rr": s.get("structural_overlay_rr"),
"structural_overlay_sources": s.get("structural_overlay_sources"),
"structural_overlay_gate_level_count": s.get(
"structural_overlay_gate_level_count"
),
"outcome": outcome,
"target_hit": target_hit,
"realized_r": realized_r,
@@ -740,6 +819,8 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
raw_counts: list[int] = []
gate_counts: list[int] = []
range_logs: list[float] = []
overlay_rows = 0
overlay_pass = 0
for cand in candidates:
sources = list(cand.get("primary_sources") or [])
for source in sources:
@@ -752,6 +833,9 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
raw_counts.append(int(cand.get("raw_level_count", 0) or 0))
gate_counts.append(int(cand.get("gate_level_count", 0) or 0))
range_logs.append(float(cand.get("range_504_log", 0.0) or 0.0))
if cand.get("structural_overlay_pass") is not None:
overlay_rows += 1
overlay_pass += int(bool(cand["structural_overlay_pass"]))
def avg(values: list[float] | list[int]) -> float | None:
return round(sum(values) / len(values), 3) if values else None
@@ -771,6 +855,11 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
"range_factor_pass": sum(
1 for value in range_logs if value >= RANGE_FACTOR_MIN_LOG
),
"structural_overlay_rows": overlay_rows,
"structural_overlay_pass": overlay_pass,
"structural_overlay_weight": (
STRUCTURAL_OVERLAY_WEIGHT if overlay_rows else None
),
}
@@ -798,6 +887,25 @@ def _sr_candidate_audit(candidates: list[dict], min_percentile: float) -> list[d
"meets_core": bool(cand.get("meets_core")),
"momentum_percentile": round(float(percentile), 6),
"strategy_rank": round(float(cand.get(RESIDUAL_HIGH_VOL_BLEND_KEY, 0.0) or 0.0), 6),
"production_rank": round(
float(cand.get(RESIDUAL_HIGH_VOL_BLEND_80_20_KEY, 0.0) or 0.0),
6,
),
"structural_overlay_pass": cand.get("structural_overlay_pass"),
"structural_overlay_score": (
round(float(cand[STRUCTURAL_OVERLAY_SCORE_KEY]), 6)
if cand.get(STRUCTURAL_OVERLAY_SCORE_KEY) is not None else None
),
"structural_overlay_rr": (
round(float(cand["structural_overlay_rr"]), 6)
if cand.get("structural_overlay_rr") is not None else None
),
"structural_overlay_sources": list(
cand.get("structural_overlay_sources") or []
),
"structural_overlay_gate_level_count": int(
cand.get("structural_overlay_gate_level_count", 0) or 0
),
"rr": round(float(cand.get("rr", 0.0)), 6),
"primary_prob": round(float(cand.get("primary_prob", 0.0)), 6),
"primary_sources": list(cand.get("primary_sources") or []),
@@ -1333,6 +1441,23 @@ def _assign_residual_high_vol_blend(candidates: list[dict]) -> None:
)
def _assign_structural_overlay_score(candidates: list[dict]) -> None:
"""Conservative rank nudge for production setups confirmed by clean S/R."""
for cand in candidates:
if cand.get("structural_overlay_pass") is None:
cand[STRUCTURAL_OVERLAY_SCORE_KEY] = None
continue
production_rank = cand.get(RESIDUAL_HIGH_VOL_BLEND_80_20_KEY)
if production_rank is None:
cand[STRUCTURAL_OVERLAY_SCORE_KEY] = None
continue
structural_score = 100.0 if cand["structural_overlay_pass"] else 0.0
cand[STRUCTURAL_OVERLAY_SCORE_KEY] = (
float(production_rank) * (1.0 - STRUCTURAL_OVERLAY_WEIGHT)
+ structural_score * STRUCTURAL_OVERLAY_WEIGHT
)
def _momentum_qualifies(cand: dict, threshold: float) -> bool:
"""Whether a candidate clears the floors (meets_core) and the momentum gate.
Threshold 0 disables the momentum gate (floors only). The gate is long-only:
@@ -2185,6 +2310,7 @@ PORTFOLIO_MONITOR_LOOKBACKS: tuple[dict, ...] = (
)
PRODUCTION_PORTFOLIO_STRATEGY = "residual80_highvol80_20_atr3"
STRUCTURAL_OVERLAY_PORTFOLIO_STRATEGY = "production_structural_overlay5_atr3"
PORTFOLIO_MONITOR_STRATEGIES: tuple[dict, ...] = (
{
"strategy": "legacy_residual80_hold",
@@ -2218,6 +2344,25 @@ PORTFOLIO_MONITOR_STRATEGIES: tuple[dict, ...] = (
)
def _portfolio_monitor_strategies() -> tuple[dict, ...]:
"""Add the frozen overlay only inside its explicit research arm."""
if _sr_research_variant() != STRUCTURAL_OVERLAY_VARIANT:
return PORTFOLIO_MONITOR_STRATEGIES
return PORTFOLIO_MONITOR_STRATEGIES + ({
"strategy": STRUCTURAL_OVERLAY_PORTFOLIO_STRATEGY,
"label": "Research: production gate + 5% clean-structure rank overlay",
"description": (
"Production-qualified universe and live exit, ranked by 95% current "
"80/20 strategy rank plus 5% clean structural/range confirmation."
),
"entry_variant": "residual80_highvol_blend80_20_fixed10",
"exit_policy": "atr_trail3",
"ranking_key": STRUCTURAL_OVERLAY_SCORE_KEY,
"use_live_config": True,
"is_production": False,
},)
def _entry_variant_config(variant: str) -> dict | None:
return next((cfg for cfg in STRATEGY_VARIANTS if cfg["variant"] == variant), None)
@@ -2504,16 +2649,19 @@ def _portfolio_monitor(
) -> dict:
latest_ord = max((max(cols[0]) for cols in prices.values() if cols[0]), default=None)
rows: list[dict] = []
for strategy in PORTFOLIO_MONITOR_STRATEGIES:
strategies = _portfolio_monitor_strategies()
for strategy in strategies:
entry_cfg = _entry_variant_config(str(strategy["entry_variant"]))
if entry_cfg is None:
continue
ranking_key = str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"])
# The production row must replay the LIVE configuration: the runtime
# qualification flag (Admin activation settings) instead of the frozen
# research-variant gate, and the Admin exit policy instead of the
# hardcoded 3x-trail/30d defaults. Research rows stay frozen so they
# remain comparable across runs.
ranking_key = str(
strategy.get("ranking_key")
or entry_cfg.get("ranking_key")
or entry_cfg["percentile_key"]
)
# Live-config rows replay the runtime qualification flag and Admin exit
# policy. The overlay opts into this deliberately so only ordering
# changes relative to the production row.
use_live = bool(strategy.get("use_live_config"))
exit_policy = str(strategy["exit_policy"])
row_hold_days = hold_days
@@ -2554,6 +2702,7 @@ def _portfolio_monitor(
"description": strategy["description"],
"is_production": bool(strategy.get("is_production")),
"entry_variant": strategy["entry_variant"],
"ranking_key": ranking_key,
"exit_policy": exit_policy,
"live_exit_mode": live_exit_mode,
"lookback": lookback["lookback"],
@@ -2569,7 +2718,7 @@ def _portfolio_monitor(
"description": s["description"],
"is_production": bool(s.get("is_production")),
}
for s in PORTFOLIO_MONITOR_STRATEGIES
for s in strategies
],
"lookbacks": [
{"lookback": lb["lookback"], "label": lb["label"]}
@@ -2578,7 +2727,9 @@ def _portfolio_monitor(
"runs": rows,
"note": (
"Portfolio monitor runs supported named strategies across cached lookbacks. "
"Local snapshot backtests remain the research surface for broad variant sweeps."
"The structural overlay appears only in its explicit research arm and changes "
"ordering, not production qualification. Local snapshot backtests remain the "
"research surface for broad variant sweeps."
),
}
@@ -3061,6 +3212,7 @@ async def run_backtest(
_assign_activation_momentum_percentiles(candidates)
_assign_residual_low_vol_blend(candidates)
_assign_residual_high_vol_blend(candidates)
_assign_structural_overlay_score(candidates)
current_min_pct = float(activation.get("min_momentum_percentile", 80.0))
for c in candidates:
c["qualified"] = _momentum_qualifies(c, current_min_pct)
@@ -3163,6 +3315,14 @@ async def run_backtest(
"range_factor_lookback": RANGE_FACTOR_LOOKBACK,
"range_factor_min_log": RANGE_FACTOR_MIN_LOG,
"range_factor_min_ratio": round(math.exp(RANGE_FACTOR_MIN_LOG), 4),
"structural_overlay_weight": (
STRUCTURAL_OVERLAY_WEIGHT
if _sr_research_variant() == STRUCTURAL_OVERLAY_VARIANT else None
),
"structural_overlay_source_variant": (
STRUCTURAL_OVERLAY_SOURCE_VARIANT
if _sr_research_variant() == STRUCTURAL_OVERLAY_VARIANT else None
),
"entry_start": (
_backtest_entry_bounds()[0].isoformat()
if _backtest_entry_bounds()[0] is not None else None