c34f3cb1a4
Diagnosing "no qualified signals for 5 days": setups were generated but none qualified. The gate required BOTH a high min_rr (2.0) AND a high min_target_probability (60), which became contradictory after the Jun-15 probability recalibration — probability already embeds R:R via the 1/(rr+1) ruin term, so high-R:R targets are inherently low-probability and nothing cleared both. Gate is now expected value (R): p*rr - (1-p) from the primary target's probability. R:R and confidence stay as floors; high-conviction / exclude-conflicts / min-target-probability become optional tighteners (default off). Defaults: min_expected_value=0.15, min_rr=1.2, min_confidence=55. EV is only enforced when computable. Migration 009 clears stored activation_* rows so the new defaults apply. Backtest sweeps min_expected_value instead of target probability. Scheduling: pipelines are now cron-configurable in Admin -> Jobs. daily_pipeline (full, default 0 7 * * *) plus a new light intraday_pipeline (OHLCV + outcome eval, default hourly US session) that keeps prices/live-R:R current without setup churn. Fundamentals on its own early weekly cron. Timezone configurable (default Europe/Berlin). Moving interval->CronTrigger also fixes the restart-deferral bug where an interval job's countdown resets on every process restart. 319 backend unit tests pass; frontend tsc clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
146 lines
5.5 KiB
Python
146 lines
5.5 KiB
Python
"""Unit tests for the activation qualification predicate (EV-based gate)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from app.services.qualification import (
|
|
best_target_probability,
|
|
expected_value_r,
|
|
primary_target_probability,
|
|
setup_qualifies,
|
|
)
|
|
|
|
# Default gate: expected value is the core test; conviction/conflict/target-prob
|
|
# are optional tighteners, off here.
|
|
DEFAULT_GATE = {
|
|
"min_expected_value": 0.15,
|
|
"min_rr": 1.2,
|
|
"min_confidence": 55.0,
|
|
"min_target_probability": 0.0,
|
|
"require_high_conviction": False,
|
|
"exclude_conflicts": False,
|
|
}
|
|
|
|
# Strict gate: every optional tightener turned on (the old shipped defaults).
|
|
STRICT_GATE = {
|
|
"min_expected_value": 0.0,
|
|
"min_rr": 2.0,
|
|
"min_confidence": 70.0,
|
|
"min_target_probability": 60.0,
|
|
"require_high_conviction": True,
|
|
"exclude_conflicts": True,
|
|
}
|
|
|
|
|
|
def _setup(**kwargs):
|
|
base = dict(
|
|
rr_ratio=3.0,
|
|
confidence_score=80.0,
|
|
recommended_action="LONG_HIGH",
|
|
risk_level="Low",
|
|
targets=[{"probability": 50.0, "is_primary": True}],
|
|
)
|
|
base.update(kwargs)
|
|
return SimpleNamespace(**base)
|
|
|
|
|
|
class TestExpectedValue:
|
|
def test_uses_primary_target_not_best(self):
|
|
s = _setup(
|
|
rr_ratio=1.5,
|
|
targets=[
|
|
{"probability": 80.0},
|
|
{"probability": 30.0, "is_primary": True},
|
|
],
|
|
)
|
|
# EV from the primary (30%): 0.3*1.5 - 0.7 = -0.25
|
|
assert expected_value_r(s) == -0.25
|
|
assert primary_target_probability(s) == 30.0
|
|
|
|
def test_falls_back_to_best_when_no_primary_flag(self):
|
|
s = _setup(rr_ratio=2.0, targets=[{"probability": 40.0}, {"probability": 60.0}])
|
|
assert primary_target_probability(s) == 60.0
|
|
# 0.6*2.0 - 0.4 = 0.8
|
|
assert abs(expected_value_r(s) - 0.8) < 1e-9
|
|
|
|
def test_none_when_no_targets(self):
|
|
assert expected_value_r(_setup(targets=[])) is None
|
|
assert primary_target_probability(_setup(targets=[])) is None
|
|
|
|
|
|
class TestSetupQualifies:
|
|
def test_positive_ev_setup_passes(self):
|
|
# primary 50% @ rr 3.0 → EV = 1.0
|
|
assert setup_qualifies(_setup(), DEFAULT_GATE) is True
|
|
|
|
def test_negative_ev_fails(self):
|
|
# primary 30% @ rr 1.3 → EV = -0.31, below the 0.15 floor
|
|
s = _setup(rr_ratio=1.3, targets=[{"probability": 30.0, "is_primary": True}])
|
|
assert setup_qualifies(s, DEFAULT_GATE) is False
|
|
|
|
def test_thin_positive_ev_below_floor_fails(self):
|
|
# Positive but thin: 0.45*1.3 - 0.55 = 0.035, under the 0.15 floor.
|
|
s = _setup(rr_ratio=1.3, targets=[{"probability": 45.0, "is_primary": True}])
|
|
assert setup_qualifies(s, DEFAULT_GATE) is False
|
|
|
|
def test_low_rr_floor_fails(self):
|
|
assert setup_qualifies(_setup(rr_ratio=1.0), DEFAULT_GATE) is False
|
|
|
|
def test_low_confidence_fails(self):
|
|
assert setup_qualifies(_setup(confidence_score=40.0), DEFAULT_GATE) is False
|
|
|
|
def test_no_targets_defers_to_rr_and_confidence(self):
|
|
# No probability → EV uncomputable → not blocked on EV; passes on floors.
|
|
assert setup_qualifies(_setup(targets=[]), DEFAULT_GATE) is True
|
|
# ...but still subject to the rr/confidence floors.
|
|
assert setup_qualifies(_setup(targets=[], rr_ratio=1.0), DEFAULT_GATE) is False
|
|
|
|
def test_conviction_and_conflict_ignored_by_default(self):
|
|
# Moderate action + medium risk still pass when tighteners are off.
|
|
s = _setup(recommended_action="LONG_MODERATE", risk_level="Medium")
|
|
assert setup_qualifies(s, DEFAULT_GATE) is True
|
|
|
|
def test_over_progressed_setup_fails_on_live_rr(self):
|
|
s = _setup(direction="long", target=120.0, stop_loss=95.0, current_price=117.0)
|
|
assert setup_qualifies(s, DEFAULT_GATE) is False
|
|
|
|
def test_fresh_setup_passes_live_rr(self):
|
|
s = _setup(direction="long", target=120.0, stop_loss=95.0, current_price=101.0)
|
|
assert setup_qualifies(s, DEFAULT_GATE) is True
|
|
|
|
def test_past_stop_fails_live_rr(self):
|
|
s = _setup(direction="long", target=120.0, stop_loss=95.0, current_price=94.0)
|
|
assert setup_qualifies(s, DEFAULT_GATE) is False
|
|
|
|
def test_missing_min_ev_key_skips_ev(self):
|
|
# Legacy callers without min_expected_value: EV defaults to -inf (no floor).
|
|
legacy = {k: v for k, v in DEFAULT_GATE.items() if k != "min_expected_value"}
|
|
s = _setup(rr_ratio=1.3, targets=[{"probability": 30.0, "is_primary": True}])
|
|
assert setup_qualifies(s, legacy) is True
|
|
|
|
|
|
class TestStrictTighteners:
|
|
def test_clean_high_conviction_passes(self):
|
|
assert setup_qualifies(_setup(targets=[{"probability": 65.0, "is_primary": True}]), STRICT_GATE) is True
|
|
|
|
def test_moderate_action_fails(self):
|
|
s = _setup(recommended_action="LONG_MODERATE", targets=[{"probability": 65.0, "is_primary": True}])
|
|
assert setup_qualifies(s, STRICT_GATE) is False
|
|
|
|
def test_non_low_risk_fails(self):
|
|
s = _setup(risk_level="Medium", targets=[{"probability": 65.0, "is_primary": True}])
|
|
assert setup_qualifies(s, STRICT_GATE) is False
|
|
|
|
def test_low_target_probability_fails(self):
|
|
assert setup_qualifies(_setup(targets=[{"probability": 40.0, "is_primary": True}]), STRICT_GATE) is False
|
|
|
|
|
|
class TestBestTargetProbability:
|
|
def test_returns_max(self):
|
|
s = _setup(targets=[{"probability": 40.0}, {"probability": 72.0}, {"probability": 55.0}])
|
|
assert best_target_probability(s) == 72.0
|
|
|
|
def test_empty_is_zero(self):
|
|
assert best_target_probability(_setup(targets=[])) == 0.0
|