trumpsignal-backend/app/services/bottom_indicators.py

"""Pure-price BTC bottom indicators (no on-chain data, no API keys).

Three independent, well-known bottom signals. Each is a simple, transparent
function of price history only — no Realized Cap, no MVRV, no paid feeds:

  A. AHR999       — value/DCA index. < 0.45 = deep-value bottom zone.
  B. 200-Week MA  — BTC has bottomed near its 200WMA every cycle.
  C. Pi Cycle Bot — 150d EMA vs 471d SMA × 0.745 crossover region.

The combiner fires only when at least 2 of the 3 agree ("三者为二"), which
historically pins genuine macro bottoms while rejecting single-indicator
noise. Long-only, low-frequency, stateless.
"""

from __future__ import annotations

import math
from dataclasses import dataclass
from datetime import date, datetime, timezone

# BTC genesis block: 2009-01-03.
_BTC_GENESIS = date(2009, 1, 3)

# AHR999 thresholds.
AHR999_BOTTOM = 0.45      # < this → deep-value / bottom zone (signal A)
AHR999_INVALIDATION = 1.2  # > this → no longer cheap, thesis dead

# 200WMA tolerance: price within +5% of the 200-week mean still counts.
WMA200_TOLERANCE = 1.05

# Pi Cycle Bottom multiplier (the canonical 0.745).
PI_CYCLE_MULT = 0.745
PI_CYCLE_EMA = 150
PI_CYCLE_SMA = 471


def _closes(candles: list[dict]) -> list[float]:
    return [float(c["close"]) for c in candles if c.get("close")]


def _sma(values: list[float], n: int) -> float | None:
    if len(values) < n:
        return None
    return sum(values[-n:]) / n


def _ema(values: list[float], n: int) -> float | None:
    if len(values) < n:
        return None
    k = 2.0 / (n + 1)
    # Seed with the SMA of the first n values, then walk forward.
    ema = sum(values[:n]) / n
    for v in values[n:]:
        ema = v * k + ema * (1 - k)
    return ema


def coin_age_days(today: date | None = None) -> int:
    d = today or datetime.now(timezone.utc).date()
    return max(1, (d - _BTC_GENESIS).days)


def ahr999(daily_closes: list[float], today: date | None = None) -> float | None:
    """AHR999 = (price / GM200) × (price / growth_val).

      GM200       = geometric mean of the last 200 daily closes
      growth_val  = 10 ** (5.84 * log10(coin_age_days) - 17.01)

    < 0.45 deep value · 0.45–1.2 DCA · > 1.2 expensive.
    """
    if len(daily_closes) < 200:
        return None
    price = daily_closes[-1]
    if price <= 0:
        return None
    window = daily_closes[-200:]
    # Geometric mean via log-space (avoids overflow).
    log_sum = sum(math.log(c) for c in window if c > 0)
    gm200 = math.exp(log_sum / 200)
    age = coin_age_days(today)
    growth_val = 10 ** (5.84 * math.log10(age) - 17.01)
    if gm200 <= 0 or growth_val <= 0:
        return None
    return (price / gm200) * (price / growth_val)


def below_200wma(
    weekly_closes: list[float],
    price: float | None = None,
) -> tuple[bool, float | None]:
    """True if price ≤ 200-week mean × 1.05. Returns (signal, wma200).

    `price` is the reference price to compare. Pass the latest DAILY close so
    all three indicators use the same "current price" — otherwise the last
    weekly close can be up to 7 days stale and B would disagree with A/C right
    at a fast-moving bottom. Falls back to the last weekly close if omitted.
    """
    wma = _sma(weekly_closes, 200)
    if wma is None or not weekly_closes:
        return False, wma
    px = price if price is not None else weekly_closes[-1]
    return px <= wma * WMA200_TOLERANCE, wma


def pi_cycle_bottom(daily_closes: list[float]) -> tuple[bool, float | None, float | None]:
    """Pi Cycle Bottom: 150d EMA ≤ 471d SMA × 0.745.

    Returns (signal, ema150, sma471_scaled).
    """
    ema150 = _ema(daily_closes, PI_CYCLE_EMA)
    sma471 = _sma(daily_closes, PI_CYCLE_SMA)
    if ema150 is None or sma471 is None:
        return False, ema150, None
    scaled = sma471 * PI_CYCLE_MULT
    return ema150 <= scaled, ema150, scaled


def trend_confirmed(
    daily_closes: list[float],
    daily_highs: list[float],
    price: float,
    sma_n: int = 200,
    high_lookback: int = 20,
    high_tol: float = 0.005,
) -> bool:
    """Structural confirmation that the bottom reversal has turned into an
    actual uptrend — gate for pyramiding (add-to-winner):

      price ≥ 200-day SMA            (trend regime reclaimed)
      AND price ≥ recent 20d high·(1−0.5%)   (at/near a fresh local high)

    Pure function (no I/O) so it's unit-testable; the caller fetches candles.
    Conservative: both conditions must hold, evaluated at add-on time.
    """
    sma = _sma(daily_closes, sma_n)
    if sma is None or not daily_highs:
        return False
    recent_high = max(daily_highs[-high_lookback:]) if len(daily_highs) >= 1 else None
    if recent_high is None or recent_high <= 0:
        return False
    return price >= sma and price >= recent_high * (1.0 - high_tol)


@dataclass(frozen=True)
class BottomConfluence:
    fired: bool                 # ≥ 2 of 3 true
    votes: int                  # 0..3
    ahr999: float | None
    a_value: bool               # AHR999 < 0.45
    b_200wma: bool              # price ≤ 200WMA × 1.05
    c_pi_cycle: bool            # 150 EMA ≤ 471 SMA × 0.745
    detail: dict


def bottom_confluence(
    daily_closes: list[float],
    weekly_closes: list[float],
    today: date | None = None,
) -> BottomConfluence:
    """2-of-3 bottom confluence. Pure price, stateless."""
    ah = ahr999(daily_closes, today)
    a = ah is not None and ah < AHR999_BOTTOM

    # All three indicators compare against the SAME current price (latest
    # daily close), so a stale weekly close can't make B disagree with A/C.
    cur_price = daily_closes[-1] if daily_closes else None
    b, wma200 = below_200wma(weekly_closes, cur_price)
    c, ema150, sma471s = pi_cycle_bottom(daily_closes)

    votes = int(a) + int(b) + int(c)
    detail = {
        "ahr999": round(ah, 4) if ah is not None else None,
        "ahr999_threshold": AHR999_BOTTOM,
        "price": round(daily_closes[-1], 2) if daily_closes else None,
        "wma200": round(wma200, 2) if wma200 is not None else None,
        "wma200_band": round(wma200 * WMA200_TOLERANCE, 2) if wma200 is not None else None,
        "pi_ema150": round(ema150, 2) if ema150 is not None else None,
        "pi_sma471_scaled": round(sma471s, 2) if sma471s is not None else None,
        "votes": votes,
        "signals": {"ahr999_value": a, "below_200wma": b, "pi_cycle_bottom": c},
    }
    return BottomConfluence(
        fired=votes >= 2,
        votes=votes,
        ahr999=ah,
        a_value=a,
        b_200wma=b,
        c_pi_cycle=c,
        detail=detail,
    )