diff --git a/app/schemas/admin.py b/app/schemas/admin.py
index 1bd5a3f..18760fa 100644
--- a/app/schemas/admin.py
+++ b/app/schemas/admin.py
@@ -59,7 +59,7 @@ class TickerUniverseUpdate(BaseModel):
class ActivationConfigUpdate(BaseModel):
"""Activation gate: what counts as an actionable signal."""
- min_expected_value: float | None = Field(default=None, ge=-1, le=10)
+ min_momentum_percentile: float | None = Field(default=None, ge=0, le=100)
min_rr: float | None = Field(default=None, ge=0)
min_confidence: float | None = Field(default=None, ge=0, le=100)
min_target_probability: float | None = Field(default=None, ge=0, le=100)
diff --git a/app/services/admin_service.py b/app/services/admin_service.py
index 7e85705..bea22dc 100644
--- a/app/services/admin_service.py
+++ b/app/services/admin_service.py
@@ -43,7 +43,7 @@ SUPPORTED_TICKER_UNIVERSES = {"sp500", "nasdaq100", "nasdaq_all"}
# confidence are floors; high-conviction / clean-read / target-probability are
# optional tighteners (off by default — turn on to be more selective).
_ACTIVATION_FLOAT_KEYS: dict[str, str] = {
- "min_expected_value": "activation_min_expected_value",
+ "min_momentum_percentile": "activation_min_momentum_percentile",
"min_rr": "activation_min_rr",
"min_confidence": "activation_min_confidence",
"min_target_probability": "activation_min_target_probability",
@@ -53,7 +53,7 @@ _ACTIVATION_BOOL_KEYS: dict[str, str] = {
"exclude_conflicts": "activation_exclude_conflicts",
}
ACTIVATION_DEFAULTS: dict[str, float | bool] = {
- "min_expected_value": 0.15,
+ "min_momentum_percentile": 80.0,
"min_rr": 1.2,
"min_confidence": 55.0,
"min_target_probability": 0.0,
@@ -201,8 +201,8 @@ async def update_activation_config(
db: AsyncSession, updates: dict[str, float | bool]
) -> dict[str, float | bool]:
"""Update the activation gate. Accepts public keys; only supplied keys change."""
- if "min_expected_value" in updates and not -1.0 <= updates["min_expected_value"] <= 10.0:
- raise ValidationError("min_expected_value must be between -1 and 10 (R units)")
+ if "min_momentum_percentile" in updates and not 0 <= updates["min_momentum_percentile"] <= 100:
+ raise ValidationError("min_momentum_percentile must be between 0 and 100")
if "min_rr" in updates and updates["min_rr"] < 0:
raise ValidationError("min_rr must be >= 0")
if "min_confidence" in updates and not 0 <= updates["min_confidence"] <= 100:
diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py
index a36fdf7..c5e61a6 100644
--- a/app/services/backtest_service.py
+++ b/app/services/backtest_service.py
@@ -14,6 +14,7 @@ held neutral here — this calibrates the price/S-R/probability machinery only.
from __future__ import annotations
+import asyncio
import json
import logging
import math
@@ -40,7 +41,6 @@ from app.services.outcome_service import (
from app.services.price_service import query_ohlcv
from app.services.qualification import (
best_target_probability,
- expected_value_r,
setup_qualifies,
)
from app.services.recommendation_service import (
@@ -110,6 +110,14 @@ def _window_setups(
if entry <= 0:
return []
+ # 12-1 month momentum (skip the last month) — the universe ranks on this.
+ # None until a year of history exists; such setups can't qualify on momentum.
+ mom_12_1 = (
+ closes[-22] / closes[-253] - 1.0
+ if len(closes) >= 253 and closes[-253] > 0
+ else None
+ )
+
try:
atr = compute_atr(highs, lows, closes)["atr"]
except Exception:
@@ -180,13 +188,12 @@ def _window_setups(
stop_loss=stop,
entry_price=entry,
)
- # meets_core = clears every gate EXCEPT the expected-value floor, so the
- # report can sweep the min_expected_value threshold without re-replaying.
- core_config = {**activation, "min_expected_value": float("-inf")}
+ # meets_core = clears every gate EXCEPT the cross-sectional momentum
+ # percentile, which can only be assigned once all tickers' setups for a
+ # week are known. run_backtest ranks momentum and finalizes `qualified`.
+ core_config = {**activation, "min_momentum_percentile": 0.0}
meets_core = setup_qualifies(setup_ns, core_config)
- ev = expected_value_r(setup_ns)
best_prob = best_target_probability(setup_ns)
- min_ev = float(activation.get("min_expected_value", 0.0))
out.append({
"direction": direction,
"entry": entry,
@@ -196,11 +203,10 @@ def _window_setups(
"confidence": confidences[direction],
"primary_prob": float(primary["probability"]),
"best_prob": best_prob,
- "ev": ev,
+ "momentum": mom_12_1,
"meets_core": meets_core,
"action": action,
"risk_level": risk_level,
- "qualified": meets_core and ev is not None and ev >= min_ev,
})
return out
@@ -230,17 +236,18 @@ def _replay_ticker(symbol: str, records: list, config: dict, activation: dict) -
realized_r = -1.0
else: # expired
realized_r = 0.0
+ iso = records[i].date.isocalendar()
candidates.append({
"symbol": symbol,
"date": records[i].date.isoformat(),
+ "iso_week": (iso[0], iso[1]),
"direction": s["direction"],
"rr": s["rr"],
"confidence": s["confidence"],
"primary_prob": s["primary_prob"],
"best_prob": s["best_prob"],
- "ev": s["ev"],
+ "momentum": s["momentum"],
"meets_core": s["meets_core"],
- "qualified": s["qualified"],
"outcome": outcome,
"target_hit": target_hit,
"realized_r": realized_r,
@@ -484,6 +491,49 @@ def _signal_evaluation(collected: dict) -> list[dict]:
return rows
+def _process_ticker(
+ symbol: str,
+ records: list,
+ config: dict,
+ activation: dict,
+ collected: dict,
+) -> list[dict]:
+ """The CPU-bound per-ticker work — replay + signal accumulation — bundled so
+ run_backtest can hand it to a worker thread. Mutates ``collected``."""
+ cands = _replay_ticker(symbol, records, config, activation)
+ _accumulate_signal_series(records, collected)
+ return cands
+
+
+def _assign_momentum_percentiles(candidates: list[dict]) -> None:
+ """Per ISO week, rank candidates by their ticker's 12-1 momentum and attach a
+ 0–100 ``momentum_percentile`` (100 = highest momentum in the universe that
+ week). Candidates whose momentum is unknown (insufficient lookback) get None
+ and therefore can't clear a momentum gate. Mutates ``candidates``."""
+ by_week: dict = defaultdict(list)
+ for c in candidates:
+ if c.get("momentum") is not None:
+ by_week[c["iso_week"]].append(c)
+ for group in by_week.values():
+ ordered = sorted(group, key=lambda c: c["momentum"])
+ n = len(ordered)
+ for rank, c in enumerate(ordered):
+ c["momentum_percentile"] = (rank / (n - 1) * 100.0) if n > 1 else 100.0
+ for c in candidates:
+ c.setdefault("momentum_percentile", None)
+
+
+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)."""
+ if not cand["meets_core"]:
+ return False
+ if threshold <= 0:
+ return True
+ mp = cand.get("momentum_percentile")
+ return mp is not None and mp >= threshold
+
+
async def run_backtest(
db: AsyncSession,
progress_cb: Callable[[int, int, str], None] | None = None,
@@ -504,29 +554,50 @@ async def run_backtest(
progress_cb(index, total, ticker.symbol)
try:
records = await query_ohlcv(db, ticker.symbol)
- candidates.extend(_replay_ticker(ticker.symbol, records, config, activation))
- _accumulate_signal_series(records, collected)
+ # Detach the ORM rows to plain objects in the event loop (safe to read
+ # here), then run the heavy replay in a worker thread. The compute is
+ # CPU-bound and used to block the event loop — and the API server with
+ # it — for the whole run; offloading lets CPython hand the GIL back to
+ # the loop every few ms so health checks / page loads stay responsive.
+ bars = [
+ SimpleNamespace(
+ date=r.date,
+ open=float(r.open),
+ high=float(r.high),
+ low=float(r.low),
+ close=float(r.close),
+ volume=int(r.volume),
+ )
+ for r in records
+ ]
+ cands = await asyncio.to_thread(
+ _process_ticker, ticker.symbol, bars, config, activation, collected
+ )
+ candidates.extend(cands)
except Exception:
logger.exception("Backtest replay failed for %s", ticker.symbol)
if progress_cb is not None and total:
progress_cb(total, total, "")
+ # Cross-sectional momentum: rank every week's universe, then "qualified" means
+ # floors + top ``min_momentum_percentile`` by 12-1 momentum.
+ _assign_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)
+
qualified = [c for c in candidates if c["qualified"]]
longs = [c for c in qualified if c["direction"] == "long"]
shorts = [c for c in qualified if c["direction"] == "short"]
- # Threshold sweep: re-apply the gate at several min_expected_value values
- # (holding the other conditions fixed) so the trade-off between how many
- # setups qualify and their expectancy is visible without re-replaying.
- current_min_ev = float(activation.get("min_expected_value", 0.15))
+ # Threshold sweep: re-apply the momentum gate at several percentile cutoffs
+ # (floors held fixed) so the trade-off between how many setups qualify and
+ # their expectancy is visible without re-replaying. 0 = floors only.
sweep = []
- for threshold in (0.4, 0.3, 0.25, 0.2, 0.15, 0.1, 0.05, 0.0):
- cands = [
- c for c in candidates
- if c["meets_core"] and c["ev"] is not None and c["ev"] >= threshold
- ]
- sweep.append({"min_expected_value": threshold, **_bucket_stats(cands)})
+ for threshold in (90.0, 80.0, 70.0, 60.0, 50.0, 0.0):
+ cands = [c for c in candidates if _momentum_qualifies(c, threshold)]
+ sweep.append({"min_momentum_percentile": threshold, **_bucket_stats(cands)})
return {
"generated_at": datetime.now(timezone.utc).isoformat(),
@@ -541,7 +612,7 @@ async def run_backtest(
"long": _bucket_stats(longs),
"short": _bucket_stats(shorts),
},
- "min_expected_value": current_min_ev,
+ "min_momentum_percentile": current_min_pct,
"sweep": sweep,
"calibration": _calibration(candidates),
"signal_eval": _signal_evaluation(collected),
diff --git a/app/services/qualification.py b/app/services/qualification.py
index 292fe21..54ed0fc 100644
--- a/app/services/qualification.py
+++ b/app/services/qualification.py
@@ -1,11 +1,14 @@
"""Shared definition of a 'qualified' (actionable) trade setup.
A single predicate, driven by the admin activation config, used by the
-performance stats (server) and mirrored on the frontend. The core gate is
-expected value (in R): a setup must promise positive, probability-weighted
-asymmetry, not just a fat-but-improbable target or a likely-but-thin one. R:R
-and confidence remain as floors, and conviction/conflict/target-probability
-survive as optional tighteners (off by default).
+performance stats (server) and mirrored on the frontend. The core selection is
+cross-sectional momentum: a setup's ticker must rank in the top
+``min_momentum_percentile`` of the universe by 12-1 month momentum — the one
+signal the backtest showed actually sorts forward returns. R:R and confidence
+remain as floors, and conviction/conflict/target-probability survive as optional
+tighteners (off by default). The momentum percentile is computed across the
+universe and attached to each setup upstream; when it's absent the gate falls
+back to the floors.
"""
from __future__ import annotations
@@ -22,37 +25,6 @@ def best_target_probability(setup: Any) -> float:
return max(probs, default=0.0)
-def primary_target_probability(setup: Any) -> float | None:
- """Probability of the starred primary target (the one the headline R:R refers
- to). Falls back to the best target's probability when none is flagged primary,
- and None when there are no targets at all (probability unknowable).
- """
- targets = getattr(setup, "targets", None) or []
- primary = next(
- (t for t in targets if isinstance(t, dict) and t.get("is_primary")), None
- )
- if primary is not None:
- return float(primary.get("probability", 0.0))
- probs = [float(t.get("probability", 0.0)) for t in targets if isinstance(t, dict)]
- return max(probs) if probs else None
-
-
-def expected_value_r(setup: Any) -> float | None:
- """Expected value per unit of risk, in R: ``p·(R:R) − (1 − p)``.
-
- ``p`` is the primary target's hit probability. This single number captures
- "is this worth taking": it rewards both a good payoff ratio and a likely
- target, so a fat-but-improbable target can't outrank a solid, probable one —
- and a high R:R no longer fights a high probability the way the old separate
- gates did. Returns None when no target probability is known.
- """
- p = primary_target_probability(setup)
- if p is None:
- return None
- p = p / 100.0
- return p * setup.rr_ratio - (1.0 - p)
-
-
def live_risk_reward(setup: Any, current_price: float) -> float | None:
"""R:R recomputed from the CURRENT price, not the (possibly stale) entry.
@@ -77,10 +49,10 @@ def setup_qualifies(setup: Any, config: dict) -> bool:
``setup`` is duck-typed: any object exposing rr_ratio, confidence_score,
recommended_action, risk_level and a ``targets`` list of dicts.
- Gate order: R:R floor → freshness (live R:R) → confidence floor → expected
- value (the core test) → optional conviction / conflict / target-probability
- tighteners. ``min_expected_value`` defaults to -inf for callers that pass a
- legacy config without the key, so they behave exactly as before.
+ Gate order: R:R floor → freshness (live R:R) → confidence floor → momentum
+ percentile (the core selection) → optional conviction / conflict /
+ target-probability tighteners. ``min_momentum_percentile`` defaults to 0 (off)
+ for callers that pass a legacy config without the key.
"""
if setup.rr_ratio < config["min_rr"]:
return False
@@ -94,13 +66,15 @@ def setup_qualifies(setup: Any, config: dict) -> bool:
return False
if (setup.confidence_score or 0.0) < config["min_confidence"]:
return False
- # Expected value (R): the core gate. Only enforced when computable — setups
- # without target probabilities (e.g. legacy historical rows) defer to the
- # R:R + confidence floors above rather than being silently dropped.
- min_ev = float(config.get("min_expected_value", float("-inf")))
- ev = expected_value_r(setup)
- if ev is not None and ev < min_ev:
- return False
+ # Cross-sectional momentum: the core selection. A setup's ticker must rank in
+ # the top ``min_momentum_percentile`` of the universe by 12-1 momentum. Only
+ # enforced when a percentile is attached (live setups / backtest) and a
+ # threshold is set; callers that don't attach it defer to the floors above.
+ min_pct = float(config.get("min_momentum_percentile", 0.0))
+ if min_pct > 0:
+ momentum_percentile = getattr(setup, "momentum_percentile", None)
+ if momentum_percentile is not None and momentum_percentile < min_pct:
+ return False
if config.get("require_high_conviction"):
if (setup.recommended_action or "") not in HIGH_CONVICTION_ACTIONS:
return False
diff --git a/frontend/src/components/admin/ActivationSettings.tsx b/frontend/src/components/admin/ActivationSettings.tsx
index 25d57a5..d1113c5 100644
--- a/frontend/src/components/admin/ActivationSettings.tsx
+++ b/frontend/src/components/admin/ActivationSettings.tsx
@@ -4,7 +4,7 @@ import { useActivationSettings, useUpdateActivationSettings } from '../../hooks/
import { SkeletonTable } from '../ui/Skeleton';
const DEFAULTS: ActivationConfig = {
- min_expected_value: 0.15,
+ min_momentum_percentile: 80,
min_rr: 1.2,
min_confidence: 55,
min_target_probability: 0,
@@ -40,26 +40,27 @@ export function ActivationSettings() {
Activation Gate
What counts as a signal worth acting on. Drives the Dashboard's "Qualified" metric, the
- Signals "Qualified only" view, and the Track Record's qualified stats. The core test is
- expected value — probability-weighted asymmetry —
- so R:R and target probability no longer fight each other. All setups are still evaluated
- regardless; tune the EV floor against the Track Record's EV sweep to see what actually wins.
+ Signals "Qualified only" view, and the Track Record's qualified stats. The core selection is
+ cross-sectional momentum — the ticker must rank in the
+ top slice of the universe by 12-1 month momentum, the one signal the backtest showed predicts
+ forward returns. R:R and confidence stay as floors. Tune the cutoff against the Track Record's
+ momentum sweep to see what actually wins.