Merge branch 'main' of ssh://git.thiessen.io:2266/dennisthiessen/signal-platform

# Conflicts:
#	app/services/fundamental_service.py
#	app/services/rr_scanner_service.py
#	app/services/scoring_service.py

app/services/alert_service.py

@@ -258,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
 # ---------------------------------------------------------------------------

app/services/fundamental_service.py

@@ -10,7 +10,7 @@ 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
@@ -49,25 +49,9 @@ 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:
     stmt = insert_for_session(db, FundamentalData).values(
         ticker_id=ticker.id,
         pe_ratio=pe_ratio,
@@ -78,8 +62,7 @@ async def store_fundamental(
         fetched_at=now,
         unavailable_fields_json=unavailable_fields_json,
     )
-        await db.execute(
-            stmt.on_conflict_do_update(
+    stmt = stmt.on_conflict_do_update(
         index_elements=["ticker_id"],
         set_={
             "pe_ratio": stmt.excluded.pe_ratio,
@@ -90,27 +73,21 @@ async def store_fundamental(
             "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()
+    ).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
 
 

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,
@@ -97,13 +97,7 @@ async def query_ohlcv(
     Returns records sorted by date ascending.
     Raises NotFoundError if the ticker does not exist.
     """
-    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)
+    ticker = await _get_ticker(db, symbol)
 
     stmt = select(OHLCVRecord).where(OHLCVRecord.ticker_id == ticker.id)
     if start_date is not None:
@@ -113,7 +107,4 @@ async def query_ohlcv(
     stmt = stmt.order_by(OHLCVRecord.date.asc())
 
     result = await db.execute(stmt)
-    records = list(result.scalars().all())
-    if cache is not None:
-        cache[cache_key] = records
-    return list(records)
+    return list(result.scalars().all())

app/services/rr_scanner_service.py

@@ -13,7 +13,7 @@ import logging
 from collections.abc import Callable
 from datetime import date, datetime, timedelta, timezone
 
-from sqlalchemy import and_, func, select
+from sqlalchemy import and_, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.exceptions import NotFoundError
@@ -58,6 +58,28 @@ async def _get_ticker(db: AsyncSession, symbol: str) -> Ticker:
     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,
@@ -116,8 +138,11 @@ async def _apply_live_recommendation_context(
         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)
@@ -153,6 +178,18 @@ async def _apply_live_recommendation_context(
             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:
@@ -553,12 +590,12 @@ async def scan_all_tickers(
     ``progress_callback(processed, total, current_symbol)`` is invoked as each
     ticker is scanned so callers (e.g. the scheduler) can surface live progress.
     """
-    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"] = {}
+    # Plain 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.symbol).order_by(Ticker.symbol))
+    symbols = list(result.scalars().all())
+    total = len(symbols)
 
     # Rank the universe up front so each new setup carries both the residual
     # activation gate percentile and the promoted production ordering score.
@@ -569,40 +606,47 @@ 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 = {}
 
     all_setups: list[TradeSetup] = []
-    for index, ticker in enumerate(tickers):
+    for index, symbol in enumerate(symbols):
         if progress_callback is not None:
-            progress_callback(index, total, ticker.symbol)
-        try:
+            progress_callback(index, total, symbol)
         # Refresh scores first so the scheduled scan works off current data.
-            # Nothing else marks scores stale, so without this they'd never
-            # update for tickers the user doesn't manually fetch.
+        # Nothing else marks scores stale, so without this they'd never update
+        # for tickers the user doesn't manually fetch. A refresh failure still
+        # scans the ticker: qualification re-gates on live scores at alert
+        # time, so a stale score is recoverable but a skipped scan is not.
         try:
             from app.services import scoring_service
 
-                await scoring_service.compute_all_dimensions(db, ticker.symbol)
-                await scoring_service.compute_composite_score(db, ticker.symbol)
+            await scoring_service.compute_all_dimensions(db, symbol)
+            await scoring_service.compute_composite_score(db, symbol)
+            await db.commit()
         except Exception:
-                logger.exception("Error refreshing scores for %s", ticker.symbol)
+            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, ticker.symbol, rr_threshold, atr_multiplier,
-                momentum_percentile=(ranks.get(ticker.symbol) or {}).get("momentum_percentile"),
-                strategy_rank=(ranks.get(ticker.symbol) or {}).get("strategy_rank"),
-                volatility_percentile=(ranks.get(ticker.symbol) or {}).get("volatility_percentile"),
+                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"),
             )
             all_setups.extend(setups)
         except Exception:
-            logger.exception("Error scanning ticker %s", ticker.symbol)
+            await db.rollback()
+            logger.exception("Error scanning ticker %s", 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, "")
 

app/services/scoring_service.py

@@ -748,17 +748,6 @@ 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)
-    )
-    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:
     stmt = insert_for_session(db, CompositeScore).values(
         ticker_id=ticker.id,
         score=composite,

app/services/watchlist_service.py

@@ -103,89 +103,6 @@ async def remove_entry(
     await db.commit()
 
 
-async def _enrich_entry(
-    db: AsyncSession,
-    entry: WatchlistEntry,
-    symbol: str,
-) -> dict:
-    """Build enriched watchlist entry dict with scores, R:R, SR levels, price."""
-    ticker_id = entry.ticker_id
-
-    # Composite score
-    comp_result = await db.execute(
-        select(CompositeScore).where(CompositeScore.ticker_id == ticker_id)
-    )
-    comp = comp_result.scalar_one_or_none()
-
-    # Dimension scores
-    dim_result = await db.execute(
-        select(DimensionScore).where(DimensionScore.ticker_id == ticker_id)
-    )
-    dims = [
-        {"dimension": ds.dimension, "score": ds.score}
-        for ds in dim_result.scalars().all()
-    ]
-
-    # Best trade setup (highest R:R) for this ticker
-    setup_result = await db.execute(
-        select(TradeSetup)
-        .where(TradeSetup.ticker_id == ticker_id)
-        .order_by(TradeSetup.rr_ratio.desc())
-        .limit(1)
-    )
-    setup = setup_result.scalar_one_or_none()
-
-    # Active SR levels
-    sr_result = await db.execute(
-        select(SRLevel)
-        .where(SRLevel.ticker_id == ticker_id)
-        .order_by(SRLevel.strength.desc())
-    )
-    sr_levels = [
-        {
-            "price_level": lv.price_level,
-            "type": lv.type,
-            "strength": lv.strength,
-        }
-        for lv in sr_result.scalars().all()
-    ]
-
-    # 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)
-    )
-    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
-    )
-    price_date = bars[0].date if bars else None
-
-    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,
-    }
-
-
 async def _enrich_entries(
     db: AsyncSession,
     rows: list[tuple[WatchlistEntry, str]],

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():

tests/unit/test_rr_scanner_preservation.py

@@ -689,6 +689,31 @@ async def test_live_recommendation_payload_uses_current_score_and_sentiment(
     assert persisted.reasoning == old_reasoning
 
 
+@pytest.mark.asyncio
+async def test_live_recommendation_withholds_stale_scores(
+    db_session: AsyncSession,
+):
+    stale_setup = await _seed_stale_setup_with_current_scores(db_session)
+    composite = (
+        await db_session.execute(
+            select(CompositeScore).where(CompositeScore.ticker_id == stale_setup.ticker_id)
+        )
+    ).scalar_one()
+    composite.is_stale = True
+    await db_session.flush()
+
+    rows = await get_trade_setups(
+        db_session,
+        symbol="TTWO",
+        live_recommendation=True,
+    )
+
+    assert len(rows) == 1
+    assert rows[0]["confidence_score"] is None
+    assert rows[0]["recommended_action"] == "NEUTRAL"
+    assert rows[0]["reasoning"] == "Score refresh pending; recommendation withheld."
+
+
 @pytest.mark.asyncio
 async def test_live_recommendation_filters_apply_to_live_values(
     db_session: AsyncSession,

tests/unit/test_rr_scanner_scan_all.py

@@ -0,0 +1,107 @@
+"""Tests for scan_all_tickers orchestration: error isolation between phases."""
+
+from __future__ import annotations
+
+import pytest
+from datetime import datetime, timezone
+from sqlalchemy import select
+
+from app.models.score import CompositeScore, DimensionScore
+from app.models.ticker import Ticker
+from app.services import rr_scanner_service, scoring_service
+from tests.conftest import _test_session_factory  # type: ignore
+
+
+@pytest.fixture
+async def session():
+    async with _test_session_factory() as s:
+        yield s
+
+
+async def test_scan_proceeds_when_score_refresh_fails(session, monkeypatch):
+    """A scoring failure must not skip setup detection for the ticker.
+
+    Qualification re-gates on live scores at alert time, so a stale score is
+    recoverable — a skipped scan is not.
+    """
+    session.add(Ticker(symbol="AAA"))
+    await session.commit()
+
+    async def _boom(db, symbol):
+        raise RuntimeError("scoring unavailable")
+
+    scanned: list[str] = []
+
+    async def _fake_scan_ticker(db, symbol, *args, **kwargs):
+        scanned.append(symbol)
+        return []
+
+    monkeypatch.setattr(scoring_service, "compute_all_dimensions", _boom)
+    monkeypatch.setattr(rr_scanner_service, "scan_ticker", _fake_scan_ticker)
+
+    setups = await rr_scanner_service.scan_all_tickers(session)
+
+    assert scanned == ["AAA"]
+    assert setups == []
+
+
+async def test_scan_marks_scores_stale_when_refresh_fails(session, monkeypatch):
+    ticker = Ticker(symbol="AAA")
+    session.add(ticker)
+    await session.flush()
+    now = datetime.now(timezone.utc)
+    session.add_all([
+        DimensionScore(
+            ticker_id=ticker.id,
+            dimension="technical",
+            score=70.0,
+            is_stale=False,
+            computed_at=now,
+        ),
+        CompositeScore(
+            ticker_id=ticker.id,
+            score=70.0,
+            is_stale=False,
+            weights_json="{}",
+            computed_at=now,
+        ),
+    ])
+    await session.commit()
+
+    async def _boom(db, symbol):
+        raise RuntimeError("scoring unavailable")
+
+    async def _fake_scan_ticker(db, symbol, *args, **kwargs):
+        comp = (
+            await db.execute(select(CompositeScore).where(CompositeScore.ticker_id == ticker.id))
+        ).scalar_one()
+        dimensions = (
+            await db.execute(select(DimensionScore).where(DimensionScore.ticker_id == ticker.id))
+        ).scalars().all()
+        assert comp.is_stale is True
+        assert all(score.is_stale for score in dimensions)
+        return []
+
+    monkeypatch.setattr(scoring_service, "compute_all_dimensions", _boom)
+    monkeypatch.setattr(rr_scanner_service, "scan_ticker", _fake_scan_ticker)
+
+    await rr_scanner_service.scan_all_tickers(session)
+
+
+async def test_scan_error_does_not_stop_later_tickers(session, monkeypatch):
+    session.add_all([Ticker(symbol="AAA"), Ticker(symbol="BBB")])
+    await session.commit()
+
+    scanned: list[str] = []
+
+    async def _fake_scan_ticker(db, symbol, *args, **kwargs):
+        scanned.append(symbol)
+        if symbol == "AAA":
+            raise RuntimeError("scan blew up")
+        return []
+
+    monkeypatch.setattr(rr_scanner_service, "scan_ticker", _fake_scan_ticker)
+
+    await rr_scanner_service.scan_all_tickers(session)
+
+    assert scanned == ["AAA", "BBB"]