Make live signal reads non-mutating
This commit is contained in:
@@ -54,7 +54,7 @@ async def read_score(
|
||||
_user=Depends(require_access),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
) -> APIEnvelope:
|
||||
"""Get composite + dimension scores for a symbol. Recomputes stale scores."""
|
||||
"""Get the latest persisted composite + dimension scores for a symbol."""
|
||||
result = await get_score(db, symbol)
|
||||
|
||||
data = ScoreResponse(
|
||||
@@ -94,6 +94,7 @@ async def read_rankings(
|
||||
RankingEntry(
|
||||
symbol=r["symbol"],
|
||||
composite_score=r["composite_score"],
|
||||
composite_stale=r.get("composite_stale", False),
|
||||
dimensions=[
|
||||
DimensionScoreResponse(**d) for d in r["dimensions"]
|
||||
],
|
||||
|
||||
@@ -97,7 +97,6 @@ async def get_ticker_trade_setups(
|
||||
db,
|
||||
symbol=symbol,
|
||||
live_recommendation=True,
|
||||
recompute_scores=True,
|
||||
)
|
||||
data = []
|
||||
for row in rows:
|
||||
|
||||
@@ -78,6 +78,7 @@ class RankingEntry(BaseModel):
|
||||
|
||||
symbol: str
|
||||
composite_score: float
|
||||
composite_stale: bool = False
|
||||
dimensions: list[DimensionScoreResponse] = []
|
||||
|
||||
|
||||
|
||||
@@ -26,6 +26,14 @@ class RecommendationSummaryResponse(BaseModel):
|
||||
composite_score: float
|
||||
|
||||
|
||||
class TradeSetupContextAsOfResponse(BaseModel):
|
||||
setup_detected_at: datetime
|
||||
score_computed_at: datetime | None = None
|
||||
sentiment_at: datetime | None = None
|
||||
price_date: date | None = None
|
||||
price_updated_at: datetime | None = None
|
||||
|
||||
|
||||
class TradeSetupResponse(BaseModel):
|
||||
"""A single trade setup detected by the R:R scanner."""
|
||||
|
||||
@@ -49,4 +57,5 @@ class TradeSetupResponse(BaseModel):
|
||||
evaluated_at: datetime | None = None
|
||||
current_price: float | None = None
|
||||
momentum_percentile: float | None = None
|
||||
context_as_of: TradeSetupContextAsOfResponse | None = None
|
||||
recommendation_summary: RecommendationSummaryResponse | None = None
|
||||
|
||||
@@ -28,6 +28,7 @@ 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.recommendation_service import (
|
||||
_risk_level_from_conflicts,
|
||||
build_recommendation_snapshot,
|
||||
enhance_trade_setup,
|
||||
get_recommendation_config,
|
||||
@@ -84,29 +85,6 @@ async def _get_latest_sentiment(db: AsyncSession, ticker_id: int) -> str | None:
|
||||
return row.classification if row else None
|
||||
|
||||
|
||||
async def _refresh_score_context_for_symbols(
|
||||
db: AsyncSession,
|
||||
symbols: set[str],
|
||||
) -> None:
|
||||
"""Refresh provider-free scores so live recommendation summaries match the page."""
|
||||
if not symbols:
|
||||
return
|
||||
|
||||
from app.services import scoring_service
|
||||
|
||||
refreshed = False
|
||||
for symbol in sorted(symbols):
|
||||
try:
|
||||
await scoring_service.compute_all_dimensions(db, symbol)
|
||||
await scoring_service.compute_composite_score(db, symbol)
|
||||
refreshed = True
|
||||
except Exception:
|
||||
logger.exception("Error refreshing live score context for %s", symbol)
|
||||
|
||||
if refreshed:
|
||||
await db.commit()
|
||||
|
||||
|
||||
async def _apply_live_recommendation_context(
|
||||
db: AsyncSession,
|
||||
setup_rows: list[tuple[TradeSetup, str]],
|
||||
@@ -122,10 +100,7 @@ async def _apply_live_recommendation_context(
|
||||
|
||||
ticker_ids = {setup.ticker_id for setup, _ in setup_rows}
|
||||
setups_by_id = {setup.id: setup for setup, _ in setup_rows}
|
||||
|
||||
directions_by_ticker: dict[int, set[str]] = {}
|
||||
for setup, _ in setup_rows:
|
||||
directions_by_ticker.setdefault(setup.ticker_id, set()).add(setup.direction.lower())
|
||||
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))
|
||||
@@ -166,10 +141,13 @@ async def _apply_live_recommendation_context(
|
||||
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
|
||||
|
||||
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:
|
||||
sentiment = sentiments.get(ticker_id)
|
||||
snapshot = build_recommendation_snapshot(
|
||||
dimension_scores=dimension_scores,
|
||||
sentiment_classification=sentiment.classification if sentiment else None,
|
||||
@@ -181,13 +159,59 @@ async def _apply_live_recommendation_context(
|
||||
live_row["confidence_score"] = round(float(snapshot[confidence_key]), 2)
|
||||
live_row["recommended_action"] = snapshot["action"]
|
||||
live_row["reasoning"] = snapshot["reasoning"]
|
||||
live_row["risk_level"] = snapshot["risk_level"]
|
||||
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()
|
||||
@@ -550,7 +574,6 @@ async def get_trade_setups(
|
||||
recommended_action: str | None = None,
|
||||
symbol: str | None = None,
|
||||
live_recommendation: bool = False,
|
||||
recompute_scores: bool = False,
|
||||
) -> list[dict]:
|
||||
"""Get latest stored trade setups, optionally filtered."""
|
||||
stmt = (
|
||||
@@ -589,12 +612,7 @@ async def get_trade_setups(
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
if recompute_scores:
|
||||
await _refresh_score_context_for_symbols(
|
||||
db, {ticker_symbol for _, ticker_symbol in latest_rows}
|
||||
)
|
||||
|
||||
prices = await _latest_closes(db, {s.ticker_id for s, _ in latest_rows})
|
||||
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
|
||||
@@ -623,8 +641,8 @@ async def get_trade_setups(
|
||||
return rows_out
|
||||
|
||||
|
||||
async def _latest_closes(db: AsyncSession, ticker_ids: set[int]) -> dict[int, float]:
|
||||
"""Most recent close per ticker — used to judge a setup's current relevance."""
|
||||
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 = (
|
||||
@@ -633,7 +651,12 @@ async def _latest_closes(db: AsyncSession, ticker_ids: set[int]) -> dict[int, fl
|
||||
.group_by(OHLCVRecord.ticker_id)
|
||||
.subquery()
|
||||
)
|
||||
stmt = select(OHLCVRecord.ticker_id, OHLCVRecord.close).join(
|
||||
stmt = select(
|
||||
OHLCVRecord.ticker_id,
|
||||
OHLCVRecord.close,
|
||||
OHLCVRecord.date,
|
||||
OHLCVRecord.created_at,
|
||||
).join(
|
||||
latest,
|
||||
and_(
|
||||
OHLCVRecord.ticker_id == latest.c.ticker_id,
|
||||
@@ -641,7 +664,23 @@ async def _latest_closes(db: AsyncSession, ticker_ids: set[int]) -> dict[int, fl
|
||||
),
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
return {tid: float(close) for tid, close in result.all()}
|
||||
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(
|
||||
@@ -658,16 +697,28 @@ async def get_trade_setup_history(
|
||||
result = await db.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
prices = await _latest_closes(db, {s.ticker_id for s, _ in rows})
|
||||
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, current_price: float | None = None) -> dict:
|
||||
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:
|
||||
@@ -706,4 +757,5 @@ def _trade_setup_to_dict(setup: TradeSetup, symbol: str, current_price: float |
|
||||
"evaluated_at": setup.evaluated_at,
|
||||
"current_price": current_price,
|
||||
"momentum_percentile": setup.momentum_percentile,
|
||||
"context_as_of": context_as_of,
|
||||
}
|
||||
|
||||
@@ -2,8 +2,8 @@
|
||||
|
||||
Computes dimension scores (technical, sr_quality, sentiment, fundamental,
|
||||
momentum) each 0-100, composite score as weighted average of available
|
||||
dimensions with re-normalized weights, staleness marking/recomputation
|
||||
on demand, and weight update triggers full recomputation.
|
||||
dimensions with re-normalized weights, staleness marking, explicit refresh
|
||||
paths, and weight update triggers full recomputation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -765,73 +765,37 @@ async def compute_composite_score(
|
||||
async def get_score(
|
||||
db: AsyncSession, symbol: str
|
||||
) -> dict:
|
||||
"""Get composite + all dimension scores for a ticker.
|
||||
"""Read composite + dimension scores for a ticker without recomputing.
|
||||
|
||||
Recomputes stale dimensions on demand, then recomputes composite.
|
||||
Returns a dict suitable for ScoreResponse, including dimension breakdowns
|
||||
and composite breakdown with re-normalization info.
|
||||
GET endpoints use this path, so it must not mutate persisted score context.
|
||||
Scheduled/manual write paths are responsible for refreshing scores.
|
||||
"""
|
||||
ticker = await _get_ticker(db, symbol)
|
||||
weights = await _get_weights(db)
|
||||
|
||||
# Check for stale dimension scores and recompute them
|
||||
result = await db.execute(
|
||||
select(DimensionScore).where(DimensionScore.ticker_id == ticker.id)
|
||||
)
|
||||
dim_scores = {ds.dimension: ds for ds in result.scalars().all()}
|
||||
|
||||
for dim in DIMENSIONS:
|
||||
ds = dim_scores.get(dim)
|
||||
if ds is None or ds.is_stale:
|
||||
await compute_dimension_score(db, symbol, dim)
|
||||
|
||||
# Check composite staleness
|
||||
comp_result = await db.execute(
|
||||
select(CompositeScore).where(CompositeScore.ticker_id == ticker.id)
|
||||
)
|
||||
comp = comp_result.scalar_one_or_none()
|
||||
|
||||
if comp is None or comp.is_stale:
|
||||
await compute_composite_score(db, symbol, weights)
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Re-fetch everything fresh
|
||||
result = await db.execute(
|
||||
select(DimensionScore).where(DimensionScore.ticker_id == ticker.id)
|
||||
)
|
||||
dim_scores_list = list(result.scalars().all())
|
||||
dim_scores = {ds.dimension: ds for ds in dim_scores_list}
|
||||
|
||||
comp_result = await db.execute(
|
||||
select(CompositeScore).where(CompositeScore.ticker_id == ticker.id)
|
||||
)
|
||||
comp = comp_result.scalar_one_or_none()
|
||||
|
||||
# Compute breakdowns for each dimension by calling the dimension computers
|
||||
breakdowns: dict[str, dict | None] = {}
|
||||
for dim in DIMENSIONS:
|
||||
try:
|
||||
raw_result = await _DIMENSION_COMPUTERS[dim](db, symbol)
|
||||
if isinstance(raw_result, tuple) and len(raw_result) == 2:
|
||||
breakdowns[dim] = raw_result[1]
|
||||
else:
|
||||
breakdowns[dim] = None
|
||||
except Exception:
|
||||
breakdowns[dim] = None
|
||||
|
||||
# Build dimension entries with breakdowns
|
||||
dimensions = []
|
||||
missing = []
|
||||
available_dims: list[str] = []
|
||||
for dim in DIMENSIONS:
|
||||
found = next((ds for ds in dim_scores_list if ds.dimension == dim), None)
|
||||
found = dim_scores.get(dim)
|
||||
if found is not None and not found.is_stale and found.score is not None:
|
||||
dimensions.append({
|
||||
"dimension": found.dimension,
|
||||
"score": found.score,
|
||||
"is_stale": found.is_stale,
|
||||
"computed_at": found.computed_at,
|
||||
"breakdown": breakdowns.get(dim),
|
||||
"breakdown": None,
|
||||
})
|
||||
w = weights.get(dim, 0.0)
|
||||
if w > 0:
|
||||
@@ -845,7 +809,7 @@ async def get_score(
|
||||
"score": found.score,
|
||||
"is_stale": found.is_stale,
|
||||
"computed_at": found.computed_at,
|
||||
"breakdown": breakdowns.get(dim),
|
||||
"breakdown": None,
|
||||
})
|
||||
|
||||
# Build composite breakdown: the non-sentiment base (re-normalized weighted
|
||||
@@ -925,31 +889,13 @@ async def get_rankings(db: AsyncSession) -> dict:
|
||||
dims[ds.ticker_id][ds.dimension] = ds
|
||||
return comps, dims
|
||||
|
||||
# Two bulk reads instead of ~4 queries per ticker.
|
||||
comps, dims_by_ticker = await _load_scores()
|
||||
|
||||
# Lazily recompute any stale/missing scores (kept fresh by the daily scan;
|
||||
# this self-heals tickers that aged out between scans), committing once.
|
||||
recomputed = False
|
||||
for ticker in tickers:
|
||||
comp = comps.get(ticker.id)
|
||||
if comp is None or comp.is_stale:
|
||||
dim_scores = dims_by_ticker.get(ticker.id, {})
|
||||
for dim in DIMENSIONS:
|
||||
ds = dim_scores.get(dim)
|
||||
if ds is None or ds.is_stale:
|
||||
await compute_dimension_score(db, ticker.symbol, dim)
|
||||
await compute_composite_score(db, ticker.symbol, weights)
|
||||
recomputed = True
|
||||
|
||||
if recomputed:
|
||||
await db.commit()
|
||||
comps, dims_by_ticker = await _load_scores()
|
||||
|
||||
rankings = [
|
||||
{
|
||||
"symbol": ticker.symbol,
|
||||
"composite_score": comp.score,
|
||||
"composite_stale": comp.is_stale,
|
||||
"dimensions": [
|
||||
{
|
||||
"dimension": ds.dimension,
|
||||
|
||||
Reference in New Issue
Block a user