diff --git a/app/services/fundamentals_peers.py b/app/services/fundamentals_peers.py new file mode 100644 index 0000000..c30c970 --- /dev/null +++ b/app/services/fundamentals_peers.py @@ -0,0 +1,85 @@ +"""Pure peer comparison for fundamentals (read-time). + +Peers are tracked-universe issuers sharing the **first two SIC digits**, +deduplicated by CIK (GOOG/GOOGL are one issuer, one observation). This module is +the pure statistics core: given a subject value and the peer group's values for a +metric, it returns median + polarity-aware favorable percentile + peer_count, or +None when there are fewer than the minimum valid peers (the caller then omits the +industry object entirely rather than show a misleading comparison). + +Grouping (which issuers share a 2-digit SIC, CIK-dedup) is the API's job; this +module only does the math. **Absolute net_debt is size-dependent and must not get +a peer percentile** — leverage is compared via net_debt_to_ebitda. +""" + +from __future__ import annotations + +import statistics +from dataclasses import dataclass + +MIN_PEERS = 5 + +# Metric -> is a higher value more favorable? (Peer-eligible metrics only; +# absolute net_debt is intentionally absent — size-dependent.) +HIGHER_IS_BETTER: dict[str, bool] = { + "revenue_growth_yoy": True, + "eps_growth_yoy": True, + "operating_margin": True, + "fcf_margin": True, + "fcf_yield": True, + "net_debt_to_ebitda": False, # lower leverage is better + "pe": False, # cheaper is better + "share_count_change_yoy": False, # dilution is bad +} + + +@dataclass +class PeerStat: + median: float + favorable_percentile: int # 0-100, polarity-aware (higher = more favorable) + peer_count: int # valid issuers in the group + + +def peer_stat( + subject: float | None, + group_values: list[float | None], + *, + higher_is_better: bool, + min_peers: int = MIN_PEERS, +) -> PeerStat | None: + """Median + favorable percentile for ``subject`` within its group. + + ``group_values`` is every issuer's value for the metric (including the + subject), CIK-deduplicated by the caller. Nulls are excluded. Returns None + when fewer than ``min_peers`` valid values exist, or the subject is null. + """ + valid = [v for v in group_values if v is not None] + if subject is None or len(valid) < min_peers: + return None + median = statistics.median(valid) + if higher_is_better: + favorable = sum(1 for v in valid if v <= subject) + else: + favorable = sum(1 for v in valid if v >= subject) + percentile = round(favorable / len(valid) * 100) + return PeerStat(median=median, favorable_percentile=percentile, peer_count=len(valid)) + + +def peer_stat_for( + metric_key: str, subject: float | None, group_values: list[float | None], **kwargs +) -> PeerStat | None: + """Convenience wrapper that looks up polarity by metric key. Returns None for + metrics not eligible for peer comparison (e.g. absolute net_debt).""" + if metric_key not in HIGHER_IS_BETTER: + return None + return peer_stat( + subject, group_values, higher_is_better=HIGHER_IS_BETTER[metric_key], **kwargs + ) + + +def two_digit_sic(sic: str | None) -> str | None: + """The 2-digit SIC prefix used for grouping, or None if unusable.""" + if not sic: + return None + digits = str(sic).strip() + return digits[:2] if len(digits) >= 2 and digits[:2].isdigit() else None diff --git a/app/services/fundamentals_reads.py b/app/services/fundamentals_reads.py new file mode 100644 index 0000000..60d80bb --- /dev/null +++ b/app/services/fundamentals_reads.py @@ -0,0 +1,104 @@ +"""Deterministic text 'reads' for the fundamentals panel (pure, one rule set). + +The tape reads and the header sentence use identical outputs — no LLM, no new +composite score. Thresholds are tunable named constants, not scattered literals +(plan: ±2pp growth, ±1pp margins, ±1% dilution, 60/40 peer bands, ≥3 periods). + +Consumers pass metric series (value + dated history, from +``fundamentals_derivation``) and peer percentiles; these functions return short +strings or None (render "—", no read). +""" + +from __future__ import annotations + +from statistics import mean +from typing import Any + +MIN_PERIODS = 3 +GROWTH_ACCEL_PP = 2.0 +MARGIN_MOVE_PP = 1.0 +SHARE_DILUTION_PCT = 1.0 +PEER_FAVORABLE = 60 +PEER_ADVERSE = 40 + + +def _history_values(history: list[Any]) -> list[float]: + return [p.value for p in history if p.value is not None] + + +def growth_read(history: list[Any]) -> str | None: + """Change in a YoY-growth series: latest − prior. Needs >= 3 periods.""" + vals = _history_values(history) + if len(vals) < MIN_PERIODS: + return None + delta = vals[-1] - vals[-2] + if delta >= GROWTH_ACCEL_PP: + return "accelerating" + if delta <= -GROWTH_ACCEL_PP: + return "decelerating" + return "steady" + + +def margin_read(history: list[Any]) -> str | None: + """Latest margin vs the mean of prior periods (pp). Needs >= 3 periods.""" + vals = _history_values(history) + if len(vals) < MIN_PERIODS: + return None + delta = vals[-1] - mean(vals[:-1]) + if delta >= MARGIN_MOVE_PP: + return "improving" + if delta <= -MARGIN_MOVE_PP: + return "deteriorating" + return "stable" + + +def share_count_read(value: float | None) -> str | None: + """Share-count YoY %: >+1% dilution, <-1% buying back, else flat.""" + if value is None: + return None + if value > SHARE_DILUTION_PCT: + return f"{value:.1f}% dilution" + if value < -SHARE_DILUTION_PCT: + return "buying back" + return "flat" + + +def peer_read(metric_key: str, favorable_percentile: int | None) -> str | None: + """Peer-relative read for a metric, polarity already baked into the + percentile (higher = more favorable).""" + if favorable_percentile is None: + return None + if favorable_percentile >= PEER_FAVORABLE: + return _FAVORABLE.get(metric_key, "above peers") + if favorable_percentile <= PEER_ADVERSE: + return _ADVERSE.get(metric_key, "below peers") + return "in line" + + +_FAVORABLE = { + "pe": "attractively valued", + "fcf_yield": "above peers", + "net_debt_to_ebitda": "conservative leverage", +} +_ADVERSE = { + "pe": "priced above peers", + "fcf_yield": "below peers", + "net_debt_to_ebitda": "elevated leverage", +} + + +def header_sentence( + growth: str | None, margin: str | None, valuation: str | None +) -> str: + """Join the growth / margin / peer-valuation reads with ' · ', omitting + segments with no read. Segment sources are fixed by the caller (growth = + revenue-growth read, margin = operating-margin read, valuation = P/E peer + read falling back to FCF yield).""" + parts = [] + if growth: + parts.append(f"growth {growth}") + if margin: + parts.append(f"margins {margin}") + if valuation: + parts.append(f"valuation {valuation}") + return " · ".join(parts) diff --git a/tests/unit/test_fundamentals_peers.py b/tests/unit/test_fundamentals_peers.py new file mode 100644 index 0000000..6702d02 --- /dev/null +++ b/tests/unit/test_fundamentals_peers.py @@ -0,0 +1,50 @@ +"""Tests for pure peer statistics.""" + +from __future__ import annotations + +from app.services import fundamentals_peers as pr + + +def test_peer_stat_higher_is_better_percentile_and_median(): + s = pr.peer_stat(3, [1, 2, 3, 4, 5], higher_is_better=True) + assert s.median == 3 + assert s.favorable_percentile == 60 # beats/ties 3 of 5 + assert s.peer_count == 5 + + +def test_peer_stat_lower_is_better_flips_direction(): + s = pr.peer_stat(3, [1, 2, 3, 4, 5], higher_is_better=False) + assert s.favorable_percentile == 60 # 3 of 5 are >= 3 + + +def test_peer_stat_top_and_bottom(): + assert pr.peer_stat(5, [1, 2, 3, 4, 5], higher_is_better=True).favorable_percentile == 100 + assert pr.peer_stat(1, [1, 2, 3, 4, 5], higher_is_better=True).favorable_percentile == 20 + + +def test_peer_stat_requires_min_valid_peers(): + assert pr.peer_stat(3, [1, 2, 3, None], higher_is_better=True) is None # 3 valid < 5 + assert pr.peer_stat(None, [1, 2, 3, 4, 5], higher_is_better=True) is None # null subject + + +def test_peer_stat_excludes_nulls_from_group(): + s = pr.peer_stat(3, [1, 2, 3, 4, 5, None, None], higher_is_better=True) + assert s.peer_count == 5 # nulls dropped + + +def test_net_debt_is_not_peer_eligible(): + assert pr.peer_stat_for("net_debt", 100, [10, 20, 30, 40, 50]) is None # size-dependent + assert pr.peer_stat_for("net_debt_to_ebitda", 1.0, [1, 2, 3, 4, 5]) is not None + + +def test_peer_stat_for_uses_polarity(): + # pe is lower-is-better: a low pe beats most peers + s = pr.peer_stat_for("pe", 10, [10, 20, 30, 40, 50]) + assert s.favorable_percentile == 100 + + +def test_two_digit_sic(): + assert pr.two_digit_sic("7372") == "73" + assert pr.two_digit_sic("3571") == "35" + assert pr.two_digit_sic(None) is None + assert pr.two_digit_sic("x") is None diff --git a/tests/unit/test_fundamentals_reads.py b/tests/unit/test_fundamentals_reads.py new file mode 100644 index 0000000..23da0e5 --- /dev/null +++ b/tests/unit/test_fundamentals_reads.py @@ -0,0 +1,53 @@ +"""Tests for deterministic text reads, incl. threshold boundaries.""" + +from __future__ import annotations + +from types import SimpleNamespace + +from app.services import fundamentals_reads as rd + + +def _hist(*values): + return [SimpleNamespace(value=v, period_end=None) for v in values] + + +def test_growth_read_boundaries(): + assert rd.growth_read(_hist(5, 6, 8)) == "accelerating" # +2.0 exactly (>=) + assert rd.growth_read(_hist(5, 6, 7.9)) == "steady" # +1.9 < 2.0 + assert rd.growth_read(_hist(10, 9, 7)) == "decelerating" # -2.0 exactly + assert rd.growth_read(_hist(5, 6)) is None # < 3 periods + + +def test_margin_read_vs_mean_of_prior(): + # prior mean = (19+20)/2 = 19.5; latest 21 -> +1.5 -> improving + assert rd.margin_read(_hist(19, 20, 21)) == "improving" + # latest exactly +1.0 over prior mean -> improving + assert rd.margin_read(_hist(20, 20, 21)) == "improving" + # within band + assert rd.margin_read(_hist(20, 20, 20.5)) == "stable" + assert rd.margin_read(_hist(21, 20)) is None # < 3 periods + + +def test_share_count_read(): + assert rd.share_count_read(1.8) == "1.8% dilution" + assert rd.share_count_read(-2.0) == "buying back" + assert rd.share_count_read(1.0) == "flat" # boundary: not > 1.0 + assert rd.share_count_read(None) is None + + +def test_peer_read_bands_and_polarity_phrasing(): + assert rd.peer_read("operating_margin", 60) == "above peers" # boundary favorable + assert rd.peer_read("operating_margin", 40) == "below peers" # boundary adverse + assert rd.peer_read("operating_margin", 50) == "in line" + assert rd.peer_read("pe", 65) == "attractively valued" + assert rd.peer_read("pe", 30) == "priced above peers" + assert rd.peer_read("net_debt_to_ebitda", 20) == "elevated leverage" + assert rd.peer_read("pe", None) is None + + +def test_header_sentence_omits_missing_segments(): + assert rd.header_sentence("accelerating", "stable", "priced above peers") == ( + "growth accelerating · margins stable · valuation priced above peers" + ) + assert rd.header_sentence(None, "improving", None) == "margins improving" + assert rd.header_sentence(None, None, None) == ""