signal-platform/tests/unit/test_fundamental_data_refresh.py

"""A5 activation: local candidate derivation and compat-cache refresh."""

from __future__ import annotations

import json
from datetime import date, datetime, timedelta, timezone

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.database import Base
from app.models.earnings_event import EarningsEvent
from app.models.fundamental import FundamentalData
from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.ohlcv import OHLCVRecord
from app.models.score import CompositeScore, DimensionScore
from app.models.ticker import Ticker
from app.services import fundamentals_candidate_service as candidates
from app.services import fundamentals_derivation as deriv
from app.services import fundamental_data_refresh_service as refresh_service


UTC = timezone.utc
NOW = datetime(2026, 7, 24, 10, 0, tzinfo=UTC)
TODAY = date(2026, 7, 24)

_engine = create_async_engine("sqlite+aiosqlite://", echo=False)
_session_factory = async_sessionmaker(
    _engine, class_=AsyncSession, expire_on_commit=False
)


@pytest.fixture(autouse=True)
async def _setup_tables():
    async with _engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    yield
    async with _engine.begin() as connection:
        await connection.run_sync(Base.metadata.drop_all)


@pytest.fixture
async def session() -> AsyncSession:
    async with _session_factory() as db:
        yield db


def _snapshot_rows(cik: str) -> list[FundamentalSnapshot]:
    rows: list[FundamentalSnapshot] = []
    periods = ("Q1", "Q2", "Q3", "FY")
    months = (3, 6, 9, 12)
    for fiscal_year, multiplier in ((2025, 1.0), (2026, 1.1)):
        revenue = [100 * multiplier, 110 * multiplier, 120 * multiplier, 130 * multiplier]
        eps = [1.0 * multiplier, 1.1 * multiplier, 1.2 * multiplier, 1.3 * multiplier]
        for index, fiscal_period in enumerate(periods):
            period_end = date(fiscal_year, months[index], 28)
            rows.append(
                FundamentalSnapshot(
                    cik=cik,
                    accession=f"{cik}-{fiscal_year}-{fiscal_period}",
                    form="10-K" if fiscal_period == "FY" else "10-Q",
                    filed_date=period_end,
                    accepted_at=datetime(
                        fiscal_year, months[index], 28, tzinfo=UTC
                    ),
                    period_end=period_end,
                    fiscal_year=fiscal_year,
                    fiscal_period=fiscal_period,
                    revenue=sum(revenue[: index + 1]),
                    diluted_eps=sum(eps[: index + 1]),
                    shares_outstanding=1_000,
                )
            )
    return rows


async def test_refresh_updates_all_fields_and_invalidates_scores(
    session: AsyncSession,
):
    first = Ticker(symbol="AAA", cik="0000000001")
    second = Ticker(symbol="AAB", cik="0000000001")
    session.add_all([first, second])
    await session.flush()
    session.add_all(_snapshot_rows(first.cik))
    session.add_all(
        [
            OHLCVRecord(
                ticker_id=first.id,
                date=TODAY - timedelta(days=1),
                open=100,
                high=100,
                low=100,
                close=100,
                volume=100,
            ),
            OHLCVRecord(
                ticker_id=second.id,
                date=TODAY - timedelta(days=1),
                open=200,
                high=200,
                low=200,
                close=200,
                volume=100,
            ),
            EarningsEvent(
                ticker_id=first.id,
                announce_date=TODAY - timedelta(days=10),
                session="amc",
                eps_estimate=2,
                eps_actual=2.2,
                source="dolt_earnings",
            ),
            EarningsEvent(
                ticker_id=first.id,
                announce_date=TODAY,
                session="amc",
                source="dolt_earnings",
            ),
        ]
    )
    for ticker in (first, second):
        session.add(
            FundamentalData(
                ticker_id=ticker.id,
                pe_ratio=1,
                revenue_growth=1,
                earnings_surprise=1,
                market_cap=1,
                fetched_at=NOW - timedelta(days=1),
            )
        )
        session.add(
            DimensionScore(
                ticker_id=ticker.id,
                dimension="fundamental",
                score=50,
                is_stale=False,
                computed_at=NOW,
            )
        )
        session.add(
            CompositeScore(
                ticker_id=ticker.id,
                score=50,
                is_stale=False,
                weights_json="{}",
                computed_at=NOW,
            )
        )
    await session.commit()

    summary = await refresh_service.refresh(
        session, now=NOW, today=TODAY
    )

    stored = {
        row.ticker_id: row
        for row in (
            await session.execute(select(FundamentalData))
        ).scalars()
    }
    assert summary["refreshed"] == 2
    assert summary["score_inputs_changed"] == 2
    assert stored[first.id].pe_ratio == pytest.approx(100 / 5.06)
    assert stored[second.id].pe_ratio == pytest.approx(200 / 5.06)
    assert stored[first.id].revenue_growth == pytest.approx(10)
    assert stored[first.id].earnings_surprise == pytest.approx(10)
    assert stored[first.id].market_cap == pytest.approx(100_000)
    assert stored[first.id].next_earnings_date == TODAY
    metadata = json.loads(stored[first.id].unavailable_fields_json)
    assert metadata["source_pe_ratio"] == "sec_facts+ohlcv_records"
    assert metadata["source_next_earnings_date"] == "dolt_earnings"

    dimensions = (
        await session.execute(select(DimensionScore))
    ).scalars().all()
    composites = (
        await session.execute(select(CompositeScore))
    ).scalars().all()
    assert all(row.is_stale for row in dimensions)
    assert all(row.is_stale for row in composites)

    for row in (*dimensions, *composites):
        row.is_stale = False
    await session.commit()
    unchanged = await refresh_service.refresh(
        session, now=NOW + timedelta(hours=1), today=TODAY
    )
    assert unchanged["score_inputs_changed"] == 0
    assert not any(
        (await session.execute(select(DimensionScore.is_stale))).scalars()
    )
    assert not any(
        (await session.execute(select(CompositeScore.is_stale))).scalars()
    )


async def test_candidate_uses_guarded_derive_outputs_and_share_fallback(
    session: AsyncSession, monkeypatch
):
    ticker = Ticker(symbol="GUARD", cik="0000000002")
    session.add(ticker)
    await session.flush()
    session.add(
        FundamentalSnapshot(
            cik=ticker.cik,
            accession="raw-accession",
            form="10-Q",
            filed_date=TODAY,
            accepted_at=NOW,
            period_end=TODAY,
            fiscal_year=2026,
            fiscal_period="Q2",
            diluted_eps=99,
            shares_outstanding=999,
        )
    )
    session.add(
        OHLCVRecord(
            ticker_id=ticker.id,
            date=TODAY,
            open=50,
            high=50,
            low=50,
            close=50,
            volume=100,
        )
    )
    await session.commit()

    def guarded(_rows):
        return deriv.DerivedFundamentals(
            metrics={
                "revenue_growth_yoy": deriv.MetricSeries(value=7)
            },
            ttm_diluted_eps=None,
            ttm_diluted_eps_caveat="split guard applied",
            shares_outstanding=123,
            shares_outstanding_estimated=True,
            latest_period_end=TODAY,
            latest_filed_date=TODAY,
        )

    monkeypatch.setattr(candidates.deriv, "derive", guarded)
    candidate = (await candidates.build_candidates(session, today=TODAY))[0]

    assert candidate.pe_ratio is None
    assert candidate.market_cap == 50 * 123
    assert candidate.revenue_growth == 7
    assert candidate.unavailable_fields["pe_ratio"] == "split guard applied"
    assert "weighted-average" in candidate.unavailable_fields["market_cap_estimated"]