signal-platform/app/services/fundamental_data_refresh_service.py

"""A5 activation: refresh the legacy fundamentals cache from local bulk data."""

from __future__ import annotations

import json
from datetime import date, datetime, timezone
from typing import Any

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import insert_for_session
from app.models.fundamental import FundamentalData
from app.models.score import CompositeScore, DimensionScore
from app.services import fundamentals_candidate_service, settings_store


# Absence is deliberately false. Production activation therefore requires one
# explicit, durable SystemSetting change after the A5 evidence is approved.
ACTIVATION_KEY = "fundamental_data_sec_dolt_cutover_enabled"
_SCORE_FIELDS = ("pe_ratio", "revenue_growth", "earnings_surprise")


async def is_enabled(db: AsyncSession) -> bool:
    raw = await settings_store.get_value(db, ACTIVATION_KEY, "false")
    return str(raw).strip().lower() == "true"


async def refresh_if_enabled(
    db: AsyncSession,
    *,
    now: datetime | None = None,
    today: date | None = None,
) -> dict[str, Any]:
    """Refresh atomically when activated; otherwise perform no writes."""
    if not await is_enabled(db):
        return {
            "enabled": False,
            "refreshed": 0,
            "score_inputs_changed": 0,
            "dimension_scores_staled": 0,
            "composite_scores_staled": 0,
        }
    return await refresh(db, now=now, today=today)


async def refresh(
    db: AsyncSession,
    *,
    now: datetime | None = None,
    today: date | None = None,
) -> dict[str, Any]:
    """Replace every ticker's compat-cache row in one database transaction.

    Candidate values are assembled before the first write and use only local
    PostgreSQL tables. A failure rolls the whole refresh back. Only changes to
    the three scoring inputs invalidate cached scores; market cap and the next
    earnings date are display-only.
    """
    refreshed_at = now or datetime.now(timezone.utc)
    candidates = await fundamentals_candidate_service.build_candidates(
        db, today=today
    )
    ticker_ids = [candidate.ticker_id for candidate in candidates]
    existing = await _existing_by_ticker(db, ticker_ids)
    changed_ids = {
        candidate.ticker_id
        for candidate in candidates
        if _score_inputs_changed(existing.get(candidate.ticker_id), candidate)
    }

    for candidate in candidates:
        unavailable_json = json.dumps(
            candidate.unavailable_fields, sort_keys=True
        )
        stmt = insert_for_session(db, FundamentalData).values(
            ticker_id=candidate.ticker_id,
            pe_ratio=candidate.pe_ratio,
            revenue_growth=candidate.revenue_growth,
            earnings_surprise=candidate.earnings_surprise,
            market_cap=candidate.market_cap,
            next_earnings_date=candidate.next_earnings_date,
            fetched_at=refreshed_at,
            unavailable_fields_json=unavailable_json,
        )
        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
                    ),
                },
            )
        )

    dimension_ids = await _fundamental_dimension_ids(db, changed_ids)
    composite_ids = await _composite_ids(db, changed_ids)
    if dimension_ids:
        await db.execute(
            update(DimensionScore)
            .where(DimensionScore.ticker_id.in_(dimension_ids))
            .values(is_stale=True)
        )
    if composite_ids:
        await db.execute(
            update(CompositeScore)
            .where(CompositeScore.ticker_id.in_(composite_ids))
            .values(is_stale=True)
        )

    await db.commit()
    return {
        "enabled": True,
        "refreshed": len(candidates),
        "score_inputs_changed": len(changed_ids),
        "dimension_scores_staled": len(dimension_ids),
        "composite_scores_staled": len(composite_ids),
    }


async def _existing_by_ticker(
    db: AsyncSession, ticker_ids: list[int]
) -> dict[int, FundamentalData]:
    if not ticker_ids:
        return {}
    rows = (
        await db.execute(
            select(FundamentalData).where(
                FundamentalData.ticker_id.in_(ticker_ids)
            )
        )
    ).scalars()
    return {row.ticker_id: row for row in rows}


async def _fundamental_dimension_ids(
    db: AsyncSession, ticker_ids: set[int]
) -> set[int]:
    if not ticker_ids:
        return set()
    rows = await db.execute(
        select(DimensionScore.ticker_id).where(
            DimensionScore.ticker_id.in_(ticker_ids),
            DimensionScore.dimension == "fundamental",
        )
    )
    return set(rows.scalars())


async def _composite_ids(
    db: AsyncSession, ticker_ids: set[int]
) -> set[int]:
    if not ticker_ids:
        return set()
    rows = await db.execute(
        select(CompositeScore.ticker_id).where(
            CompositeScore.ticker_id.in_(ticker_ids)
        )
    )
    return set(rows.scalars())


def _score_inputs_changed(
    existing: FundamentalData | None,
    candidate: fundamentals_candidate_service.CandidateFundamentals,
) -> bool:
    if existing is None:
        return True
    return any(
        getattr(existing, field) != getattr(candidate, field)
        for field in _SCORE_FIELDS
    )