fix: align production defaults and close review parity gaps

Ship greenfield min_rr=2.0 and conf=0, read-only Structural S/R, indicator
cache invalidation, and UI/gate language that treats GTL as screening not exit.
Align strategy_rank missing-vol fallback live vs backtest, single-source
PRIMARY_TARGET_MIN_RR, expand prod parity tests, and drop dead FE clients.
This commit is contained in:
2026-07-18 13:03:22 +02:00
parent e07da0f8f0
commit b0e33e1606
25 changed files with 429 additions and 221 deletions
+2 -2
View File
@@ -26,7 +26,7 @@ class TestActivationConfig:
config = await get_activation_config(session)
assert config == {
"min_momentum_percentile": 80.0,
"min_rr": 1.2,
"min_rr": 2.0,
"min_confidence": 0.0, # off — the July 2026 ablation showed it adds nothing
"require_high_conviction": False,
"exclude_conflicts": False,
@@ -47,7 +47,7 @@ class TestActivationConfig:
async def test_partial_update_keeps_other_value(self, session: AsyncSession):
await update_activation_config(session, {"min_confidence": 80.0})
config = await get_activation_config(session)
assert config["min_rr"] == 1.2 # default untouched
assert config["min_rr"] == 2.0 # default untouched
assert config["min_confidence"] == 80.0
async def test_rejects_out_of_range_momentum_percentile(self, session: AsyncSession):
+14 -3
View File
@@ -71,10 +71,13 @@ async def test_ranks_universe_into_raw_percentiles_when_benchmark_missing(sessio
await _seed(session, "MID", rate=1.002)
await _seed(session, "LOW", rate=0.999) # declining → bottom momentum
ranks = await ms.compute_activation_ranks(session)
assert ranks["HIGH"]["momentum_percentile"] == 100.0
assert ranks["MID"]["momentum_percentile"] == 50.0
assert ranks["LOW"]["momentum_percentile"] == 0.0
# Thin momentum-only view stays aligned with the production ranker.
pct = await ms.compute_momentum_percentiles(session)
assert pct["HIGH"] == 100.0
assert pct["MID"] == 50.0
assert pct["LOW"] == 0.0
assert pct == {s: ranks[s]["momentum_percentile"] for s in pct}
async def test_ranks_universe_into_residual_percentiles_when_benchmark_available(session, monkeypatch):
@@ -91,6 +94,10 @@ async def test_ranks_universe_into_residual_percentiles_when_benchmark_available
await _seed_closes(session, "BETA", market)
await _seed_closes(session, "LAG", [market[i] * (0.9992 ** i) for i in range(n)])
ranks = await ms.compute_activation_ranks(session)
assert ranks["DRIFT"]["momentum_percentile"] == 100.0
assert ranks["BETA"]["momentum_percentile"] == 50.0
assert ranks["LAG"]["momentum_percentile"] == 0.0
pct = await ms.compute_momentum_percentiles(session)
assert pct["DRIFT"] == 100.0
assert pct["BETA"] == 50.0
@@ -105,6 +112,9 @@ async def test_short_history_ticker_is_unranked(session, monkeypatch):
await _seed(session, "LONG", rate=1.005)
await _seed(session, "SHORTHX", rate=1.005, n=100) # < 1y → no momentum
ranks = await ms.compute_activation_ranks(session)
assert "LONG" in ranks and ranks["LONG"]["momentum_percentile"] is not None
assert "SHORTHX" not in ranks or ranks["SHORTHX"]["momentum_percentile"] is None
pct = await ms.compute_momentum_percentiles(session)
assert "LONG" in pct
assert "SHORTHX" not in pct
@@ -115,4 +125,5 @@ async def test_empty_universe_returns_empty(session, monkeypatch):
return {}
monkeypatch.setattr(ms, "_load_activation_benchmark", no_benchmark)
assert await ms.compute_activation_ranks(session) == {}
assert await ms.compute_momentum_percentiles(session) == {}
+150
View File
@@ -26,7 +26,11 @@ from app.services.backtest_service import (
from app.services.momentum_service import (
STRATEGY_RANK_MOMENTUM_WEIGHT,
STRATEGY_RANK_VOL_WEIGHT,
blend_strategy_rank,
)
from app.services.qualification import MIN_TARGET_PROBABILITY
from app.services.recommendation_service import PRIMARY_TARGET_MIN_RR
from app.services import rr_scanner_service
def _production_monitor_row() -> dict:
@@ -103,3 +107,149 @@ def test_live_gate_equals_the_production_variant_gate() -> None:
assert _momentum_qualifies(cand, cutoff) == _qualifies_strategy_variant(
cand, entry_cfg
), cand
def test_activation_defaults_match_promoted_production_gate() -> None:
"""Greenfield Admin must ship the researched gate, not the old trough defaults."""
assert float(ACTIVATION_DEFAULTS["min_rr"]) == 2.0
assert float(ACTIVATION_DEFAULTS["min_confidence"]) == 0.0
assert float(ACTIVATION_DEFAULTS["min_momentum_percentile"]) == 80.0
assert ACTIVATION_DEFAULTS["exclude_neutral"] is True
def test_primary_target_rr_floor_is_single_sourced() -> None:
assert rr_scanner_service.PRIMARY_TARGET_MIN_RR == PRIMARY_TARGET_MIN_RR
assert PRIMARY_TARGET_MIN_RR == 1.5
assert MIN_TARGET_PROBABILITY == 20.0
def test_strategy_rank_falls_back_to_momentum_when_vol_missing() -> None:
"""Live and backtest must not bury a name solely because vol history is short."""
assert blend_strategy_rank(80.0, 60.0) == 76.0
assert blend_strategy_rank(80.0, None) == 80.0
assert blend_strategy_rank(None, 60.0) is None
assert blend_strategy_rank(None, None) is None
from app.services import backtest_service as bt
cands = [
{bt.PRODUCTION_PERCENTILE_KEY: 80.0, bt.VOL_PERCENTILE_KEY: None},
{bt.PRODUCTION_PERCENTILE_KEY: 70.0, bt.VOL_PERCENTILE_KEY: 50.0},
]
bt._assign_residual_high_vol_blend(cands)
assert cands[0][bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY] == 80.0
assert cands[1][bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY] == 66.0
@pytest.mark.asyncio
async def test_live_scan_and_backtest_window_share_gtl_primary() -> None:
"""Same OHLCV + dims: live scan_ticker primary ≡ backtest _window_setups.
No gate_levels_override — both paths build the production GTL from bars.
Dimension scores are seeded to the values the backtest window computes so
probability ranking cannot diverge for that reason alone.
"""
from datetime import date, datetime, timedelta, timezone
from app.models.ohlcv import OHLCVRecord
from app.models.score import DimensionScore
from app.models.ticker import Ticker
from app.services import backtest_service as bt
from app.services.recommendation_service import DEFAULT_RECOMMENDATION_CONFIG
from app.services.rr_scanner_service import scan_ticker
from app.services.scoring_service import (
compute_momentum_from_closes,
compute_technical_from_arrays,
)
from tests.conftest import _test_session_factory
n = 120
base = date(2024, 1, 1)
# Oscillating range so GTL finds traffic-backed proposals above/below spot.
closes: list[float] = []
highs: list[float] = []
lows: list[float] = []
volumes: list[int] = []
price = 100.0
for i in range(n):
phase = i % 30
if phase < 12:
price = price + (94.0 - price) * 0.2
elif phase < 24:
price = price + (108.0 - price) * 0.2
else:
price = 100.0 + (i % 5) * 0.3
high = price + 1.2
low = price - 1.2
close = price
closes.append(close)
highs.append(high)
lows.append(low)
volumes.append(100_000 + i * 10)
tech = (compute_technical_from_arrays(highs, lows, closes, volumes)[0]) or 50.0
mom = (compute_momentum_from_closes(closes)[0]) or 50.0
async with _test_session_factory() as session:
ticker = Ticker(symbol="GTLPAR")
session.add(ticker)
await session.flush()
bars = [
OHLCVRecord(
ticker_id=ticker.id,
date=base + timedelta(days=i),
open=closes[i] - 0.2,
high=highs[i],
low=lows[i],
close=closes[i],
volume=volumes[i],
)
for i in range(n)
]
session.add_all(bars)
now = datetime.now(timezone.utc)
session.add_all([
DimensionScore(
ticker_id=ticker.id, dimension="technical", score=float(tech),
is_stale=False, computed_at=now,
),
DimensionScore(
ticker_id=ticker.id, dimension="momentum", score=float(mom),
is_stale=False, computed_at=now,
),
])
await session.commit()
live = await scan_ticker(session, "GTLPAR", rr_threshold=1.5, atr_multiplier=1.5)
# Re-load bars as plain ORM list for the pure backtest window path.
from sqlalchemy import select
records = list(
(
await session.execute(
select(OHLCVRecord)
.where(OHLCVRecord.ticker_id == ticker.id)
.order_by(OHLCVRecord.date.asc())
)
).scalars().all()
)
config = dict(DEFAULT_RECOMMENDATION_CONFIG)
activation = dict(ACTIVATION_DEFAULTS)
sim = bt._window_setups(records, config, activation)
live_by_dir = {s.direction: s for s in live}
sim_by_dir = {s["direction"]: s for s in sim}
assert set(live_by_dir) == set(sim_by_dir), (
f"direction mismatch live={set(live_by_dir)} sim={set(sim_by_dir)}"
)
assert live_by_dir, "expected at least one directional setup from GTL"
for direction, live_setup in live_by_dir.items():
sim_setup = sim_by_dir[direction]
assert live_setup.target == pytest.approx(float(sim_setup["target"]), abs=0.05), (
f"{direction}: live target {live_setup.target} != sim {sim_setup['target']}"
)
assert live_setup.rr_ratio == pytest.approx(float(sim_setup["rr"]), abs=0.05), (
f"{direction}: live rr {live_setup.rr_ratio} != sim {sim_setup['rr']}"
)
+9 -12
View File
@@ -1,11 +1,8 @@
"""Bug-condition exploration tests for R:R scanner target quality.
"""Regression: scanner must not headline the most distant (max raw R:R) level.
These tests confirm the bug described in bugfix.md: the old code always selected
the most distant S/R level (highest raw R:R) regardless of strength or proximity.
The fix replaces max-R:R selection with quality-score selection.
Since the code is already fixed, these tests PASS on the current codebase.
On the unfixed code they would FAIL, confirming the bug.
Historical bug: provisional candidate pick used max R:R / quality only. Production
headline is probability-based primary after enhance_trade_setup — near levels
with real reach-probability beat far lotteries.
**Validates: Requirements 1.1, 1.3, 1.4, 2.1, 2.3, 2.4**
"""
@@ -76,10 +73,7 @@ def _make_ohlcv_bars(
@pytest.mark.asyncio
async def test_long_prefers_strong_near_over_weak_far(scan_session: AsyncSession):
"""With a strong nearby resistance and a weak distant resistance, the
scanner should pick the strong nearby one — NOT the most distant.
On unfixed code this would fail because max-R:R always picks the
farthest level.
probability primary should be the nearby level — NOT the far lottery.
"""
ticker = Ticker(symbol="EXPLR")
scan_session.add(ticker)
@@ -126,8 +120,11 @@ async def test_long_prefers_strong_near_over_weak_far(scan_session: AsyncSession
"Bug: scanner picked the weak distant level (130) instead of the "
"strong nearby level (105)"
)
# It should pick the strong nearby level
# Probability primary should pick the strong nearby level
assert selected_target == pytest.approx(105.0, abs=0.01)
primaries = [t for t in long_setups[0].targets if t.get("is_primary")]
assert len(primaries) == 1
assert primaries[0]["price"] == pytest.approx(105.0, abs=0.01)
# ---------------------------------------------------------------------------
+21 -18
View File
@@ -1,8 +1,8 @@
"""Fix-checking tests for R:R scanner quality-score selection.
"""Fix-checking tests for R:R scanner probability-based primary selection.
Verify that the fixed scan_ticker selects the candidate with the highest
quality score among all candidates meeting the R:R threshold, for both
long and short setups.
Verify that after enhance_trade_setup the headline target is the most likely
worthwhile primary (R:R + probability floors), for both long and short setups.
The pre-enhance quality loop only seeds a provisional target.
**Validates: Requirements 2.1, 2.2, 2.3, 2.4**
"""
@@ -22,9 +22,7 @@ from app.services.rr_scanner_service import scan_ticker
def _assert_primary_is_most_likely_worthwhile(setup) -> None:
"""The persisted headline target must equal the starred primary in the
targets table, and that primary must be the highest-probability target
with R:R >= 1.5 (fallback: highest R:R)."""
"""Headline = starred primary = max(probability, rr) among floor-clearing targets."""
targets = setup.targets
assert targets, "expected generated targets"
primaries = [t for t in targets if t.get("is_primary")]
@@ -32,7 +30,11 @@ def _assert_primary_is_most_likely_worthwhile(setup) -> None:
primary = primaries[0]
assert setup.target == pytest.approx(primary["price"], abs=0.01)
worthwhile = [t for t in targets if t["rr_ratio"] >= 1.5]
# Mirrors recommendation_service._select_primary_target floors.
worthwhile = [
t for t in targets
if float(t["rr_ratio"]) >= 1.5 and float(t["probability"]) >= 20.0
]
pool = worthwhile or targets
best = max(pool, key=lambda t: (t["probability"], t["rr_ratio"]))
assert primary["price"] == pytest.approx(best["price"], abs=0.01)
@@ -122,7 +124,7 @@ def short_candidate_levels(draw: st.DrawFn) -> list[dict]:
# ---------------------------------------------------------------------------
# Property test: long setup selects highest quality score candidate
# Property test: long setup selects probability-based primary
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@@ -132,14 +134,14 @@ def short_candidate_levels(draw: st.DrawFn) -> list[dict]:
deadline=None,
suppress_health_check=[HealthCheck.function_scoped_fixture],
)
async def test_property_long_selects_highest_quality(
async def test_property_long_selects_probability_primary(
levels: list[dict],
scan_session: AsyncSession,
):
"""**Validates: Requirements 2.1, 2.3, 2.4**
Property: when multiple resistance levels meet the R:R threshold,
the fixed scan_ticker selects the one with the highest quality score.
the headline after enhance is the probability-based primary.
"""
from tests.conftest import _test_engine, _test_session_factory
from app.database import Base
@@ -183,7 +185,7 @@ async def test_property_long_selects_highest_quality(
# ---------------------------------------------------------------------------
# Property test: short setup selects highest quality score candidate
# Property test: short setup selects probability-based primary
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@@ -193,14 +195,14 @@ async def test_property_long_selects_highest_quality(
deadline=None,
suppress_health_check=[HealthCheck.function_scoped_fixture],
)
async def test_property_short_selects_highest_quality(
async def test_property_short_selects_probability_primary(
levels: list[dict],
scan_session: AsyncSession,
):
"""**Validates: Requirements 2.2, 2.3, 2.4**
Property: when multiple support levels meet the R:R threshold,
the fixed scan_ticker selects the one with the highest quality score.
the headline after enhance is the probability-based primary.
"""
from tests.conftest import _test_engine, _test_session_factory
from app.database import Base
@@ -303,9 +305,10 @@ async def test_deterministic_long_three_levels(scan_session: AsyncSession):
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
# Level A (105, strength=90) should win with highest quality
_assert_primary_is_most_likely_worthwhile(long_setups[0])
# Near/strong level A wins on reach-probability over far lottery C.
assert long_setups[0].target == pytest.approx(105.0, abs=0.01), (
f"Expected target=105.0 (highest quality), got {long_setups[0].target}"
f"Expected primary=105.0 (near, high reach-prob), got {long_setups[0].target}"
)
@@ -366,7 +369,7 @@ async def test_deterministic_short_three_levels(scan_session: AsyncSession):
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, "Expected exactly one short setup"
# Level A (95, strength=85) should win with highest quality
_assert_primary_is_most_likely_worthwhile(short_setups[0])
assert short_setups[0].target == pytest.approx(95.0, abs=0.01), (
f"Expected target=95.0 (highest quality), got {short_setups[0].target}"
f"Expected primary=95.0 (near, high reach-prob), got {short_setups[0].target}"
)
+37 -21
View File
@@ -1,7 +1,8 @@
"""Integration tests for R:R scanner full flow with quality-based target selection.
"""Integration tests for R:R scanner full flow with probability-based primary.
Verifies the complete scan_ticker pipeline: quality-based S/R level selection,
correct TradeSetup field population, and database persistence.
Verifies scan_ticker → enhance_trade_setup: headline target is the primary
selected by probability floors (not the pre-enhance quality candidate loop),
TradeSetup fields, and persistence.
**Validates: Requirements 2.1, 2.2, 2.3, 2.4, 3.4**
"""
@@ -63,35 +64,52 @@ def _make_ohlcv_bars(
# ===========================================================================
# 8.1 Integration test: full scan_ticker flow with quality-based selection,
# 8.1 Integration test: full scan_ticker flow with probability primary,
# correct TradeSetup fields, and database persistence
# ===========================================================================
def _assert_headline_is_probability_primary(setup: TradeSetup) -> None:
"""Headline target/rr must match the starred primary from _select_primary_target."""
targets = setup.targets or []
assert targets, "expected generated targets after enhance"
primaries = [t for t in targets if t.get("is_primary")]
assert len(primaries) == 1, "exactly one primary target expected"
primary = primaries[0]
assert setup.target == pytest.approx(float(primary["price"]), abs=0.01)
assert setup.rr_ratio == pytest.approx(float(primary["rr_ratio"]), abs=0.01)
worthwhile = [
t for t in targets
if float(t.get("rr_ratio", 0.0)) >= 1.5 and float(t.get("probability", 0.0)) >= 20.0
]
pool = worthwhile or targets
best = max(pool, key=lambda t: (float(t["probability"]), float(t["rr_ratio"])))
assert primary["price"] == pytest.approx(float(best["price"]), abs=0.01)
@pytest.mark.asyncio
async def test_scan_ticker_full_flow_quality_selection_and_persistence(
async def test_scan_ticker_full_flow_probability_primary_and_persistence(
scan_session: AsyncSession,
):
"""Integration test for the complete scan_ticker pipeline.
"""Integration test for the complete scan_ticker → enhance pipeline.
Scenario:
- Entry ≈ 100, ATR ≈ 2.0, risk ≈ 3.0 (atr_multiplier=1.5)
- 3 resistance levels above (long candidates):
A: price=105, strength=90 (strong, near) → highest quality
A: price=105, strength=90 (strong, near) → typically highest reach-prob
B: price=115, strength=40 (medium, mid)
C: price=135, strength=5 (weak, far)
C: price=135, strength=5 (weak, far / lottery)
- 3 support levels below (short candidates):
D: price=95, strength=85 (strong, near) → highest quality
D: price=95, strength=85 (strong, near)
E: price=85, strength=35 (medium, mid)
F: price=65, strength=8 (weak, far)
- CompositeScore: 72.5
Verifies:
1. Both long and short setups are produced
2. Long target = Level A (highest quality, not most distant)
3. Short target = Level D (highest quality, not most distant)
4. All TradeSetup fields are correct and rounded to 4 decimals
5. rr_ratio is the actual R:R of the selected level
6. Old setups are deleted, new ones persisted
2. Headline is the probability-based primary (not a distant lottery)
3. Near/strong levels win over far/weak when they clear floors
4. rr_ratio matches the selected primary's R:R
5. Old setups are deleted, new ones persisted
"""
# -- Setup: create ticker --
ticker = Ticker(symbol="INTEG")
@@ -172,16 +190,14 @@ async def test_scan_ticker_full_flow_quality_selection_and_persistence(
long_setup = long_setups[0]
short_setup = short_setups[0]
# -- Assert: long target is Level A (highest quality, not most distant) --
# Level A: price=105 (strong, near) should beat Level C: price=135 (weak, far)
# -- Assert: headline is probability primary; near/strong beats far lottery --
_assert_headline_is_probability_primary(long_setup)
_assert_headline_is_probability_primary(short_setup)
assert long_setup.target == pytest.approx(105.0, abs=0.01), (
f"Long target should be 105.0 (highest quality), got {long_setup.target}"
f"Long primary should be 105.0 (near, high reach-prob), got {long_setup.target}"
)
# -- Assert: short target is Level D (highest quality, not most distant) --
# Level D: price=95 (strong, near) should beat Level F: price=65 (weak, far)
assert short_setup.target == pytest.approx(95.0, abs=0.01), (
f"Short target should be 95.0 (highest quality), got {short_setup.target}"
f"Short primary should be 95.0 (near, high reach-prob), got {short_setup.target}"
)
# -- Assert: entry_price is the last close (≈ 100) --