"""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, which this snapshot-only scorer doesn't have yet. Return None (NOT 0.0) so this indicator is EXCLUDED from the weighted sum and its weight is renormalised away — exactly the "a dead indicator must not drag the score toward zero" rule stated in the module docstring. Returning 0.0 would keep its 0.05 weight in the denominator and silently compress every other indicator's contribution. Wire in a trend lookup to re-activate it.""" return None 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. Return None (NOT 0.0) until we have a trend window, so this indicator is excluded + its weight renormalised away rather than diluting every other indicator. (Same reasoning as _stablecoin_supply_signal.)""" return None # 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