"""Integration tests for R:R scanner full flow with quality-based target selection. Verifies the complete scan_ticker pipeline: quality-based S/R level selection, correct TradeSetup field population, and database persistence. **Validates: Requirements 2.1, 2.2, 2.3, 2.4, 3.4** """ from __future__ import annotations from datetime import date, datetime, timedelta, timezone import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.ohlcv import OHLCVRecord from app.models.score import CompositeScore from app.models.sr_level import SRLevel from app.models.ticker import Ticker from app.models.trade_setup import TradeSetup from app.services.rr_scanner_service import scan_ticker, _compute_quality_score # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @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 # =========================================================================== # 8.1 Integration test: full scan_ticker flow with quality-based selection, # correct TradeSetup fields, and database persistence # =========================================================================== @pytest.mark.asyncio async def test_scan_ticker_full_flow_quality_selection_and_persistence( scan_session: AsyncSession, ): """Integration test for the complete scan_ticker pipeline. Scenario: - Entry ≈ 100, ATR ≈ 2.0, risk ≈ 3.0 (atr_multiplier=1.5) - 3 resistance levels above (long candidates): A: price=105, strength=90 (strong, near) → highest quality B: price=115, strength=40 (medium, mid) C: price=135, strength=5 (weak, far) - 3 support levels below (short candidates): D: price=95, strength=85 (strong, near) → highest quality E: price=85, strength=35 (medium, mid) F: price=65, strength=8 (weak, far) - CompositeScore: 72.5 Verifies: 1. Both long and short setups are produced 2. Long target = Level A (highest quality, not most distant) 3. Short target = Level D (highest quality, not most distant) 4. All TradeSetup fields are correct and rounded to 4 decimals 5. rr_ratio is the actual R:R of the selected level 6. Old setups are deleted, new ones persisted """ # -- Setup: create ticker -- ticker = Ticker(symbol="INTEG") scan_session.add(ticker) await scan_session.flush() # -- Setup: OHLCV bars (20 bars, close ≈ 100, ATR ≈ 2.0) -- bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0) scan_session.add_all(bars) # -- Setup: S/R levels -- sr_levels = [ # Long candidates (resistance above entry) SRLevel(ticker_id=ticker.id, price_level=105.0, type="resistance", strength=90, detection_method="volume_profile"), SRLevel(ticker_id=ticker.id, price_level=115.0, type="resistance", strength=40, detection_method="volume_profile"), SRLevel(ticker_id=ticker.id, price_level=135.0, type="resistance", strength=5, detection_method="pivot_point"), # Short candidates (support below entry) SRLevel(ticker_id=ticker.id, price_level=95.0, type="support", strength=85, detection_method="volume_profile"), SRLevel(ticker_id=ticker.id, price_level=85.0, type="support", strength=35, detection_method="pivot_point"), SRLevel(ticker_id=ticker.id, price_level=65.0, type="support", strength=8, detection_method="volume_profile"), ] scan_session.add_all(sr_levels) # -- Setup: CompositeScore -- comp = CompositeScore( ticker_id=ticker.id, score=72.5, is_stale=False, weights_json="{}", computed_at=datetime.now(timezone.utc), ) scan_session.add(comp) # -- Setup: dummy old setups that should be deleted -- old_setup = TradeSetup( ticker_id=ticker.id, direction="long", entry_price=99.0, stop_loss=96.0, target=120.0, rr_ratio=7.0, composite_score=50.0, detected_at=datetime(2024, 1, 1, tzinfo=timezone.utc), ) scan_session.add(old_setup) await scan_session.commit() # Verify old setup exists before scan pre_result = await scan_session.execute( select(TradeSetup).where(TradeSetup.ticker_id == ticker.id) ) pre_setups = list(pre_result.scalars().all()) 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) # -- Assert: both directions produced -- assert len(setups) == 2, f"Expected 2 setups (long + short), got {len(setups)}" long_setups = [s for s in setups if s.direction == "long"] short_setups = [s for s in setups if s.direction == "short"] assert len(long_setups) == 1, f"Expected 1 long setup, got {len(long_setups)}" assert len(short_setups) == 1, f"Expected 1 short setup, got {len(short_setups)}" long_setup = long_setups[0] short_setup = short_setups[0] # -- Assert: long target is Level A (highest quality, not most distant) -- # Level A: price=105 (strong, near) should beat Level C: price=135 (weak, far) assert long_setup.target == pytest.approx(105.0, abs=0.01), ( f"Long target should be 105.0 (highest quality), got {long_setup.target}" ) # -- Assert: short target is Level D (highest quality, not most distant) -- # Level D: price=95 (strong, near) should beat Level F: price=65 (weak, far) assert short_setup.target == pytest.approx(95.0, abs=0.01), ( f"Short target should be 95.0 (highest quality), got {short_setup.target}" ) # -- Assert: entry_price is the last close (≈ 100) -- # Last bar: index 19, close = 100 + (19 % 3 - 1) * 0.5 = 100 + 0*0.5 = 100.0 expected_entry = 100.0 assert long_setup.entry_price == pytest.approx(expected_entry, abs=0.5) assert short_setup.entry_price == pytest.approx(expected_entry, abs=0.5) entry = long_setup.entry_price # actual entry for R:R calculations # -- Assert: stop_loss values -- # ATR ≈ 2.0, risk = ATR × 1.5 = 3.0 # Long stop = entry - risk, Short stop = entry + risk risk = long_setup.entry_price - long_setup.stop_loss assert risk > 0, "Long risk must be positive" assert short_setup.stop_loss > short_setup.entry_price, "Short stop must be above entry" # -- Assert: rr_ratio is the actual R:R of the selected level -- long_reward = long_setup.target - long_setup.entry_price long_expected_rr = round(long_reward / risk, 4) assert long_setup.rr_ratio == pytest.approx(long_expected_rr, abs=0.01), ( f"Long rr_ratio should be actual R:R={long_expected_rr}, got {long_setup.rr_ratio}" ) short_risk = short_setup.stop_loss - short_setup.entry_price short_reward = short_setup.entry_price - short_setup.target short_expected_rr = round(short_reward / short_risk, 4) assert short_setup.rr_ratio == pytest.approx(short_expected_rr, abs=0.01), ( f"Short rr_ratio should be actual R:R={short_expected_rr}, got {short_setup.rr_ratio}" ) # -- Assert: composite_score matches -- assert long_setup.composite_score == pytest.approx(72.5, abs=0.01) assert short_setup.composite_score == pytest.approx(72.5, abs=0.01) # -- Assert: ticker_id is correct -- assert long_setup.ticker_id == ticker.id assert short_setup.ticker_id == ticker.id # -- Assert: detected_at is set -- assert long_setup.detected_at is not None assert short_setup.detected_at is not None # -- Assert: fields are rounded to 4 decimal places -- for setup in [long_setup, short_setup]: for field_name in ("entry_price", "stop_loss", "target", "rr_ratio", "composite_score"): val = getattr(setup, field_name) rounded = round(val, 4) assert val == pytest.approx(rounded, abs=1e-6), ( f"{setup.direction} {field_name}={val} not rounded to 4 decimals" ) # -- Assert: database persistence -- # History is preserved: old setup remains, 2 new setups are appended db_result = await scan_session.execute( select(TradeSetup).where(TradeSetup.ticker_id == ticker.id) ) persisted = list(db_result.scalars().all()) assert len(persisted) == 3, ( f"Expected 3 persisted setups (1 old + 2 new), got {len(persisted)}" ) persisted_directions = sorted(s.direction for s in persisted) assert persisted_directions == ["long", "long", "short"], ( f"Expected ['long', 'long', 'short'] persisted, got {persisted_directions}" ) # Verify latest persisted records match returned setups persisted_long = max((s for s in persisted if s.direction == "long"), key=lambda s: s.id) persisted_short = max((s for s in persisted if s.direction == "short"), key=lambda s: s.id) assert persisted_long.target == long_setup.target assert persisted_long.rr_ratio == long_setup.rr_ratio assert persisted_long.entry_price == long_setup.entry_price assert persisted_long.stop_loss == long_setup.stop_loss assert persisted_long.composite_score == long_setup.composite_score assert persisted_short.target == short_setup.target assert persisted_short.rr_ratio == short_setup.rr_ratio 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