signal-platform/tests/unit/test_fundamentals_derivation.py

"""Tests for pure read-time derivation of fundamentals from YTD snapshots."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import date, datetime, timezone

import pytest

from app.services import fundamentals_derivation as fd

UTC = timezone.utc


@dataclass
class Snap:
    fiscal_year: int
    fiscal_period: str
    period_end: date
    filed_date: date
    accepted_at: datetime
    revenue: float | None = None
    net_income: float | None = None
    operating_income: float | None = None
    diluted_eps: float | None = None
    cfo: float | None = None
    capex: float | None = None
    depreciation_amortization: float | None = None
    cash_and_st_investments: float | None = None
    total_debt: float | None = None
    shares_outstanding: float | None = None
    weighted_avg_diluted_shares: float | None = None


_FP = ["Q1", "Q2", "Q3", "FY"]
_ENDS = {  # period_end per (fy, quarter index 0..3)
    2025: [date(2024, 12, 31), date(2025, 3, 31), date(2025, 6, 30), date(2025, 9, 30)],
    2026: [date(2025, 12, 31), date(2026, 3, 31), date(2026, 6, 30), date(2026, 9, 30)],
}


def _year(fy, discretes: dict[str, list[float]], instants: dict[str, list] | None = None):
    """Build 4 snapshot rows (Q1,Q2,Q3,FY) with YTD-cumulative flow fields from the
    given per-quarter discrete values; instants set as-is per quarter."""
    rows = []
    for i, fp in enumerate(_FP):
        r = Snap(fy, fp, _ENDS[fy][i], _ENDS[fy][i], datetime(fy, 1 + i, 1, tzinfo=UTC))
        for fname, ds in discretes.items():
            setattr(r, fname, round(sum(ds[: i + 1]), 4))  # cumulative YTD
        for fname, vals in (instants or {}).items():
            setattr(r, fname, vals[i])
        rows.append(r)
    return rows


def _two_years():
    rev25 = [100, 110, 120, 130]
    rev26 = [110, 121, 132, 143]  # +10% each quarter YoY
    rows = _year(2025, {
        "revenue": rev25,
        "operating_income": [x * 0.2 for x in rev25],
        "diluted_eps": [1.0, 1.1, 1.2, 1.3],
        "cfo": [x * 0.25 for x in rev25],
        "capex": [x * 0.05 for x in rev25],
        "depreciation_amortization": [x * 0.05 for x in rev25],
    }, instants={"shares_outstanding": [1000, 1000, 1000, 1000], "cash_and_st_investments": [40] * 4, "total_debt": [140] * 4})
    rows += _year(2026, {
        "revenue": rev26,
        "operating_income": [x * 0.2 for x in rev26],
        "diluted_eps": [1.1, 1.21, 1.32, 1.43],
        "cfo": [x * 0.25 for x in rev26],
        "capex": [x * 0.05 for x in rev26],
        "depreciation_amortization": [x * 0.05 for x in rev26],
    }, instants={"shares_outstanding": [900, 900, 900, 900], "cash_and_st_investments": [50] * 4, "total_debt": [150] * 4})
    return rows


def test_revenue_growth_yoy_and_q4_derivation():
    d = fd.derive(_two_years())
    # TTM revenue FY2026 = 110+121+132+143 = 506; FY2025 = 460 -> +10%
    assert d.metrics["revenue_growth_yoy"].value == pytest.approx(10.0, abs=1e-6)
    # latest period is FY2026
    assert d.latest_period_end == date(2026, 9, 30)
    # tape has 4 points, newest last, each carrying a period_end
    hist = d.metrics["revenue_growth_yoy"].history
    assert len(hist) == 4 and hist[-1].period_end == date(2026, 9, 30)


def test_operating_and_fcf_margin():
    d = fd.derive(_two_years())
    assert d.metrics["operating_margin"].value == pytest.approx(20.0, abs=1e-6)
    # FCF margin = (TTM cfo - TTM capex)/TTM rev = (0.25 - 0.05) = 20%
    assert d.metrics["fcf_margin"].value == pytest.approx(20.0, abs=1e-6)


def test_net_debt_leverage_and_share_dilution():
    d = fd.derive(_two_years())
    # net debt = total_debt - cash = 150 - 50 = 100 (latest instant)
    assert d.metrics["net_debt"].value == pytest.approx(100.0)
    # EBITDA TTM = TTM operating_income + TTM D&A; net_debt/ebitda
    op_ttm = 506 * 0.2       # 101.2
    da_ttm = 506 * 0.05      # 25.3
    assert d.metrics["net_debt_to_ebitda"].value == pytest.approx(100.0 / (op_ttm + da_ttm), rel=1e-6)
    # shares 900 vs 1000 a year earlier -> -10% (buyback)
    assert d.metrics["share_count_change_yoy"].value == pytest.approx(-10.0, abs=1e-6)


def test_split_suspect_share_move_suppresses_share_and_eps_comparisons():
    rows = _two_years()
    for row in rows:
        if row.fiscal_year == 2026:
            row.shares_outstanding = 2000  # +100% resembles an unadjusted 2-for-1 split

    d = fd.derive(rows)

    for key in ("share_count_change_yoy", "eps_growth_yoy"):
        series = d.metrics[key]
        assert series.value is None
        assert series.history[-1].value is None
        assert "possible split" in series.caveat
    assert d.metrics["revenue_growth_yoy"].value == pytest.approx(10.0)


def test_valuation_inputs():
    d = fd.derive(_two_years())
    # TTM diluted EPS FY2026 = 1.1+1.21+1.32+1.43 = 5.06
    assert d.ttm_diluted_eps == pytest.approx(5.06, abs=1e-6)
    # TTM FCF = TTM cfo - TTM capex = 506*0.25 - 506*0.05 = 101.2
    assert d.ttm_fcf == pytest.approx(506 * 0.20, abs=1e-6)
    assert d.shares_outstanding == 900


def test_missing_period_yields_null_never_partial():
    rows = _two_years()
    # drop FY2026 Q3 -> discrete Q3 and Q4 (needs YTD Q3) become underivable,
    # so TTM at FY2026 is null -> revenue growth null (not a partial sum)
    rows = [r for r in rows if not (r.fiscal_year == 2026 and r.fiscal_period == "Q3")]
    d = fd.derive(rows)
    assert d.metrics["revenue_growth_yoy"].value is None
    assert d.ttm_diluted_eps is None


def test_net_debt_requires_both_components():
    rows = _two_years()
    for r in rows:  # drop debt on the latest year -> can't form net debt
        if r.fiscal_year == 2026:
            r.total_debt = None
    d = fd.derive(rows)
    assert d.metrics["net_debt"].value is None
    assert d.metrics["net_debt_to_ebitda"].value is None  # net debt null -> leverage null


def test_leverage_null_when_ebitda_nonpositive():
    rows = _two_years()
    for r in rows:  # negative operating income -> TTM EBITDA <= 0
        r.operating_income = -abs(r.revenue)
        r.depreciation_amortization = 1
    d = fd.derive(rows)
    assert d.metrics["net_debt"].value == pytest.approx(100.0)  # net debt still valid
    assert d.metrics["net_debt_to_ebitda"].value is None        # but leverage nulled


def test_tape_stops_at_a_gap():
    rows = [r for r in _two_years() if not (r.fiscal_year == 2026 and r.fiscal_period == "Q1")]
    d = fd.derive(rows)
    hist = d.metrics["operating_margin"].history
    # consecutive suffix ending at FY2026: Q2, Q3, FY (not compressed across the Q1 gap)
    assert [p.period_end for p in hist] == [date(2026, 3, 31), date(2026, 6, 30), date(2026, 9, 30)]


def test_yoy_growth_null_when_prior_nonpositive():
    rows = _two_years()
    for r in rows:  # prior-year TTM EPS becomes negative
        if r.fiscal_year == 2025:
            r.diluted_eps = -abs(r.diluted_eps)
    d = fd.derive(rows)
    assert d.metrics["eps_growth_yoy"].value is None  # loss->profit is not a %


def test_amendment_selection_newest_accepted_wins():
    rows = _two_years()
    # an amendment to FY2026 FY restates revenue YTD higher, accepted later
    amended = Snap(2026, "FY", date(2026, 9, 30), date(2026, 11, 1),
                   datetime(2027, 1, 1, tzinfo=UTC), revenue=999999,
                   operating_income=100, diluted_eps=1.43, cfo=100, capex=10,
                   depreciation_amortization=25, shares_outstanding=900,
                   cash_and_st_investments=50, total_debt=150)
    d = fd.derive(rows + [amended])
    # Q4 revenue discrete now uses the amended YTD(FY)=999999 minus YTD(Q3)=363
    # so TTM/growth reflects the amendment, proving newest accepted_at won.
    assert d.metrics["revenue_growth_yoy"].value != pytest.approx(10.0, abs=1e-6)


# -- partial amendments (A5 parity findings) ---------------------------------

def test_partial_amendment_does_not_blank_the_period():
    # DVN's FY2025 10-K/A carries no financial facts at the report date. Taking
    # the newest accession wholesale nulled the period, and with it the quarter
    # chain, TTM and YoY.
    rows = _two_years()
    part_iii_only = Snap(2026, "FY", date(2026, 9, 30), date(2026, 11, 1),
                         datetime(2027, 1, 1, tzinfo=UTC))
    baseline = fd.derive(rows)
    d = fd.derive(rows + [part_iii_only])
    assert d.ttm_diluted_eps == pytest.approx(baseline.ttm_diluted_eps)
    assert d.metrics["revenue_growth_yoy"].value == pytest.approx(
        baseline.metrics["revenue_growth_yoy"].value
    )


def test_amendment_restating_one_field_leaves_the_others_intact():
    rows = _two_years()
    revenue_only = Snap(2026, "FY", date(2026, 9, 30), date(2026, 11, 1),
                        datetime(2027, 1, 1, tzinfo=UTC), revenue=999999)
    baseline = fd.derive(rows)
    d = fd.derive(rows + [revenue_only])
    assert d.metrics["revenue_growth_yoy"].value != pytest.approx(
        baseline.metrics["revenue_growth_yoy"].value
    )
    assert d.ttm_diluted_eps == pytest.approx(baseline.ttm_diluted_eps)  # fell back


def test_same_key_row_for_a_different_period_is_never_merged():
    # SEC labels two different year-ends with one fiscal_year for some filers
    # (FRT, CRM). That is a mislabelled filing, not an amendment -- merging the
    # two would silently blend fiscal years.
    rows = _two_years()
    mislabelled = Snap(2026, "FY", date(2027, 9, 30), date(2027, 11, 1),
                       datetime(2027, 12, 1, tzinfo=UTC), revenue=999999)
    selected = fd._select_latest_per_period(rows + [mislabelled])
    assert selected[(2026, "FY")] is mislabelled


# -- split safety for the TTM EPS scalar (A5 parity findings) ----------------

def _split_rows():
    """Two years where the share count jumps ~25x at the latest quarter, as
    BKNG's did (31.7M -> 774.9M) when its split landed mid-window."""
    rows = _two_years()
    for row in rows:
        if (row.fiscal_year, row.fiscal_period) == (2026, "FY"):
            row.shares_outstanding = 25000.0  # vs 1000 a year earlier
    return rows


def test_split_suppresses_ttm_diluted_eps():
    # TTM sums four quarters of per-share values; a split inside the window
    # mixes units. Unguarded this produced BKNG's P/E of 1.10, which clamps to a
    # *perfect* fundamental sub-score -- worse than having no value at all.
    d = fd.derive(_split_rows())
    assert d.ttm_diluted_eps is None
    assert d.ttm_diluted_eps_caveat == fd.SPLIT_SENSITIVE_CAVEAT


def test_ttm_diluted_eps_survives_when_no_split_is_suspected():
    d = fd.derive(_two_years())
    assert d.ttm_diluted_eps is not None
    assert d.ttm_diluted_eps_caveat is None


def test_split_guard_leaves_dollar_scalars_alone():
    # Only per-share values are split-sensitive; FCF is in dollars.
    baseline = fd.derive(_two_years())
    d = fd.derive(_split_rows())
    assert d.ttm_fcf == pytest.approx(baseline.ttm_fcf)


# -- multi-class share-count fallback (A5 parity findings) -------------------

def test_shares_fall_back_to_weighted_average_when_cover_page_count_is_absent():
    # META/CMCSA/BRK-B/CHTR report the cover-page count per share class, which is
    # dimensional and therefore absent from companyfacts -- silently removing
    # market cap and FCF yield for some of the largest issuers.
    rows = _two_years()
    for row in rows:
        row.shares_outstanding = None
        row.weighted_avg_diluted_shares = 2_564_000_000.0
    d = fd.derive(rows)
    assert d.shares_outstanding == 2_564_000_000.0
    assert d.shares_outstanding_estimated is True


def test_point_in_time_share_count_is_preferred_and_not_flagged():
    baseline = fd.derive(_two_years()).shares_outstanding
    assert baseline is not None, "fixture should carry a cover-page count"
    rows = _two_years()
    for row in rows:
        row.weighted_avg_diluted_shares = 1.0  # must lose to the real count
    d = fd.derive(rows)
    assert d.shares_outstanding == baseline
    assert d.shares_outstanding_estimated is False


def test_weighted_average_fallback_survives_a_partial_amendment():
    # A Part-III-only 10-K/A on a multi-class issuer's latest period: the merged
    # row must keep the weighted-average count, or market cap silently vanishes.
    rows = _two_years()
    for row in rows:
        row.shares_outstanding = None
        row.weighted_avg_diluted_shares = 2_564_000_000.0
    part_iii_only = Snap(2026, "FY", date(2026, 9, 30), date(2026, 11, 1),
                         datetime(2027, 1, 1, tzinfo=UTC))
    d = fd.derive(rows + [part_iii_only])
    assert d.shares_outstanding == 2_564_000_000.0
    assert d.shares_outstanding_estimated is True


def test_no_share_count_at_all_stays_none_and_unflagged():
    rows = _two_years()
    for row in rows:
        row.shares_outstanding = None
    d = fd.derive(rows)
    assert d.shares_outstanding is None
    assert d.shares_outstanding_estimated is False


def test_merge_lists_cover_every_parser_field():
    """_MERGED_FIELDS/_CARRIED_FIELDS are hand-maintained, and _merge_amendments
    builds the merged row from them alone — so a parser field missing from both
    is not merely stale on a merged period, it is *absent*, and callers using
    getattr(row, name, None) read None. That is how weighted_avg_diluted_shares
    silently lost market cap for multi-class issuers with a partial amendment.
    Adding a column to SnapshotRow must fail here rather than lose data quietly.
    """
    import dataclasses

    from app.services.sec_facts_parser import SnapshotRow

    parser_fields = {f.name for f in dataclasses.fields(SnapshotRow)}
    covered = set(fd._MERGED_FIELDS) | set(fd._CARRIED_FIELDS)
    assert not parser_fields - covered, (
        f"parser fields not merged or carried: {sorted(parser_fields - covered)}"
    )