260 lines
10 KiB
Python
260 lines
10 KiB
Python
"""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
|