1. Peer percentile is now a tie-aware rank against the OTHER issuers ((worse + 0.5*tied)/(peers-1)): an all-equal group maps to 50 (not 100), the median maps to 50, a unique best to 100, a unique worst to 0. 2. Deterministic reads use the consecutive non-null suffix ending at the latest point (>=3 values): a null latest or an internal gap yields no read, so a read never reflects a period displayed as n/a. 3. Peer filtering excludes non-finite (NaN/±inf) as well as null, including an invalid subject. Tests updated + added (all-equal, median rank, non-finite, latest-null history, internal gap). 15 passed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
"""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 _latest_run(history: list[Any]) -> list[float]:
|
||
"""The consecutive non-null values ending at the latest point (oldest->newest).
|
||
A null latest, or an internal gap, truncates the run — so a read never reflects
|
||
a period whose displayed value is n/a."""
|
||
run: list[float] = []
|
||
for p in reversed(history):
|
||
if p.value is None:
|
||
break
|
||
run.append(p.value)
|
||
run.reverse()
|
||
return run
|
||
|
||
|
||
def growth_read(history: list[Any]) -> str | None:
|
||
"""Change in a YoY-growth series: latest − prior. Needs >= 3 consecutive
|
||
non-null values ending at the latest point."""
|
||
vals = _latest_run(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 consecutive
|
||
non-null values ending at the latest point."""
|
||
vals = _latest_run(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)
|