trumpsignal-backend/app/services/macro/scoring.py

"""Composite "regime" score over the macro indicators.

Goal: one -100..+100 number that says whether the overall macro setup tilts
risk-on (positive) or risk-off (negative). Used purely as advisory display on
the BTC page — does NOT (yet) feed sizing or trade decisions.

Design choices:
  * Each indicator contributes a per-indicator signal in [-1, +1].
  * Weights sum to 1.0 across the indicators we got. Missing indicators are
    excluded and the remaining weights are renormalised — so a single dead
    API doesn't drag the whole score toward zero.
  * The contrarian indicators (fear/greed, AHR999, funding) are intentionally
    inverted: extreme fear / cheap AHR999 / extreme funding all add positive
    score (buy-the-fear logic).

The weights here are seeded by hand from rough conviction. Re-tune after a
few weeks of live data — see `composite_score` history vs realised forward
returns.
"""
from __future__ import annotations

from typing import Optional

# (weight, signal_function) per indicator. signal_function takes the raw value
# and returns a number in [-1, +1].
#
# Inputs may be None if the upstream fetch failed; the orchestrator filters
# them out and renormalises before summing.


def _ahr999_signal(v: Optional[float]) -> Optional[float]:
    """< 0.45 cheap → +1; 0.45–1.2 neutral; > 1.2 expensive → -1."""
    if v is None: return None
    if v < 0.45: return 1.0
    if v > 1.2:  return -1.0
    # Linear interpolation between 0.45 and 1.2 → +1 to -1.
    return 1.0 - 2 * ((v - 0.45) / (1.2 - 0.45))


def _altseason_signal(v: Optional[float]) -> Optional[float]:
    """High altseason index = risk-on (+1), low = BTC season (defensive but
    not bearish, so a mild positive)."""
    if v is None: return None
    if v >= 75:  return 0.7    # altseason — generally bullish risk
    if v <= 25:  return 0.3    # BTC-only — defensive but not a bear signal
    return (v - 50) / 50       # linear, -0.5 to +0.5


def _fear_greed_signal(v: Optional[int]) -> Optional[float]:
    """Contrarian: extreme fear → buy (+1), extreme greed → sell (-1)."""
    if v is None: return None
    # Map 0..100 → +1..-1 (inverted, contrarian).
    return (50 - v) / 50


def _btc_dominance_signal(v: Optional[float]) -> Optional[float]:
    """Hard to read in isolation — only inform on extremes.
    Very high dominance often signals fear (risk-off into BTC) → mild bearish.
    Very low = altseason froth → also mild bearish (cycle late). Mid = neutral."""
    if v is None: return None
    if v >= 60:  return -0.3
    if v <= 40:  return -0.3
    return 0.0


def _eth_btc_signal(v: Optional[float]) -> Optional[float]:
    """Rising ETH/BTC = risk-on. No persistent absolute level matters; this is
    really a trend indicator. We approximate with absolute thresholds for the
    current cycle (2025-2026): > 0.04 risk-on, < 0.025 risk-off."""
    if v is None: return None
    if v >= 0.04:  return 0.7
    if v <= 0.025: return -0.7
    return (v - 0.0325) / 0.0075 * 0.7    # linear in the middle


def _stablecoin_supply_signal(v: Optional[float]) -> Optional[float]:
    """Absolute supply tells us little day-over-day; we need the delta. Since
    this scorer sees only the snapshot, we treat presence as 0 and let the
    visual chart show the trend. Returns 0 if we have any value at all."""
    if v is None: return None
    return 0.0    # contribution = 0 until we wire in a trend lookup


def _etf_flow_signal(v: Optional[float]) -> Optional[float]:
    """Net inflow = institutional bid → +1, outflow → -1. Scale by magnitude."""
    if v is None: return None
    # Daily prints over $200M are notable; over $500M unusually large.
    abs_v = abs(v)
    sign = 1 if v > 0 else (-1 if v < 0 else 0)
    if abs_v >= 500_000_000:  return 1.0 * sign
    if abs_v >= 200_000_000:  return 0.7 * sign
    if abs_v >= 50_000_000:   return 0.4 * sign
    return 0.1 * sign


def _open_interest_signal(v: Optional[float]) -> Optional[float]:
    """OI in isolation doesn't tell us direction — we'd need OI vs price
    correlation. Until we have a trend window, contribute 0."""
    if v is None: return None
    return 0.0


# Weights (sum to 1.0 across all). When an indicator is missing, we drop its
# weight and renormalise the rest.
WEIGHTS = {
    "ahr999":              0.20,
    "altcoin_season":      0.10,
    "fear_greed":          0.20,
    "btc_dominance":       0.05,
    "eth_btc":             0.15,
    "stablecoin_supply":   0.05,
    "etf_flow":            0.20,
    "btc_open_interest":   0.05,
}


def compute_composite(values: dict) -> tuple[Optional[float], Optional[str]]:
    """Return (score in [-100, +100], regime_label) or (None, None) if there
    isn't enough data to score.

    `values` keys must match WEIGHTS keys (without "_signal" suffix).
    """
    pairs = [
        ("ahr999",            _ahr999_signal(values.get("ahr999"))),
        ("altcoin_season",    _altseason_signal(values.get("altcoin_season_index"))),
        ("fear_greed",        _fear_greed_signal(values.get("fear_greed"))),
        ("btc_dominance",     _btc_dominance_signal(values.get("btc_dominance_pct"))),
        ("eth_btc",           _eth_btc_signal(values.get("eth_btc_ratio"))),
        ("stablecoin_supply", _stablecoin_supply_signal(values.get("stablecoin_supply_usd"))),
        ("etf_flow",          _etf_flow_signal(values.get("etf_flow_net_usd_1d"))),
        ("btc_open_interest", _open_interest_signal(values.get("btc_open_interest_usd"))),
    ]
    alive = [(k, s) for k, s in pairs if s is not None]
    if not alive:
        return None, None

    total_w = sum(WEIGHTS[k] for k, _ in alive)
    if total_w <= 0:
        return None, None

    score = sum(WEIGHTS[k] * s for k, s in alive) / total_w * 100

    if score >=  60: label = "BULL"
    elif score >= 20: label = "BULLISH"
    elif score > -20: label = "NEUTRAL"
    elif score > -60: label = "BEARISH"
    else:             label = "BEAR"

    return round(score, 1), label