"""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 import fundamentals_quality_service, system_event_service 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 import settings_store from app.services.trade_policy import ( MANUAL_BOOK, SHADOW_BOOK, get_reentry_gate_locks, observe_reentry_gate_transitions, ) from app.services.recommendation_service import ( PRIMARY_TARGET_MIN_RR, _risk_level_from_conflicts, build_recommendation_snapshot, enhance_trade_setup, get_recommendation_config, ) logger = logging.getLogger(__name__) # Markers of the most recent *successful* scan, written together only when # scan_all_tickers completes. COMPLETED gives its freshness; RUN_ID identifies # the run — the same id stamped on every setup row it produced. The shadow book # matches RUN_ID exactly and then selects setups by that id, so neither a # concurrent manual scan nor a stale prior run can be mistaken for it. KEY_LAST_SCAN_COMPLETED = "last_scan_run_completed_at" KEY_LAST_SCAN_RUN_ID = "last_scan_run_id" STRATEGY_VERSION = "residual_highvol_80_20_atr_trail3_v1" # 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, scan_run_id: str | None = None, fundamentals_eligible: bool | 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 fundamentals_eligible is None: fundamentals_eligible = await fundamentals_quality_service.ticker_is_eligible( db, ticker.id ) if not fundamentals_eligible: logger.info( "Skipping %s: unresolved or unavailable SEC fundamentals", ticker.symbol, ) return [] 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: # Stamp identity after enhancement so it survives regardless of how the # enhancer rebuilds the row; the shadow book selects its batch by this. setup.scan_run_id = scan_run_id 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) # Data-quality failures are not weak signals: they make a ticker ineligible. # Resolve once for the universe scan and pass the decision into scan_ticker. try: fundamentals_blocked_ids = ( await fundamentals_quality_service.blocked_ticker_ids(db) ) except Exception: await db.rollback() logger.exception( "Could not resolve fundamentals quality; blocking this scan closed" ) await system_event_service.log_event_standalone( severity="error", source="rr_scanner", code="fundamentals_quality_unavailable", message=( "The fundamentals quality gate could not be evaluated; the " "universe scan was blocked to avoid issuing unchecked setups." ), dedup_key="rr_scanner:fundamentals_quality_unavailable", ) fundamentals_blocked_ids = {ticker_id for ticker_id, _ in 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) # One id for the whole run: stamped on every setup row and written to the # completion marker, so the shadow book can select this run's batch by # identity. From the pipeline when run as its scan step; a fresh id (never # matching any pipeline's) when triggered standalone. from app.services import pipeline_run scan_run_id = pipeline_run.current() or pipeline_run.new_run_id() for index, (ticker_id, symbol) in enumerate(ticker_rows): if progress_callback is not None: progress_callback(index, total, symbol) if ticker_id in fundamentals_blocked_ids: logger.info( "Skipping %s: unresolved or unavailable SEC fundamentals", symbol, ) continue # Refresh Structural S/R once, then scores. get_sr_levels is read-only; # without this recalculate the score path would see yesterday's zones. # 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, sr_service await sr_service.recalculate_sr_levels(db, symbol) 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, scan_run_id=scan_run_id, fundamentals_eligible=True, ) 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: # Both books, from the same observation: gate-reset state is per book, # so observing only the manual book would leave shadow stop-outs stuck # with a fail timestamp that never requalifies — permanently ineligible. transitioned_ticker_ids: set[int] = set() for book in (MANUAL_BOOK, SHADOW_BOOK): 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, book=book, ) 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, "") # Publish the run markers only now that the scan has completed: COMPLETED for # freshness and RUN_ID (the same id stamped on this run's setup rows) for # identity, in one commit. A hard failure above leaves the previous, # now-superseded, markers in place — so the shadow book will not match. await settings_store.upsert_setting( db, KEY_LAST_SCAN_COMPLETED, datetime.now(timezone.utc).isoformat() ) await settings_store.upsert_setting(db, KEY_LAST_SCAN_RUN_ID, scan_run_id) await db.commit() 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_open_trade_user_id: int | None = None, 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] = {} try: excluded_ticker_ids.update( await fundamentals_quality_service.blocked_ticker_ids(db) ) except Exception: await db.rollback() logger.exception( "Could not resolve fundamentals quality; hiding actionable setups" ) await system_event_service.log_event_standalone( severity="error", source="rr_scanner", code="fundamentals_quality_unavailable", message=( "The fundamentals quality gate could not be evaluated; actionable " "setups were hidden until the metadata check recovers." ), dedup_key="rr_scanner:fundamentals_quality_unavailable", ) return [] if exclude_open_trade_tickers: # Manual book only. The shadow book holds the *top-ranked* names by # construction, so letting its positions hide setups would leave the # discretionary list picking over leftovers — and would bias the very # shadow-vs-manual comparison the shadow book exists to measure. open_trade_stmt = ( select(PaperTrade.ticker_id) .where(PaperTrade.status == "open", PaperTrade.book == MANUAL_BOOK) .distinct() ) # Scope to one user for the personal setup list (don't hide a name just # because someone else holds it); leave it global for the Telegram # broadcast, which has no single owner. if exclude_open_trade_user_id is not None: open_trade_stmt = open_trade_stmt.where( PaperTrade.user_id == exclude_open_trade_user_id ) open_trade_result = await db.execute(open_trade_stmt) 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, }