app.py

"""
PolyBloom ML Service — Space A v20.0
PRICE MEMORY TIER: Chronos (zero-shot T5 foundation model)

Tier role: reads closes + price-derived indicators only.
No orderflow (CVD, OB, funding). No sentiment (polymarket, fear/greed).

Endpoints:
  GET  /          health check
  POST /predict   returns single Chronos prediction
"""

import os
import math
import time
import logging
import numpy as np

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

logging.basicConfig(level=logging.INFO, format="%(levelname)s  %(name)s  %(message)s")
logger = logging.getLogger("space-a")

app = FastAPI(title="PolyBloom Space A — Price Memory", version="20.0.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# ─────────────────────────────────────────────────────────────────────────────
# Global state
# ─────────────────────────────────────────────────────────────────────────────

_model = None
_chronos_loaded = False

# ─────────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────────

def _tanh(x: float) -> float:
    return math.tanh(x)

def _clamp(v: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, v))

def _slope(arr: np.ndarray, p: int) -> float:
    if len(arr) < p + 1:
        return 0.0
    b = arr[-p - 1]
    return float((arr[-1] - b) / b) if b != 0 else 0.0

def _ema(arr: np.ndarray, p: int) -> float:
    k = 2.0 / (p + 1)
    e = float(arr[0])
    for v in arr[1:]:
        e = float(v) * k + e * (1.0 - k)
    return e

def _score_to_prediction(score: float, reasoning: str) -> dict:
    direction  = "UP" if score > 0.05 else ("DOWN" if score < -0.05 else "NEUTRAL")
    confidence = _clamp(45.0 + abs(score) * 55.0, 40.0, 95.0)
    return {
        "name":       "Chronos",
        "direction":  direction,
        "confidence": round(confidence),
        "slope_pct":  round(score * 0.5, 6),
        "reasoning":  reasoning,
        "tier":       "price_memory",
    }

# ─────────────────────────────────────────────────────────────────────────────
# Startup
# ─────────────────────────────────────────────────────────────────────────────

@app.on_event("startup")
async def startup_event():
    global _model, _chronos_loaded
    try:
        from chronos import ChronosPipeline
        import torch
        model_name = os.environ.get("CHRONOS_MODEL", "amazon/chronos-t5-tiny")
        logger.info("Chronos: loading %s …", model_name)
        _model = ChronosPipeline.from_pretrained(
            model_name,
            device_map="cpu",
            torch_dtype=torch.float32,
        )
        _chronos_loaded = True
        logger.info("Chronos loaded ✓")
    except Exception as exc:
        logger.warning("Chronos not loaded: %s", exc)

# ─────────────────────────────────────────────────────────────────────────────
# Inference
# ─────────────────────────────────────────────────────────────────────────────

def _predict_real(closes: list, feats: dict) -> dict:
    import torch
    c       = np.array(closes[-64:], dtype=np.float32)
    context = torch.tensor(c).unsqueeze(0)
    with torch.no_grad():
        forecast = _model.predict(context, prediction_length=2)
    median = float(forecast[0].median(dim=0).values.mean().item())
    last   = float(c[-1])
    score  = _tanh((median / last - 1.0) * 20.0) if last != 0 else 0.0
    return _score_to_prediction(score, f"Chronos: median={median:.2f} last={last:.2f}")


def _predict_bootstrap(closes: list, feats: dict) -> dict:
    c    = np.array(closes, dtype=np.float64)
    n    = len(c)
    last = float(c[-1]) if n > 0 else 1.0
    e20  = _ema(c[-max(40, n):], 20)
    e50  = _ema(c[-max(80, n):], min(50, max(n // 2, 5)))
    sl20 = _slope(c, min(20, n - 1))
    rsi  = float(feats.get("rsi") or 50.0)
    score = (
        math.copysign(min(1.0, abs(sl20) * 150.0), sl20) * 0.45 +
        math.copysign(min(1.0, abs((e20 - e50) / last) * 200.0), (e20 - e50)) * 0.35 +
        ((rsi - 50.0) / 50.0) * 0.20
    )
    return _score_to_prediction(
        score,
        f"Chronos bootstrap: sl20={sl20*100:.4f}% ema_diff={((e20-e50)/last)*100:.4f}%",
    )

# ─────────────────────────────────────────────────────────────────────────────
# Endpoints
# ─────────────────────────────────────────────────────────────────────────────

@app.get("/")
async def health():
    return {
        "ok":      True,
        "version": "20.0.0",
        "space":   "A",
        "tier":    "price_memory",
        "models":  ["Chronos"],
        "chronos": _chronos_loaded,
    }


@app.post("/predict")
async def predict(request: Request):
    t0   = time.perf_counter()
    body = await request.json()

    closes = list(body.get("closes") or [])
    if len(closes) < 16:
        return JSONResponse(status_code=422, content={"error": "closes too short — need 16+"})

    # Price-memory tier: only price-derived features allowed
    allowed = {"rsi", "stoch_rsi_k", "stoch_rsi_d", "vwap_dev", "atr_pct_of_spot",
               "atr", "bollinger_squeeze", "vol_regime", "vol_percentile",
               "garman_klass_vol", "market_regime", "htf_trend_15m", "htf_trend_1h",
               "candle_pattern", "nearest_support", "nearest_resistance", "session"}
    feats = {k: v for k, v in body.items() if k != "closes" and k in allowed}

    if _chronos_loaded and _model is not None:
        try:
            pred = _predict_real(closes, feats)
        except Exception as exc:
            logger.warning("Chronos real inference failed: %s", exc)
            pred = _predict_bootstrap(closes, feats)
    else:
        pred = _predict_bootstrap(closes, feats)

    pred["latency_ms"] = round((time.perf_counter() - t0) * 1000)
    return pred