test: drop redundant scanner primary-target suites
test_rr_scanner_bug_exploration.py and test_rr_scanner_fix_check.py both assert one invariant: the headline target is the probability-based near level, not the far max-R:R lottery. That is already covered directly by test_recommendation_service.py's _select_primary_target tests, which also reach cases these never did (empty list, probability floor, activation vs scanner floor), and end to end by test_rr_scanner_integration.py's full-flow test -- a strict superset of their deterministic cases: three resistance and three support levels, both directions, plus persistence and rr_ratio consistency. The two files also duplicated each other, and their docstrings had gone stale: test_deterministic_long_three_levels documented a hand-computed _compute_quality_score winner even though the assertion is about the probability primary that supersedes it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -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']}"
|
|
||||||
)
|
|
||||||
@@ -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}"
|
|
||||||
)
|
|
||||||
Reference in New Issue
Block a user