diff --git a/alembic/versions/019_enforce_score_snapshot_invariants.py b/alembic/versions/019_enforce_score_snapshot_invariants.py new file mode 100644 index 0000000..8537d72 --- /dev/null +++ b/alembic/versions/019_enforce_score_snapshot_invariants.py @@ -0,0 +1,61 @@ +"""Enforce singleton score and fundamental snapshots. + +Revision ID: 019 +Revises: 018 +Create Date: 2026-07-11 00:00:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "019" +down_revision = "018" +branch_labels = None +depends_on = None + + +def _remove_duplicates(table: str, partition_by: str, order_by: str) -> None: + op.execute( + sa.text( + f""" + DELETE FROM {table} + WHERE id IN ( + SELECT id FROM ( + SELECT id, ROW_NUMBER() OVER ( + PARTITION BY {partition_by} + ORDER BY {order_by} DESC, id DESC + ) AS row_number + FROM {table} + ) AS ranked + WHERE row_number > 1 + ) + """ + ) + ) + + +def upgrade() -> None: + _remove_duplicates("dimension_scores", "ticker_id, dimension", "computed_at") + _remove_duplicates("composite_scores", "ticker_id", "computed_at") + _remove_duplicates("fundamental_data", "ticker_id", "fetched_at") + + op.create_unique_constraint( + "uq_dimension_score_ticker_dimension", + "dimension_scores", + ["ticker_id", "dimension"], + ) + op.create_unique_constraint("uq_composite_score_ticker", "composite_scores", ["ticker_id"]) + op.create_unique_constraint("uq_fundamental_data_ticker", "fundamental_data", ["ticker_id"]) + op.create_index("ix_sr_levels_ticker_id", "sr_levels", ["ticker_id"]) + op.create_index("ix_trade_setups_ticker_rr", "trade_setups", ["ticker_id", "rr_ratio"]) + + +def downgrade() -> None: + op.drop_index("ix_trade_setups_ticker_rr", table_name="trade_setups") + op.drop_index("ix_sr_levels_ticker_id", table_name="sr_levels") + op.drop_constraint("uq_fundamental_data_ticker", "fundamental_data", type_="unique") + op.drop_constraint("uq_composite_score_ticker", "composite_scores", type_="unique") + op.drop_constraint("uq_dimension_score_ticker_dimension", "dimension_scores", type_="unique") diff --git a/app/database.py b/app/database.py index 2ba0f3c..bd6b294 100644 --- a/app/database.py +++ b/app/database.py @@ -1,5 +1,8 @@ from collections.abc import AsyncGenerator +from typing import Any +from sqlalchemy.dialects.postgresql import insert as postgresql_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, @@ -28,6 +31,13 @@ class Base(DeclarativeBase): pass +def insert_for_session(session: AsyncSession, table: Any) -> Any: + """Build a dialect-native INSERT that supports conflict handling.""" + if session.get_bind().dialect.name == "postgresql": + return postgresql_insert(table) + return sqlite_insert(table) + + async def get_session() -> AsyncGenerator[AsyncSession, None]: async with async_session_factory() as session: yield session diff --git a/app/models/fundamental.py b/app/models/fundamental.py index d2af7bf..4f77e6a 100644 --- a/app/models/fundamental.py +++ b/app/models/fundamental.py @@ -1,6 +1,6 @@ from datetime import date, datetime -from sqlalchemy import Date, DateTime, Float, ForeignKey, Text +from sqlalchemy import Date, DateTime, Float, ForeignKey, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base @@ -8,6 +8,9 @@ from app.database import Base class FundamentalData(Base): __tablename__ = "fundamental_data" + __table_args__ = ( + UniqueConstraint("ticker_id", name="uq_fundamental_data_ticker"), + ) id: Mapped[int] = mapped_column(primary_key=True) ticker_id: Mapped[int] = mapped_column( diff --git a/app/models/score.py b/app/models/score.py index e0758c4..1d84766 100644 --- a/app/models/score.py +++ b/app/models/score.py @@ -1,6 +1,6 @@ from datetime import datetime -from sqlalchemy import Boolean, DateTime, Float, ForeignKey, String, Text +from sqlalchemy import Boolean, DateTime, Float, ForeignKey, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base @@ -8,6 +8,9 @@ from app.database import Base class DimensionScore(Base): __tablename__ = "dimension_scores" + __table_args__ = ( + UniqueConstraint("ticker_id", "dimension", name="uq_dimension_score_ticker_dimension"), + ) id: Mapped[int] = mapped_column(primary_key=True) ticker_id: Mapped[int] = mapped_column( @@ -25,6 +28,9 @@ class DimensionScore(Base): class CompositeScore(Base): __tablename__ = "composite_scores" + __table_args__ = ( + UniqueConstraint("ticker_id", name="uq_composite_score_ticker"), + ) id: Mapped[int] = mapped_column(primary_key=True) ticker_id: Mapped[int] = mapped_column( diff --git a/app/models/sr_level.py b/app/models/sr_level.py index 32d01c2..6787a18 100644 --- a/app/models/sr_level.py +++ b/app/models/sr_level.py @@ -1,6 +1,6 @@ from datetime import datetime -from sqlalchemy import DateTime, Float, ForeignKey, Integer, String +from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base @@ -8,6 +8,7 @@ from app.database import Base class SRLevel(Base): __tablename__ = "sr_levels" + __table_args__ = (Index("ix_sr_levels_ticker_id", "ticker_id"),) id: Mapped[int] = mapped_column(primary_key=True) ticker_id: Mapped[int] = mapped_column( diff --git a/app/models/trade_setup.py b/app/models/trade_setup.py index f389b4a..0375364 100644 --- a/app/models/trade_setup.py +++ b/app/models/trade_setup.py @@ -2,7 +2,7 @@ from datetime import date, datetime import json -from sqlalchemy import Date, DateTime, Float, ForeignKey, String, Text +from sqlalchemy import Date, DateTime, Float, ForeignKey, Index, String, Text from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base @@ -10,6 +10,7 @@ from app.database import Base class TradeSetup(Base): __tablename__ = "trade_setups" + __table_args__ = (Index("ix_trade_setups_ticker_rr", "ticker_id", "rr_ratio"),) id: Mapped[int] = mapped_column(primary_key=True) ticker_id: Mapped[int] = mapped_column( diff --git a/app/services/alert_service.py b/app/services/alert_service.py index 1a4ccbf..9e4f4ea 100644 --- a/app/services/alert_service.py +++ b/app/services/alert_service.py @@ -17,11 +17,12 @@ from __future__ import annotations import logging import math +from collections import defaultdict from datetime import datetime, timedelta, timezone from types import SimpleNamespace import httpx -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings @@ -257,17 +258,6 @@ def _log_alert(db: AsyncSession, alert_type: str, key: str, value: float | None ) -async def _watermark(db: AsyncSession, symbol: str) -> float | None: - result = await db.execute( - select(AlertLog.value) - .where(AlertLog.alert_type == WATERMARK_TYPE, AlertLog.dedup_key == symbol) - .order_by(AlertLog.created_at.desc()) - .limit(1) - ) - row = result.first() - return row[0] if row else None - - # --------------------------------------------------------------------------- # Trigger collectors # --------------------------------------------------------------------------- @@ -407,17 +397,49 @@ async def _collect_sr_proximity(db: AsyncSession) -> list[tuple[str, str]]: single alert. Scoped to the watchlist only — qualified tickers already get their own 'qualified setup' alert, so S/R on them would be redundant. """ + watchlist = await _watchlist_tickers(db) + if not watchlist: + return [] + + ticker_ids = [ticker_id for ticker_id, _ in watchlist] + latest_dates = ( + select( + OHLCVRecord.ticker_id, + func.max(OHLCVRecord.date).label("latest_date"), + ) + .where(OHLCVRecord.ticker_id.in_(ticker_ids)) + .group_by(OHLCVRecord.ticker_id) + .subquery() + ) + prices_result = await db.execute( + select(OHLCVRecord.ticker_id, OHLCVRecord.close).join( + latest_dates, + (OHLCVRecord.ticker_id == latest_dates.c.ticker_id) + & (OHLCVRecord.date == latest_dates.c.latest_date), + ) + ) + prices = {ticker_id: float(close) for ticker_id, close in prices_result.all()} + + levels_result = await db.execute( + select(SRLevel).where(SRLevel.ticker_id.in_(ticker_ids)) + ) + levels_by_ticker: dict[int, list[dict]] = defaultdict(list) + for level in levels_result.scalars(): + levels_by_ticker[level.ticker_id].append( + { + "price_level": level.price_level, + "strength": level.strength, + "type": level.type, + } + ) + out: list[tuple[str, str]] = [] - for tid, symbol in await _watchlist_tickers(db): - price = await _latest_close(db, tid) + for tid, symbol in watchlist: + price = prices.get(tid) if not price: continue - levels_result = await db.execute(select(SRLevel).where(SRLevel.ticker_id == tid)) - levels = [ - {"price_level": lv.price_level, "strength": lv.strength, "type": lv.type} - for lv in levels_result.scalars().all() - ] + levels = levels_by_ticker[tid] if not levels: continue @@ -445,17 +467,54 @@ async def _collect_score_drops(db: AsyncSession) -> list[tuple[str, str]]: doesn't re-fire; let the watermark rise with the score so the next drop is measured from the new high. """ - out: list[tuple[str, str]] = [] - for tid, symbol in await _watchlist_tickers(db): - comp_result = await db.execute( - select(CompositeScore.score).where(CompositeScore.ticker_id == tid) - ) - row = comp_result.first() - if row is None or row[0] is None: - continue - current = float(row[0]) + watchlist = await _watchlist_tickers(db) + if not watchlist: + return [] - base = await _watermark(db, symbol) + ticker_ids = [ticker_id for ticker_id, _ in watchlist] + symbols = [symbol for _, symbol in watchlist] + scores_result = await db.execute( + select(CompositeScore.ticker_id, CompositeScore.score).where( + CompositeScore.ticker_id.in_(ticker_ids) + ) + ) + scores = {ticker_id: float(score) for ticker_id, score in scores_result.all()} + + ranked_watermarks = ( + select( + AlertLog.dedup_key, + AlertLog.value, + func.row_number() + .over( + partition_by=AlertLog.dedup_key, + order_by=(AlertLog.created_at.desc(), AlertLog.id.desc()), + ) + .label("rank"), + ) + .where( + AlertLog.alert_type == WATERMARK_TYPE, + AlertLog.dedup_key.in_(symbols), + ) + .subquery() + ) + watermarks_result = await db.execute( + select(ranked_watermarks.c.dedup_key, ranked_watermarks.c.value).where( + ranked_watermarks.c.rank == 1 + ) + ) + watermarks = { + symbol: float(value) + for symbol, value in watermarks_result.all() + if value is not None + } + + out: list[tuple[str, str]] = [] + for tid, symbol in watchlist: + current = scores.get(tid) + if current is None: + continue + + base = watermarks.get(symbol) if base is None: _log_alert(db, WATERMARK_TYPE, symbol, value=current) # seed, no alert continue diff --git a/app/services/fundamental_service.py b/app/services/fundamental_service.py index edcf6a2..234d249 100644 --- a/app/services/fundamental_service.py +++ b/app/services/fundamental_service.py @@ -10,9 +10,10 @@ import json import logging from datetime import datetime, timezone -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession +from app.database import insert_for_session from app.exceptions import NotFoundError from app.models.fundamental import FundamentalData from app.models.score import DimensionScore @@ -48,51 +49,45 @@ async def store_fundamental( """ ticker = await _get_ticker(db, symbol) - # Check for existing record - result = await db.execute( - select(FundamentalData).where(FundamentalData.ticker_id == ticker.id) - ) - existing = result.scalar_one_or_none() - now = datetime.now(timezone.utc) unavailable_fields_json = json.dumps(unavailable_fields or {}) - if existing is not None: - existing.pe_ratio = pe_ratio - existing.revenue_growth = revenue_growth - existing.earnings_surprise = earnings_surprise - existing.market_cap = market_cap - existing.next_earnings_date = next_earnings_date - existing.fetched_at = now - existing.unavailable_fields_json = unavailable_fields_json - record = existing - else: - record = FundamentalData( - ticker_id=ticker.id, - pe_ratio=pe_ratio, - revenue_growth=revenue_growth, - earnings_surprise=earnings_surprise, - market_cap=market_cap, - next_earnings_date=next_earnings_date, - fetched_at=now, - unavailable_fields_json=unavailable_fields_json, - ) - db.add(record) + stmt = insert_for_session(db, FundamentalData).values( + ticker_id=ticker.id, + pe_ratio=pe_ratio, + revenue_growth=revenue_growth, + earnings_surprise=earnings_surprise, + market_cap=market_cap, + next_earnings_date=next_earnings_date, + fetched_at=now, + unavailable_fields_json=unavailable_fields_json, + ) + stmt = stmt.on_conflict_do_update( + index_elements=["ticker_id"], + set_={ + "pe_ratio": stmt.excluded.pe_ratio, + "revenue_growth": stmt.excluded.revenue_growth, + "earnings_surprise": stmt.excluded.earnings_surprise, + "market_cap": stmt.excluded.market_cap, + "next_earnings_date": stmt.excluded.next_earnings_date, + "fetched_at": stmt.excluded.fetched_at, + "unavailable_fields_json": stmt.excluded.unavailable_fields_json, + }, + ).returning(FundamentalData) + record = (await db.execute(stmt)).scalar_one() # Mark fundamental dimension score as stale if it exists # TODO: Use DimensionScore service when built - dim_result = await db.execute( - select(DimensionScore).where( + await db.execute( + update(DimensionScore) + .where( DimensionScore.ticker_id == ticker.id, DimensionScore.dimension == "fundamental", ) + .values(is_stale=True) ) - dim_score = dim_result.scalar_one_or_none() - if dim_score is not None: - dim_score.is_stale = True await db.commit() - await db.refresh(record) return record diff --git a/app/services/price_service.py b/app/services/price_service.py index 0b68a07..e9c0ad1 100644 --- a/app/services/price_service.py +++ b/app/services/price_service.py @@ -3,9 +3,9 @@ from datetime import date, datetime from sqlalchemy import select -from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession +from app.database import insert_for_session from app.exceptions import NotFoundError, ValidationError from app.models.ohlcv import OHLCVRecord from app.models.ticker import Ticker @@ -53,7 +53,7 @@ async def upsert_ohlcv( _validate_ohlcv(high, low, open_, close, volume, record_date) ticker = await _get_ticker(db, symbol) - stmt = pg_insert(OHLCVRecord).values( + stmt = insert_for_session(db, OHLCVRecord).values( ticker_id=ticker.id, date=record_date, open=open_, @@ -64,7 +64,7 @@ async def upsert_ohlcv( created_at=datetime.utcnow(), ) stmt = stmt.on_conflict_do_update( - constraint="uq_ohlcv_ticker_date", + index_elements=["ticker_id", "date"], set_={ "open": stmt.excluded.open, "high": stmt.excluded.high, diff --git a/app/services/rr_scanner_service.py b/app/services/rr_scanner_service.py index e2fd188..4026e19 100644 --- a/app/services/rr_scanner_service.py +++ b/app/services/rr_scanner_service.py @@ -566,6 +566,7 @@ async def scan_all_tickers( ranks = await momentum_service.compute_activation_ranks(db) except Exception: + await db.rollback() logger.exception("Activation ranking refresh failed") ranks = {} @@ -573,19 +574,21 @@ async def scan_all_tickers( for index, ticker in enumerate(tickers): if progress_callback is not None: progress_callback(index, total, ticker.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. try: - # 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. - try: - from app.services import scoring_service + from app.services import scoring_service - await scoring_service.compute_all_dimensions(db, ticker.symbol) - await scoring_service.compute_composite_score(db, ticker.symbol) - await db.commit() - except Exception: - logger.exception("Error refreshing scores for %s", ticker.symbol) + await scoring_service.compute_all_dimensions(db, ticker.symbol) + await scoring_service.compute_composite_score(db, ticker.symbol) + await db.commit() + except Exception: + await db.rollback() + logger.exception("Error refreshing scores for %s", ticker.symbol) + continue + try: setups = await scan_ticker( db, ticker.symbol, rr_threshold, atr_multiplier, momentum_percentile=(ranks.get(ticker.symbol) or {}).get("momentum_percentile"), @@ -594,6 +597,7 @@ async def scan_all_tickers( ) all_setups.extend(setups) except Exception: + await db.rollback() logger.exception("Error scanning ticker %s", ticker.symbol) if progress_callback is not None and total: diff --git a/app/services/scoring_service.py b/app/services/scoring_service.py index f40d24d..a6b3e67 100644 --- a/app/services/scoring_service.py +++ b/app/services/scoring_service.py @@ -16,6 +16,7 @@ from datetime import datetime, timezone from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.database import insert_for_session from app.exceptions import NotFoundError, ValidationError from app.models.score import CompositeScore, DimensionScore from app.models.ticker import Ticker @@ -661,14 +662,23 @@ async def compute_dimension_score( # Can't compute — mark stale existing.is_stale = True elif score_val is not None: - dim = DimensionScore( + stmt = insert_for_session(db, DimensionScore).values( ticker_id=ticker.id, dimension=dimension, score=score_val, is_stale=False, computed_at=now, ) - db.add(dim) + await db.execute( + stmt.on_conflict_do_update( + index_elements=["ticker_id", "dimension"], + set_={ + "score": stmt.excluded.score, + "is_stale": False, + "computed_at": stmt.excluded.computed_at, + }, + ) + ) return score_val @@ -738,25 +748,24 @@ async def compute_composite_score( # Persist composite score now = datetime.now(timezone.utc) - comp_result = await db.execute( - select(CompositeScore).where(CompositeScore.ticker_id == ticker.id) + stmt = insert_for_session(db, CompositeScore).values( + ticker_id=ticker.id, + score=composite, + is_stale=False, + weights_json=json.dumps(weights), + computed_at=now, ) - existing = comp_result.scalar_one_or_none() - - if existing is not None: - existing.score = composite - existing.is_stale = False - existing.weights_json = json.dumps(weights) - existing.computed_at = now - else: - comp = CompositeScore( - ticker_id=ticker.id, - score=composite, - is_stale=False, - weights_json=json.dumps(weights), - computed_at=now, + await db.execute( + stmt.on_conflict_do_update( + index_elements=["ticker_id"], + set_={ + "score": stmt.excluded.score, + "is_stale": False, + "weights_json": stmt.excluded.weights_json, + "computed_at": stmt.excluded.computed_at, + }, ) - db.add(comp) + ) return composite, missing diff --git a/app/services/watchlist_service.py b/app/services/watchlist_service.py index 296d59e..257bf0d 100644 --- a/app/services/watchlist_service.py +++ b/app/services/watchlist_service.py @@ -8,6 +8,7 @@ best trade setup, active S/R levels, and latest price + day-over-day move. from __future__ import annotations import logging +from collections import defaultdict from datetime import datetime, timezone from sqlalchemy import func, select @@ -102,87 +103,122 @@ async def remove_entry( await db.commit() -async def _enrich_entry( +async def _enrich_entries( db: AsyncSession, - entry: WatchlistEntry, - symbol: str, -) -> dict: - """Build enriched watchlist entry dict with scores, R:R, SR levels, price.""" - ticker_id = entry.ticker_id + rows: list[tuple[WatchlistEntry, str]], +) -> list[dict]: + """Build watchlist rows from a fixed set of bulk lookups.""" + if not rows: + return [] - # Composite score - comp_result = await db.execute( - select(CompositeScore).where(CompositeScore.ticker_id == ticker_id) + ticker_ids = [entry.ticker_id for entry, _ in rows] + comps_result = await db.execute( + select(CompositeScore).where(CompositeScore.ticker_id.in_(ticker_ids)) ) - comp = comp_result.scalar_one_or_none() + comps = {score.ticker_id: score for score in comps_result.scalars()} - # Dimension scores - dim_result = await db.execute( - select(DimensionScore).where(DimensionScore.ticker_id == ticker_id) + dims_result = await db.execute( + select(DimensionScore).where(DimensionScore.ticker_id.in_(ticker_ids)) ) - dims = [ - {"dimension": ds.dimension, "score": ds.score} - for ds in dim_result.scalars().all() - ] + dims_by_ticker: dict[int, list[dict]] = defaultdict(list) + for score in dims_result.scalars(): + dims_by_ticker[score.ticker_id].append( + {"dimension": score.dimension, "score": score.score} + ) - # Best trade setup (highest R:R) for this ticker + ranked_setups = ( + select( + TradeSetup.id, + func.row_number() + .over( + partition_by=TradeSetup.ticker_id, + order_by=TradeSetup.rr_ratio.desc(), + ) + .label("rank"), + ) + .where(TradeSetup.ticker_id.in_(ticker_ids)) + .subquery() + ) setup_result = await db.execute( select(TradeSetup) - .where(TradeSetup.ticker_id == ticker_id) - .order_by(TradeSetup.rr_ratio.desc()) - .limit(1) + .join(ranked_setups, TradeSetup.id == ranked_setups.c.id) + .where(ranked_setups.c.rank == 1) ) - setup = setup_result.scalar_one_or_none() + best_setups = {setup.ticker_id: setup for setup in setup_result.scalars()} - # Active SR levels - sr_result = await db.execute( + levels_result = await db.execute( select(SRLevel) - .where(SRLevel.ticker_id == ticker_id) - .order_by(SRLevel.strength.desc()) + .where(SRLevel.ticker_id.in_(ticker_ids)) + .order_by(SRLevel.ticker_id, SRLevel.strength.desc()) ) - sr_levels = [ - { - "price_level": lv.price_level, - "type": lv.type, - "strength": lv.strength, - } - for lv in sr_result.scalars().all() - ] + levels_by_ticker: dict[int, list[dict]] = defaultdict(list) + for level in levels_result.scalars(): + levels_by_ticker[level.ticker_id].append( + { + "price_level": level.price_level, + "type": level.type, + "strength": level.strength, + } + ) - # Latest two daily closes → current price + day-over-day move - price_result = await db.execute( - select(OHLCVRecord.close, OHLCVRecord.date) - .where(OHLCVRecord.ticker_id == ticker_id) - .order_by(OHLCVRecord.date.desc()) - .limit(2) + ranked_prices = ( + select( + OHLCVRecord.ticker_id, + OHLCVRecord.close, + OHLCVRecord.date, + func.row_number() + .over( + partition_by=OHLCVRecord.ticker_id, + order_by=OHLCVRecord.date.desc(), + ) + .label("rank"), + ) + .where(OHLCVRecord.ticker_id.in_(ticker_ids)) + .subquery() ) - bars = price_result.all() - last_close = bars[0].close if bars else None - prev_close = bars[1].close if len(bars) > 1 else None - change_pct = ( - (last_close - prev_close) / prev_close * 100 - if last_close is not None and prev_close - else None + prices_result = await db.execute( + select( + ranked_prices.c.ticker_id, + ranked_prices.c.close, + ranked_prices.c.date, + ) + .where(ranked_prices.c.rank <= 2) + .order_by(ranked_prices.c.ticker_id, ranked_prices.c.rank) ) - price_date = bars[0].date if bars else None + prices_by_ticker: dict[int, list[tuple[float, datetime]]] = defaultdict(list) + for ticker_id, close, price_date in prices_result.all(): + prices_by_ticker[ticker_id].append((close, price_date)) - return { - "symbol": symbol, - "entry_type": entry.entry_type, - "composite_score": comp.score if comp else None, - "dimensions": dims, - "rr_ratio": setup.rr_ratio if setup else None, - "rr_direction": setup.direction if setup else None, - # Residual 12-1 activation percentile gates qualification; strategy_rank - # is the promoted top-pick ordering score. - "momentum_percentile": setup.momentum_percentile if setup else None, - "strategy_rank": setup.strategy_rank if setup else None, - "sr_levels": sr_levels, - "last_close": last_close, - "change_pct": change_pct, - "price_date": price_date, - "added_at": entry.added_at, - } + entries: list[dict] = [] + for entry, symbol in rows: + ticker_id = entry.ticker_id + comp = comps.get(ticker_id) + setup = best_setups.get(ticker_id) + bars = prices_by_ticker[ticker_id] + last_close = bars[0][0] if bars else None + prev_close = bars[1][0] if len(bars) > 1 else None + entries.append( + { + "symbol": symbol, + "entry_type": entry.entry_type, + "composite_score": comp.score if comp else None, + "dimensions": dims_by_ticker[ticker_id], + "rr_ratio": setup.rr_ratio if setup else None, + "rr_direction": setup.direction if setup else None, + "momentum_percentile": setup.momentum_percentile if setup else None, + "strategy_rank": setup.strategy_rank if setup else None, + "sr_levels": levels_by_ticker[ticker_id], + "last_close": last_close, + "change_pct": ( + (last_close - prev_close) / prev_close * 100 + if last_close is not None and prev_close + else None + ), + "price_date": bars[0][1] if bars else None, + "added_at": entry.added_at, + } + ) + return entries async def get_watchlist( @@ -203,10 +239,7 @@ async def get_watchlist( result = await db.execute(stmt) rows = result.all() - entries: list[dict] = [] - for entry, symbol in rows: - enriched = await _enrich_entry(db, entry, symbol) - entries.append(enriched) + entries = await _enrich_entries(db, rows) # Sort if sort_by == "composite": diff --git a/tests/unit/test_alert_service.py b/tests/unit/test_alert_service.py index 69368e3..b4ad09a 100644 --- a/tests/unit/test_alert_service.py +++ b/tests/unit/test_alert_service.py @@ -95,7 +95,17 @@ async def test_score_drop_seeds_then_alerts(session): msgs = await svc._collect_score_drops(session) await session.commit() assert msgs == [] - assert await svc._watermark(session, "AAA") == 80.0 + watermarks = ( + await session.execute( + select(AlertLog.value) + .where( + AlertLog.alert_type == svc.WATERMARK_TYPE, + AlertLog.dedup_key == "AAA", + ) + .order_by(AlertLog.created_at.desc(), AlertLog.id.desc()) + ) + ).scalars().all() + assert watermarks == [80.0] # Drop the composite well past the threshold row = (await session.execute( @@ -111,7 +121,71 @@ async def test_score_drop_seeds_then_alerts(session): assert key == "scoredrop:AAA" assert "AAA" in text # rebaselined to the new (lower) level - assert await svc._watermark(session, "AAA") == 60.0 + watermarks = ( + await session.execute( + select(AlertLog.value) + .where( + AlertLog.alert_type == svc.WATERMARK_TYPE, + AlertLog.dedup_key == "AAA", + ) + .order_by(AlertLog.created_at.desc(), AlertLog.id.desc()) + ) + ).scalars().all() + assert watermarks[0] == 60.0 + + +async def test_score_drop_uses_latest_watermark_when_timestamps_tie(session): + await _seed_watchlisted_ticker(session, "AAA", 50.0) + timestamp = datetime.now(timezone.utc) + session.add_all([ + AlertLog( + alert_type=svc.WATERMARK_TYPE, + dedup_key="AAA", + value=90.0, + created_at=timestamp, + ), + AlertLog( + alert_type=svc.WATERMARK_TYPE, + dedup_key="AAA", + value=70.0, + created_at=timestamp, + ), + ]) + await session.commit() + + msgs = await svc._collect_score_drops(session) + + assert len(msgs) == 1 + assert "from 70" in msgs[0][1] + + +async def test_score_drop_seeds_when_latest_watermark_has_no_value(session): + await _seed_watchlisted_ticker(session, "AAA", 80.0) + session.add( + AlertLog( + alert_type=svc.WATERMARK_TYPE, + dedup_key="AAA", + value=None, + created_at=datetime.now(timezone.utc), + ) + ) + await session.commit() + + msgs = await svc._collect_score_drops(session) + await session.commit() + + assert msgs == [] + values = ( + await session.execute( + select(AlertLog.value) + .where( + AlertLog.alert_type == svc.WATERMARK_TYPE, + AlertLog.dedup_key == "AAA", + ) + .order_by(AlertLog.created_at.desc(), AlertLog.id.desc()) + ) + ).scalars().all() + assert values[0] == 80.0 def test_format_qualified_includes_current_price_and_target_move():