src/score.py

"""Decoupled candidate score — the SCREEN, not the impact estimator.

Per docs/METHODOLOGY.md §3 / docs/STRATEGY.md §4.7:

    Score = 0.30·LiquidityRise
          + 0.20·InstitutionalProxy
          + 0.15·IndexInclusion
          + 0.15·BrokerNamed
          + 0.20·FilingPresent

The single biggest mistake to avoid (CLAUDE.md):
    *Returns* — total_return, momentum, drawdown — are NOT inputs. They
    appear only on the right-hand side of the impact regression elsewhere.
    A test asserts this: holding the *flow* signals fixed, the score must
    not move when we vary the cumulative return outcome.
"""

from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd

from src.constants import SCORE_WEIGHTS


@dataclass(frozen=True, slots=True)
class TickerFeatures:
    """All five signals for one ticker, each in [0, 1] before weighting."""

    ticker: str
    liquidity_rise: float
    institutional_proxy: float
    index_inclusion: float  # 0 or 1
    broker_named: float
    filing_present: float  # 0 or 1


@dataclass(frozen=True, slots=True)
class ScoreBreakdown:
    """Per-signal raw value + weighted total. Total ∈ [0, 1] given weights sum to 1.

    Per-signal fields hold the *raw* signal value in [0, 1] (matching the
    candidate_scores DB schema), not the post-weight contribution. ``total``
    is the weighted sum.
    """

    ticker: str
    liquidity_rise: float
    institutional_proxy: float
    index_inclusion: float
    broker_named: float
    filing_present: float
    total: float

    def as_db_row(self) -> dict[str, float | int | str]:
        return {
            "ticker": self.ticker,
            "liquidity_rise": float(self.liquidity_rise),
            "institutional_proxy": float(self.institutional_proxy),
            "index_inclusion": int(round(self.index_inclusion)),
            "broker_named": float(self.broker_named),
            "filing_present": int(round(self.filing_present)),
            "total_score": float(self.total),
        }


# ---------------------------------------------------------------------------
# Score function
# ---------------------------------------------------------------------------


def candidate_score(
    feat: TickerFeatures, weights: dict[str, float] = SCORE_WEIGHTS
) -> ScoreBreakdown:
    """Pure function: weight-and-sum the five signals.

    Each signal must be pre-normalised to [0, 1]. The function does not look
    at price, return, or volume directly — only at the upstream feature row.
    """
    if not _approximately_one(sum(weights.values())):
        raise ValueError(f"score weights must sum to 1.0, got {sum(weights.values())}")

    total = float(
        weights["liquidity_rise"] * feat.liquidity_rise
        + weights["institutional_proxy"] * feat.institutional_proxy
        + weights["index_inclusion"] * feat.index_inclusion
        + weights["broker_named"] * feat.broker_named
        + weights["filing_present"] * feat.filing_present
    )
    return ScoreBreakdown(
        ticker=feat.ticker,
        liquidity_rise=feat.liquidity_rise,
        institutional_proxy=feat.institutional_proxy,
        index_inclusion=feat.index_inclusion,
        broker_named=feat.broker_named,
        filing_present=feat.filing_present,
        total=total,
    )


def _approximately_one(x: float, tol: float = 1e-9) -> bool:
    return abs(x - 1.0) < tol


# ---------------------------------------------------------------------------
# Signal computation helpers
# ---------------------------------------------------------------------------


def rank_normalise(values: pd.Series) -> pd.Series:
    """Map a Series to [0, 1] by rank percentile.

    NaN values stay NaN. Ties get the average rank (pandas default).
    A higher original value → higher rank → score closer to 1.
    """
    if values.empty:
        return values
    ranks = values.rank(method="average", na_option="keep")
    n = ranks.notna().sum()
    if n <= 1:
        return ranks.where(ranks.isna(), 0.5)
    return (ranks - 1.0) / (n - 1.0)


def liquidity_rise_from_amihud(
    amihud_60d: pd.Series, *, pre_end: pd.Timestamp, post_start: pd.Timestamp
) -> float:
    """Mean(amihud_60d) over the pre window minus mean over the post window.

    Higher Amihud = less liquid. So pre - post > 0 means liquidity improved.
    Returns the raw delta (caller should rank-normalise across universe).
    NaN if either window is empty.
    """
    if amihud_60d.empty:
        return float("nan")
    pre = amihud_60d.loc[
        (amihud_60d.index >= pre_end - pd.Timedelta(days=90))
        & (amihud_60d.index <= pre_end)
    ]
    post = amihud_60d.loc[
        (amihud_60d.index >= post_start)
        & (amihud_60d.index <= post_start + pd.Timedelta(days=90))
    ]
    pre = pre.dropna()
    post = post.dropna()
    if pre.empty or post.empty:
        return float("nan")
    return float(pre.mean() - post.mean())


def institutional_proxy_from_prices(
    prices: pd.DataFrame, *, lookback: int = 30
) -> float:
    """Fraction of the last ``lookback`` trading days where close exceeds
    the same-day rolling 30-day VWAP.

    VWAP_t = Σ(typical_price_i · volume_i) / Σ(volume_i)  for i in [t-30, t]
    typical_price = (H + L + C) / 3.

    Returns a value in [0, 1]; higher = more sustained close-above-VWAP =
    stronger sustained-accumulation signal. NaN if too little data.
    """
    needed = {"high", "low", "close", "volume"}
    if not needed.issubset(prices.columns):
        raise KeyError(f"prices missing columns: {sorted(needed - set(prices.columns))}")
    if len(prices) < lookback + 1:
        return float("nan")

    p = prices.sort_index().copy()
    p["typical"] = (p["high"] + p["low"] + p["close"]) / 3.0
    p["pv"] = p["typical"] * p["volume"]
    rolling_pv = p["pv"].rolling(window=lookback, min_periods=lookback).sum()
    rolling_v = p["volume"].rolling(window=lookback, min_periods=lookback).sum()
    safe_v = rolling_v.where(rolling_v > 0, np.nan)
    vwap = rolling_pv / safe_v

    last = p.tail(lookback)
    last_vwap = vwap.tail(lookback)
    above = (last["close"] > last_vwap).astype(float)
    above = above.where(last_vwap.notna(), np.nan)
    valid = above.dropna()
    if valid.empty:
        return float("nan")
    return float(valid.mean())


def broker_named_normalise(broker_count: int, max_count: int) -> float:
    """Linear normalise broker_named_count to [0, 1] given the universe max."""
    if max_count <= 0:
        return 0.0
    return float(broker_count) / float(max_count)


def filing_present_within(
    filings: pd.Series, *, as_of: pd.Timestamp, months: int = 12
) -> int:
    """1 if any filing date in (as_of - months, as_of], else 0.

    ``filings`` is a Series of pd.Timestamp (filing dates) for one ticker.
    """
    if filings.empty:
        return 0
    cutoff = as_of - pd.DateOffset(months=months)
    return int(((filings > cutoff) & (filings <= as_of)).any())