dennisthiessen
b0e33e1606
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.
256 lines
10 KiB
Python
256 lines
10 KiB
Python
"""Parity guards: the backtest's production strategy must equal the live setup.
|
|
|
|
The portfolio monitor's production row replays the live qualification flag and
|
|
the runtime Admin exit policy, but several constants are still defined on both
|
|
sides (defaults, trail width, ordering weights). These tests fail if the two
|
|
sides drift, so a change to the live strategy forces the backtest — and vice
|
|
versa — to move with it.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from app.services import paper_trade_service
|
|
from app.services.admin_service import ACTIVATION_DEFAULTS
|
|
from app.services.backtest_service import (
|
|
ATR_MULTIPLIER,
|
|
ATR_TRAIL_MULTIPLIER,
|
|
LIVE_EXIT_MODE_TO_SIM,
|
|
PORTFOLIO_MONITOR_STRATEGIES,
|
|
PRODUCTION_PERCENTILE_KEY,
|
|
RESIDUAL_HIGH_VOL_BLEND_80_20_KEY,
|
|
TIME_EXIT_DAYS,
|
|
_entry_variant_config,
|
|
_momentum_qualifies,
|
|
_qualifies_strategy_variant,
|
|
)
|
|
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:
|
|
return next(s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production"))
|
|
|
|
|
|
def test_exit_defaults_match_the_simulated_exit() -> None:
|
|
assert paper_trade_service.DEFAULT_EXIT_MODE == "atr_trailing"
|
|
assert LIVE_EXIT_MODE_TO_SIM[paper_trade_service.DEFAULT_EXIT_MODE] == "atr_trail3"
|
|
assert paper_trade_service.DEFAULT_ATR_MULTIPLIER == ATR_TRAIL_MULTIPLIER
|
|
assert paper_trade_service.DEFAULT_HOLD_DAYS == max(TIME_EXIT_DAYS)
|
|
|
|
|
|
def test_every_live_exit_mode_has_a_sim_mapping() -> None:
|
|
assert set(paper_trade_service._VALID_EXIT_MODES) == set(LIVE_EXIT_MODE_TO_SIM)
|
|
|
|
|
|
def test_setup_stop_width_matches_the_frontend_constant() -> None:
|
|
"""The UI recovers ATR from a setup as |entry - stop| / 1.5 to render the real
|
|
exit plan (frontend/src/lib/exitPlan.ts: SETUP_STOP_ATR_MULTIPLIER). Nothing
|
|
else transmits ATR, so if the scanner's stop width changes here the UI would
|
|
silently draw the trailing stop in the wrong place."""
|
|
import inspect
|
|
|
|
from app.services import rr_scanner_service
|
|
|
|
frontend_constant = 1.5
|
|
assert ATR_MULTIPLIER == frontend_constant
|
|
for fn in (rr_scanner_service.scan_ticker, rr_scanner_service.scan_all_tickers):
|
|
signature = inspect.signature(fn)
|
|
assert signature.parameters["atr_multiplier"].default == frontend_constant
|
|
|
|
|
|
def test_gate_default_matches_the_promoted_cutoff() -> None:
|
|
prod = _production_monitor_row()
|
|
entry_cfg = _entry_variant_config(str(prod["entry_variant"]))
|
|
assert entry_cfg is not None
|
|
assert float(entry_cfg["cutoff"]) == float(ACTIVATION_DEFAULTS["min_momentum_percentile"])
|
|
|
|
|
|
def test_production_ordering_weights_are_single_sourced() -> None:
|
|
# The promoted ordering is 80/20 momentum/vol; the backtest imports the
|
|
# weight, so equality here pins the *value* the promotion was validated at.
|
|
assert STRATEGY_RANK_MOMENTUM_WEIGHT == 0.8
|
|
assert STRATEGY_RANK_VOL_WEIGHT == pytest.approx(0.2)
|
|
prod = _production_monitor_row()
|
|
entry_cfg = _entry_variant_config(str(prod["entry_variant"]))
|
|
assert entry_cfg is not None
|
|
assert entry_cfg["ranking_key"] == RESIDUAL_HIGH_VOL_BLEND_80_20_KEY
|
|
|
|
|
|
def test_production_monitor_row_replays_the_live_config() -> None:
|
|
prod = _production_monitor_row()
|
|
assert prod.get("use_live_config") is True
|
|
assert prod["exit_policy"] == "atr_trail3"
|
|
|
|
|
|
def test_live_gate_equals_the_production_variant_gate() -> None:
|
|
"""The monitor's live-gate switch relies on the runtime `qualified` flag
|
|
(_momentum_qualifies) selecting exactly what the frozen production variant
|
|
gate selects at the default cutoff."""
|
|
prod = _production_monitor_row()
|
|
entry_cfg = _entry_variant_config(str(prod["entry_variant"]))
|
|
assert entry_cfg is not None
|
|
cutoff = float(ACTIVATION_DEFAULTS["min_momentum_percentile"])
|
|
for cand in (
|
|
{"meets_core": True, "direction": "long", PRODUCTION_PERCENTILE_KEY: 92.0},
|
|
{"meets_core": True, "direction": "long", PRODUCTION_PERCENTILE_KEY: 80.0},
|
|
{"meets_core": True, "direction": "long", PRODUCTION_PERCENTILE_KEY: 79.9},
|
|
{"meets_core": True, "direction": "long", PRODUCTION_PERCENTILE_KEY: None},
|
|
{"meets_core": True, "direction": "short", PRODUCTION_PERCENTILE_KEY: 95.0},
|
|
{"meets_core": False, "direction": "long", PRODUCTION_PERCENTILE_KEY: 95.0},
|
|
):
|
|
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']}"
|
|
)
|