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

promote residual momentum ranking
Deploy / lint (push) Successful in 7s
Details
Deploy / test (push) Successful in 1m8s
Details
Deploy / deploy (push) Successful in 35s
Details

Browse Source
This commit is contained in:
Dennis Thiessen
2026-07-02 21:00:39 +02:00
parent 849489a4b5
commit aadec7d403
21 changed files with 310 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/backtest_service.py
+65 -75
View File
@@ -93,6 +93,9 @@ ATR_MULTIPLIER = 1.5
# signal, not the outcome of a target/stop structure built on top of one.
MIN_CROSS_SECTION = 20 # min tickers present in a week to score that week
MIN_RELIABLE_PERIODS = 12 # min non-overlapping windows before a signal's IC is trusted
PRODUCTION_PERCENTILE_KEY = "activation_momentum_percentile"
RAW_PERCENTILE_KEY = "momentum_percentile"
RESIDUAL_PERCENTILE_KEY = "residual_momentum_percentile"
def _wrap_levels(level_dicts: list[dict]) -> list[Any]:
@@ -845,12 +848,26 @@ def _assign_momentum_percentiles(candidates: list[dict]) -> None:
def _assign_residual_momentum_percentiles(candidates: list[dict]) -> None:
"""Research-only residual-momentum percentile used by strategy variants."""
"""Residual-momentum percentile promoted to production activation ranking."""
_assign_signal_percentiles(
candidates, "residual_momentum", "residual_momentum_percentile"
candidates, "residual_momentum", RESIDUAL_PERCENTILE_KEY
)
def _assign_activation_momentum_percentiles(candidates: list[dict]) -> None:
"""Production activation rank: residual 12-1 when available, raw fallback.
The raw fallback mirrors the live scanner's behavior when benchmark history
is unavailable. In normal backtests, SPY is loaded and this is residual.
"""
for c in candidates:
c[PRODUCTION_PERCENTILE_KEY] = (
c.get(RESIDUAL_PERCENTILE_KEY)
if c.get(RESIDUAL_PERCENTILE_KEY) is not None
else c.get(RAW_PERCENTILE_KEY)
)
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:
@@ -861,7 +878,7 @@ def _momentum_qualifies(cand: dict, threshold: float) -> bool:
return True
if cand["direction"] == "short":
return False
mp = cand.get("momentum_percentile")
mp = cand.get(PRODUCTION_PERCENTILE_KEY)
return mp is not None and mp >= threshold
@@ -890,7 +907,7 @@ def _gate_ablation(candidates: list[dict], activation: dict, threshold: float) -
return True
if c["direction"] == "short":
return False
mp = c.get("momentum_percentile")
mp = c.get(PRODUCTION_PERCENTILE_KEY)
return mp is not None and mp >= threshold
def rr_ok(c: dict) -> bool:
@@ -965,7 +982,7 @@ def _simulate_portfolio(
hold_days: int,
*,
qualified_fn: Callable[[dict], bool] | None = None,
ranking_key: str = "momentum_percentile",
ranking_key: str = PRODUCTION_PERCENTILE_KEY,
max_positions: int = SIM_MAX_POSITIONS,
risk_per_trade: float = SIM_RISK_PER_TRADE,
) -> dict | None:
@@ -1189,9 +1206,18 @@ def _simulate_portfolio(
STRATEGY_VARIANTS: tuple[dict, ...] = (
{
"variant": "production_raw_80_fixed10",
"label": "Production raw 80 / max 10",
"percentile_key": "momentum_percentile",
"variant": "production_residual_80_fixed10",
"label": "Production residual 80 / max 10",
"percentile_key": PRODUCTION_PERCENTILE_KEY,
"cutoff": 80.0,
"max_positions": 10,
"risk_per_trade": 0.01,
"risk_scale": None,
},
{
"variant": "legacy_raw_80_fixed10",
"label": "Legacy raw 80 / max 10",
"percentile_key": RAW_PERCENTILE_KEY,
"cutoff": 80.0,
"max_positions": 10,
"risk_per_trade": 0.01,
@@ -1200,52 +1226,16 @@ STRATEGY_VARIANTS: tuple[dict, ...] = (
{
"variant": "raw_90_fixed10",
"label": "Raw 90 / max 10",
"percentile_key": "momentum_percentile",
"percentile_key": RAW_PERCENTILE_KEY,
"cutoff": 90.0,
"max_positions": 10,
"risk_per_trade": 0.01,
"risk_scale": None,
},
{
"variant": "raw_90_fixed15",
"label": "Raw 90 / max 15",
"percentile_key": "momentum_percentile",
"cutoff": 90.0,
"max_positions": 15,
"risk_per_trade": 0.01,
"risk_scale": None,
},
{
"variant": "residual_80_fixed10",
"label": "Residual 80 / max 10",
"percentile_key": "residual_momentum_percentile",
"cutoff": 80.0,
"max_positions": 10,
"risk_per_trade": 0.01,
"risk_scale": None,
},
{
"variant": "residual_80_fixed15",
"label": "Residual 80 / max 15",
"percentile_key": "residual_momentum_percentile",
"cutoff": 80.0,
"max_positions": 15,
"risk_per_trade": 0.01,
"risk_scale": None,
},
{
"variant": "residual_80_fixed20",
"label": "Residual 80 / max 20",
"percentile_key": "residual_momentum_percentile",
"cutoff": 80.0,
"max_positions": 20,
"risk_per_trade": 0.01,
"risk_scale": None,
},
{
"variant": "raw_80_fixed15",
"label": "Raw 80 / max 15",
"percentile_key": "momentum_percentile",
"label": "Residual 80 / max 15 capacity check",
"percentile_key": PRODUCTION_PERCENTILE_KEY,
"cutoff": 80.0,
"max_positions": 15,
"risk_per_trade": 0.01,
@@ -1295,7 +1285,7 @@ def _strategy_variant_sims(
rows.append({
"variant": cfg["variant"],
"label": cfg["label"],
"ranking": "residual" if "residual" in percentile_key else "raw",
"ranking": "raw" if percentile_key == RAW_PERCENTILE_KEY else "residual",
"cutoff": cutoff,
"max_positions": int(cfg["max_positions"]),
"risk_per_trade_pct": round(float(cfg["risk_per_trade"]) * 100, 2),
@@ -1312,14 +1302,12 @@ def _pct_loss(base: float | None, candidate: float | None) -> float | None:
def _build_research_recommendation(report: dict) -> dict:
"""Advisory rules for research variants. These are deliberately conservative:
production only changes later if a portfolio variant beats the baseline under
transparent drawdown/Sharpe/CAGR constraints."""
"""Advisory rules for the remaining research variants after residual promotion."""
variants = {
v.get("variant"): v
for v in (report.get("strategy_variants") or {}).get("variants", [])
}
base = variants.get("production_raw_80_fixed10")
base = variants.get("production_residual_80_fixed10")
items: list[dict] = []
if base is None:
return {
@@ -1331,25 +1319,25 @@ def _build_research_recommendation(report: dict) -> dict:
base_dd = base.get("max_drawdown_pct")
base_cagr = base.get("cagr_pct")
residuals = [
v for key, v in variants.items()
if key.startswith("residual_80_") and v.get("risk_scale") is None
]
residual = max(residuals, key=lambda v: v.get("sharpe") or -999, default=None)
capacity = variants.get("residual_80_fixed15")
if (
residual and base_sharpe is not None and residual.get("sharpe") is not None
and base_dd is not None and residual.get("max_drawdown_pct") is not None
capacity and base_sharpe is not None and base_cagr is not None
and capacity.get("sharpe") is not None and capacity.get("cagr_pct") is not None
and capacity.get("max_drawdown_pct") is not None and base_dd is not None
):
sharpe_delta = residual["sharpe"] - base_sharpe
dd_delta = residual["max_drawdown_pct"] - base_dd
candidate = sharpe_delta >= 0.10 and dd_delta <= 2.0
candidate = (
capacity["sharpe"] > base_sharpe
and capacity["cagr_pct"] > base_cagr
and capacity["max_drawdown_pct"] <= base_dd + 1.0
)
items.append({
"topic": "residual_momentum",
"topic": "capacity_15",
"candidate": candidate,
"text": (
f"Residual momentum {'is a promotion candidate' if candidate else 'stays research-only'}: "
f"{residual['label']} Sharpe {residual['sharpe']:.2f} vs {base_sharpe:.2f}, "
f"drawdown {residual['max_drawdown_pct']:.1f}% vs {base_dd:.1f}%."
f"Max-15 capacity {'is worth promoting' if candidate else 'is not needed yet'}: "
f"Sharpe {capacity['sharpe']:.2f} vs {base_sharpe:.2f}, "
f"CAGR {capacity['cagr_pct']:+.1f}% vs {base_cagr:+.1f}%, "
f"skipped {capacity.get('skipped_book_full', 0)} vs {base.get('skipped_book_full', 0)}."
),
})
@@ -1385,8 +1373,8 @@ def _build_research_recommendation(report: dict) -> dict:
return {
"items": items,
"note": (
"Advisory only. Production changes require a variant to pass the rule "
"and then be adopted explicitly in a later strategy-version change."
"Residual 12-1 momentum is now the production activation rank. "
"Remaining rows are research comparisons only."
),
}
@@ -1473,7 +1461,8 @@ def _build_recommendation(report: dict) -> dict:
"text": f"Gate: keep the {label} (worth {delta:+.2f}R/trade under the hold exit).",
})
# Momentum cutoff: best per-trade net among the active-gate sweep rows.
# Activation cutoff: best per-trade net among the promoted residual-momentum
# sweep rows.
sweep_rows = [
r for r in report.get("sweep") or []
if r.get("net_avg_r") is not None and (r.get("min_momentum_percentile") or 0) > 0
@@ -1483,7 +1472,7 @@ def _build_recommendation(report: dict) -> dict:
items.append({
"topic": "cutoff",
"text": (
f"Momentum cutoff: {best_cut['min_momentum_percentile']:.0f} has the best "
f"Residual-momentum cutoff: {best_cut['min_momentum_percentile']:.0f} has the best "
f"per-trade net ({best_cut['net_avg_r']:+.2f}R over {best_cut['total']} setups)."
),
})
@@ -1653,9 +1642,11 @@ async def run_backtest(
progress_cb(total, total, "")
# Cross-sectional momentum: rank every week's universe, then "qualified" means
# floors + top ``min_momentum_percentile`` by 12-1 momentum.
# floors + top ``min_momentum_percentile`` by promoted residual 12-1 momentum
# (raw 12-1 fallback only when benchmark data is unavailable).
_assign_momentum_percentiles(candidates)
_assign_residual_momentum_percentiles(candidates)
_assign_activation_momentum_percentiles(candidates)
current_min_pct = float(activation.get("min_momentum_percentile", 80.0))
for c in candidates:
c["qualified"] = _momentum_qualifies(c, current_min_pct)
@@ -1779,10 +1770,9 @@ async def run_backtest(
"strategy_variants": {
"variants": strategy_variant_rows,
"note": (
"Research-only hold-to-horizon portfolio variants. These compare "
"raw vs residual momentum ranking, cutoff 80 vs 90, and max 10/15/20 "
"position capacity. They do not change live "
"qualification or paper-trade behavior."
"Research-only hold-to-horizon portfolio variants. Production now "
"uses residual 12-1 momentum at cutoff 80; the remaining rows compare "
"the legacy raw rank, raw cutoff 90, and one max-15 capacity check."
),
},
"signal_eval": _signal_evaluation(collected),