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.
286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""Regression: scanner must not headline the most distant (max raw R:R) level.
|
||
|
||
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**
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date, timedelta
|
||
|
||
import pytest
|
||
from hypothesis import given, settings, HealthCheck, strategies as st
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.ohlcv import OHLCVRecord
|
||
from app.models.sr_level import SRLevel
|
||
from app.models.ticker import Ticker
|
||
from app.services.rr_scanner_service import scan_ticker
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Session fixture that allows scan_ticker to commit
|
||
# ---------------------------------------------------------------------------
|
||
# The default db_session fixture wraps in session.begin() which conflicts
|
||
# with scan_ticker's internal commit(). We use a plain session instead.
|
||
|
||
@pytest.fixture
|
||
async def scan_session() -> AsyncSession:
|
||
"""Provide a DB session compatible with scan_ticker (which commits)."""
|
||
from tests.conftest import _test_session_factory
|
||
|
||
async with _test_session_factory() as session:
|
||
yield session
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _make_ohlcv_bars(
|
||
ticker_id: int,
|
||
num_bars: int = 20,
|
||
base_close: float = 100.0,
|
||
) -> list[OHLCVRecord]:
|
||
"""Generate realistic OHLCV bars with small daily variation.
|
||
|
||
Produces bars where close ≈ base_close, with enough range for ATR
|
||
computation (needs >= 15 bars). The ATR will be roughly 2.0.
|
||
"""
|
||
bars: list[OHLCVRecord] = []
|
||
start = date(2024, 1, 1)
|
||
for i in range(num_bars):
|
||
close = base_close + (i % 3 - 1) * 0.5 # oscillate ±0.5
|
||
bars.append(OHLCVRecord(
|
||
ticker_id=ticker_id,
|
||
date=start + timedelta(days=i),
|
||
open=close - 0.3,
|
||
high=close + 1.0,
|
||
low=close - 1.0,
|
||
close=close,
|
||
volume=100_000,
|
||
))
|
||
return bars
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Deterministic test: strong-near vs weak-far (long setup)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@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
|
||
probability primary should be the nearby level — NOT the far lottery.
|
||
"""
|
||
ticker = Ticker(symbol="EXPLR")
|
||
scan_session.add(ticker)
|
||
await scan_session.flush()
|
||
|
||
# 20 bars closing around 100
|
||
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
|
||
scan_session.add_all(bars)
|
||
|
||
# With ATR=2.0 and multiplier=1.5, risk=3.0.
|
||
# R:R threshold=1.5 → min reward=4.5 → min target=104.5
|
||
# Strong nearby resistance: price=105, strength=90 (R:R≈1.67, quality≈0.66)
|
||
near_level = SRLevel(
|
||
ticker_id=ticker.id,
|
||
price_level=105.0,
|
||
type="resistance",
|
||
strength=90,
|
||
detection_method="volume_profile",
|
||
)
|
||
# Weak distant resistance: price=130, strength=5 (R:R=10, quality≈0.58)
|
||
far_level = SRLevel(
|
||
ticker_id=ticker.id,
|
||
price_level=130.0,
|
||
type="resistance",
|
||
strength=5,
|
||
detection_method="volume_profile",
|
||
)
|
||
scan_session.add_all([near_level, far_level])
|
||
await scan_session.flush()
|
||
|
||
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"
|
||
|
||
selected_target = long_setups[0].target
|
||
# The scanner must NOT pick the most distant level (130)
|
||
assert selected_target != pytest.approx(130.0, abs=0.01), (
|
||
"Bug: scanner picked the weak distant level (130) instead of the "
|
||
"strong nearby level (105)"
|
||
)
|
||
# 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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Deterministic test: strong-near vs weak-far (short setup)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_short_prefers_strong_near_over_weak_far(scan_session: AsyncSession):
|
||
"""Short-side mirror: strong nearby support should be preferred over
|
||
weak distant support.
|
||
"""
|
||
ticker = Ticker(symbol="EXPLS")
|
||
scan_session.add(ticker)
|
||
await scan_session.flush()
|
||
|
||
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
|
||
scan_session.add_all(bars)
|
||
|
||
# With ATR=2.0 and multiplier=1.5, risk=3.0.
|
||
# R:R threshold=1.5 → min reward=4.5 → min target below 95.5
|
||
# Strong nearby support: price=95, strength=85 (R:R≈1.67, quality≈0.64)
|
||
near_level = SRLevel(
|
||
ticker_id=ticker.id,
|
||
price_level=95.0,
|
||
type="support",
|
||
strength=85,
|
||
detection_method="pivot_point",
|
||
)
|
||
# Weak distant support: price=70, strength=5 (R:R=10, quality≈0.58)
|
||
far_level = SRLevel(
|
||
ticker_id=ticker.id,
|
||
price_level=70.0,
|
||
type="support",
|
||
strength=5,
|
||
detection_method="pivot_point",
|
||
)
|
||
scan_session.add_all([near_level, far_level])
|
||
await scan_session.flush()
|
||
|
||
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"
|
||
|
||
selected_target = short_setups[0].target
|
||
assert selected_target != pytest.approx(70.0, abs=0.01), (
|
||
"Bug: scanner picked the weak distant level (70) instead of the "
|
||
"strong nearby level (95)"
|
||
)
|
||
assert selected_target == pytest.approx(95.0, abs=0.01)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hypothesis property test: selection is NOT always the most distant level
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@st.composite
|
||
def strong_near_weak_far_pair(draw: st.DrawFn) -> dict:
|
||
"""Generate a (strong-near, weak-far) resistance pair above entry=100.
|
||
|
||
Guarantees:
|
||
- near_price < far_price (both above entry)
|
||
- near_strength >> far_strength
|
||
- Both meet the R:R threshold of 1.5 given typical ATR ≈ 2 → risk ≈ 3
|
||
"""
|
||
# Near level: 5–15 above entry (R:R ≈ 1.7–5.0 with risk≈3)
|
||
near_dist = draw(st.floats(min_value=5.0, max_value=15.0))
|
||
near_strength = draw(st.integers(min_value=70, max_value=100))
|
||
|
||
# Far level: 25–60 above entry (R:R ≈ 8.3–20 with risk≈3)
|
||
far_dist = draw(st.floats(min_value=25.0, max_value=60.0))
|
||
far_strength = draw(st.integers(min_value=1, max_value=15))
|
||
|
||
return {
|
||
"near_price": 100.0 + near_dist,
|
||
"near_strength": near_strength,
|
||
"far_price": 100.0 + far_dist,
|
||
"far_strength": far_strength,
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@given(pair=strong_near_weak_far_pair())
|
||
@settings(
|
||
max_examples=15,
|
||
deadline=None,
|
||
suppress_health_check=[HealthCheck.function_scoped_fixture],
|
||
)
|
||
async def test_property_scanner_does_not_always_pick_most_distant(
|
||
pair: dict,
|
||
scan_session: AsyncSession,
|
||
):
|
||
"""**Validates: Requirements 1.1, 1.3, 1.4, 2.1, 2.3, 2.4**
|
||
|
||
Property: when a strong nearby resistance exists alongside a weak distant
|
||
resistance, the scanner does NOT always select the most distant level.
|
||
|
||
On unfixed code this would fail for every example because max-R:R always
|
||
picks the farthest level.
|
||
"""
|
||
from tests.conftest import _test_engine, _test_session_factory
|
||
|
||
# Each hypothesis example needs a fresh DB state
|
||
async with _test_engine.begin() as conn:
|
||
from app.database import Base
|
||
await conn.run_sync(Base.metadata.drop_all)
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
async with _test_session_factory() as session:
|
||
ticker = Ticker(symbol="PROP")
|
||
session.add(ticker)
|
||
await session.flush()
|
||
|
||
bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
|
||
session.add_all(bars)
|
||
|
||
near_level = SRLevel(
|
||
ticker_id=ticker.id,
|
||
price_level=pair["near_price"],
|
||
type="resistance",
|
||
strength=pair["near_strength"],
|
||
detection_method="volume_profile",
|
||
)
|
||
far_level = SRLevel(
|
||
ticker_id=ticker.id,
|
||
price_level=pair["far_price"],
|
||
type="resistance",
|
||
strength=pair["far_strength"],
|
||
detection_method="volume_profile",
|
||
)
|
||
session.add_all([near_level, far_level])
|
||
await session.commit()
|
||
|
||
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"
|
||
|
||
selected_target = long_setups[0].target
|
||
most_distant = round(pair["far_price"], 4)
|
||
|
||
# The fixed scanner should prefer the strong nearby level, not the
|
||
# most distant weak one.
|
||
assert selected_target != pytest.approx(most_distant, abs=0.01), (
|
||
f"Bug: scanner picked the most distant level ({most_distant}) "
|
||
f"with strength={pair['far_strength']} over the nearby level "
|
||
f"({round(pair['near_price'], 4)}) with strength={pair['near_strength']}"
|
||
)
|