dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Finalize GTL and retire S/R research harness

Browse Source
This commit is contained in:
dennisthiessen
2026-07-13 17:58:04 +02:00
parent 9d362bd568
commit bee5a5ce89
35 changed files with 374 additions and 5385779 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_backtest_service.py
+46 -246
View File
@@ -169,40 +169,6 @@ def test_residual_high_vol_blend_is_research_only_rank():
assert cands[0][bt.RESIDUAL_HIGH_VOL_BLEND_60_40_KEY] == 72.0
def test_structural_overlay_is_a_frozen_five_percent_rank_nudge():
cands = [
{
bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: 80.0,
"structural_overlay_pass": True,
},
{
bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: 80.0,
"structural_overlay_pass": False,
},
{bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: 80.0},
]
bt._assign_structural_overlay_score(cands)
assert bt.STRUCTURAL_OVERLAY_WEIGHT == 0.05
assert cands[0][bt.STRUCTURAL_OVERLAY_SCORE_KEY] == pytest.approx(81.0)
assert cands[1][bt.STRUCTURAL_OVERLAY_SCORE_KEY] == pytest.approx(76.0)
assert cands[2][bt.STRUCTURAL_OVERLAY_SCORE_KEY] is None
def test_structural_overlay_monitor_strategy_is_opt_in(monkeypatch):
monkeypatch.setenv("BACKTEST_SR_VARIANT", "production_control")
assert bt._portfolio_monitor_strategies() == bt.PORTFOLIO_MONITOR_STRATEGIES
monkeypatch.setenv("BACKTEST_SR_VARIANT", bt.STRUCTURAL_OVERLAY_VARIANT)
strategies = bt._portfolio_monitor_strategies()
overlay = strategies[-1]
assert overlay["strategy"] == bt.STRUCTURAL_OVERLAY_PORTFOLIO_STRATEGY
assert overlay["ranking_key"] == bt.STRUCTURAL_OVERLAY_SCORE_KEY
assert overlay["use_live_config"] is True
assert overlay["is_production"] is False
def test_strategy_variants_keep_only_current_research_candidates():
variants = {cfg["variant"]: cfg for cfg in bt.STRATEGY_VARIANTS}
@@ -638,35 +604,6 @@ class TestSimulatePortfolio:
def test_nothing_qualified_returns_none(self):
assert bt._simulate_portfolio([], {}, None, "hold", 30) is None
def test_configured_entry_end_truncates_flat_calendar_tail(self, monkeypatch):
closes = [100.0 + i for i in range(100)]
prices = {"AAA": _sim_prices(self.ORD, closes)}
cand = _sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=130.0)
monkeypatch.setenv(
"BACKTEST_ENTRY_END", date.fromordinal(self.ORD).isoformat()
)
sim = bt._simulate_portfolio([cand], prices, None, "hold", 3)
assert sim is not None
assert sim["end_date"] == date.fromordinal(self.ORD + 3).isoformat()
def test_configured_entry_start_aligns_book_calendar(self, monkeypatch):
closes = [100.0 + i for i in range(10)]
prices = {"AAA": _sim_prices(self.ORD, closes)}
cand = _sim_cand(
"AAA", self.ORD + 1, entry=101.0, stop=96.0, target=130.0
)
monkeypatch.setenv(
"BACKTEST_ENTRY_START", date.fromordinal(self.ORD).isoformat()
)
sim = bt._simulate_portfolio([cand], prices, None, "hold", 3)
assert sim is not None
assert sim["start_date"] == date.fromordinal(self.ORD).isoformat()
def test_bucket_stats_counts_and_expectancy():
cands = [
_cand(70, OUTCOME_TARGET_HIT, 3.0), # +3R win
@@ -792,138 +729,15 @@ def test_window_setups_too_short_returns_empty():
assert bt._window_setups([], {}, {}) == []
def test_sr_research_variant_is_explicit_and_validated(monkeypatch):
monkeypatch.delenv("BACKTEST_SR_VARIANT", raising=False)
assert bt._sr_research_variant() == bt.EXPLICIT_TARGET_LADDER_VARIANT
for variant in (
"production_control",
"legacy_range_grid_touch",
"legacy_range_grid_neutral",
"production_range504",
"rewrite_range504_legacy_primary",
"rewrite_range504_structural_legacy_primary",
"rewrite_range504_structural_primary2",
"production_structural_overlay",
"explicit_target_ladder",
):
monkeypatch.setenv("BACKTEST_SR_VARIANT", variant)
assert bt._sr_research_variant() == variant
monkeypatch.setenv("BACKTEST_SR_VARIANT", "not-a-variant")
with pytest.raises(ValueError, match="Unknown BACKTEST_SR_VARIANT"):
bt._sr_research_variant()
def test_backtest_target_model_is_small_and_validated():
assert bt.validate_backtest_target_model(" PRODUCTION_GTL ") == "production_gtl"
assert bt.validate_backtest_target_model("structural_sr") == "structural_sr"
with pytest.raises(ValueError, match="Unknown backtest target model"):
bt.validate_backtest_target_model("legacy_range_grid_touch")
def test_range_factor_detector_mapping_is_explicit():
assert bt._sr_detector_variant("production_range504") == "production_control"
assert bt._sr_detector_variant("rewrite_range504_legacy_primary") == "rewrite"
assert bt._sr_detector_variant(
"rewrite_range504_structural_legacy_primary"
) == "rewrite"
assert bt._sr_detector_variant("rewrite_range504_structural_primary2") == "rewrite"
assert bt._sr_detector_variant("production_structural_overlay") == "production_control"
assert bt._sr_detector_variant("soft_zones_legacy_primary") == "soft_zones"
def test_range_504_log_uses_only_the_bounded_window():
highs = [1_000.0, *([100.0] * bt.RANGE_FACTOR_LOOKBACK)]
lows = [10.0, *([50.0] * bt.RANGE_FACTOR_LOOKBACK)]
assert bt._range_504_log(highs, lows) == pytest.approx(math.log(2.0))
def test_range_factor_gate_is_research_arm_only():
below = bt.RANGE_FACTOR_MIN_LOG - 0.01
assert not bt._range_factor_allows("production_range504", below)
assert not bt._range_factor_allows("rewrite_range504_legacy_primary", below)
assert not bt._range_factor_allows(
"rewrite_range504_structural_primary2",
below,
)
assert bt._range_factor_allows(
"production_range504",
bt.RANGE_FACTOR_MIN_LOG,
)
assert bt._range_factor_allows("production_control", below)
def test_residual_arms_change_only_the_primary_rr_floor():
activation = {"min_rr": 2.0}
assert bt._primary_min_rr_for_variant(
"rewrite_range504_structural_legacy_primary",
activation,
) == 1.5
assert bt._primary_min_rr_for_variant(
"rewrite_range504_structural_primary2",
activation,
) == 2.0
assert bt._primary_min_rr_for_variant(
"production_structural_overlay",
activation,
) == 1.5
assert bt._primary_min_rr_for_variant(
"explicit_target_ladder",
activation,
) == 1.5
def test_structural_overlay_tags_production_geometry_without_replacing_it(monkeypatch):
production = {
"direction": "long",
"meets_core": True,
"rr": 2.2,
"target": 111.0,
"primary_sources": ["volume_profile"],
"gate_level_count": 53,
}
structural = {
"direction": "long",
"meets_core": True,
"rr": 2.8,
"target": 114.0,
"primary_sources": ["pivot_point"],
"gate_level_count": 14,
}
def fake_window_setups(*args, sr_variant=None, **kwargs):
if sr_variant == "production_control":
return [production]
assert sr_variant == bt.STRUCTURAL_OVERLAY_SOURCE_VARIANT
return [structural]
monkeypatch.setattr(bt, "_window_setups", fake_window_setups)
rows = bt._structural_overlay_window_setups([], {}, {})
assert len(rows) == 1
assert rows[0]["target"] == production["target"]
assert rows[0]["rr"] == production["rr"]
assert rows[0]["structural_overlay_pass"] is True
assert rows[0]["structural_overlay_rr"] == structural["rr"]
assert rows[0]["structural_overlay_sources"] == ["pivot_point"]
assert rows[0]["structural_overlay_gate_level_count"] == 14
assert production.get("structural_overlay_pass") is None
@pytest.mark.parametrize(
("variant", "neutral_strength"),
[
("legacy_range_grid_touch", False),
("legacy_range_grid_neutral", True),
],
)
def test_window_setups_routes_explicit_range_grid(
monkeypatch,
variant,
neutral_strength,
):
captured = {}
def fake_detector(*args, **kwargs):
captured.update(kwargs)
return []
monkeypatch.setenv("BACKTEST_SR_VARIANT", variant)
monkeypatch.setattr(bt, "detect_sr_levels_legacy", fake_detector)
records = [
def _flat_window_records():
return [
SimpleNamespace(
date=date(2024, 1, 1) + timedelta(days=i),
open=100.0,
@@ -934,40 +748,17 @@ def test_window_setups_routes_explicit_range_grid(
)
for i in range(bt.MIN_LOOKBACK)
]
assert bt._window_setups(records, {}, {}) == []
assert captured == {
"include_pivots": False,
"neutral_strength": neutral_strength,
"explicit_range_grid": True,
}
def test_window_setups_routes_full_explicit_target_ladder(monkeypatch):
def test_window_setups_routes_production_gtl_by_default(monkeypatch):
captured = {}
def fake_detector(highs, lows, closes):
captured.update({
"highs": highs,
"lows": lows,
"closes": closes,
})
captured.update({"highs": highs, "lows": lows, "closes": closes})
return []
monkeypatch.setenv("BACKTEST_SR_VARIANT", bt.EXPLICIT_TARGET_LADDER_VARIANT)
monkeypatch.setattr(bt, "detect_gate_target_ladder", fake_detector)
records = [
SimpleNamespace(
date=date(2024, 1, 1) + timedelta(days=i),
open=100.0,
high=101.0,
low=99.0,
close=100.0,
volume=1_000_000,
)
for i in range(bt.MIN_LOOKBACK)
]
assert bt._window_setups(records, {}, {}) == []
assert bt._window_setups(_flat_window_records(), {}, {}) == []
assert captured == {
"highs": [101.0] * bt.MIN_LOOKBACK,
"lows": [99.0] * bt.MIN_LOOKBACK,
@@ -975,32 +766,41 @@ def test_window_setups_routes_full_explicit_target_ladder(monkeypatch):
}
@pytest.mark.parametrize(
"variant",
["legacy_geometry_neutral", "legacy_range_grid_neutral"],
)
def test_neutral_strength_is_enforced_after_zone_clustering(variant):
levels = [
SimpleNamespace(strength=100),
SimpleNamespace(strength=75),
]
result = bt._apply_zone_strength_variant(levels, variant)
assert [level.strength for level in result] == [50, 50]
def test_window_setups_routes_structural_comparison(monkeypatch):
captured = {}
def fake_detector(highs, lows, closes, volumes):
captured.update({
"highs": highs,
"lows": lows,
"closes": closes,
"volumes": volumes,
})
return []
monkeypatch.setattr(bt, "detect_sr_levels", fake_detector)
assert bt._window_setups(
_flat_window_records(),
{},
{},
target_model=bt.STRUCTURAL_SR_TARGET_MODEL,
) == []
assert captured == {
"highs": [101.0] * bt.MIN_LOOKBACK,
"lows": [99.0] * bt.MIN_LOOKBACK,
"closes": [100.0] * bt.MIN_LOOKBACK,
"volumes": [1_000_000] * bt.MIN_LOOKBACK,
}
def test_touch_strength_survives_post_cluster_research_hook():
levels = [SimpleNamespace(strength=82), SimpleNamespace(strength=37)]
result = bt._apply_zone_strength_variant(levels, "legacy_range_grid_touch")
assert [level.strength for level in result] == [82, 37]
def test_backtest_entry_bounds_validate_dates(monkeypatch):
monkeypatch.setenv("BACKTEST_ENTRY_START", "2024-07-01")
monkeypatch.setenv("BACKTEST_ENTRY_END", "2024-12-31")
assert bt._backtest_entry_bounds() == (date(2024, 7, 1), date(2024, 12, 31))
monkeypatch.setenv("BACKTEST_ENTRY_START", "2025-01-01")
with pytest.raises(ValueError, match="on or before"):
bt._backtest_entry_bounds()
def test_window_setups_rejects_removed_research_arm():
with pytest.raises(ValueError, match="Unknown backtest target model"):
bt._window_setups(
_flat_window_records(),
{},
{},
target_model="production_control",
)
def test_replay_ticker_candidates_carry_gate_fields():
@@ -1028,9 +828,7 @@ def test_replay_ticker_candidates_carry_gate_fields():
for c in cands:
assert c.get("action") is not None
assert "risk_level" in c
assert c["range_504_log"] >= 0.0
assert c["range_504_ratio"] >= 1.0
assert isinstance(c["range_factor_pass"], bool)
assert c["target_model"] == bt.PRODUCTION_GTL_TARGET_MODEL
async def _seed_oscillating_ticker(session, symbol: str, n: int = 160) -> None:
@@ -1070,6 +868,8 @@ async def test_run_backtest_smoke(session):
# cost assumption is reported, and every bucket carries net numbers
assert report["params"]["cost_per_side_pct"] == pytest.approx(bt.COST_PER_SIDE * 100)
assert report["params"]["target_model"] == bt.PRODUCTION_GTL_TARGET_MODEL
assert report["params"]["is_production_target_model"] is True
assert "net_avg_r" in report["overall_all"]
# ablation baseline reproduces the qualified set exactly, and every row
tests/unit/test_detect_sr_levels.py
+21 -131
View File
@@ -6,14 +6,12 @@ from app.services.sr_service import (
MAX_LEVELS,
_bar_respect_weight,
_cap_levels,
_legacy_range_grid_nodes,
_legacy_volume_profile_nodes,
_gate_target_range_centers,
_merge_levels,
_round_number_candidates,
_strength_from_respects,
detect_gate_target_ladder,
detect_sr_levels,
detect_sr_levels_legacy,
)
@@ -237,134 +235,26 @@ class TestDetectSrLevels:
pinned = sum(1 for lvl in levels if lvl["strength"] == 100)
assert pinned < len(levels)
def test_legacy_control_retains_old_uncapped_grid(self):
highs, lows, closes, volumes = _make_series(n=500)
levels = detect_sr_levels_legacy(highs, lows, closes, volumes)
assert levels
assert all(level["sources"] for level in levels)
# The research control intentionally keeps the deployed detector's much
# denser output instead of borrowing the rewrite's presentation cap.
def test_gate_target_range_centers_cover_the_observed_range(self):
highs, lows, closes, _ = _make_series(n=500)
centers = _gate_target_range_centers(highs, lows, closes)
assert len(centers) == 20
assert centers == sorted(centers)
assert min(lows) < centers[0] < centers[-1] < max(highs)
def test_gate_target_ladder_is_dense_transient_price_traffic(self):
highs, lows, closes, _ = _make_series(n=500)
levels = detect_gate_target_ladder(highs, lows, closes)
assert len(levels) > MAX_LEVELS
assert any("range_grid" in level["sources"] for level in levels)
assert all("volume_profile" not in level["sources"] for level in levels)
assert all(0 <= level["strength"] <= 100 for level in levels)
assert all(level["rejection_count"] >= 0 for level in levels)
def test_legacy_geometry_neutral_changes_only_strength(self):
highs, lows, closes, volumes = _make_series(n=500)
control = detect_sr_levels_legacy(highs, lows, closes, volumes)
neutral = detect_sr_levels_legacy(
highs, lows, closes, volumes, neutral_strength=True
def test_gate_target_ladder_is_deterministic(self):
highs, lows, closes, _ = _make_series(n=500)
assert detect_gate_target_ladder(highs, lows, closes) == (
detect_gate_target_ladder(list(highs), list(lows), list(closes))
)
assert [level["price_level"] for level in neutral] == [
level["price_level"] for level in sorted(
control, key=lambda row: row["price_level"]
)
]
assert all(level["strength"] == 50 for level in neutral)
def test_legacy_source_ablation_isolates_candidates(self):
highs, lows, closes, volumes = _make_series(n=500)
pivots = detect_sr_levels_legacy(
highs, lows, closes, volumes, include_volume_profile=False
)
traffic = detect_sr_levels_legacy(
highs, lows, closes, volumes, include_pivots=False
)
assert pivots and traffic
assert all(level["sources"] == ["pivot_point"] for level in pivots)
assert all(level["sources"] == ["volume_profile"] for level in traffic)
def test_legacy_profile_union_matches_explicit_range_grid(self):
highs, lows, closes, volumes = _make_series(n=500)
deployed = _legacy_volume_profile_nodes(highs, lows, closes, volumes)
explicit = _legacy_range_grid_nodes(highs, lows, closes)
assert sorted(deployed) == explicit
assert len(explicit) == 20
def test_explicit_target_ladder_preserves_full_legacy_mechanics(self):
highs, lows, closes, volumes = _make_series(n=500)
deployed = detect_sr_levels_legacy(highs, lows, closes, volumes)
explicit = detect_sr_levels_legacy(
highs,
lows,
closes,
volumes,
explicit_range_grid=True,
)
def mechanics(level):
return {
key: value
for key, value in level.items()
if key not in {"detection_method", "sources"}
}
assert [mechanics(level) for level in explicit] == [
mechanics(level) for level in deployed
]
assert any("range_grid" in level["sources"] for level in explicit)
assert all("volume_profile" not in level["sources"] for level in explicit)
def test_gate_target_ladder_is_the_explicit_volume_free_detector(self):
highs, lows, closes, volumes = _make_series(n=500)
expected = detect_sr_levels_legacy(
highs,
lows,
closes,
volumes,
explicit_range_grid=True,
)
assert detect_gate_target_ladder(highs, lows, closes) == expected
def test_explicit_range_grid_is_volume_independent(self):
highs, lows, closes, volumes = _make_series(n=500)
shifted_volumes = [volume * (i + 1) for i, volume in enumerate(volumes)]
baseline = detect_sr_levels_legacy(
highs,
lows,
closes,
volumes,
include_pivots=False,
explicit_range_grid=True,
)
shifted = detect_sr_levels_legacy(
highs,
lows,
closes,
shifted_volumes,
include_pivots=False,
explicit_range_grid=True,
)
assert shifted == baseline
def test_explicit_range_grid_neutral_changes_only_strength(self):
highs, lows, closes, volumes = _make_series(n=500)
touch = detect_sr_levels_legacy(
highs,
lows,
closes,
volumes,
include_pivots=False,
explicit_range_grid=True,
)
neutral = detect_sr_levels_legacy(
highs,
lows,
closes,
volumes,
include_pivots=False,
neutral_strength=True,
explicit_range_grid=True,
)
touch_by_price = {level["price_level"]: level for level in touch}
neutral_by_price = {level["price_level"]: level for level in neutral}
assert neutral_by_price.keys() == touch_by_price.keys()
for price, neutral_level in neutral_by_price.items():
assert neutral_level["strength"] == 50
assert {
key: value
for key, value in neutral_level.items()
if key != "strength"
} == {
key: value
for key, value in touch_by_price[price].items()
if key != "strength"
}
tests/unit/test_scheduler.py
+13 -1
View File
@@ -3,15 +3,27 @@
import pytest
from app.scheduler import (
_is_job_enabled,
_consume_backtest_target_model,
_parse_frequency,
_resume_tickers,
_last_successful,
configure_scheduler,
queue_backtest_target_model,
scheduler,
)
def test_manual_backtest_target_model_is_one_shot():
assert queue_backtest_target_model("structural_sr") == "structural_sr"
assert _consume_backtest_target_model() == "structural_sr"
assert _consume_backtest_target_model() == "production_gtl"
def test_manual_backtest_target_model_rejects_removed_research_arms():
with pytest.raises(ValueError, match="Unknown backtest target model"):
queue_backtest_target_model("production_control")
class TestParseFrequency:
def test_hourly(self):
assert _parse_frequency("hourly") == {"hours": 1}