Quality-grade entry gate: trade fewer, higher-quality setups (config, off by default)
Forensic on 2026-06-30: of the bot's 8 traded names, SEVEN graded F (<0.50) and one B — it sprayed low-quality chop (SLS F0.12 -$412, RIVN F0.19, BB F0.21, CCXI F0.26 +$374, NVDX F0.33, NOK F0.48, RDW F0.49; GVH B0.71). The trader made +$616 on SVRE — which the bot SAW but its 35% gap cap excluded; and even graded, SVRE is F (0.35): low computed RVOL + no clean pullback structure. So the grade cuts chop but does NOT recognise a vertical catalyst RUNNER. Adds ExecutionSettings.min_quality_score (env TRADING_MIN_QUALITY_SCORE, default 0.0 = off) and a gate in request_approvals_for_ready_signals mirroring the VWAP gate: a signal scoring below the threshold is skipped and logged rule_type=quality_below; fail-open on a missing score. 3 tests (skip-low-when-enforced / allow-when-off / allow-at-threshold); 44/44 trading-execution tests green. Honest scope: this is the "trade less chop" lever (evidence: the bot's own labeled data has q>=0.55 win ~71% vs q<0.55 ~36%). It will NOT catch SVRE-class leading gainers — the grade rewards pullback structure the runners lack, and the gap cap excludes them upstream. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
1 parent
0870bb780f
commit
a114579cc1
2 files changed
+63
-2
No files matched your search
@@ -67,11 +67,11 @@ class Clock:
|
||||
|
||||
|
||||
def emit_signal(store, symbol="GOOD", entry=14.0, stop=13.45,
|
||||
above_vwap=None, vwap=None, day_open=None, cum_volume=None):
|
||||
above_vwap=None, vwap=None, day_open=None, cum_volume=None, quality_score=0.7):
|
||||
signal_data = {
|
||||
"entry_price": entry,
|
||||
"stop_loss_price": stop,
|
||||
"quality_score": 0.7,
|
||||
"quality_score": quality_score,
|
||||
}
|
||||
if above_vwap is not None:
|
||||
signal_data["above_vwap"] = above_vwap
|
||||
@@ -240,6 +240,29 @@ def test_vwap_gate_shadow_logs_but_allows_when_off(store):
|
||||
assert len(vb) == 1 and vb[0]["action_taken"] == "shadow_logged"
|
||||
|
||||
|
||||
def test_quality_gate_skips_low_grade_when_enforced(store):
|
||||
emit_signal(store, symbol="CHOP", quality_score=0.30) # F-grade chop
|
||||
service = make_service(store, min_quality_score=0.50)
|
||||
created = service.request_approvals_for_ready_signals()
|
||||
assert created == [] # entry skipped
|
||||
assert query_approval_queue(store) == []
|
||||
qb = [json.loads(e["payload_json"]) for e in store.query_events(event_type="risk_rule_triggered")
|
||||
if json.loads(e["payload_json"])["rule_type"] == "quality_below"]
|
||||
assert len(qb) == 1 and qb[0]["action_taken"] == "skipped_entry"
|
||||
|
||||
|
||||
def test_quality_gate_off_allows_low_grade(store):
|
||||
emit_signal(store, symbol="CHOP", quality_score=0.30)
|
||||
created = make_service(store, min_quality_score=0.0).request_approvals_for_ready_signals()
|
||||
assert len(created) == 1 # gate off -> still enters
|
||||
|
||||
|
||||
def test_quality_gate_allows_at_or_above_threshold(store):
|
||||
emit_signal(store, symbol="GOOD", quality_score=0.72) # B-grade
|
||||
created = make_service(store, min_quality_score=0.50).request_approvals_for_ready_signals()
|
||||
assert len(created) == 1
|
||||
|
||||
|
||||
def test_vwap_gate_allows_above_vwap(store):
|
||||
emit_signal(store, symbol="HIV", entry=14.0, stop=13.45, above_vwap=True, vwap=13.0)
|
||||
service = make_service(store, require_above_vwap=True)
|
||||
|
||||
@@ -87,6 +87,15 @@ class ExecutionSettings:
|
||||
# measurement; entries are only SKIPPED when this is True. Fail-open: a signal
|
||||
# with above_vwap=None (missing the field) is never blocked.
|
||||
require_above_vwap: bool = False
|
||||
# QUALITY-GRADE GATE — "trade fewer, higher-quality setups". The per-signal setup
|
||||
# grade (A>=0.80, B>=0.65, C>=0.50, else F) is on the ready-signal snapshot; a signal
|
||||
# scoring below this is a low-quality/chop setup. Every sub-threshold signal is
|
||||
# shadow-logged (rule_type=quality_below) for measurement; entries are only SKIPPED
|
||||
# when min_quality_score > 0. Fail-open: a signal with no quality_score is never
|
||||
# blocked. NOTE: the grade rewards a clean impulse-pullback structure + RVOL, so it
|
||||
# does NOT recognise a vertical catalyst RUNNER (those grade F) — this gate cuts
|
||||
# chop, it does not select leading gainers.
|
||||
min_quality_score: float = 0.0
|
||||
# UNIFIED ENTRY: run the fast path's shared anti-chase gates (over_extended,
|
||||
# day-extension, halt) on the LIVE auto path too. These previously ran ONLY on
|
||||
# the fast trigger path, leaving live auto entries unprotected (see the
|
||||
@@ -210,6 +219,12 @@ class ExecutionSettings:
|
||||
def flag(key: str, default: str) -> bool:
|
||||
return values.get(key, default).strip().lower() in {"1", "true", "yes", "on"}
|
||||
|
||||
def _qfloat(v: str) -> float:
|
||||
try:
|
||||
return max(0.0, min(1.0, float(v)))
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
order_type = values.get("TRADING_ENTRY_ORDER_TYPE", "limit").strip().lower()
|
||||
if order_type not in {"limit", "market"}:
|
||||
order_type = "limit"
|
||||
@@ -240,6 +255,7 @@ class ExecutionSettings:
|
||||
concentrate_top_n=int(values.get("TRADING_CONCENTRATE_TOP_N", "0")),
|
||||
max_fresh_entries_per_day=int(values.get("TRADING_MAX_FRESH_ENTRIES_PER_DAY", "0")),
|
||||
require_above_vwap=flag("TRADING_REQUIRE_ABOVE_VWAP", "1"),
|
||||
min_quality_score=_qfloat(values.get("TRADING_MIN_QUALITY_SCORE", "0.0")),
|
||||
unified_entry=flag("TRADING_UNIFIED_ENTRY", "1"),
|
||||
entry_fill_model=values.get("TRADING_ENTRY_FILL_MODEL", "resting"),
|
||||
liquidity_max_volume_pct=float(
|
||||
@@ -696,6 +712,28 @@ class TradingExecutionService:
|
||||
self._requested_symbols.add(symbol) # don't re-evaluate every tick
|
||||
continue
|
||||
|
||||
# QUALITY-GRADE gate — trade fewer, higher-quality setups. Shadow-logged
|
||||
# always; entries only SKIPPED when min_quality_score > 0. Fail-open: a
|
||||
# signal with no quality_score is never blocked.
|
||||
qs = signal.get("quality_score")
|
||||
if qs is not None and self.settings.min_quality_score > 0 and qs < self.settings.min_quality_score:
|
||||
self.store.emit(
|
||||
RiskRuleTriggeredEvent(
|
||||
timestamp=datetime.now(),
|
||||
mode=self.mode,
|
||||
correlation_id=self.session_id,
|
||||
message=(f"{symbol} ready below quality gate "
|
||||
f"(score {qs:.2f} < {self.settings.min_quality_score:.2f}) — skipped"),
|
||||
rule_type="quality_below",
|
||||
rule_value=float(qs),
|
||||
current_state={"symbol": symbol, "quality_score": float(qs),
|
||||
"min": self.settings.min_quality_score},
|
||||
action_taken="skipped_entry",
|
||||
)
|
||||
)
|
||||
self._requested_symbols.add(symbol) # don't re-evaluate every tick
|
||||
continue
|
||||
|
||||
# Anti-chase / halt gates on the LIVE auto path (the entry-path
|
||||
# unification — these ran ONLY on the fast trigger path before). The
|
||||
# auto path rests a limit AT the trigger, so it can't chase above it
|
||||
|
||||
Reference in new issue
Block a user