Completes the pure read-time core. fundamentals_peers.py: median + polarity-aware favorable percentile + peer_count for a subject within its SIC group (CIK-deduped by the caller); returns None below MIN_PEERS=5 so the caller omits the industry object. Absolute net_debt is intentionally NOT peer-eligible (size-dependent) — leverage compares via net_debt_to_ebitda. HIGHER_IS_BETTER polarity map + two_digit_sic() grouping key. fundamentals_reads.py: one shared deterministic rule set (no LLM): growth_read (+-2pp), margin_read (latest vs mean-of-prior, +-1pp), share_count_read (+-1%), peer_read (60/40 bands, polarity-aware phrasing per metric), header_sentence (growth · margins · valuation, omitting empty). Tunable named constants; >=3 periods required for a series read. Tests: 8 peer + 5 reads, anchored on the boundary cases (exactly +2.0pp, exactly 60th percentile, exactly +1.0pp margin). 13 passed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""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
|