Files
signal-platform/tests/unit/test_fundamentals_derivation.py
T

239 lines
8.2 KiB
Python

"""Tests for pure read-time derivation of fundamentals from YTD snapshots."""
from __future__ import annotations
from dataclasses import dataclass, replace
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
_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_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_derive_as_of_excludes_future_amendment():
rows = _two_years()
original = next(
row for row in rows if row.fiscal_year == 2026 and row.fiscal_period == "FY"
)
amendment = replace(
original,
accepted_at=datetime(2027, 1, 1, tzinfo=UTC),
revenue=999999,
)
before = fd.derive_as_of([*rows, amendment], datetime(2026, 12, 31, tzinfo=UTC))
after = fd.derive_as_of([*rows, amendment], datetime(2027, 1, 2, tzinfo=UTC))
assert before.metrics["revenue_growth_yoy"].value == pytest.approx(10.0)
assert after.metrics["revenue_growth_yoy"].value != pytest.approx(10.0)
def test_derive_as_of_treats_sqlite_naive_acceptance_as_utc():
rows = _two_years()
rows[0].accepted_at = datetime(2025, 1, 1)
result = fd.derive_as_of(rows, datetime(2027, 1, 1, tzinfo=UTC))
assert result.latest_period_end == date(2026, 9, 30)
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)