diff --git a/tests/unit/test_rr_scanner_bug_exploration.py b/tests/unit/test_rr_scanner_bug_exploration.py deleted file mode 100644 index 933b1b3..0000000 --- a/tests/unit/test_rr_scanner_bug_exploration.py +++ /dev/null @@ -1,285 +0,0 @@ -"""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']}" - ) diff --git a/tests/unit/test_rr_scanner_fix_check.py b/tests/unit/test_rr_scanner_fix_check.py deleted file mode 100644 index 5ab1e49..0000000 --- a/tests/unit/test_rr_scanner_fix_check.py +++ /dev/null @@ -1,375 +0,0 @@ -"""Fix-checking tests for R:R scanner probability-based primary selection. - -Verify that after enhance_trade_setup the headline target is the most likely -worthwhile primary (R:R + probability floors), for both long and short setups. -The pre-enhance quality loop only seeds a provisional target. - -**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: - """Headline = starred primary = max(probability, rr) among floor-clearing targets.""" - 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) - - # Mirrors recommendation_service._select_primary_target floors. - worthwhile = [ - t for t in targets - if float(t["rr_ratio"]) >= 1.5 and float(t["probability"]) >= 20.0 - ] - 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 probability-based primary -# --------------------------------------------------------------------------- - -@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_probability_primary( - 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 headline after enhance is the probability-based primary. - """ - 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, - 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" - - _assert_primary_is_most_likely_worthwhile(long_setups[0]) - - -# --------------------------------------------------------------------------- -# Property test: short setup selects probability-based primary -# --------------------------------------------------------------------------- - -@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_probability_primary( - 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 headline after enhance is the probability-based primary. - """ - 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, - 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" - - _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, - 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" - - _assert_primary_is_most_likely_worthwhile(long_setups[0]) - # Near/strong level A wins on reach-probability over far lottery C. - assert long_setups[0].target == pytest.approx(105.0, abs=0.01), ( - f"Expected primary=105.0 (near, high reach-prob), 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, - 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" - - _assert_primary_is_most_likely_worthwhile(short_setups[0]) - assert short_setups[0].target == pytest.approx(95.0, abs=0.01), ( - f"Expected primary=95.0 (near, high reach-prob), got {short_setups[0].target}" - )