1043 lines
38 KiB
Python
1043 lines
38 KiB
Python
"""R:R scanner service.
|
|
|
|
Scans tracked tickers for asymmetric risk-reward trade setups. Candidate
|
|
targets come from a transient, volume-free proposal ladder; persisted S/R is
|
|
reserved for human-facing charts and alerts. Stops remain ATR-based.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from collections.abc import Callable
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from sqlalchemy import and_, func, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.exceptions import NotFoundError
|
|
from app.models.fundamental import FundamentalData
|
|
from app.models.ohlcv import OHLCVRecord
|
|
from app.models.paper_trade import PaperTrade
|
|
from app.models.score import CompositeScore, DimensionScore
|
|
from app.models.sentiment import SentimentScore
|
|
from app.models.signal_context_snapshot import SignalContextSnapshot
|
|
from app.models.ticker import Ticker
|
|
from app.models.trade_setup import TradeSetup
|
|
from app.services.indicator_service import _extract_ohlcv, compute_atr
|
|
from app.services.price_service import query_ohlcv
|
|
from app.services.qualification import setup_qualifies
|
|
from app.services.sr_service import detect_gate_target_ladder
|
|
from app.services.trade_policy import (
|
|
get_reentry_gate_locks,
|
|
observe_reentry_gate_transitions,
|
|
)
|
|
from app.services.recommendation_service import (
|
|
_risk_level_from_conflicts,
|
|
build_recommendation_snapshot,
|
|
enhance_trade_setup,
|
|
get_recommendation_config,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
STRATEGY_VERSION = "residual_highvol_80_20_atr_trail3_v1"
|
|
PRIMARY_TARGET_MIN_RR = 1.5
|
|
|
|
# A setup counts as live only while the daily scan keeps re-emitting it. The
|
|
# scan runs every day (07:00 UTC cron), so anything older than this was NOT
|
|
# re-confirmed — typically because no level clears the R:R threshold from the
|
|
# current price anymore. Without this cutoff such rows stay "latest" forever
|
|
# (the scanner never writes a replacement) and keep surfacing on the live
|
|
# views. 3 days buffers a missed pipeline run or two; history endpoints are
|
|
# unaffected.
|
|
LIVE_SETUP_MAX_AGE_DAYS = 3
|
|
|
|
|
|
def _materialize_gate_target_levels(
|
|
highs: list[float],
|
|
lows: list[float],
|
|
closes: list[float],
|
|
) -> list[Any]:
|
|
"""Create transient level objects for target generation, never persistence."""
|
|
detected = detect_gate_target_ladder(highs, lows, closes)
|
|
return [
|
|
SimpleNamespace(
|
|
id=-(index + 1),
|
|
price_level=float(level["price_level"]),
|
|
type=str(level["type"]),
|
|
strength=int(level["strength"]),
|
|
detection_method=str(level.get("detection_method", "range_grid")),
|
|
sources=list(level.get("sources") or ["range_grid"]),
|
|
rejection_count=int(level.get("rejection_count", 0) or 0),
|
|
last_rejection_age=level.get("last_rejection_age"),
|
|
)
|
|
for index, level in enumerate(detected)
|
|
]
|
|
|
|
|
|
async def _get_ticker(db: AsyncSession, symbol: str) -> Ticker:
|
|
normalised = symbol.strip().upper()
|
|
result = await db.execute(select(Ticker).where(Ticker.symbol == normalised))
|
|
ticker = result.scalar_one_or_none()
|
|
if ticker is None:
|
|
raise NotFoundError(f"Ticker not found: {normalised}")
|
|
return ticker
|
|
|
|
|
|
async def _mark_ticker_scores_stale(db: AsyncSession, symbol: str) -> None:
|
|
"""Prevent a failed refresh from being presented as a current signal."""
|
|
result = await db.execute(
|
|
select(Ticker.id).where(Ticker.symbol == symbol.strip().upper())
|
|
)
|
|
ticker_id = result.scalar_one_or_none()
|
|
if ticker_id is None:
|
|
raise NotFoundError(f"Ticker not found: {symbol.strip().upper()}")
|
|
|
|
await db.execute(
|
|
update(DimensionScore)
|
|
.where(DimensionScore.ticker_id == ticker_id)
|
|
.values(is_stale=True)
|
|
)
|
|
await db.execute(
|
|
update(CompositeScore)
|
|
.where(CompositeScore.ticker_id == ticker_id)
|
|
.values(is_stale=True)
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
def _compute_quality_score(
|
|
rr: float,
|
|
strength: int,
|
|
distance: float,
|
|
entry_price: float,
|
|
*,
|
|
w_rr: float = 0.35,
|
|
w_strength: float = 0.35,
|
|
w_proximity: float = 0.30,
|
|
rr_cap: float = 10.0,
|
|
) -> float:
|
|
"""Compute a quality score for a candidate S/R level."""
|
|
norm_rr = min(rr / rr_cap, 1.0)
|
|
norm_strength = strength / 100.0
|
|
norm_proximity = 1.0 - min(distance / entry_price, 1.0)
|
|
return w_rr * norm_rr + w_strength * norm_strength + w_proximity * norm_proximity
|
|
|
|
|
|
async def _get_dimension_scores(db: AsyncSession, ticker_id: int) -> dict[str, float]:
|
|
result = await db.execute(
|
|
select(DimensionScore).where(DimensionScore.ticker_id == ticker_id)
|
|
)
|
|
rows = result.scalars().all()
|
|
return {row.dimension: float(row.score) for row in rows}
|
|
|
|
|
|
async def _get_latest_sentiment(db: AsyncSession, ticker_id: int) -> str | None:
|
|
result = await db.execute(
|
|
select(SentimentScore)
|
|
.where(SentimentScore.ticker_id == ticker_id)
|
|
.order_by(SentimentScore.timestamp.desc())
|
|
.limit(1)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
return row.classification if row else None
|
|
|
|
|
|
async def _apply_live_recommendation_context(
|
|
db: AsyncSession,
|
|
setup_rows: list[tuple[TradeSetup, str]],
|
|
rows: list[dict],
|
|
) -> list[dict]:
|
|
"""Decorate latest setup rows with current score/sentiment recommendation data.
|
|
|
|
This intentionally updates only the API payload. Stored trade setups and
|
|
history remain point-in-time records for outcome analysis.
|
|
"""
|
|
if not rows or not setup_rows:
|
|
return rows
|
|
|
|
ticker_ids = {setup.ticker_id for setup, _ in setup_rows}
|
|
setups_by_id = {setup.id: setup for setup, _ in setup_rows}
|
|
directions_by_ticker = await _latest_available_directions_by_ticker(db, ticker_ids)
|
|
|
|
dim_result = await db.execute(
|
|
select(DimensionScore).where(DimensionScore.ticker_id.in_(ticker_ids))
|
|
)
|
|
dims_by_ticker: dict[int, dict[str, float]] = {}
|
|
stale_score_ticker_ids: set[int] = set()
|
|
for ds in dim_result.scalars().all():
|
|
dims_by_ticker.setdefault(ds.ticker_id, {})[ds.dimension] = float(ds.score)
|
|
if ds.is_stale:
|
|
stale_score_ticker_ids.add(ds.ticker_id)
|
|
|
|
comp_result = await db.execute(
|
|
select(CompositeScore)
|
|
.where(CompositeScore.ticker_id.in_(ticker_ids))
|
|
.order_by(CompositeScore.ticker_id, CompositeScore.computed_at.desc())
|
|
)
|
|
composites: dict[int, CompositeScore] = {}
|
|
for comp in comp_result.scalars().all():
|
|
composites.setdefault(comp.ticker_id, comp)
|
|
|
|
sent_result = await db.execute(
|
|
select(SentimentScore)
|
|
.where(SentimentScore.ticker_id.in_(ticker_ids))
|
|
.order_by(SentimentScore.ticker_id, SentimentScore.timestamp.desc())
|
|
)
|
|
sentiments: dict[int, SentimentScore] = {}
|
|
for sent in sent_result.scalars().all():
|
|
sentiments.setdefault(sent.ticker_id, sent)
|
|
|
|
config = await get_recommendation_config(db)
|
|
live_rows: list[dict] = []
|
|
for row in rows:
|
|
setup = setups_by_id.get(row["id"])
|
|
if setup is None:
|
|
live_rows.append(row)
|
|
continue
|
|
|
|
ticker_id = setup.ticker_id
|
|
live_row = dict(row)
|
|
|
|
comp = composites.get(ticker_id)
|
|
if comp is not None:
|
|
live_row["composite_score"] = float(comp.score)
|
|
live_row["context_as_of"]["score_computed_at"] = comp.computed_at
|
|
|
|
if (
|
|
comp is None
|
|
or comp.is_stale
|
|
or ticker_id in stale_score_ticker_ids
|
|
):
|
|
live_row["confidence_score"] = None
|
|
live_row["recommended_action"] = "NEUTRAL"
|
|
live_row["reasoning"] = "Score refresh pending; recommendation withheld."
|
|
live_row["risk_level"] = "High"
|
|
live_rows.append(live_row)
|
|
continue
|
|
|
|
dimension_scores = dims_by_ticker.get(ticker_id)
|
|
sentiment = sentiments.get(ticker_id)
|
|
if sentiment is not None:
|
|
live_row["context_as_of"]["sentiment_at"] = sentiment.timestamp
|
|
if dimension_scores:
|
|
snapshot = build_recommendation_snapshot(
|
|
dimension_scores=dimension_scores,
|
|
sentiment_classification=sentiment.classification if sentiment else None,
|
|
config=config,
|
|
available_directions=directions_by_ticker.get(ticker_id),
|
|
)
|
|
direction = setup.direction.lower()
|
|
confidence_key = "long_confidence" if direction == "long" else "short_confidence"
|
|
live_row["confidence_score"] = round(float(snapshot[confidence_key]), 2)
|
|
live_row["recommended_action"] = snapshot["action"]
|
|
live_row["reasoning"] = snapshot["reasoning"]
|
|
setup_conflicts = _setup_specific_conflicts(live_row.get("conflict_flags", []))
|
|
live_conflicts = [str(item) for item in snapshot["conflicts"]]
|
|
live_row["conflict_flags"] = live_conflicts + setup_conflicts
|
|
live_row["risk_level"] = _risk_level_from_conflicts(live_row["conflict_flags"])
|
|
|
|
live_rows.append(live_row)
|
|
|
|
return live_rows
|
|
|
|
|
|
def _setup_specific_conflicts(conflicts: list[str]) -> list[str]:
|
|
signal_prefixes = (
|
|
"sentiment-technical:",
|
|
"sentiment-momentum:",
|
|
"momentum-technical:",
|
|
"fundamental-technical:",
|
|
)
|
|
return [
|
|
str(conflict)
|
|
for conflict in conflicts
|
|
if not str(conflict).startswith(signal_prefixes)
|
|
]
|
|
|
|
|
|
async def _latest_available_directions_by_ticker(
|
|
db: AsyncSession,
|
|
ticker_ids: set[int],
|
|
) -> dict[int, set[str]]:
|
|
if not ticker_ids:
|
|
return {}
|
|
|
|
result = await db.execute(
|
|
select(TradeSetup)
|
|
.where(TradeSetup.ticker_id.in_(ticker_ids))
|
|
.order_by(
|
|
TradeSetup.ticker_id,
|
|
TradeSetup.direction,
|
|
TradeSetup.detected_at.desc(),
|
|
TradeSetup.id.desc(),
|
|
)
|
|
)
|
|
latest_by_key: set[tuple[int, str]] = set()
|
|
directions: dict[int, set[str]] = {}
|
|
for setup in result.scalars().all():
|
|
direction = setup.direction.lower()
|
|
key = (setup.ticker_id, direction)
|
|
if key in latest_by_key:
|
|
continue
|
|
latest_by_key.add(key)
|
|
directions.setdefault(setup.ticker_id, set()).add(direction)
|
|
return directions
|
|
|
|
|
|
def _json_default(value):
|
|
if isinstance(value, (datetime, date)):
|
|
return value.isoformat()
|
|
return str(value)
|
|
|
|
|
|
async def _create_signal_context_snapshots(
|
|
db: AsyncSession,
|
|
setups: list[TradeSetup],
|
|
*,
|
|
strategy_version: str = STRATEGY_VERSION,
|
|
) -> None:
|
|
"""Capture point-in-time discretionary context for freshly generated setups.
|
|
|
|
The scanner stores the setup itself first so each snapshot can be keyed by
|
|
``trade_setup_id``. This is intentionally forward-only: old sentiment,
|
|
fundamentals and composite scores are not reconstructed from today's data.
|
|
"""
|
|
if not setups:
|
|
return
|
|
|
|
ticker_ids = {s.ticker_id for s in setups}
|
|
|
|
dims: dict[int, dict[str, dict]] = {}
|
|
dim_rows = (
|
|
await db.execute(select(DimensionScore).where(DimensionScore.ticker_id.in_(ticker_ids)))
|
|
).scalars().all()
|
|
for row in dim_rows:
|
|
dims.setdefault(row.ticker_id, {})[row.dimension] = {
|
|
"score": float(row.score),
|
|
"is_stale": bool(row.is_stale),
|
|
"computed_at": row.computed_at,
|
|
}
|
|
|
|
composites: dict[int, CompositeScore] = {}
|
|
comp_rows = (
|
|
await db.execute(
|
|
select(CompositeScore)
|
|
.where(CompositeScore.ticker_id.in_(ticker_ids))
|
|
.order_by(CompositeScore.ticker_id, CompositeScore.computed_at.desc())
|
|
)
|
|
).scalars().all()
|
|
for row in comp_rows:
|
|
composites.setdefault(row.ticker_id, row)
|
|
|
|
sentiments: dict[int, SentimentScore] = {}
|
|
sent_rows = (
|
|
await db.execute(
|
|
select(SentimentScore)
|
|
.where(SentimentScore.ticker_id.in_(ticker_ids))
|
|
.order_by(SentimentScore.ticker_id, SentimentScore.timestamp.desc())
|
|
)
|
|
).scalars().all()
|
|
for row in sent_rows:
|
|
sentiments.setdefault(row.ticker_id, row)
|
|
|
|
fundamentals: dict[int, FundamentalData] = {}
|
|
fund_rows = (
|
|
await db.execute(
|
|
select(FundamentalData)
|
|
.where(FundamentalData.ticker_id.in_(ticker_ids))
|
|
.order_by(FundamentalData.ticker_id, FundamentalData.fetched_at.desc())
|
|
)
|
|
).scalars().all()
|
|
for row in fund_rows:
|
|
fundamentals.setdefault(row.ticker_id, row)
|
|
|
|
now = datetime.now(timezone.utc)
|
|
for setup in setups:
|
|
comp = composites.get(setup.ticker_id)
|
|
sent = sentiments.get(setup.ticker_id)
|
|
fund = fundamentals.get(setup.ticker_id)
|
|
score_context = {
|
|
"composite_score": float(comp.score) if comp else float(setup.composite_score),
|
|
"composite_is_stale": bool(comp.is_stale) if comp else None,
|
|
"composite_computed_at": comp.computed_at if comp else None,
|
|
"momentum_percentile": (
|
|
float(setup.momentum_percentile)
|
|
if setup.momentum_percentile is not None
|
|
else None
|
|
),
|
|
"volatility_percentile": (
|
|
float(setup.volatility_percentile)
|
|
if setup.volatility_percentile is not None
|
|
else None
|
|
),
|
|
"strategy_rank": (
|
|
float(setup.strategy_rank)
|
|
if setup.strategy_rank is not None
|
|
else None
|
|
),
|
|
"dimensions": dims.get(setup.ticker_id, {}),
|
|
}
|
|
sentiment_context = (
|
|
{
|
|
"classification": sent.classification,
|
|
"confidence": int(sent.confidence),
|
|
"recommendation": sent.recommendation,
|
|
"timestamp": sent.timestamp,
|
|
"source": sent.source,
|
|
}
|
|
if sent
|
|
else {}
|
|
)
|
|
fundamental_context = (
|
|
{
|
|
"pe_ratio": fund.pe_ratio,
|
|
"revenue_growth": fund.revenue_growth,
|
|
"earnings_surprise": fund.earnings_surprise,
|
|
"market_cap": fund.market_cap,
|
|
"next_earnings_date": fund.next_earnings_date,
|
|
"fetched_at": fund.fetched_at,
|
|
}
|
|
if fund
|
|
else {}
|
|
)
|
|
db.add(
|
|
SignalContextSnapshot(
|
|
trade_setup_id=setup.id,
|
|
ticker_id=setup.ticker_id,
|
|
detected_at=setup.detected_at,
|
|
created_at=now,
|
|
strategy_version=strategy_version,
|
|
direction=setup.direction,
|
|
entry_price=float(setup.entry_price),
|
|
stop_loss=float(setup.stop_loss),
|
|
target=float(setup.target),
|
|
rr_ratio=float(setup.rr_ratio),
|
|
confidence_score=(
|
|
float(setup.confidence_score) if setup.confidence_score is not None else None
|
|
),
|
|
recommended_action=setup.recommended_action,
|
|
risk_level=setup.risk_level,
|
|
momentum_percentile=(
|
|
float(setup.momentum_percentile)
|
|
if setup.momentum_percentile is not None
|
|
else None
|
|
),
|
|
score_context_json=json.dumps(score_context, default=_json_default),
|
|
sentiment_context_json=json.dumps(sentiment_context, default=_json_default),
|
|
fundamental_context_json=json.dumps(fundamental_context, default=_json_default),
|
|
)
|
|
)
|
|
|
|
|
|
async def resolve_activation_ranks_for_symbol(
|
|
db: AsyncSession,
|
|
symbol: str,
|
|
) -> dict[str, float | None]:
|
|
"""Universe activation ranks for one symbol (manual single-ticker scans).
|
|
|
|
The daily ``scan_all_tickers`` path ranks the whole universe once and passes
|
|
percentiles into ``scan_ticker``. Manual refresh must do the same: without
|
|
``momentum_percentile`` the activation gate treats the setup as unranked and
|
|
it silently drops out of qualified trades.
|
|
|
|
Prefer a fresh cross-sectional rank; if ranking fails or the symbol is
|
|
missing from the universe slice, fall back to the most recent prior setup
|
|
that still carries ranks so a refresh never zeroes the gate inputs.
|
|
"""
|
|
symbol_u = symbol.strip().upper()
|
|
empty: dict[str, float | None] = {
|
|
"momentum_percentile": None,
|
|
"strategy_rank": None,
|
|
"volatility_percentile": None,
|
|
}
|
|
|
|
try:
|
|
from app.services import momentum_service
|
|
|
|
ranks = await momentum_service.compute_activation_ranks(db)
|
|
hit = ranks.get(symbol_u)
|
|
if hit is not None and hit.get("momentum_percentile") is not None:
|
|
return {
|
|
"momentum_percentile": hit.get("momentum_percentile"),
|
|
"strategy_rank": hit.get("strategy_rank"),
|
|
"volatility_percentile": hit.get("volatility_percentile"),
|
|
}
|
|
except Exception:
|
|
logger.exception(
|
|
"Activation ranking failed for single-ticker scan of %s", symbol_u
|
|
)
|
|
|
|
ticker_result = await db.execute(
|
|
select(Ticker.id).where(Ticker.symbol == symbol_u)
|
|
)
|
|
ticker_id = ticker_result.scalar_one_or_none()
|
|
if ticker_id is None:
|
|
return empty
|
|
|
|
prev_result = await db.execute(
|
|
select(TradeSetup)
|
|
.where(
|
|
TradeSetup.ticker_id == ticker_id,
|
|
TradeSetup.momentum_percentile.is_not(None),
|
|
)
|
|
.order_by(TradeSetup.detected_at.desc(), TradeSetup.id.desc())
|
|
.limit(1)
|
|
)
|
|
prev = prev_result.scalar_one_or_none()
|
|
if prev is None:
|
|
return empty
|
|
|
|
return {
|
|
"momentum_percentile": (
|
|
float(prev.momentum_percentile) if prev.momentum_percentile is not None else None
|
|
),
|
|
"strategy_rank": (
|
|
float(prev.strategy_rank) if prev.strategy_rank is not None else None
|
|
),
|
|
"volatility_percentile": (
|
|
float(prev.volatility_percentile) if prev.volatility_percentile is not None else None
|
|
),
|
|
}
|
|
|
|
|
|
async def scan_ticker(
|
|
db: AsyncSession,
|
|
symbol: str,
|
|
rr_threshold: float = 1.5,
|
|
atr_multiplier: float = 1.5,
|
|
momentum_percentile: float | None = None,
|
|
strategy_rank: float | None = None,
|
|
volatility_percentile: float | None = None,
|
|
primary_min_rr: float | None = None,
|
|
gate_levels_override: list[Any] | None = None,
|
|
) -> list[TradeSetup]:
|
|
"""Scan a single ticker for trade setups meeting the R:R threshold.
|
|
|
|
``momentum_percentile`` is the ticker's residual 12-1 momentum activation
|
|
rank across the universe (computed by the caller), stored on each setup so
|
|
the activation gate can select the top slice. ``strategy_rank`` is the
|
|
production ordering score used for top-pick ranking.
|
|
|
|
``primary_min_rr`` controls target selection only. Its 1.5 default is
|
|
intentionally independent of the later activation floor (2.0 in the live
|
|
Admin configuration). ``gate_levels_override`` is dependency injection for
|
|
deterministic scanner tests; production builds the transient ladder from
|
|
the ticker's OHLCV window.
|
|
"""
|
|
ticker = await _get_ticker(db, symbol)
|
|
|
|
if primary_min_rr is None:
|
|
primary_min_rr = PRIMARY_TARGET_MIN_RR
|
|
|
|
records = await query_ohlcv(db, symbol)
|
|
if not records or len(records) < 15:
|
|
logger.info(
|
|
"Skipping %s: insufficient OHLCV data (%d bars, need 15+)",
|
|
symbol, len(records),
|
|
)
|
|
return []
|
|
|
|
_, highs, lows, closes, _ = _extract_ohlcv(records)
|
|
entry_price = closes[-1]
|
|
|
|
try:
|
|
atr_result = compute_atr(highs, lows, closes)
|
|
atr_value = atr_result["atr"]
|
|
except Exception:
|
|
logger.info("Skipping %s: cannot compute ATR", symbol)
|
|
return []
|
|
|
|
if atr_value <= 0:
|
|
logger.info("Skipping %s: ATR is zero or negative", symbol)
|
|
return []
|
|
|
|
gate_levels = (
|
|
list(gate_levels_override)
|
|
if gate_levels_override is not None
|
|
else _materialize_gate_target_levels(highs, lows, closes)
|
|
)
|
|
|
|
if not gate_levels:
|
|
logger.info("Skipping %s: no gate target levels available", symbol)
|
|
return []
|
|
|
|
levels_above = sorted(
|
|
[lv for lv in gate_levels if lv.price_level > entry_price],
|
|
key=lambda lv: lv.price_level,
|
|
)
|
|
levels_below = sorted(
|
|
[lv for lv in gate_levels if lv.price_level < entry_price],
|
|
key=lambda lv: lv.price_level,
|
|
reverse=True,
|
|
)
|
|
|
|
comp_result = await db.execute(
|
|
select(CompositeScore).where(CompositeScore.ticker_id == ticker.id)
|
|
)
|
|
comp = comp_result.scalar_one_or_none()
|
|
composite_score = comp.score if comp else 0.0
|
|
|
|
dimension_scores = await _get_dimension_scores(db, ticker.id)
|
|
sentiment_classification = await _get_latest_sentiment(db, ticker.id)
|
|
|
|
now = datetime.now(timezone.utc)
|
|
setups: list[TradeSetup] = []
|
|
|
|
if levels_above:
|
|
stop = entry_price - (atr_value * atr_multiplier)
|
|
risk = entry_price - stop
|
|
if risk > 0:
|
|
best_quality = 0.0
|
|
best_candidate_rr = 0.0
|
|
best_candidate_target = 0.0
|
|
for lv in levels_above:
|
|
reward = lv.price_level - entry_price
|
|
if reward <= 0:
|
|
continue
|
|
rr = reward / risk
|
|
if rr < rr_threshold:
|
|
continue
|
|
distance = lv.price_level - entry_price
|
|
quality = _compute_quality_score(rr, lv.strength, distance, entry_price)
|
|
if quality > best_quality:
|
|
best_quality = quality
|
|
best_candidate_rr = rr
|
|
best_candidate_target = lv.price_level
|
|
|
|
if best_candidate_rr > 0:
|
|
setups.append(TradeSetup(
|
|
ticker_id=ticker.id,
|
|
direction="long",
|
|
entry_price=round(entry_price, 4),
|
|
stop_loss=round(stop, 4),
|
|
target=round(best_candidate_target, 4),
|
|
rr_ratio=round(best_candidate_rr, 4),
|
|
composite_score=round(composite_score, 4),
|
|
detected_at=now,
|
|
momentum_percentile=momentum_percentile,
|
|
strategy_rank=strategy_rank,
|
|
volatility_percentile=volatility_percentile,
|
|
))
|
|
|
|
if levels_below:
|
|
stop = entry_price + (atr_value * atr_multiplier)
|
|
risk = stop - entry_price
|
|
if risk > 0:
|
|
best_quality = 0.0
|
|
best_candidate_rr = 0.0
|
|
best_candidate_target = 0.0
|
|
for lv in levels_below:
|
|
reward = entry_price - lv.price_level
|
|
if reward <= 0:
|
|
continue
|
|
rr = reward / risk
|
|
if rr < rr_threshold:
|
|
continue
|
|
distance = entry_price - lv.price_level
|
|
quality = _compute_quality_score(rr, lv.strength, distance, entry_price)
|
|
if quality > best_quality:
|
|
best_quality = quality
|
|
best_candidate_rr = rr
|
|
best_candidate_target = lv.price_level
|
|
|
|
if best_candidate_rr > 0:
|
|
setups.append(TradeSetup(
|
|
ticker_id=ticker.id,
|
|
direction="short",
|
|
entry_price=round(entry_price, 4),
|
|
stop_loss=round(stop, 4),
|
|
target=round(best_candidate_target, 4),
|
|
rr_ratio=round(best_candidate_rr, 4),
|
|
composite_score=round(composite_score, 4),
|
|
detected_at=now,
|
|
momentum_percentile=momentum_percentile,
|
|
strategy_rank=strategy_rank,
|
|
volatility_percentile=volatility_percentile,
|
|
))
|
|
|
|
available_directions = {s.direction for s in setups}
|
|
enhanced_setups: list[TradeSetup] = []
|
|
for setup in setups:
|
|
try:
|
|
enhanced = await enhance_trade_setup(
|
|
db=db,
|
|
ticker=ticker,
|
|
setup=setup,
|
|
dimension_scores=dimension_scores,
|
|
sr_levels=gate_levels,
|
|
sentiment_classification=sentiment_classification,
|
|
atr_value=atr_value,
|
|
primary_min_rr=primary_min_rr,
|
|
available_directions=available_directions,
|
|
)
|
|
enhanced_setups.append(enhanced)
|
|
except Exception:
|
|
logger.exception("Error enhancing setup for %s (%s)", ticker.symbol, setup.direction)
|
|
enhanced_setups.append(setup)
|
|
|
|
for setup in enhanced_setups:
|
|
db.add(setup)
|
|
|
|
await db.commit()
|
|
|
|
for s in enhanced_setups:
|
|
await db.refresh(s)
|
|
|
|
await _create_signal_context_snapshots(db, enhanced_setups)
|
|
await db.commit()
|
|
|
|
return enhanced_setups
|
|
|
|
|
|
async def scan_all_tickers(
|
|
db: AsyncSession,
|
|
rr_threshold: float = 1.5,
|
|
atr_multiplier: float = 1.5,
|
|
progress_callback: Callable[[int, int, str], None] | None = None,
|
|
) -> list[TradeSetup]:
|
|
"""Scan all tracked tickers for trade setups.
|
|
|
|
``progress_callback(processed, total, current_symbol)`` is invoked as each
|
|
ticker is scanned so callers (e.g. the scheduler) can surface live progress.
|
|
"""
|
|
# Plain ids/strings, not Ticker instances: the rollbacks below expire any
|
|
# ORM objects held across them, and touching an expired attribute afterwards
|
|
# triggers sync lazy-loading, which raises on an AsyncSession.
|
|
result = await db.execute(select(Ticker.id, Ticker.symbol).order_by(Ticker.symbol))
|
|
ticker_rows = [(int(ticker_id), symbol) for ticker_id, symbol in result.all()]
|
|
total = len(ticker_rows)
|
|
|
|
# Gate-reset observations must use the same runtime activation settings as
|
|
# the live setup list. If the config cannot be loaded, scan normally but do
|
|
# not mutate reset state from an evaluation whose rules are unknown.
|
|
activation: dict | None = None
|
|
try:
|
|
from app.services.admin_service import get_activation_config
|
|
|
|
activation = await get_activation_config(db)
|
|
except Exception:
|
|
await db.rollback()
|
|
logger.exception("Activation config load for re-entry gate reset failed")
|
|
|
|
# Rank the universe up front so each new setup carries both the residual
|
|
# activation gate percentile and the promoted production ordering score.
|
|
# Best-effort; the ranker falls back to raw 12-1 momentum only if benchmark
|
|
# data is unavailable.
|
|
try:
|
|
from app.services import momentum_service
|
|
|
|
ranks = await momentum_service.compute_activation_ranks(db)
|
|
except Exception:
|
|
await db.rollback()
|
|
logger.exception("Activation ranking refresh failed")
|
|
ranks = {}
|
|
|
|
all_setups: list[TradeSetup] = []
|
|
evaluated_ticker_ids: set[int] = set()
|
|
qualified_ticker_ids: set[int] = set()
|
|
gate_observation_started_at = datetime.now(timezone.utc)
|
|
for index, (ticker_id, symbol) in enumerate(ticker_rows):
|
|
if progress_callback is not None:
|
|
progress_callback(index, total, symbol)
|
|
# Refresh scores first so the scheduled scan works off current data.
|
|
# Nothing else marks scores stale, so without this they'd never update
|
|
# for tickers the user doesn't manually fetch. A refresh failure still
|
|
# scans the ticker: qualification re-gates on live scores at alert
|
|
# time, so a stale score is recoverable but a skipped scan is not.
|
|
try:
|
|
from app.services import scoring_service
|
|
|
|
await scoring_service.compute_all_dimensions(db, symbol)
|
|
await scoring_service.compute_composite_score(db, symbol)
|
|
await db.commit()
|
|
except Exception:
|
|
await db.rollback()
|
|
logger.exception("Error refreshing scores for %s", symbol)
|
|
try:
|
|
await _mark_ticker_scores_stale(db, symbol)
|
|
except Exception:
|
|
await db.rollback()
|
|
logger.exception("Could not mark scores stale for %s", symbol)
|
|
continue
|
|
|
|
try:
|
|
setups = await scan_ticker(
|
|
db, symbol, rr_threshold, atr_multiplier,
|
|
momentum_percentile=(ranks.get(symbol) or {}).get("momentum_percentile"),
|
|
strategy_rank=(ranks.get(symbol) or {}).get("strategy_rank"),
|
|
volatility_percentile=(ranks.get(symbol) or {}).get("volatility_percentile"),
|
|
primary_min_rr=PRIMARY_TARGET_MIN_RR,
|
|
)
|
|
all_setups.extend(setups)
|
|
if activation is not None:
|
|
try:
|
|
if any(setup_qualifies(setup, activation) for setup in setups):
|
|
qualified_ticker_ids.add(ticker_id)
|
|
evaluated_ticker_ids.add(ticker_id)
|
|
except Exception:
|
|
logger.exception(
|
|
"Gate-reset qualification observation failed for %s", symbol
|
|
)
|
|
except Exception:
|
|
await db.rollback()
|
|
logger.exception("Error scanning ticker %s", symbol)
|
|
|
|
if activation is not None:
|
|
transitioned_ticker_ids = await observe_reentry_gate_transitions(
|
|
db,
|
|
evaluated_ticker_ids=evaluated_ticker_ids,
|
|
qualified_ticker_ids=qualified_ticker_ids,
|
|
observed_at=gate_observation_started_at,
|
|
)
|
|
await db.commit()
|
|
if transitioned_ticker_ids:
|
|
logger.info(
|
|
"Updated post-stop gate-reset state for %d ticker(s)",
|
|
len(transitioned_ticker_ids),
|
|
)
|
|
|
|
if progress_callback is not None and total:
|
|
progress_callback(total, total, "")
|
|
|
|
return all_setups
|
|
|
|
|
|
async def get_trade_setups(
|
|
db: AsyncSession,
|
|
direction: str | None = None,
|
|
min_confidence: float | None = None,
|
|
recommended_action: str | None = None,
|
|
symbol: str | None = None,
|
|
live_recommendation: bool = False,
|
|
exclude_open_trade_tickers: bool = False,
|
|
exclude_reentry_gate_locked_tickers: bool = False,
|
|
include_reentry_gate_lock: bool = False,
|
|
) -> list[dict]:
|
|
"""Get latest stored trade setups, optionally filtered.
|
|
|
|
Only setups the daily scan re-emitted within ``LIVE_SETUP_MAX_AGE_DAYS``
|
|
are returned — an older "latest" row means the scanner no longer finds a
|
|
valid setup for that ticker, so it must not surface as current.
|
|
"""
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=LIVE_SETUP_MAX_AGE_DAYS)
|
|
stmt = (
|
|
select(TradeSetup, Ticker.symbol)
|
|
.join(Ticker, TradeSetup.ticker_id == Ticker.id)
|
|
.where(TradeSetup.detected_at >= cutoff)
|
|
)
|
|
if direction is not None:
|
|
stmt = stmt.where(TradeSetup.direction == direction.lower())
|
|
if symbol is not None:
|
|
stmt = stmt.where(Ticker.symbol == symbol.strip().upper())
|
|
# With live_recommendation these fields are overlaid with current values
|
|
# below, so filtering happens there instead of against the stored columns.
|
|
if min_confidence is not None and not live_recommendation:
|
|
stmt = stmt.where(TradeSetup.confidence_score >= min_confidence)
|
|
if recommended_action is not None and not live_recommendation:
|
|
stmt = stmt.where(TradeSetup.recommended_action == recommended_action)
|
|
excluded_ticker_ids: set[int] = set()
|
|
reentry_gate_locks: dict[int, datetime] = {}
|
|
if exclude_open_trade_tickers:
|
|
open_trade_result = await db.execute(
|
|
select(PaperTrade.ticker_id)
|
|
.where(PaperTrade.status == "open")
|
|
.distinct()
|
|
)
|
|
excluded_ticker_ids.update(
|
|
ticker_id for ticker_id, in open_trade_result.all()
|
|
)
|
|
if exclude_reentry_gate_locked_tickers or include_reentry_gate_lock:
|
|
reentry_gate_locks = await get_reentry_gate_locks(db)
|
|
if exclude_reentry_gate_locked_tickers:
|
|
excluded_ticker_ids.update(reentry_gate_locks)
|
|
if excluded_ticker_ids:
|
|
stmt = stmt.where(~TradeSetup.ticker_id.in_(excluded_ticker_ids))
|
|
|
|
stmt = stmt.order_by(TradeSetup.detected_at.desc(), TradeSetup.id.desc())
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
latest_by_key: dict[tuple[str, str], tuple[TradeSetup, str]] = {}
|
|
for setup, ticker_symbol in rows:
|
|
dedupe_key = (ticker_symbol, setup.direction)
|
|
if dedupe_key not in latest_by_key:
|
|
latest_by_key[dedupe_key] = (setup, ticker_symbol)
|
|
|
|
latest_rows = list(latest_by_key.values())
|
|
latest_rows.sort(
|
|
key=lambda row: (
|
|
row[0].strategy_rank if row[0].strategy_rank is not None else -1.0,
|
|
row[0].momentum_percentile if row[0].momentum_percentile is not None else -1.0,
|
|
row[0].confidence_score if row[0].confidence_score is not None else -1.0,
|
|
row[0].rr_ratio,
|
|
row[0].composite_score,
|
|
),
|
|
reverse=True,
|
|
)
|
|
|
|
prices = await _latest_price_context(db, {s.ticker_id for s, _ in latest_rows})
|
|
rows_out = [
|
|
_trade_setup_to_dict(setup, ticker_symbol, prices.get(setup.ticker_id))
|
|
for setup, ticker_symbol in latest_rows
|
|
]
|
|
if live_recommendation:
|
|
rows_out = await _apply_live_recommendation_context(db, latest_rows, rows_out)
|
|
if min_confidence is not None:
|
|
rows_out = [
|
|
row for row in rows_out
|
|
if row["confidence_score"] is not None
|
|
and row["confidence_score"] >= min_confidence
|
|
]
|
|
if recommended_action is not None:
|
|
rows_out = [
|
|
row for row in rows_out
|
|
if row["recommended_action"] == recommended_action
|
|
]
|
|
rows_out.sort(
|
|
key=lambda row: (
|
|
row["strategy_rank"] if row["strategy_rank"] is not None else -1.0,
|
|
row["momentum_percentile"] if row["momentum_percentile"] is not None else -1.0,
|
|
row["confidence_score"] if row["confidence_score"] is not None else -1.0,
|
|
row["rr_ratio"],
|
|
row["composite_score"],
|
|
),
|
|
reverse=True,
|
|
)
|
|
if include_reentry_gate_lock:
|
|
ticker_by_setup_id = {
|
|
setup.id: setup.ticker_id for setup, _ in latest_rows
|
|
}
|
|
for row in rows_out:
|
|
ticker_id = ticker_by_setup_id.get(row["id"])
|
|
row["reentry_gate_reset_required"] = (
|
|
ticker_id in reentry_gate_locks if ticker_id is not None else False
|
|
)
|
|
return rows_out
|
|
|
|
|
|
async def _latest_price_context(db: AsyncSession, ticker_ids: set[int]) -> dict[int, dict]:
|
|
"""Most recent daily OHLCV row per ticker for live price context."""
|
|
if not ticker_ids:
|
|
return {}
|
|
latest = (
|
|
select(OHLCVRecord.ticker_id, func.max(OHLCVRecord.date).label("md"))
|
|
.where(OHLCVRecord.ticker_id.in_(ticker_ids))
|
|
.group_by(OHLCVRecord.ticker_id)
|
|
.subquery()
|
|
)
|
|
stmt = select(
|
|
OHLCVRecord.ticker_id,
|
|
OHLCVRecord.close,
|
|
OHLCVRecord.date,
|
|
OHLCVRecord.created_at,
|
|
).join(
|
|
latest,
|
|
and_(
|
|
OHLCVRecord.ticker_id == latest.c.ticker_id,
|
|
OHLCVRecord.date == latest.c.md,
|
|
),
|
|
)
|
|
result = await db.execute(stmt)
|
|
return {
|
|
tid: {
|
|
"current_price": float(close),
|
|
"price_date": price_date,
|
|
"price_updated_at": created_at,
|
|
}
|
|
for tid, close, price_date, created_at in result.all()
|
|
}
|
|
|
|
|
|
async def _latest_closes(db: AsyncSession, ticker_ids: set[int]) -> dict[int, float]:
|
|
"""Most recent close per ticker, kept for callers that only need price."""
|
|
price_context = await _latest_price_context(db, ticker_ids)
|
|
return {
|
|
ticker_id: context["current_price"]
|
|
for ticker_id, context in price_context.items()
|
|
}
|
|
|
|
|
|
async def get_trade_setup_history(
|
|
db: AsyncSession,
|
|
symbol: str,
|
|
) -> list[dict]:
|
|
"""Get full recommendation history for a symbol (newest first)."""
|
|
stmt = (
|
|
select(TradeSetup, Ticker.symbol)
|
|
.join(Ticker, TradeSetup.ticker_id == Ticker.id)
|
|
.where(Ticker.symbol == symbol.strip().upper())
|
|
.order_by(TradeSetup.detected_at.desc(), TradeSetup.id.desc())
|
|
)
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
prices = await _latest_price_context(db, {s.ticker_id for s, _ in rows})
|
|
return [
|
|
_trade_setup_to_dict(setup, ticker_symbol, prices.get(setup.ticker_id))
|
|
for setup, ticker_symbol in rows
|
|
]
|
|
|
|
|
|
def _trade_setup_to_dict(setup: TradeSetup, symbol: str, price_context: dict | None = None) -> dict:
|
|
targets: list[dict] = []
|
|
conflicts: list[str] = []
|
|
current_price = (
|
|
float(price_context["current_price"])
|
|
if price_context and price_context.get("current_price") is not None
|
|
else None
|
|
)
|
|
context_as_of = {
|
|
"setup_detected_at": setup.detected_at,
|
|
"score_computed_at": None,
|
|
"sentiment_at": None,
|
|
"price_date": price_context.get("price_date") if price_context else None,
|
|
"price_updated_at": price_context.get("price_updated_at") if price_context else None,
|
|
}
|
|
|
|
if setup.targets_json:
|
|
try:
|
|
parsed_targets = json.loads(setup.targets_json)
|
|
if isinstance(parsed_targets, list):
|
|
targets = parsed_targets
|
|
except (TypeError, ValueError):
|
|
targets = []
|
|
|
|
if setup.conflict_flags_json:
|
|
try:
|
|
parsed_conflicts = json.loads(setup.conflict_flags_json)
|
|
if isinstance(parsed_conflicts, list):
|
|
conflicts = [str(item) for item in parsed_conflicts]
|
|
except (TypeError, ValueError):
|
|
conflicts = []
|
|
|
|
return {
|
|
"id": setup.id,
|
|
"symbol": symbol,
|
|
"direction": setup.direction,
|
|
"entry_price": setup.entry_price,
|
|
"stop_loss": setup.stop_loss,
|
|
"target": setup.target,
|
|
"rr_ratio": setup.rr_ratio,
|
|
"composite_score": setup.composite_score,
|
|
"detected_at": setup.detected_at,
|
|
"confidence_score": setup.confidence_score,
|
|
"targets": targets,
|
|
"conflict_flags": conflicts,
|
|
"recommended_action": setup.recommended_action,
|
|
"reasoning": setup.reasoning,
|
|
"risk_level": setup.risk_level,
|
|
"actual_outcome": setup.actual_outcome,
|
|
"outcome_date": setup.outcome_date,
|
|
"evaluated_at": setup.evaluated_at,
|
|
"current_price": current_price,
|
|
"momentum_percentile": setup.momentum_percentile,
|
|
"strategy_rank": setup.strategy_rank,
|
|
"volatility_percentile": setup.volatility_percentile,
|
|
"context_as_of": context_as_of,
|
|
}
|