dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: record fundamentals research decision and clean up

Browse Source
This commit is contained in:
dennisthiessen
2026-07-23 17:50:33 +02:00
parent dba7ea739b
commit 361cfd7883
21 changed files with 102 additions and 209972 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/fundamentals_derivation.py
+16 -66
View File
@@ -20,7 +20,7 @@ Rules:
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from datetime import date
from typing import Any, Iterable
_FP_TO_Q = {"Q1": 1, "Q2": 2, "Q3": 3, "FY": 4}
@@ -30,12 +30,7 @@ TAPE_LEN = 4 # quarter-tape length
# Duration (flow) fields differenced from YTD into discrete quarters + summed to TTM.
_FLOW_FIELDS = (
"revenue",
"net_income",
"operating_income",
"diluted_eps",
"cfo",
"capex",
"revenue", "net_income", "operating_income", "diluted_eps", "cfo", "capex",
"depreciation_amortization",
)
@@ -49,9 +44,7 @@ class MetricPoint:
@dataclass
class MetricSeries:
value: float | None = None
history: list[MetricPoint] = field(
default_factory=list
) # oldest -> newest, <= TAPE_LEN
history: list[MetricPoint] = field(default_factory=list) # oldest -> newest, <= TAPE_LEN
period_end: date | None = None
filed_date: date | None = None
@@ -89,9 +82,7 @@ def derive(snapshots: Iterable[Any]) -> DerivedFundamentals:
result.ttm_diluted_eps = _ttm(discrete["diluted_eps"], *latest)
ttm_cfo = _ttm(discrete["cfo"], *latest)
ttm_capex = _ttm(discrete["capex"], *latest)
result.ttm_fcf = (
None if ttm_cfo is None or ttm_capex is None else ttm_cfo - ttm_capex
)
result.ttm_fcf = None if ttm_cfo is None or ttm_capex is None else ttm_cfo - ttm_capex
# tape = the CONSECUTIVE run of up to TAPE_LEN quarters ending at the latest,
# stopping at a gap — so trend text never compares non-adjacent periods.
@@ -99,9 +90,7 @@ def derive(snapshots: Iterable[Any]) -> DerivedFundamentals:
result.metrics = {
"revenue_growth_yoy": _yoy_growth_series(discrete["revenue"], selected, tape),
"eps_growth_yoy": _yoy_growth_series(discrete["diluted_eps"], selected, tape),
"operating_margin": _margin_series(
discrete["operating_income"], discrete["revenue"], selected, tape
),
"operating_margin": _margin_series(discrete["operating_income"], discrete["revenue"], selected, tape),
"fcf_margin": _fcf_margin_series(discrete, selected, tape),
"net_debt": _instant_series(selected, tape, _net_debt),
"net_debt_to_ebitda": _leverage_series(selected, discrete, tape),
@@ -113,27 +102,8 @@ def derive(snapshots: Iterable[Any]) -> DerivedFundamentals:
return result
def derive_as_of(snapshots: Iterable[Any], as_of: datetime) -> DerivedFundamentals:
"""Derive using only SEC filings accepted by the historical cutoff."""
cutoff = _utc_datetime(as_of)
visible = (
row
for row in snapshots
if (accepted := getattr(row, "accepted_at", None)) is not None
and _utc_datetime(accepted) <= cutoff
)
return derive(visible)
def _utc_datetime(value: datetime) -> datetime:
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
# -- period selection --------------------------------------------------------
def _select_latest_per_period(snapshots: Iterable[Any]) -> dict[tuple[int, str], Any]:
best: dict[tuple[int, str], Any] = {}
for row in snapshots:
@@ -156,9 +126,7 @@ def _ordered_quarters(selected: dict[tuple[int, str], Any]) -> list[tuple[int, i
return sorted((fy, _FP_TO_Q[fp]) for (fy, fp) in selected)
def _consecutive_suffix(
quarters: list[tuple[int, int]], n: int
) -> list[tuple[int, int]]:
def _consecutive_suffix(quarters: list[tuple[int, int]], n: int) -> list[tuple[int, int]]:
"""The run of up to n quarters ending at the latest, walking back only through
adjacent periods (stop at the first gap). Returned oldest -> newest."""
if not quarters:
@@ -178,10 +146,7 @@ def _consecutive_suffix(
# -- discrete + TTM ----------------------------------------------------------
def _discrete_quarters(
selected: dict[tuple[int, str], Any], field_name: str
) -> dict[tuple[int, int], float]:
def _discrete_quarters(selected: dict[tuple[int, str], Any], field_name: str) -> dict[tuple[int, int], float]:
out: dict[tuple[int, int], float] = {}
for (fy, fp), row in selected.items():
val = _discrete_value(selected, fy, fp, field_name)
@@ -224,7 +189,6 @@ def _pct_change(cur: float | None, prior: float | None) -> float | None:
# -- per-metric series (value at latest + tape history) ----------------------
def _period_end(selected, fy: int, q: int) -> date | None:
row = selected.get((fy, _Q_TO_FP[q]))
return row.period_end if row is not None else None
@@ -232,7 +196,7 @@ def _period_end(selected, fy: int, q: int) -> date | None:
def _yoy_growth_series(dq, selected, tape) -> MetricSeries:
pts = []
for fy, q in tape:
for (fy, q) in tape:
cur, prior = _ttm(dq, fy, q), _ttm(dq, fy - 1, q)
pts.append(MetricPoint(_period_end(selected, fy, q), _pct_change(cur, prior)))
return _series(pts)
@@ -240,7 +204,7 @@ def _yoy_growth_series(dq, selected, tape) -> MetricSeries:
def _margin_series(num_dq, den_dq, selected, tape) -> MetricSeries:
pts = []
for fy, q in tape:
for (fy, q) in tape:
num, den = _ttm(num_dq, fy, q), _ttm(den_dq, fy, q)
val = None if num is None or not den else num / den * 100.0
pts.append(MetricPoint(_period_end(selected, fy, q), val))
@@ -249,38 +213,24 @@ def _margin_series(num_dq, den_dq, selected, tape) -> MetricSeries:
def _fcf_margin_series(discrete, selected, tape) -> MetricSeries:
pts = []
for fy, q in tape:
cfo, capex, rev = (
_ttm(discrete["cfo"], fy, q),
_ttm(discrete["capex"], fy, q),
_ttm(discrete["revenue"], fy, q),
)
val = (
None
if cfo is None or capex is None or not rev
else (cfo - capex) / rev * 100.0
)
for (fy, q) in tape:
cfo, capex, rev = _ttm(discrete["cfo"], fy, q), _ttm(discrete["capex"], fy, q), _ttm(discrete["revenue"], fy, q)
val = None if cfo is None or capex is None or not rev else (cfo - capex) / rev * 100.0
pts.append(MetricPoint(_period_end(selected, fy, q), val))
return _series(pts)
def _instant_series(selected, tape, fn) -> MetricSeries:
pts = [
MetricPoint(_period_end(selected, fy, q), fn(selected.get((fy, _Q_TO_FP[q]))))
for (fy, q) in tape
]
pts = [MetricPoint(_period_end(selected, fy, q), fn(selected.get((fy, _Q_TO_FP[q])))) for (fy, q) in tape]
return _series(pts)
def _leverage_series(selected, discrete, tape) -> MetricSeries:
pts = []
for fy, q in tape:
for (fy, q) in tape:
row = selected.get((fy, _Q_TO_FP[q]))
nd = _net_debt(row)
op, da = (
_ttm(discrete["operating_income"], fy, q),
_ttm(discrete["depreciation_amortization"], fy, q),
)
op, da = _ttm(discrete["operating_income"], fy, q), _ttm(discrete["depreciation_amortization"], fy, q)
ebitda = None if op is None or da is None else op + da
# Null when EBITDA <= 0: a negative denominator would flip polarity and a
# "lower is better" read would rank a distressed issuer as favorable.
@@ -291,7 +241,7 @@ def _leverage_series(selected, discrete, tape) -> MetricSeries:
def _share_change_series(selected, tape) -> MetricSeries:
pts = []
for fy, q in tape:
for (fy, q) in tape:
cur = _shares(selected.get((fy, _Q_TO_FP[q])))
prior = _shares(selected.get((fy - 1, _Q_TO_FP[q])))
pts.append(MetricPoint(_period_end(selected, fy, q), _pct_change(cur, prior)))
app/services/fundamentals_research.py
-176
View File
@@ -1,176 +0,0 @@
"""Pure scoring helpers for point-in-time fundamentals research.
The runner converts a CIK-deduplicated cross-section into favorable 0..100
factor ranks and three deliberately small composites. Historical valuation is
absent: stored bars are split-adjusted, while filing-time EPS and share counts
are not guaranteed to be on today's split basis.
"""
from __future__ import annotations
import math
from collections.abc import Mapping
from typing import Any
MIN_CROSS_SECTION = 5
FACTOR_POLARITY: dict[str, bool] = {
"revenue_growth_yoy": True,
"eps_growth_yoy": True,
"operating_margin": True,
"fcf_margin": True,
"net_debt_to_ebitda": False,
"share_count_change_yoy": False,
}
QUALITY_FACTORS = (
"operating_margin",
"fcf_margin",
"net_debt_to_ebitda",
"share_count_change_yoy",
)
GROWTH_FACTORS = ("revenue_growth_yoy", "eps_growth_yoy")
SPLIT_SAFE_FACTOR_POLARITY: dict[str, bool] = {
"revenue_growth_yoy": True,
"operating_margin": True,
"fcf_margin": True,
"net_debt_to_ebitda": False,
}
SPLIT_SAFE_QUALITY_FACTORS = (
"operating_margin",
"fcf_margin",
"net_debt_to_ebitda",
)
SPLIT_SAFE_GROWTH_FACTORS = ("revenue_growth_yoy",)
COMPOSITE_KEYS = ("quality", "growth", "balanced")
def raw_features(derived: Any) -> dict[str, float | None]:
"""Extract the six research-eligible values from derived fundamentals."""
metrics = getattr(derived, "metrics", {}) or {}
return {
key: _finite_or_none(getattr(metrics.get(key), "value", None))
for key in FACTOR_POLARITY
}
def cross_section_scores(
features_by_issuer: Mapping[str, Mapping[str, Any]],
*,
min_cross_section: int = MIN_CROSS_SECTION,
split_safe: bool = False,
) -> dict[str, dict[str, float | None]]:
"""Return favorable factor ranks and composites for every issuer.
The default reproduces the original registered experiment. ``split_safe``
excludes diluted-EPS growth and share-count change because filing-time
values are not comparable across stock splits without point-in-time split
factors. Its quality score needs two of three remaining inputs and its
growth score is revenue growth. Balanced always weights the two sub-scores
equally.
"""
factor_polarity = SPLIT_SAFE_FACTOR_POLARITY if split_safe else FACTOR_POLARITY
quality_factors = SPLIT_SAFE_QUALITY_FACTORS if split_safe else QUALITY_FACTORS
growth_factors = SPLIT_SAFE_GROWTH_FACTORS if split_safe else GROWTH_FACTORS
result = {
str(issuer): {
**{key: None for key in factor_polarity},
**{key: None for key in COMPOSITE_KEYS},
}
for issuer in features_by_issuer
}
for factor, higher_is_better in factor_polarity.items():
values = {
str(issuer): _finite_or_none(features.get(factor))
for issuer, features in features_by_issuer.items()
}
ranks = favorable_percentiles(
values,
higher_is_better=higher_is_better,
min_count=min_cross_section,
)
for issuer, rank in ranks.items():
result[issuer][factor] = rank
for scores in result.values():
quality_values = _available(scores, quality_factors)
growth_values = _available(scores, growth_factors)
if len(quality_values) >= 2:
scores["quality"] = _mean(quality_values)
if growth_values:
scores["growth"] = _mean(growth_values)
if scores["quality"] is not None and scores["growth"] is not None:
scores["balanced"] = _mean(
[float(scores["quality"]), float(scores["growth"])]
)
return result
def favorable_percentiles(
values_by_issuer: Mapping[str, Any],
*,
higher_is_better: bool,
min_count: int = MIN_CROSS_SECTION,
) -> dict[str, float | None]:
"""Tie-aware favorable percentile for a deduplicated cross-section."""
valid = {
str(issuer): float(value)
for issuer, value in values_by_issuer.items()
if _finite_or_none(value) is not None
}
result: dict[str, float | None] = {str(issuer): None for issuer in values_by_issuer}
if len(valid) < min_count:
return result
for issuer, subject in valid.items():
others = [value for key, value in valid.items() if key != issuer]
if higher_is_better:
worse = sum(value < subject for value in others)
else:
worse = sum(value > subject for value in others)
tied = sum(value == subject for value in others)
result[issuer] = round(
(worse + 0.5 * tied) / len(others) * 100.0,
4,
)
return result
def overlay_rank(
strategy_rank: Any,
fundamental_score: Any,
weight: float,
*,
missing_score: float = 50.0,
) -> float | None:
"""Blend production rank with fundamentals without changing the gate."""
base = _finite_or_none(strategy_rank)
if base is None:
return None
if not 0.0 <= weight <= 1.0:
raise ValueError("weight must be between 0 and 1")
score = _finite_or_none(fundamental_score)
if score is None:
score = float(missing_score)
return round((1.0 - weight) * base + weight * score, 4)
def _available(scores: Mapping[str, Any], keys: tuple[str, ...]) -> list[float]:
return [
value for key in keys if (value := _finite_or_none(scores.get(key))) is not None
]
def _mean(values: list[float]) -> float:
return round(sum(values) / len(values), 4)
def _finite_or_none(value: Any) -> float | None:
if (
isinstance(value, (int, float))
and not isinstance(value, bool)
and math.isfinite(value)
):
return float(value)
return None