Tighten qualified signal gate
Deploy / lint (push) Successful in 8s
Deploy / test (push) Successful in 1m18s
Deploy / deploy (push) Successful in 40s

This commit is contained in:
2026-07-04 13:51:44 +02:00
parent 23d1db1f30
commit 65335cf1f3
4 changed files with 74 additions and 20 deletions
+25 -11
View File
@@ -5,9 +5,9 @@ performance stats (server) and mirrored on the frontend. The core selection is
residual cross-sectional momentum: a setup's ticker must rank in the top residual cross-sectional momentum: a setup's ticker must rank in the top
``min_momentum_percentile`` of the universe by beta-adjusted 12-1 month momentum. ``min_momentum_percentile`` of the universe by beta-adjusted 12-1 month momentum.
R:R and confidence remain as floors, and conviction/conflict survive as optional R:R and confidence remain as floors, and conviction/conflict survive as optional
tighteners (off by default). The activation percentile is computed across the tighteners (off by default). Qualified setups must also have a probability-backed
universe and attached to each setup upstream; when it's absent the gate falls target; otherwise a mathematically high R:R can be driven by a fragile target
back to the floors. with no independent validation.
""" """
from __future__ import annotations from __future__ import annotations
@@ -34,6 +34,18 @@ def best_target_probability(setup: Any) -> float:
return max(probs, default=0.0) return max(probs, default=0.0)
def primary_target_probability(setup: Any) -> float | None:
"""Probability of the primary/headline target, falling back to best target."""
targets = getattr(setup, "targets", None) or []
for target in targets:
if not isinstance(target, dict) or not target.get("is_primary"):
continue
probability = target.get("probability")
return float(probability) if probability is not None else None
best = best_target_probability(setup)
return best if best > 0 else None
def live_risk_reward(setup: Any, current_price: float) -> float | None: def live_risk_reward(setup: Any, current_price: float) -> float | None:
"""R:R recomputed from the CURRENT price, not the (possibly stale) entry. """R:R recomputed from the CURRENT price, not the (possibly stale) entry.
@@ -58,10 +70,10 @@ def setup_qualifies(setup: Any, config: dict) -> bool:
``setup`` is duck-typed: any object exposing rr_ratio, confidence_score, ``setup`` is duck-typed: any object exposing rr_ratio, confidence_score,
recommended_action, risk_level and a ``targets`` list of dicts. recommended_action, risk_level and a ``targets`` list of dicts.
Gate order: R:R floor freshness (live R:R) → confidence floor → momentum Gate order: R:R floor, freshness (live R:R), target probability, confidence
percentile (the core selection) optional conviction / conflict tighteners. floor, momentum percentile (the core selection), then optional conviction /
``min_momentum_percentile`` defaults to 0 (off) for callers that pass a legacy conflict tighteners. ``min_momentum_percentile`` defaults to 0 (off) for
config without the key. callers that pass a legacy config without the key.
""" """
if setup.rr_ratio < config["min_rr"]: if setup.rr_ratio < config["min_rr"]:
return False return False
@@ -73,20 +85,22 @@ def setup_qualifies(setup: Any, config: dict) -> bool:
live_rr = live_risk_reward(setup, float(current_price)) live_rr = live_risk_reward(setup, float(current_price))
if live_rr is not None and live_rr < config["min_rr"]: if live_rr is not None and live_rr < config["min_rr"]:
return False return False
if primary_target_probability(setup) is None:
return False
if (setup.confidence_score or 0.0) < config["min_confidence"]: if (setup.confidence_score or 0.0) < config["min_confidence"]:
return False return False
# Residual cross-sectional momentum: the core selection. A setup's ticker # Residual cross-sectional momentum: the core selection. A setup's ticker
# must rank in the top ``min_momentum_percentile`` of the universe by # must rank in the top ``min_momentum_percentile`` of the universe by
# beta-adjusted 12-1 momentum. The validated edge is long-only, so while the # beta-adjusted 12-1 momentum. The validated edge is long-only, so while the
# gate is active shorts (which fight the trend) never qualify. The percentile # gate is active shorts (which fight the trend) never qualify. Missing ranks
# floor is only enforced when a percentile is attached (live setups / # do not qualify because the production edge depends on this cross-sectional
# backtest); callers that don't attach it defer to the floors above. # selection.
min_pct = float(config.get("min_momentum_percentile", 0.0)) min_pct = float(config.get("min_momentum_percentile", 0.0))
if min_pct > 0: if min_pct > 0:
if (getattr(setup, "direction", "long") or "long") == "short": if (getattr(setup, "direction", "long") or "long") == "short":
return False return False
momentum_percentile = getattr(setup, "momentum_percentile", None) momentum_percentile = getattr(setup, "momentum_percentile", None)
if momentum_percentile is not None and momentum_percentile < min_pct: if momentum_percentile is None or momentum_percentile < min_pct:
return False return False
# A setup is actionable only when the live ticker action points in the same # A setup is actionable only when the live ticker action points in the same
# direction. NEUTRAL means no clear signal; an opposite action means the # direction. NEUTRAL means no clear signal; an opposite action means the
+5 -3
View File
@@ -39,13 +39,15 @@ export function qualifiesSetup(setup: TradeSetup, config: ActivationConfig): boo
if (setup.current_price != null && liveRiskReward(setup, setup.current_price) < config.min_rr) { if (setup.current_price != null && liveRiskReward(setup, setup.current_price) < config.min_rr) {
return false; return false;
} }
const targetProbability = primaryTargetProbability(setup);
if (targetProbability == null || targetProbability <= 0) return false;
if ((setup.confidence_score ?? 0) < config.min_confidence) return false; if ((setup.confidence_score ?? 0) < config.min_confidence) return false;
// Residual cross-sectional momentum is the core selection (long-only). While // Residual cross-sectional momentum is the core selection (long-only). While
// the gate is active, shorts never qualify; the percentile floor is enforced // the gate is active, shorts never qualify; missing ranks do not qualify
// only when a percentile is attached, otherwise defer to the floors. // because the production edge depends on this cross-sectional selection.
if (config.min_momentum_percentile > 0) { if (config.min_momentum_percentile > 0) {
if (setup.direction === 'short') return false; if (setup.direction === 'short') return false;
if (setup.momentum_percentile != null && setup.momentum_percentile < config.min_momentum_percentile) { if (setup.momentum_percentile == null || setup.momentum_percentile < config.min_momentum_percentile) {
return false; return false;
} }
} }
+11
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import json
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
import pytest import pytest
@@ -127,6 +128,15 @@ def _make_setup(
detected: datetime | None = None, detected: datetime | None = None,
**kwargs, **kwargs,
) -> TradeSetup: ) -> TradeSetup:
targets_json = kwargs.pop(
"targets_json",
json.dumps([{
"price": target,
"rr_ratio": rr,
"probability": 50.0,
"is_primary": True,
}]),
)
return TradeSetup( return TradeSetup(
ticker_id=ticker.id, ticker_id=ticker.id,
direction=direction, direction=direction,
@@ -136,6 +146,7 @@ def _make_setup(
rr_ratio=rr, rr_ratio=rr,
composite_score=50.0, composite_score=50.0,
detected_at=detected or datetime(2026, 1, 2, 21, 0, tzinfo=timezone.utc), detected_at=detected or datetime(2026, 1, 2, 21, 0, tzinfo=timezone.utc),
targets_json=targets_json,
**kwargs, **kwargs,
) )
+33 -6
View File
@@ -4,10 +4,15 @@ from __future__ import annotations
from types import SimpleNamespace from types import SimpleNamespace
from app.services.qualification import best_target_probability, setup_qualifies from app.services.qualification import (
best_target_probability,
primary_target_probability,
setup_qualifies,
)
# Default gate: floors only; the momentum selection is off (0). Conviction / # Default gate: floors only; the momentum selection is off (0). Conviction /
# conflict / target-probability are optional tighteners, off here. # conflict are optional tighteners, off here. Target probability is always
# required because qualified means the headline target is evidence-backed.
DEFAULT_GATE = { DEFAULT_GATE = {
"min_momentum_percentile": 0.0, "min_momentum_percentile": 0.0,
"min_rr": 1.2, "min_rr": 1.2,
@@ -68,6 +73,15 @@ class TestFloors:
s = _setup(direction="long", target=120.0, stop_loss=95.0, current_price=94.0) s = _setup(direction="long", target=120.0, stop_loss=95.0, current_price=94.0)
assert setup_qualifies(s, DEFAULT_GATE) is False assert setup_qualifies(s, DEFAULT_GATE) is False
def test_missing_target_probability_fails(self):
assert setup_qualifies(_setup(targets=[]), DEFAULT_GATE) is False
def test_non_primary_target_probability_still_passes(self):
assert setup_qualifies(
_setup(targets=[{"probability": 42.0}]),
DEFAULT_GATE,
) is True
class TestMomentumGate: class TestMomentumGate:
def test_top_momentum_passes(self): def test_top_momentum_passes(self):
@@ -76,15 +90,17 @@ class TestMomentumGate:
def test_below_threshold_fails(self): def test_below_threshold_fails(self):
assert setup_qualifies(_setup(momentum_percentile=50.0), MOMENTUM_GATE) is False assert setup_qualifies(_setup(momentum_percentile=50.0), MOMENTUM_GATE) is False
def test_missing_percentile_defers_to_floors(self): def test_missing_percentile_fails_when_gate_active(self):
# No percentile attached (e.g. production not yet wired) → the momentum # No residual rank means the production momentum edge was not measured.
# gate is skipped and the setup still clears on the floors. assert setup_qualifies(_setup(), MOMENTUM_GATE) is False
assert setup_qualifies(_setup(), MOMENTUM_GATE) is True
def test_threshold_zero_disables_gate(self): def test_threshold_zero_disables_gate(self):
# min_momentum_percentile 0 → a low-momentum name still passes. # min_momentum_percentile 0 → a low-momentum name still passes.
assert setup_qualifies(_setup(momentum_percentile=10.0), DEFAULT_GATE) is True assert setup_qualifies(_setup(momentum_percentile=10.0), DEFAULT_GATE) is True
def test_threshold_zero_allows_missing_percentile(self):
assert setup_qualifies(_setup(), DEFAULT_GATE) is True
def test_missing_key_defaults_off(self): def test_missing_key_defaults_off(self):
legacy = {k: v for k, v in DEFAULT_GATE.items() if k != "min_momentum_percentile"} legacy = {k: v for k, v in DEFAULT_GATE.items() if k != "min_momentum_percentile"}
assert setup_qualifies(_setup(momentum_percentile=10.0), legacy) is True assert setup_qualifies(_setup(momentum_percentile=10.0), legacy) is True
@@ -146,3 +162,14 @@ class TestBestTargetProbability:
def test_empty_is_zero(self): def test_empty_is_zero(self):
assert best_target_probability(_setup(targets=[])) == 0.0 assert best_target_probability(_setup(targets=[])) == 0.0
def test_primary_probability_prefers_starred_target(self):
s = _setup(targets=[
{"probability": 70.0},
{"probability": 45.0, "is_primary": True},
])
assert primary_target_probability(s) == 45.0
def test_primary_probability_falls_back_to_best(self):
s = _setup(targets=[{"probability": 40.0}, {"probability": 72.0}])
assert primary_target_probability(s) == 72.0