Add blue-sky projected targets and played-out setup UX
Fixes stale below-price setups showing as current recommendations. Three distinct causes share the symptom (get_trade_setups returns the latest stored setup per direction and never expires it): - Genuine blue-sky (no overhead S/R): scanner + TargetGenerator now project a measured-move target (entry +/- 3*ATR, ~2:1 R:R), flagged projected with a low sr_strength probability haircut. Overhead check keys on level tag OR price so it never projects through a straddling resistance cluster. - Projected targets clear a stricter activation bar (long-only, momentum >= 90, confidence >= min+10), independent of the general momentum gate. Mirrored in frontend qualification.ts. - Played-out UX (fixes the reported TTWO case, which is R:R-starved under a resistance cluster, not blue-sky): when price is at/past target or through the stop, RecommendationPanel shows a "No current setup" state and softens the stale ticker-level header/reasoning, instead of a stale actionable card. No migration: the projected flag rides in existing targets_json. 504 backend unit tests pass; frontend typechecks. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,201 @@
|
||||
"""Tests for blue-sky projected targets.
|
||||
|
||||
When a ticker has NO S/R level overhead in the trade direction (e.g. a stock at
|
||||
all-time highs), the scanner would otherwise produce no setup and the last stale
|
||||
one would linger. Instead we project a measured-move (ATR) target so a breakout
|
||||
name still yields a setup — one that faces a stricter activation bar. These
|
||||
tests cover the target generator's projection rule and the scanner emission.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
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.services.recommendation_service import (
|
||||
PROJECTED_TARGET_ATR_MULTIPLE,
|
||||
PROJECTED_TARGET_STRENGTH,
|
||||
target_generator,
|
||||
)
|
||||
from app.services.rr_scanner_service import scan_ticker
|
||||
|
||||
|
||||
def _lvl(price: float, type_: str, strength: int = 50, id_: int = 1) -> SimpleNamespace:
|
||||
return SimpleNamespace(id=id_, price_level=price, type=type_, strength=strength)
|
||||
|
||||
|
||||
class TestGenerateTargetsProjection:
|
||||
"""target_generator.generate_targets(direction, entry, stop, sr_levels, atr)."""
|
||||
|
||||
def test_blue_sky_long_projects_target(self):
|
||||
# Entry 100, stop 97 (risk 3), ATR 2 — no resistance overhead at all.
|
||||
targets = target_generator.generate_targets(
|
||||
direction="long",
|
||||
entry_price=100.0,
|
||||
stop_loss=97.0,
|
||||
sr_levels=[_lvl(90.0, "support"), _lvl(85.0, "support")],
|
||||
atr_value=2.0,
|
||||
)
|
||||
assert len(targets) == 1
|
||||
t = targets[0]
|
||||
assert t["projected"] is True
|
||||
assert t["sr_level_id"] == -1
|
||||
assert t["sr_strength"] == PROJECTED_TARGET_STRENGTH
|
||||
# 100 + 3 ATR (=6) = 106; reward 6 / risk 3 = 2.0 R:R
|
||||
assert t["price"] == pytest.approx(100.0 + PROJECTED_TARGET_ATR_MULTIPLE * 2.0)
|
||||
assert t["rr_ratio"] == pytest.approx(2.0)
|
||||
|
||||
def test_blue_sky_with_no_levels_at_all_projects(self):
|
||||
targets = target_generator.generate_targets(
|
||||
direction="long", entry_price=100.0, stop_loss=97.0, sr_levels=[], atr_value=2.0,
|
||||
)
|
||||
assert len(targets) == 1 and targets[0]["projected"] is True
|
||||
|
||||
def test_overhead_resistance_too_close_does_not_project(self):
|
||||
# A resistance 0.5 ATR above (< 1.0 ATR min distance) is filtered out as a
|
||||
# candidate — but it IS real overhead, so we must NOT project through it.
|
||||
targets = target_generator.generate_targets(
|
||||
direction="long",
|
||||
entry_price=100.0,
|
||||
stop_loss=97.0,
|
||||
sr_levels=[_lvl(101.0, "resistance")],
|
||||
atr_value=2.0,
|
||||
)
|
||||
assert targets == []
|
||||
|
||||
def test_normal_overhead_resistance_is_not_projected(self):
|
||||
targets = target_generator.generate_targets(
|
||||
direction="long",
|
||||
entry_price=100.0,
|
||||
stop_loss=97.0,
|
||||
sr_levels=[_lvl(106.0, "resistance", strength=80)],
|
||||
atr_value=2.0,
|
||||
)
|
||||
assert len(targets) == 1
|
||||
assert not targets[0].get("projected")
|
||||
assert targets[0]["sr_level_id"] == 1
|
||||
|
||||
def test_resistance_tagged_straddle_does_not_project(self):
|
||||
# A resistance-tagged zone rep whose near edge sits just BELOW entry still
|
||||
# means real overhead — never project through it, even though its price is
|
||||
# under entry and it isn't a valid candidate.
|
||||
targets = target_generator.generate_targets(
|
||||
direction="long",
|
||||
entry_price=100.0,
|
||||
stop_loss=97.0,
|
||||
sr_levels=[_lvl(99.5, "resistance"), _lvl(90.0, "support")],
|
||||
atr_value=2.0,
|
||||
)
|
||||
assert targets == []
|
||||
|
||||
def test_blue_sky_short_projects_below(self):
|
||||
targets = target_generator.generate_targets(
|
||||
direction="short",
|
||||
entry_price=100.0,
|
||||
stop_loss=103.0,
|
||||
sr_levels=[_lvl(110.0, "resistance")],
|
||||
atr_value=2.0,
|
||||
)
|
||||
assert len(targets) == 1
|
||||
assert targets[0]["projected"] is True
|
||||
assert targets[0]["price"] == pytest.approx(100.0 - PROJECTED_TARGET_ATR_MULTIPLE * 2.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Scanner emission
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
async def scan_session() -> AsyncSession:
|
||||
from tests.conftest import _test_session_factory
|
||||
|
||||
async with _test_session_factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
def _make_bars(ticker_id: int, num_bars: int = 20, base_close: float = 100.0):
|
||||
bars = []
|
||||
start = date(2024, 1, 1)
|
||||
for i in range(num_bars):
|
||||
close = base_close + (i % 3 - 1) * 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
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scan_emits_projected_long_when_blue_sky(scan_session: AsyncSession):
|
||||
"""A ticker with only support below entry (no overhead) yields a projected long."""
|
||||
ticker = Ticker(symbol="BLUESKY")
|
||||
scan_session.add(ticker)
|
||||
await scan_session.flush()
|
||||
|
||||
scan_session.add_all(_make_bars(ticker.id, num_bars=20, base_close=100.0))
|
||||
# Only support levels below entry — nothing overhead.
|
||||
scan_session.add_all([
|
||||
SRLevel(ticker_id=ticker.id, price_level=95.0, type="support", strength=80,
|
||||
detection_method="pivot_point"),
|
||||
SRLevel(ticker_id=ticker.id, price_level=90.0, type="support", strength=60,
|
||||
detection_method="pivot_point"),
|
||||
])
|
||||
scan_session.add(CompositeScore(
|
||||
ticker_id=ticker.id, score=70.0, is_stale=False, weights_json="{}",
|
||||
computed_at=datetime.now(timezone.utc),
|
||||
))
|
||||
await scan_session.commit()
|
||||
|
||||
setups = await scan_ticker(scan_session, "BLUESKY", rr_threshold=1.5, atr_multiplier=1.5)
|
||||
|
||||
long_setups = [s for s in setups if s.direction == "long"]
|
||||
assert len(long_setups) == 1, "blue-sky ticker should still yield a long setup"
|
||||
long_setup = long_setups[0]
|
||||
|
||||
# Target is the measured-move projection, well above entry, R:R ~2.0.
|
||||
assert long_setup.target > long_setup.entry_price
|
||||
assert long_setup.rr_ratio == pytest.approx(2.0, abs=0.05)
|
||||
|
||||
primary = [t for t in long_setup.targets if t.get("is_primary")]
|
||||
assert primary and primary[0]["projected"] is True
|
||||
assert primary[0]["sr_level_id"] == -1
|
||||
assert any("projected-target" in c for c in long_setup.conflict_flags)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scan_does_not_project_when_resistance_overhead(scan_session: AsyncSession):
|
||||
"""With a normal resistance overhead, the long target is that S/R level, not a projection."""
|
||||
ticker = Ticker(symbol="CAPPED")
|
||||
scan_session.add(ticker)
|
||||
await scan_session.flush()
|
||||
|
||||
scan_session.add_all(_make_bars(ticker.id, num_bars=20, base_close=100.0))
|
||||
scan_session.add_all([
|
||||
SRLevel(ticker_id=ticker.id, price_level=106.0, type="resistance", strength=80,
|
||||
detection_method="pivot_point"),
|
||||
SRLevel(ticker_id=ticker.id, price_level=95.0, type="support", strength=60,
|
||||
detection_method="pivot_point"),
|
||||
])
|
||||
scan_session.add(CompositeScore(
|
||||
ticker_id=ticker.id, score=70.0, is_stale=False, weights_json="{}",
|
||||
computed_at=datetime.now(timezone.utc),
|
||||
))
|
||||
await scan_session.commit()
|
||||
|
||||
setups = await scan_ticker(scan_session, "CAPPED", rr_threshold=1.5, atr_multiplier=1.5)
|
||||
long_setups = [s for s in setups if s.direction == "long"]
|
||||
assert len(long_setups) == 1
|
||||
primary = [t for t in long_setups[0].targets if t.get("is_primary")]
|
||||
assert primary and not primary[0].get("projected")
|
||||
@@ -6,6 +6,7 @@ from types import SimpleNamespace
|
||||
|
||||
from app.services.qualification import (
|
||||
best_target_probability,
|
||||
primary_target_is_projected,
|
||||
primary_target_probability,
|
||||
setup_qualifies,
|
||||
)
|
||||
@@ -155,6 +156,54 @@ class TestExcludeNeutral:
|
||||
assert setup_qualifies(_setup(recommended_action="NEUTRAL"), DEFAULT_GATE) is True
|
||||
|
||||
|
||||
def _projected_setup(**kwargs):
|
||||
"""A setup whose primary (headline) target is a blue-sky projection."""
|
||||
base = dict(
|
||||
direction="long",
|
||||
momentum_percentile=92.0,
|
||||
confidence_score=80.0,
|
||||
targets=[{"probability": 30.0, "is_primary": True, "projected": True}],
|
||||
)
|
||||
base.update(kwargs)
|
||||
return _setup(**base)
|
||||
|
||||
|
||||
class TestProjectedTargetGate:
|
||||
"""Projected targets clear a stricter bar, independent of the momentum gate."""
|
||||
|
||||
def test_projected_passes_stricter_bar(self):
|
||||
# DEFAULT_GATE has the momentum selection OFF, yet the projected block
|
||||
# still requires strong momentum + higher confidence — and this one clears.
|
||||
assert setup_qualifies(_projected_setup(), DEFAULT_GATE) is True
|
||||
|
||||
def test_projected_fails_below_momentum_floor(self):
|
||||
assert setup_qualifies(_projected_setup(momentum_percentile=85.0), DEFAULT_GATE) is False
|
||||
|
||||
def test_projected_fails_missing_momentum(self):
|
||||
assert setup_qualifies(_projected_setup(momentum_percentile=None), DEFAULT_GATE) is False
|
||||
|
||||
def test_projected_fails_below_raised_confidence_floor(self):
|
||||
# min_confidence 55 + 10 margin = 65; a 60% confidence projected setup fails
|
||||
# even though it would clear the plain 55 floor.
|
||||
assert setup_qualifies(_projected_setup(confidence_score=60.0), DEFAULT_GATE) is False
|
||||
|
||||
def test_projected_short_never_qualifies(self):
|
||||
s = _projected_setup(direction="short", recommended_action="SHORT_HIGH")
|
||||
assert setup_qualifies(s, DEFAULT_GATE) is False
|
||||
|
||||
def test_sr_anchored_setup_unaffected_by_projected_bar(self):
|
||||
# A normal (non-projected) setup with modest momentum still passes DEFAULT.
|
||||
assert setup_qualifies(_setup(momentum_percentile=10.0), DEFAULT_GATE) is True
|
||||
|
||||
def test_primary_target_is_projected_helper(self):
|
||||
assert primary_target_is_projected(_projected_setup()) is True
|
||||
assert primary_target_is_projected(_setup()) is False
|
||||
|
||||
def test_projected_flag_falls_back_when_no_primary(self):
|
||||
s = _setup(targets=[{"probability": 30.0, "projected": True}])
|
||||
assert primary_target_is_projected(s) is True
|
||||
|
||||
|
||||
class TestBestTargetProbability:
|
||||
def test_returns_max(self):
|
||||
s = _setup(targets=[{"probability": 40.0}, {"probability": 72.0}, {"probability": 55.0}])
|
||||
|
||||
Reference in New Issue
Block a user