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

Separate chart S/R from gate target ladder

Browse Source
This commit is contained in:
dennisthiessen
2026-07-13 11:17:03 +02:00
parent 995a0380c3
commit 8161c352a0
12 changed files with 277 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_backtest_service.py
+12 -4
View File
@@ -942,12 +942,16 @@ def test_window_setups_routes_explicit_range_grid(
def test_window_setups_routes_full_explicit_target_ladder(monkeypatch):
captured = {}
def fake_detector(*args, **kwargs):
captured.update(kwargs)
def fake_detector(highs, lows, closes):
captured.update({
"highs": highs,
"lows": lows,
"closes": closes,
})
return []
monkeypatch.setenv("BACKTEST_SR_VARIANT", bt.EXPLICIT_TARGET_LADDER_VARIANT)
monkeypatch.setattr(bt, "detect_sr_levels_legacy", fake_detector)
monkeypatch.setattr(bt, "detect_gate_target_ladder", fake_detector)
records = [
SimpleNamespace(
date=date(2024, 1, 1) + timedelta(days=i),
@@ -961,7 +965,11 @@ def test_window_setups_routes_full_explicit_target_ladder(monkeypatch):
]
assert bt._window_setups(records, {}, {}) == []
assert captured == {"explicit_range_grid": True}
assert captured == {
"highs": [101.0] * bt.MIN_LOOKBACK,
"lows": [99.0] * bt.MIN_LOOKBACK,
"closes": [100.0] * bt.MIN_LOOKBACK,
}
@pytest.mark.parametrize(
tests/unit/test_detect_sr_levels.py
+13
View File
@@ -11,6 +11,7 @@ from app.services.sr_service import (
_merge_levels,
_round_number_candidates,
_strength_from_respects,
detect_gate_target_ladder,
detect_sr_levels,
detect_sr_levels_legacy,
)
@@ -301,6 +302,18 @@ class TestDetectSrLevels:
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)]
tests/unit/test_rr_scanner_bug_exploration.py
+18 -3
View File
@@ -110,7 +110,12 @@ async def test_long_prefers_strong_near_over_weak_far(scan_session: AsyncSession
scan_session.add_all([near_level, far_level])
await scan_session.flush()
setups = await scan_ticker(scan_session, "EXPLR", rr_threshold=1.5)
setups = await scan_ticker(
scan_session,
"EXPLR",
rr_threshold=1.5,
gate_levels_override=[near_level, far_level],
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
@@ -162,7 +167,12 @@ async def test_short_prefers_strong_near_over_weak_far(scan_session: AsyncSessio
scan_session.add_all([near_level, far_level])
await scan_session.flush()
setups = await scan_ticker(scan_session, "EXPLS", rr_threshold=1.5)
setups = await scan_ticker(
scan_session,
"EXPLS",
rr_threshold=1.5,
gate_levels_override=[near_level, far_level],
)
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, "Expected exactly one short setup"
@@ -256,7 +266,12 @@ async def test_property_scanner_does_not_always_pick_most_distant(
session.add_all([near_level, far_level])
await session.commit()
setups = await scan_ticker(session, "PROP", rr_threshold=1.5)
setups = await scan_ticker(
session,
"PROP",
rr_threshold=1.5,
gate_levels_override=[near_level, far_level],
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
tests/unit/test_rr_scanner_fix_check.py
+24 -4
View File
@@ -169,7 +169,12 @@ async def test_property_long_selects_highest_quality(
session.add_all(sr_levels)
await session.commit()
setups = await scan_ticker(session, "FIXL", rr_threshold=1.5)
setups = await scan_ticker(
session,
"FIXL",
rr_threshold=1.5,
gate_levels_override=sr_levels,
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
@@ -225,7 +230,12 @@ async def test_property_short_selects_highest_quality(
session.add_all(sr_levels)
await session.commit()
setups = await scan_ticker(session, "FIXS", rr_threshold=1.5)
setups = await scan_ticker(
session,
"FIXS",
rr_threshold=1.5,
gate_levels_override=sr_levels,
)
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, "Expected exactly one short setup"
@@ -283,7 +293,12 @@ async def test_deterministic_long_three_levels(scan_session: AsyncSession):
scan_session.add_all([level_a, level_b, level_c])
await scan_session.flush()
setups = await scan_ticker(scan_session, "DET3L", rr_threshold=1.5)
setups = await scan_ticker(
scan_session,
"DET3L",
rr_threshold=1.5,
gate_levels_override=[level_a, level_b, level_c],
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, "Expected exactly one long setup"
@@ -341,7 +356,12 @@ async def test_deterministic_short_three_levels(scan_session: AsyncSession):
scan_session.add_all([level_a, level_b, level_c])
await scan_session.flush()
setups = await scan_ticker(scan_session, "DET3S", rr_threshold=1.5)
setups = await scan_ticker(
scan_session,
"DET3S",
rr_threshold=1.5,
gate_levels_override=[level_a, level_b, level_c],
)
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, "Expected exactly one short setup"
tests/unit/test_rr_scanner_integration.py
+66 -1
View File
@@ -8,6 +8,7 @@ correct TradeSetup field population, and database persistence.
from __future__ import annotations
import json
from datetime import date, datetime, timedelta, timezone
import pytest
@@ -152,7 +153,13 @@ async def test_scan_ticker_full_flow_quality_selection_and_persistence(
assert len(pre_setups) == 1, "Dummy old setup should exist before scan"
# -- Act: run scan_ticker --
setups = await scan_ticker(scan_session, "INTEG", rr_threshold=1.5, atr_multiplier=1.5)
setups = await scan_ticker(
scan_session,
"INTEG",
rr_threshold=1.5,
atr_multiplier=1.5,
gate_levels_override=sr_levels,
)
# -- Assert: both directions produced --
assert len(setups) == 2, f"Expected 2 setups (long + short), got {len(setups)}"
@@ -255,3 +262,61 @@ async def test_scan_ticker_full_flow_quality_selection_and_persistence(
assert persisted_short.entry_price == short_setup.entry_price
assert persisted_short.stop_loss == short_setup.stop_loss
assert persisted_short.composite_score == short_setup.composite_score
@pytest.mark.asyncio
async def test_scan_ticker_uses_transient_ladder_not_persisted_chart_levels(
scan_session: AsyncSession,
monkeypatch,
):
ticker = Ticker(symbol="DUAL")
scan_session.add(ticker)
await scan_session.flush()
scan_session.add_all(_make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0))
scan_session.add(SRLevel(
ticker_id=ticker.id,
price_level=130.0,
type="resistance",
strength=100,
detection_method="pivot_point",
))
await scan_session.commit()
ladder = [
{
"price_level": 105.0,
"type": "resistance",
"strength": 90,
"detection_method": "range_grid",
"sources": ["range_grid"],
"rejection_count": 5,
"last_rejection_age": None,
},
{
"price_level": 95.0,
"type": "support",
"strength": 85,
"detection_method": "range_grid",
"sources": ["range_grid"],
"rejection_count": 4,
"last_rejection_age": None,
},
]
monkeypatch.setattr(
"app.services.rr_scanner_service.detect_gate_target_ladder",
lambda highs, lows, closes: ladder,
)
setups = await scan_ticker(
scan_session,
"DUAL",
rr_threshold=1.5,
)
long_setup = next(setup for setup in setups if setup.direction == "long")
assert long_setup.target == pytest.approx(105.0, abs=0.01)
assert long_setup.target != pytest.approx(130.0, abs=0.01)
targets = json.loads(long_setup.targets_json or "[]")
assert targets
assert all(target["sr_level_id"] < 0 for target in targets)
assert all(target["sr_sources"] == ["range_grid"] for target in targets)
tests/unit/test_rr_scanner_preservation.py
+38 -9
View File
@@ -197,17 +197,25 @@ async def test_property_zero_candidates_produce_no_setup(
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
session.add_all(bars)
gate_levels = []
for lv_data in scenario.get("levels", []):
session.add(SRLevel(
level = SRLevel(
ticker_id=ticker.id,
price_level=lv_data["price"],
type=lv_data["type"],
strength=lv_data["strength"],
detection_method="volume_profile",
))
)
session.add(level)
gate_levels.append(level)
await session.commit()
setups = await scan_ticker(session, "PRSV0", rr_threshold=1.5)
setups = await scan_ticker(
session,
"PRSV0",
rr_threshold=1.5,
gate_levels_override=gate_levels,
)
assert setups == [], (
f"Expected no setups for zero-candidate scenario "
@@ -247,16 +255,22 @@ async def test_property_single_candidate_selected_unchanged(
session.add_all(bars)
lv = scenario["level"]
session.add(SRLevel(
level = SRLevel(
ticker_id=ticker.id,
price_level=lv["price"],
type=lv["type"],
strength=lv["strength"],
detection_method="volume_profile",
))
)
session.add(level)
await session.commit()
setups = await scan_ticker(session, "PRSV1", rr_threshold=1.5)
setups = await scan_ticker(
session,
"PRSV1",
rr_threshold=1.5,
gate_levels_override=[level],
)
direction = scenario["direction"]
dir_setups = [s for s in setups if s.direction == direction]
@@ -292,7 +306,12 @@ async def test_no_sr_levels_produces_no_setup(scan_session: AsyncSession):
scan_session.add_all(bars)
await scan_session.flush()
setups = await scan_ticker(scan_session, "NOSRL", rr_threshold=1.5)
setups = await scan_ticker(
scan_session,
"NOSRL",
rr_threshold=1.5,
gate_levels_override=[],
)
assert setups == [], (
f"Expected no setups when no SR levels exist, got {len(setups)}"
@@ -329,7 +348,12 @@ async def test_single_resistance_above_threshold_selected(scan_session: AsyncSes
scan_session.add(level)
await scan_session.flush()
setups = await scan_ticker(scan_session, "SINGL", rr_threshold=1.5)
setups = await scan_ticker(
scan_session,
"SINGL",
rr_threshold=1.5,
gate_levels_override=[level],
)
long_setups = [s for s in setups if s.direction == "long"]
assert len(long_setups) == 1, (
@@ -366,7 +390,12 @@ async def test_single_support_below_threshold_selected(scan_session: AsyncSessio
scan_session.add(level)
await scan_session.flush()
setups = await scan_ticker(scan_session, "SINGS", rr_threshold=1.5)
setups = await scan_ticker(
scan_session,
"SINGS",
rr_threshold=1.5,
gate_levels_override=[level],
)
short_setups = [s for s in setups if s.direction == "short"]
assert len(short_setups) == 1, (
tests/unit/test_rr_scanner_scan_all.py
+3
View File
@@ -31,9 +31,11 @@ async def test_scan_proceeds_when_score_refresh_fails(session, monkeypatch):
raise RuntimeError("scoring unavailable")
scanned: list[str] = []
primary_floors: list[float] = []
async def _fake_scan_ticker(db, symbol, *args, **kwargs):
scanned.append(symbol)
primary_floors.append(kwargs["primary_min_rr"])
return []
monkeypatch.setattr(scoring_service, "compute_all_dimensions", _boom)
@@ -42,6 +44,7 @@ async def test_scan_proceeds_when_score_refresh_fails(session, monkeypatch):
setups = await rr_scanner_service.scan_all_tickers(session)
assert scanned == ["AAA"]
assert primary_floors == [rr_scanner_service.PRIMARY_TARGET_MIN_RR]
assert setups == []