signal-platform/app/services/fundamentals_research.py

"""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")
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,
) -> dict[str, dict[str, float | None]]:
    """Return favorable factor ranks and composites for every issuer.

    Quality needs two of four inputs; growth needs one of two. Balanced requires
    both sub-scores and weights them equally, so quality's four inputs do not
    mechanically dominate growth's two inputs.
    """
    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