"""Fix-checking tests for R:R scanner quality-score 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. **Validates: Requirements 2.1, 2.2, 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 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).""" targets = setup.targets assert targets, "expected generated targets" 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(primary["price"], abs=0.01) worthwhile = [t for t in targets if t["rr_ratio"] >= 1.5] 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) # --------------------------------------------------------------------------- # Session fixture (plain session, not wrapped in begin()) # --------------------------------------------------------------------------- @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 OHLCV bars closing around base_close with ATR ≈ 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 # --------------------------------------------------------------------------- # Hypothesis strategy: multiple resistance levels above entry for longs # --------------------------------------------------------------------------- @st.composite def long_candidate_levels(draw: st.DrawFn) -> list[dict]: """Generate 2-5 resistance levels above entry_price=100. All levels meet the R:R threshold of 1.5 given ATR≈2, risk≈3, so min reward=4.5, min target=104.5. """ num_levels = draw(st.integers(min_value=2, max_value=5)) levels = [] for _ in range(num_levels): # Distance from entry: 5 to 50 (all above 4.5 threshold) distance = draw(st.floats(min_value=5.0, max_value=50.0)) strength = draw(st.integers(min_value=0, max_value=100)) levels.append({ "price": 100.0 + distance, "strength": strength, }) return levels @st.composite def short_candidate_levels(draw: st.DrawFn) -> list[dict]: """Generate 2-5 support levels below entry_price=100. All levels meet the R:R threshold of 1.5 given ATR≈2, risk≈3, so min reward=4.5, max target=95.5. """ num_levels = draw(st.integers(min_value=2, max_value=5)) levels = [] for _ in range(num_levels): # Distance below entry: 5 to 50 (all above 4.5 threshold) distance = draw(st.floats(min_value=5.0, max_value=50.0)) strength = draw(st.integers(min_value=0, max_value=100)) levels.append({ "price": 100.0 - distance, "strength": strength, }) return levels # --------------------------------------------------------------------------- # Property test: long setup selects highest quality score candidate # --------------------------------------------------------------------------- @pytest.mark.asyncio @given(levels=long_candidate_levels()) @settings( max_examples=20, deadline=None, suppress_health_check=[HealthCheck.function_scoped_fixture], ) async def test_property_long_selects_highest_quality( 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. """ from tests.conftest import _test_engine, _test_session_factory from app.database import Base # Fresh DB state per hypothesis example async with _test_engine.begin() as conn: 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="FIXL") session.add(ticker) await session.flush() bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0) session.add_all(bars) sr_levels = [] for lv in levels: sr_levels.append(SRLevel( ticker_id=ticker.id, price_level=lv["price"], type="resistance", strength=lv["strength"], detection_method="volume_profile", )) session.add_all(sr_levels) await session.commit() setups = await scan_ticker(session, "FIXL", rr_threshold=1.5) long_setups = [s for s in setups if s.direction == "long"] assert len(long_setups) == 1, "Expected exactly one long setup" _assert_primary_is_most_likely_worthwhile(long_setups[0]) # --------------------------------------------------------------------------- # Property test: short setup selects highest quality score candidate # --------------------------------------------------------------------------- @pytest.mark.asyncio @given(levels=short_candidate_levels()) @settings( max_examples=20, deadline=None, suppress_health_check=[HealthCheck.function_scoped_fixture], ) async def test_property_short_selects_highest_quality( 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. """ from tests.conftest import _test_engine, _test_session_factory from app.database import Base # Fresh DB state per hypothesis example async with _test_engine.begin() as conn: 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="FIXS") session.add(ticker) await session.flush() bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0) session.add_all(bars) sr_levels = [] for lv in levels: sr_levels.append(SRLevel( ticker_id=ticker.id, price_level=lv["price"], type="support", strength=lv["strength"], detection_method="pivot_point", )) session.add_all(sr_levels) await session.commit() setups = await scan_ticker(session, "FIXS", rr_threshold=1.5) short_setups = [s for s in setups if s.direction == "short"] assert len(short_setups) == 1, "Expected exactly one short setup" _assert_primary_is_most_likely_worthwhile(short_setups[0]) # --------------------------------------------------------------------------- # Deterministic test: 3 levels with known quality scores (long) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_deterministic_long_three_levels(scan_session: AsyncSession): """**Validates: Requirements 2.1, 2.3, 2.4** Concrete example with 3 resistance levels of known quality scores. Entry=100, ATR≈2, risk≈3. Level A: price=105, strength=90 → rr=5/3≈1.67, dist=5 quality = 0.35*(1.67/10) + 0.35*(90/100) + 0.30*(1-5/100) = 0.35*0.167 + 0.35*0.9 + 0.30*0.95 = 0.0585 + 0.315 + 0.285 = 0.6585 Level B: price=112, strength=50 → rr=12/3=4.0, dist=12 quality = 0.35*(4/10) + 0.35*(50/100) + 0.30*(1-12/100) = 0.35*0.4 + 0.35*0.5 + 0.30*0.88 = 0.14 + 0.175 + 0.264 = 0.579 Level C: price=130, strength=10 → rr=30/3=10.0, dist=30 quality = 0.35*(10/10) + 0.35*(10/100) + 0.30*(1-30/100) = 0.35*1.0 + 0.35*0.1 + 0.30*0.7 = 0.35 + 0.035 + 0.21 = 0.595 Expected winner: Level A (quality=0.6585) """ ticker = Ticker(symbol="DET3L") 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) level_a = SRLevel( ticker_id=ticker.id, price_level=105.0, type="resistance", strength=90, detection_method="volume_profile", ) level_b = SRLevel( ticker_id=ticker.id, price_level=112.0, type="resistance", strength=50, detection_method="volume_profile", ) level_c = SRLevel( ticker_id=ticker.id, price_level=130.0, type="resistance", strength=10, detection_method="volume_profile", ) 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) 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 long_setups[0].target == pytest.approx(105.0, abs=0.01), ( f"Expected target=105.0 (highest quality), got {long_setups[0].target}" ) # --------------------------------------------------------------------------- # Deterministic test: 3 levels with known quality scores (short) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_deterministic_short_three_levels(scan_session: AsyncSession): """**Validates: Requirements 2.2, 2.3, 2.4** Concrete example with 3 support levels of known quality scores. Entry=100, ATR≈2, risk≈3. Level A: price=95, strength=85 → rr=5/3≈1.67, dist=5 quality = 0.35*(1.67/10) + 0.35*(85/100) + 0.30*(1-5/100) = 0.0585 + 0.2975 + 0.285 = 0.641 Level B: price=88, strength=45 → rr=12/3=4.0, dist=12 quality = 0.35*(4/10) + 0.35*(45/100) + 0.30*(1-12/100) = 0.14 + 0.1575 + 0.264 = 0.5615 Level C: price=70, strength=8 → rr=30/3=10.0, dist=30 quality = 0.35*(10/10) + 0.35*(8/100) + 0.30*(1-30/100) = 0.35 + 0.028 + 0.21 = 0.588 Expected winner: Level A (quality=0.641) """ ticker = Ticker(symbol="DET3S") 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) level_a = SRLevel( ticker_id=ticker.id, price_level=95.0, type="support", strength=85, detection_method="pivot_point", ) level_b = SRLevel( ticker_id=ticker.id, price_level=88.0, type="support", strength=45, detection_method="pivot_point", ) level_c = SRLevel( ticker_id=ticker.id, price_level=70.0, type="support", strength=8, detection_method="pivot_point", ) 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) 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 short_setups[0].target == pytest.approx(95.0, abs=0.01), ( f"Expected target=95.0 (highest quality), got {short_setups[0].target}" )