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..fede2ac 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 @@ -407,17 +408,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 +478,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..432269f 100644 --- a/app/services/fundamental_service.py +++ b/app/services/fundamental_service.py @@ -13,6 +13,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 from app.models.fundamental import FundamentalData from app.models.score import DimensionScore @@ -67,7 +68,7 @@ async def store_fundamental( existing.unavailable_fields_json = unavailable_fields_json record = existing else: - record = FundamentalData( + stmt = insert_for_session(db, FundamentalData).values( ticker_id=ticker.id, pe_ratio=pe_ratio, revenue_growth=revenue_growth, @@ -77,7 +78,24 @@ async def store_fundamental( fetched_at=now, unavailable_fields_json=unavailable_fields_json, ) - db.add(record) + await db.execute( + 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, + }, + ) + ) + result = await db.execute( + select(FundamentalData).where(FundamentalData.ticker_id == ticker.id) + ) + record = result.scalar_one() # Mark fundamental dimension score as stale if it exists # TODO: Use DimensionScore service when built diff --git a/app/services/price_service.py b/app/services/price_service.py index 0b68a07..9258bbc 100644 --- a/app/services/price_service.py +++ b/app/services/price_service.py @@ -97,7 +97,13 @@ async def query_ohlcv( Returns records sorted by date ascending. Raises NotFoundError if the ticker does not exist. """ - ticker = await _get_ticker(db, symbol) + normalised = symbol.strip().upper() + cache = db.info.get("ohlcv_cache") + cache_key = (normalised, start_date, end_date) + if cache is not None and cache_key in cache: + return list(cache[cache_key]) + + ticker = await _get_ticker(db, normalised) stmt = select(OHLCVRecord).where(OHLCVRecord.ticker_id == ticker.id) if start_date is not None: @@ -107,4 +113,7 @@ async def query_ohlcv( stmt = stmt.order_by(OHLCVRecord.date.asc()) result = await db.execute(stmt) - return list(result.scalars().all()) + records = list(result.scalars().all()) + if cache is not None: + cache[cache_key] = records + return list(records) diff --git a/app/services/rr_scanner_service.py b/app/services/rr_scanner_service.py index e2fd188..84d0fec 100644 --- a/app/services/rr_scanner_service.py +++ b/app/services/rr_scanner_service.py @@ -556,6 +556,9 @@ async def scan_all_tickers( result = await db.execute(select(Ticker).order_by(Ticker.symbol)) tickers = list(result.scalars().all()) total = len(tickers) + # Ranking, score refresh, and setup detection repeatedly read the same + # immutable OHLCV series during one scan. Scope the cache to this run only. + db.info["ohlcv_cache"] = {} # Rank the universe up front so each new setup carries both the residual # activation gate percentile and the promoted production ordering score. @@ -582,7 +585,6 @@ async def scan_all_tickers( 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) @@ -596,6 +598,11 @@ async def scan_all_tickers( except Exception: logger.exception("Error scanning ticker %s", ticker.symbol) + # scan_ticker commits successful setup writes. This final commit persists + # refreshed scores for tickers that produced no setup or hit a scan error. + await db.commit() + + db.info.pop("ohlcv_cache", None) if progress_callback is not None and total: progress_callback(total, total, "") diff --git a/app/services/scoring_service.py b/app/services/scoring_service.py index f40d24d..cef3f7b 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 @@ -749,14 +759,24 @@ async def compute_composite_score( existing.weights_json = json.dumps(weights) existing.computed_at = now else: - comp = CompositeScore( + stmt = insert_for_session(db, CompositeScore).values( ticker_id=ticker.id, score=composite, is_stale=False, weights_json=json.dumps(weights), computed_at=now, ) - db.add(comp) + 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, + }, + ) + ) return composite, missing diff --git a/app/services/watchlist_service.py b/app/services/watchlist_service.py index 296d59e..96aefe2 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 @@ -185,6 +186,124 @@ async def _enrich_entry( } +async def _enrich_entries( + db: AsyncSession, + rows: list[tuple[WatchlistEntry, str]], +) -> list[dict]: + """Build watchlist rows from a fixed set of bulk lookups.""" + if not rows: + return [] + + ticker_ids = [entry.ticker_id for entry, _ in rows] + comps_result = await db.execute( + select(CompositeScore).where(CompositeScore.ticker_id.in_(ticker_ids)) + ) + comps = {score.ticker_id: score for score in comps_result.scalars()} + + dims_result = await db.execute( + select(DimensionScore).where(DimensionScore.ticker_id.in_(ticker_ids)) + ) + 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} + ) + + 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) + .join(ranked_setups, TradeSetup.id == ranked_setups.c.id) + .where(ranked_setups.c.rank == 1) + ) + best_setups = {setup.ticker_id: setup for setup in setup_result.scalars()} + + levels_result = await db.execute( + select(SRLevel) + .where(SRLevel.ticker_id.in_(ticker_ids)) + .order_by(SRLevel.ticker_id, SRLevel.strength.desc()) + ) + 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, + } + ) + + 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() + ) + 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) + ) + 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)) + + 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( db: AsyncSession, user_id: int, @@ -203,10 +322,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":