Files
dennisthiessenandClaude Opus 4.8 979b4047dc fix(fundamentals): pure-core review — tie-aware percentile, null-safe reads
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>
2026-07-22 20:41:18 +02:00

116 lines
3.6 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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)