Deploy forecast-service to Hugging Face Space

.dockerignore

@@ -0,0 +1,9 @@
+__pycache__/
+*.pyc
+.venv/
+venv/
+.hf_cache/
+db.sqlite
+*.log
+.git/
+.env

.gitignore

@@ -0,0 +1,6 @@
+# forecast-service local artifacts
+__pycache__/
+*.pyc
+db.sqlite
+.venv/
+venv/

DEPLOY.md

@@ -0,0 +1,51 @@
+# Deploying the forecast-service
+
+The Next.js app calls this service via `FORECAST_URL`. It can't run on Vercel (it loads ~2GB of
+torch models), so host the container somewhere always-on and point `FORECAST_URL` at it.
+
+The repo has a `Dockerfile` that builds + runs it (binds `0.0.0.0`, honours the platform `$PORT`,
+falls back to `8008`). First boot downloads model weights (~2GB) — slow once, then cached.
+
+## Option A — Hugging Face Spaces (free CPU, recommended)
+1. Create a new **Space** → SDK: **Docker** → blank.
+2. Push the contents of `forecast-service/` to the Space repo (the `Dockerfile` is at its root).
+3. In the Space **README.md** frontmatter, set the port:
+   ```yaml
+   ---
+   title: FutureQuery forecast-service
+   sdk: docker
+   app_port: 8008
+   ---
+   ```
+4. The Space builds + boots (first boot is slow while it downloads models). Its URL is
+   `https://<user>-<space>.hf.space`.
+5. (Optional) add `OPENROUTER_API_KEY` as a Space **Secret** if you want the LangGraph synthesis.
+
+## Option B — Render / Railway
+1. New **Web Service** → connect this repo → root directory `forecast-service` → runtime **Docker**.
+2. They inject `$PORT` automatically (the Dockerfile uses it). No port config needed.
+3. Render free tier **sleeps when idle** → the first request after idle is slow (cold model load).
+   Bump to a paid instance (~$7/mo) to keep it warm.
+
+## Option C — Fly.io / a small VM
+- `fly launch` in `forecast-service/` (it detects the Dockerfile), or run the image on any VM with
+  ≥4GB RAM: `docker build -t forecast . && docker run -p 8008:8008 forecast`.
+
+## Wire it to the app (all hosts)
+1. Verify it's up: `curl https://<your-host>/health` → shows which models loaded.
+2. In **Vercel → futurequery-vercel → Settings → Environment Variables**, add
+   `FORECAST_URL=https://<your-host>` for **Production + Preview + Development**, then **redeploy**.
+3. The `/pipeline` page becomes usable. In `/api/predict`, the **Momentum (TimesFM 2.5 /
+   Chronos-2)** row can turn green only when an exact/likely Polymarket CLOB market has enough
+   price history and the service returns `use_forecast: true`; it remains a supporting signal and
+   never moves the headline probability.
+4. (Optional) for the separate `edge-finder/` PaperBrain crons, wire `FORECAST_URL` into the
+   workflow `env` before relying on cron momentum; the bundled workflows currently do not pass it,
+   so adding a secret alone does not activate that signal.
+
+## Notes
+- CPU inference is fine for demo volume (a few seconds/call); a GPU host (Modal/Replicate) is only
+  needed for heavy traffic.
+- The service degrades gracefully: if a model fails to load it still starts and returns a naive
+  forecast, so a misconfigured host won't hard-fail the web app (Momentum stays empty/warn with
+  clear copy).

Dockerfile

@@ -0,0 +1,29 @@
+# forecast-service — TimesFM 2.x + Chronos-2 ensemble API.
+# Builds a container any host can run (Hugging Face Spaces / Render / Railway / Fly / a VM),
+# so the Vercel app can reach it via FORECAST_URL.
+#
+# First boot downloads ~2GB of model weights from HuggingFace into HF_HOME (expected, once).
+# Binds 0.0.0.0 (FORECAST_HOST) so the host can route to it; honours the host's $PORT
+# (Render/Railway/Fly inject one) and otherwise serves on 8008.
+FROM python:3.12-slim
+
+# build-essential for any source-built wheels; libgomp1 for torch; git in case a dep needs it.
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        build-essential git libgomp1 \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Install deps first so Docker caches this layer across code changes.
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+ENV FORECAST_HOST=0.0.0.0 \
+    HF_HOME=/app/.hf_cache \
+    PYTHONUNBUFFERED=1
+
+EXPOSE 8008
+# ${PORT:-8008}: use the platform-injected port if present, else 8008.
+CMD ["sh", "-c", "FORECAST_PORT=${PORT:-8008} python main.py"]

README.md

@@ -1,10 +1,129 @@
 ---
-title: Futurequery Forecast Service
-emoji: 🌖
-colorFrom: gray
-colorTo: red
+title: FutureQuery forecast-service
 sdk: docker
-pinned: false
+app_port: 8008
 ---
+# FutureQuery `forecast-service`
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+An ensemble time-series forecasting service for prediction-market probabilities.
+It blends **Google TimesFM 2.x** and **Amazon Chronos-2**, applies
+superforecasting-style calibration, tracks its own Brier score over time, and
+exposes a small FastAPI surface plus a LangGraph research pipeline.
+
+> Replaces the old single-model `timesfm-service/`.
+
+## Quick start
+
+```bash
+cd forecast-service
+pip install -r requirements.txt
+python main.py        # starts on http://localhost:8008
+```
+
+The first run downloads model weights from HuggingFace (**~2GB** — this is
+normal and only happens once). If a model fails to load (offline, no GPU, OOM)
+the service **still starts** and falls back to a naive forecast, so the API
+never hard-fails.
+
+## API
+
+### `POST /forecast`
+
+```jsonc
+// request
+{
+  "prices": [0.48, 0.50, 0.51, 0.52, ...],  // historical yes_prices
+  "covariates": [[...], ...],                // optional related series
+  "question_type": "event",                  // "event" | "numeric"
+  "horizon": 5,
+  "question": "Will X happen by 2027?"       // optional, stored for calibration
+}
+```
+
+```jsonc
+// response
+{
+  "timesfm":  { "forecast": [...], "trend": "up" },
+  "chronos2": { "forecast": [...], "trend": "up" },
+  "ensemble": { "forecast": [...], "trend": "up", "brier_estimate": 0.21 },
+  "use_forecast": true,        // false if event w/ no price history, or series < 10 pts
+  "warning": null,             // e.g. "Series too short for reliable forecast"
+  "id": "…"                    // calibration row id
+}
+```
+
+**Ensemble logic** (`ensemble.py`)
+
+| Step | Rule |
+|------|------|
+| Length guard | `< 10` points → `use_forecast=false` + warning |
+| Event blend | `0.5·TimesFM + 0.5·Chronos-2` (equal weight — both models show negative R² on binary event markets, so neither dominates) |
+| Numeric blend | `0.4·TimesFM + 0.6·Chronos-2` (Chronos-2 wins on fev-bench) |
+| Base-rate prior | if `|last − 0.5| ≤ 0.1`, pull ensemble **15% toward 0.5** |
+| Extremise | only if confidence `> 0.70`: `0.5 + 1.3·(p − 0.5)` |
+
+### `GET /calibration`
+
+Returns the realised **Brier score**, a **reliability curve** (decile bins of
+predicted vs. observed frequency), and **directional accuracy** over all
+resolved forecasts — so you can benchmark against superforecasters (~0.15
+Brier) over time. Data is stored in `db.sqlite`.
+
+### `POST /resolve`
+
+```jsonc
+{ "id": "<forecast id>", "outcome": 1.0 }   // 1.0 = YES happened, 0.0 = NO
+```
+
+Records a market's actual outcome so it feeds the calibration scoring.
+
+### `GET /health`
+
+Liveness + which models actually loaded.
+
+## LangGraph pipeline (Phase 3)
+
+`pipeline.py` runs the full **SCAN → TRIAGE → SYNTHESISE** loop across
+Polymarket, Kalshi, Metaculus, and Manifold, calls `/forecast`, and pauses at a
+human checkpoint before writing the result to `db.sqlite`.
+
+```bash
+python pipeline.py "Will UK interest rates fall below 4% by Jan 2027?"
+```
+
+At the checkpoint, press **Enter** to accept or type **`override 0.42`** to set
+your own probability.
+
+## Files
+
+| File | Purpose |
+|------|---------|
+| `main.py` | FastAPI app, startup model loading, endpoints |
+| `models.py` | TimesFM / Chronos-2 wrappers with graceful fallback |
+| `ensemble.py` | Blending, base-rate prior, extremisation (numpy-only, testable) |
+| `calibration.py` | SQLite store + Brier / reliability / directional scoring |
+| `pipeline.py` | LangGraph SCAN→TRIAGE→SYNTHESISE workflow |
+| `requirements.txt` | Python dependencies |
+
+## Notes
+
+* `FORECAST_PORT` overrides the default port (`8008`).
+* The Next.js app calls `/forecast` only when `FORECAST_URL` is set and an exact/likely
+  Polymarket market with a CLOB token was selected.
+* Returned ensembles are supporting Momentum signals only; they do not blend into the headline
+  probability, and the app degrades gracefully when the service is offline.
+
+## Browser pipeline (no terminal)
+
+The same SCAN → TRIAGE → ENSEMBLE → SYNTHESISE → CHECKPOINT loop is exposed over
+HTTP so it can be driven entirely from the web app:
+
+- `POST /pipeline/start` — `{ "question": "...", "question_type": "event" }` runs
+  to the human checkpoint and returns the full state plus a `thread_id`.
+- `POST /pipeline/resume` — `{ "thread_id": "...", "resume": "" }` to accept, or
+  `{ "resume": "override 0.42" }` to override.
+
+In the Next.js app open **`/pipeline`** (e.g. http://localhost:3000/pipeline):
+type a question, watch the scan / triage / arbitrage / ensemble panels populate,
+then **Approve** or **Override** the synthesised probability. Requires
+`FORECAST_URL` to be set so the app can reach this service.

calibration.py

@@ -0,0 +1,200 @@
+"""Calibration store + scoring for the FutureQuery forecaster.
+
+Every ``/forecast`` call is logged to a local SQLite database
+(``forecast-service/db.sqlite``). When a market resolves, call
+``record_resolution`` (or POST ``/resolve``) with the binary outcome. The
+``/calibration`` endpoint then reports, over all resolved forecasts:
+
+* **Brier score**          — mean squared error of the probabilities.
+* **reliability curve**    — predicted vs. observed frequency per decile bin
+  (the data for a calibration / reliability diagram).
+* **directional accuracy** — how often ``p >= 0.5`` agreed with the outcome.
+
+This is what lets us benchmark the ensemble against superforecaster Brier
+scores over time. Only ``numpy``/``sqlite3`` are required; ``properscoring``
+and ``scikit-learn`` are used when present but are not hard dependencies.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import time
+import uuid
+from typing import Dict, List, Optional
+
+import numpy as np
+
+DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "db.sqlite")
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS forecasts (
+    id            TEXT PRIMARY KEY,
+    created_at    REAL NOT NULL,
+    question      TEXT,
+    question_type TEXT,
+    horizon       INTEGER,
+    p_forecast    REAL,          -- headline ensemble probability
+    p_market      REAL,          -- market price at forecast time (prices[-1])
+    trend         TEXT,
+    brier_estimate REAL,
+    use_forecast  INTEGER,
+    resolved      INTEGER DEFAULT 0,
+    outcome       REAL,          -- 1.0 = YES happened, 0.0 = NO
+    resolved_at   REAL
+);
+"""
+
+
+def _connect(db_path: Optional[str] = None) -> sqlite3.Connection:
+    conn = sqlite3.connect(db_path or DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def init_db(db_path: Optional[str] = None) -> None:
+    with _connect(db_path) as conn:
+        conn.executescript(_SCHEMA)
+        conn.commit()
+
+
+def record_forecast(
+    *,
+    question: str,
+    question_type: str,
+    horizon: int,
+    p_forecast: Optional[float],
+    p_market: Optional[float],
+    trend: Optional[str],
+    brier_estimate: Optional[float],
+    use_forecast: bool,
+    db_path: Optional[str] = None,
+) -> str:
+    """Persist a forecast (pending resolution). Returns the new row id."""
+    init_db(db_path)
+    fid = uuid.uuid4().hex
+    with _connect(db_path) as conn:
+        conn.execute(
+            """
+            INSERT INTO forecasts (id, created_at, question, question_type,
+                horizon, p_forecast, p_market, trend, brier_estimate,
+                use_forecast, resolved)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0)
+            """,
+            (
+                fid,
+                time.time(),
+                question,
+                question_type,
+                int(horizon),
+                _nullable_float(p_forecast),
+                _nullable_float(p_market),
+                trend,
+                _nullable_float(brier_estimate),
+                1 if use_forecast else 0,
+            ),
+        )
+        conn.commit()
+    return fid
+
+
+def record_resolution(
+    forecast_id: str, outcome: float, db_path: Optional[str] = None
+) -> bool:
+    """Record the realised binary outcome (1.0 YES / 0.0 NO). Returns found?"""
+    init_db(db_path)
+    with _connect(db_path) as conn:
+        cur = conn.execute(
+            "UPDATE forecasts SET resolved = 1, outcome = ?, resolved_at = ? "
+            "WHERE id = ?",
+            (float(outcome), time.time(), forecast_id),
+        )
+        conn.commit()
+        return cur.rowcount > 0
+
+
+def compute_calibration(db_path: Optional[str] = None) -> Dict:
+    """Aggregate Brier score, reliability curve and directional accuracy."""
+    init_db(db_path)
+    with _connect(db_path) as conn:
+        rows = conn.execute(
+            "SELECT p_forecast, outcome FROM forecasts "
+            "WHERE resolved = 1 AND p_forecast IS NOT NULL AND outcome IS NOT NULL"
+        ).fetchall()
+        n_total = conn.execute("SELECT COUNT(*) AS c FROM forecasts").fetchone()["c"]
+
+    n_resolved = len(rows)
+    if n_resolved == 0:
+        return {
+            "brier_score": None,
+            "directional_accuracy": None,
+            "reliability_curve": [],
+            "n_resolved": 0,
+            "n_total": int(n_total),
+            "note": "No resolved forecasts yet — record outcomes via POST /resolve.",
+        }
+
+    preds = np.array([float(r["p_forecast"]) for r in rows], dtype=float)
+    outcomes = np.array([float(r["outcome"]) for r in rows], dtype=float)
+    preds = np.clip(preds, 0.0, 1.0)
+    outcomes = np.clip(outcomes, 0.0, 1.0)
+
+    brier = _brier_score(outcomes, preds)
+    directional = float(np.mean((preds >= 0.5) == (outcomes >= 0.5)))
+    reliability = _reliability_curve(preds, outcomes)
+
+    return {
+        "brier_score": round(float(brier), 4),
+        "directional_accuracy": round(directional, 4),
+        "reliability_curve": reliability,
+        "n_resolved": int(n_resolved),
+        "n_total": int(n_total),
+        # Tetlock's superforecasters land near ~0.14–0.18 on hard questions.
+        "benchmark": {"superforecaster_brier": 0.15, "always_0.5_brier": 0.25},
+    }
+
+
+# --- internals -----------------------------------------------------------
+
+def _brier_score(outcomes: np.ndarray, preds: np.ndarray) -> float:
+    try:
+        import properscoring as ps  # type: ignore
+
+        return float(np.mean(ps.brier_score(outcomes, preds)))
+    except Exception:
+        return float(np.mean((preds - outcomes) ** 2))
+
+
+def _reliability_curve(preds: np.ndarray, outcomes: np.ndarray) -> List[Dict]:
+    """Decile reliability bins (predicted vs. observed frequency)."""
+    edges = np.linspace(0.0, 1.0, 11)
+    curve: List[Dict] = []
+    for lo, hi in zip(edges[:-1], edges[1:]):
+        # include the right edge in the final bin
+        if hi >= 1.0:
+            mask = (preds >= lo) & (preds <= hi)
+        else:
+            mask = (preds >= lo) & (preds < hi)
+        count = int(np.sum(mask))
+        if count == 0:
+            continue
+        curve.append(
+            {
+                "bin_lower": round(float(lo), 2),
+                "bin_upper": round(float(hi), 2),
+                "predicted": round(float(np.mean(preds[mask])), 4),
+                "observed": round(float(np.mean(outcomes[mask])), 4),
+                "count": count,
+            }
+        )
+    return curve
+
+
+def _nullable_float(value: Optional[float]) -> Optional[float]:
+    if value is None:
+        return None
+    try:
+        f = float(value)
+    except (TypeError, ValueError):
+        return None
+    return f if np.isfinite(f) else None

ensemble.py

@@ -0,0 +1,176 @@
+"""Ensemble blending + calibration nudges for the FutureQuery forecaster.
+
+This module is deliberately dependency-light (numpy only) so the blending
+rules can be unit-tested without loading TimesFM / Chronos. ``main.py`` runs
+the two models and hands their raw forecast arrays to :func:`full_response`;
+the short-series / no-history guards live in :func:`short_series_response`.
+
+Rules (per the Nov-2025 research notes baked into the task):
+
+* < 10 points                     -> use_forecast = False, warning set.
+* question_type == "event"        -> ensemble = 0.5*TimesFM + 0.5*Chronos-2
+  (equal weight: the models showed negative R^2 on binary event markets, so
+  we do not let either dominate).
+* question_type == "numeric"      -> Chronos-2 0.6 / TimesFM 0.4 (Chronos-2
+  beats TimesFM on fev-bench for numeric trajectories).
+* base-rate prior                 -> if the latest price is within 0.1 of 0.5,
+  pull the ensemble 15% back toward 0.5 (regression to the mean).
+* extremise (events only)         -> if confidence > 0.70, sharpen with the
+  standard superforecasting transform 0.5 + 1.3*(p - 0.5).
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, Optional, Sequence
+
+import numpy as np
+
+# --- weights & thresholds (single source of truth) -----------------------
+MIN_POINTS = 10
+EVENT_WEIGHTS = (0.5, 0.5)        # (timesfm, chronos2)
+NUMERIC_WEIGHTS = (0.4, 0.6)      # (timesfm, chronos2) -> Chronos-2 favoured
+BASE_RATE_BAND = 0.1              # |last - 0.5| <= 0.1 triggers the prior
+BASE_RATE_PULL = 0.15            # move 15% toward 0.5
+EXTREMISE_GATE = 0.70            # only extremise confident forecasts
+EXTREMISE_K = 1.3               # superforecasting extremisation factor
+TREND_EPS = 0.02                 # 2 percentage points = flat band for events
+
+
+def _clamp01(arr: np.ndarray) -> np.ndarray:
+    return np.clip(arr, 0.0, 1.0)
+
+
+def _trend(forecast: np.ndarray, reference: float, numeric: bool) -> str:
+    """up / down / flat relative to the last observed value."""
+    if forecast.size == 0:
+        return "flat"
+    delta = float(forecast[-1]) - float(reference)
+    eps = max(TREND_EPS * abs(reference), 1e-9) if numeric else TREND_EPS
+    if delta > eps:
+        return "up"
+    if delta < -eps:
+        return "down"
+    return "flat"
+
+
+def _persistence(prices: np.ndarray, horizon: int, event: bool) -> np.ndarray:
+    """Flat continuation used when we cannot really forecast."""
+    if prices.size:
+        return np.full(int(horizon), float(prices[-1]), dtype=float)
+    return np.full(int(horizon), 0.5 if event else 0.0, dtype=float)
+
+
+def _block(forecast: np.ndarray, reference: float, numeric: bool) -> Dict:
+    return {
+        "forecast": [round(float(x), 4) for x in forecast],
+        "trend": _trend(forecast, reference, numeric),
+    }
+
+
+def blend(
+    prices: Sequence[float],
+    timesfm_fc: Sequence[float],
+    chronos_fc: Sequence[float],
+    question_type: str,
+    horizon: int,
+) -> Dict:
+    """Blend two model forecasts into the ensemble (the heart of Phase 1)."""
+    price_arr = np.asarray(prices, dtype=float)
+    last = float(price_arr[-1]) if price_arr.size else 0.5
+    event = question_type == "event"
+    numeric = not event
+
+    tf = np.asarray(timesfm_fc, dtype=float)
+    ch = np.asarray(chronos_fc, dtype=float)
+    n = min(tf.size, ch.size, int(horizon))
+    if n == 0:
+        n = int(horizon)
+    tf = tf[:n]
+    ch = ch[:n]
+
+    if event:
+        tf = _clamp01(tf)
+        ch = _clamp01(ch)
+        w_tf, w_ch = EVENT_WEIGHTS
+    else:
+        w_tf, w_ch = NUMERIC_WEIGHTS
+
+    ensemble = w_tf * tf + w_ch * ch
+
+    if event:
+        # 1) base-rate prior: regress near-coin-flip forecasts toward 0.5
+        if abs(last - 0.5) <= BASE_RATE_BAND:
+            ensemble = ensemble + BASE_RATE_PULL * (0.5 - ensemble)
+        # 2) extremise only confident forecasts
+        headline = float(np.mean(ensemble))
+        confidence = max(headline, 1.0 - headline)
+        if confidence > EXTREMISE_GATE:
+            ensemble = 0.5 + EXTREMISE_K * (ensemble - 0.5)
+        ensemble = _clamp01(ensemble)
+
+    brier = _brier_estimate(ensemble) if event else None
+
+    return {
+        "timesfm": _block(tf, last, numeric),
+        "chronos2": _block(ch, last, numeric),
+        "ensemble": {
+            **_block(ensemble, last, numeric),
+            "brier_estimate": brier,
+        },
+    }
+
+
+def _brier_estimate(ensemble: np.ndarray) -> Optional[float]:
+    """Expected Brier score if the outcome were Bernoulli(p): p*(1-p).
+
+    This is a self-consistency proxy reported at forecast time (we don't yet
+    know the resolution). It peaks at 0.25 for p=0.5 and shrinks as the
+    forecast gets confident. The realised Brier score is tracked separately in
+    calibration.py once markets resolve.
+    """
+    if ensemble.size == 0:
+        return None
+    p = float(np.clip(ensemble[-1], 0.0, 1.0))
+    return round(p * (1.0 - p), 2)
+
+
+def short_series_response(
+    prices: Sequence[float],
+    horizon: int,
+    question_type: str,
+    warning: str,
+) -> Dict:
+    """Schema-valid response when we refuse to forecast (too short / no data)."""
+    event = question_type == "event"
+    price_arr = np.asarray(prices, dtype=float)
+    flat = _persistence(price_arr, horizon, event)
+    last = float(price_arr[-1]) if price_arr.size else (0.5 if event else 0.0)
+    block = _block(flat, last, not event)
+    return {
+        "timesfm": dict(block),
+        "chronos2": dict(block),
+        "ensemble": {**block, "brier_estimate": None},
+        "use_forecast": False,
+        "warning": warning,
+    }
+
+
+def full_response(
+    prices: Sequence[float],
+    timesfm_fc: Sequence[float],
+    chronos_fc: Sequence[float],
+    question_type: str,
+    horizon: int,
+) -> Dict:
+    payload = blend(prices, timesfm_fc, chronos_fc, question_type, horizon)
+    payload["use_forecast"] = True
+    payload["warning"] = None
+    return payload
+
+
+def headline_probability(payload: Dict) -> Optional[float]:
+    """Representative probability of an ensemble payload (last horizon point)."""
+    forecast: List[float] = payload.get("ensemble", {}).get("forecast", [])
+    if not forecast:
+        return None
+    return float(np.clip(forecast[-1], 0.0, 1.0))

main.py

@@ -0,0 +1,213 @@
+"""FutureQuery forecast-service — FastAPI ensemble API (TimesFM + Chronos-2).
+
+Run it::
+
+    cd forecast-service
+    pip install -r requirements.txt
+    python main.py        # serves http://localhost:8008
+
+Endpoints
+---------
+POST /forecast        -> ensemble forecast (TimesFM + Chronos-2) for a price series
+GET  /calibration     -> Brier score, reliability curve, directional accuracy
+POST /resolve         -> record a market's realised outcome for calibration
+GET  /health          -> model availability / liveness
+POST /pipeline/start  -> run the full scan->triage->ensemble->synthesise loop
+POST /pipeline/resume -> approve / override the human checkpoint (browser-driven)
+
+Models are loaded once on startup (not per request). The first run downloads
+weights from HuggingFace (~2GB) — expected. If a model can't be loaded the
+service still starts and falls back to a naive forecast, so the API never hard
+-fails (see models.py).
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from contextlib import asynccontextmanager
+from typing import List, Literal, Optional
+
+import numpy as np
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+
+import calibration
+import ensemble
+from models import ChronosModel, TimesFMModel
+from pipeline_api import router as pipeline_router
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+)
+log = logging.getLogger("forecast.main")
+
+PORT = int(os.environ.get("FORECAST_PORT", "8008"))
+
+# Holds the loaded models for the lifetime of the process.
+STATE: dict = {"timesfm": None, "chronos": None}
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    log.info("Loading forecasting models (first run downloads ~2GB)...")
+    STATE["timesfm"] = TimesFMModel()
+    STATE["chronos"] = ChronosModel()
+    calibration.init_db()
+    log.info(
+        "Models ready — TimesFM available=%s, Chronos available=%s",
+        STATE["timesfm"].available,
+        STATE["chronos"].available,
+    )
+    yield
+    STATE["timesfm"] = None
+    STATE["chronos"] = None
+
+
+app = FastAPI(title="FutureQuery forecast-service", version="1.0.0", lifespan=lifespan)
+
+# Allow the Next.js dev server / browser tools to call the API directly.
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(pipeline_router)
+
+
+# --- request / response models -------------------------------------------
+
+class ForecastRequest(BaseModel):
+    prices: List[float] = Field(default_factory=list, description="historical yes_prices")
+    covariates: Optional[List[List[float]]] = Field(
+        default=None, description="optional related series (past covariates)"
+    )
+    question_type: Literal["numeric", "event"] = "event"
+    horizon: int = Field(default=5, ge=1, le=64)
+    question: Optional[str] = None  # optional label, stored for calibration
+
+
+class ResolveRequest(BaseModel):
+    id: str
+    outcome: float = Field(ge=0.0, le=1.0, description="1.0 = YES happened, 0.0 = NO")
+
+
+# --- endpoints ------------------------------------------------------------
+
+@app.get("/health")
+def health() -> dict:
+    tf = STATE.get("timesfm")
+    ch = STATE.get("chronos")
+    return {
+        "status": "ok",
+        "timesfm": {
+            "available": bool(tf and tf.available),
+            "label": tf.label if tf else None,
+        },
+        "chronos2": {
+            "available": bool(ch and ch.available),
+            "label": ch.label if ch else None,
+        },
+    }
+
+
+@app.post("/forecast")
+def forecast(req: ForecastRequest) -> dict:
+    prices = [float(p) for p in (req.prices or []) if _finite(p)]
+    horizon = int(req.horizon)
+    qtype = req.question_type
+
+    # --- guards: refuse to forecast when we shouldn't --------------------
+    if len(prices) == 0:
+        warning = (
+            "No price history for event market"
+            if qtype == "event"
+            else "No data provided"
+        )
+        payload = ensemble.short_series_response(prices, horizon, qtype, warning)
+        return _finalize(payload, req, prices)
+
+    if len(prices) < ensemble.MIN_POINTS:
+        warning = (
+            f"Series too short for reliable forecast "
+            f"(have {len(prices)}, need >= {ensemble.MIN_POINTS})"
+        )
+        payload = ensemble.short_series_response(prices, horizon, qtype, warning)
+        return _finalize(payload, req, prices)
+
+    # --- run both models on startup-loaded handles -----------------------
+    tf_model = STATE.get("timesfm") or TimesFMModel()
+    ch_model = STATE.get("chronos") or ChronosModel()
+
+    timesfm_fc = tf_model.forecast(prices, horizon, req.covariates)
+    chronos_fc = ch_model.forecast(prices, horizon, req.covariates)
+
+    payload = ensemble.full_response(prices, timesfm_fc, chronos_fc, qtype, horizon)
+    return _finalize(payload, req, prices)
+
+
+@app.get("/calibration")
+def get_calibration() -> dict:
+    return calibration.compute_calibration()
+
+
+@app.post("/resolve")
+def resolve(req: ResolveRequest) -> dict:
+    found = calibration.record_resolution(req.id, req.outcome)
+    return {"ok": found, "id": req.id}
+
+
+# --- helpers --------------------------------------------------------------
+
+def _finalize(payload: dict, req: ForecastRequest, prices: List[float]) -> dict:
+    """Attach a calibration id and persist the forecast for later scoring."""
+    headline = ensemble.headline_probability(payload)
+    p_market = prices[-1] if prices else None
+    trend = payload.get("ensemble", {}).get("trend")
+    brier = payload.get("ensemble", {}).get("brier_estimate")
+    try:
+        fid = calibration.record_forecast(
+            question=req.question or "",
+            question_type=req.question_type,
+            horizon=int(req.horizon),
+            p_forecast=headline,
+            p_market=p_market,
+            trend=trend,
+            brier_estimate=brier,
+            use_forecast=bool(payload.get("use_forecast", False)),
+        )
+        payload["id"] = fid
+    except Exception as exc:  # never let logging break a forecast
+        log.warning("could not persist forecast for calibration: %s", exc)
+        payload["id"] = None
+    return payload
+
+
+def _finite(value) -> bool:
+    try:
+        return np.isfinite(float(value))
+    except (TypeError, ValueError):
+        return False
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    print("\n" + "=" * 56, flush=True)
+    print("  forecast-service is READY when the next line says", flush=True)
+    print("  'Uvicorn running on http://127.0.0.1:8008'", flush=True)
+    print("=" * 56 + "\n", flush=True)
+    # bind loopback (127.0.0.1): avoids the Windows firewall prompt and
+    # localhost/IPv6 mismatches that can stop the server from coming up.
+    # 127.0.0.1 locally; set FORECAST_HOST=0.0.0.0 in a container so the host can reach it.
+    uvicorn.run(
+        "main:app",
+        host=os.environ.get("FORECAST_HOST", "127.0.0.1"),
+        port=PORT,
+        reload=False,
+    )

models.py

@@ -0,0 +1,260 @@
+"""Model wrappers for TimesFM and Chronos-2 with graceful degradation.
+
+Both wrappers expose a uniform interface::
+
+    model.available -> bool
+    model.label     -> str   (e.g. "TimesFM 2.5", "Chronos-2")
+    model.forecast(series, horizon, covariates=None) -> np.ndarray
+
+If the heavy ML dependencies or the model weights are unavailable (not
+installed, offline, download failed, OOM, ...) the wrapper keeps
+``available = False`` and ``forecast`` returns a damped naive forecast.
+That guarantees ``main.py`` always starts and ``/forecast`` always responds,
+which is exactly what Phase 5 asks for. When the real weights are present the
+wrappers use them.
+
+The API surfaces of these libraries changed across releases, so each loader
+tries the current (2.x) API first and falls back to the legacy API. Every
+inference call is wrapped so a runtime surprise degrades to the naive path
+instead of 500-ing the request.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional, Sequence
+
+import numpy as np
+
+log = logging.getLogger("forecast.models")
+
+
+def naive_forecast(series: Sequence[float], horizon: int) -> np.ndarray:
+    """Damped-drift fallback forecast.
+
+    Continues the last observed step, damping it each period so the line does
+    not run away over the horizon. Falls back to persistence (repeat the last
+    value) when there is too little history.
+    """
+    arr = np.asarray(series, dtype=float)
+    arr = arr[np.isfinite(arr)]
+    if arr.size == 0:
+        return np.zeros(int(horizon), dtype=float)
+
+    last = float(arr[-1])
+    step = float(arr[-1] - arr[-2]) if arr.size >= 2 else 0.0
+    damp = 0.6
+    out = []
+    cur = last
+    for _ in range(int(horizon)):
+        cur = cur + step
+        step *= damp
+        out.append(cur)
+    return np.asarray(out, dtype=float)
+
+
+class TimesFMModel:
+    """Google TimesFM wrapper (tries the 2.5 API, then the legacy 2.0 API)."""
+
+    def __init__(self) -> None:
+        self.available = False
+        self.label = "TimesFM 2.0"
+        self._impl = None
+        self._mode: Optional[str] = None
+        self._load()
+
+    def _load(self) -> None:
+        try:
+            import timesfm  # type: ignore
+        except Exception as exc:  # pragma: no cover - import guard
+            log.warning("timesfm not importable (%s) — using naive fallback", exc)
+            return
+
+        # --- Preferred: TimesFM 2.5 (current PyPI API) ---------------------
+        try:
+            import torch  # type: ignore
+
+            torch.set_float32_matmul_precision("high")
+            model = timesfm.TimesFM_2p5_200M_torch.from_pretrained(
+                "google/timesfm-2.5-200m-pytorch"
+            )
+            model.compile(
+                timesfm.ForecastConfig(
+                    max_context=1024,
+                    max_horizon=256,
+                    normalize_inputs=True,
+                    use_continuous_quantile_head=True,
+                    force_flip_invariance=True,
+                    infer_is_positive=False,  # probabilities can sit near 0
+                    fix_quantile_crossing=True,
+                )
+            )
+            self._impl = model
+            self._mode = "2.5"
+            self.label = "TimesFM 2.5"
+            self.available = True
+            log.info("Loaded TimesFM 2.5 (google/timesfm-2.5-200m-pytorch)")
+            return
+        except Exception as exc:
+            log.warning("TimesFM 2.5 load failed (%s) — trying legacy 2.0 API", exc)
+
+        # --- Fallback: TimesFM 2.0 (legacy hparams/checkpoint API) ---------
+        try:
+            tfm = timesfm.TimesFm(
+                hparams=timesfm.TimesFmHparams(
+                    backend="cpu",
+                    per_core_batch_size=32,
+                    horizon_len=128,
+                    context_len=2048,
+                ),
+                checkpoint=timesfm.TimesFmCheckpoint(
+                    huggingface_repo_id="google/timesfm-2.0-500m-pytorch"
+                ),
+            )
+            self._impl = tfm
+            self._mode = "2.0"
+            self.label = "TimesFM 2.0"
+            self.available = True
+            log.info("Loaded TimesFM 2.0 (google/timesfm-2.0-500m-pytorch)")
+        except Exception as exc:
+            log.warning("TimesFM 2.0 load failed (%s) — using naive fallback", exc)
+
+    def forecast(
+        self,
+        series: Sequence[float],
+        horizon: int,
+        covariates: Optional[Sequence[Sequence[float]]] = None,
+    ) -> np.ndarray:
+        if not self.available or self._impl is None:
+            return naive_forecast(series, horizon)
+
+        arr = np.asarray(series, dtype=float)
+        try:
+            if self._mode == "2.5":
+                point, _quantiles = self._impl.forecast(horizon=int(horizon), inputs=[arr])
+                return np.asarray(point[0], dtype=float)[:horizon]
+            # legacy 2.0
+            point, _ = self._impl.forecast([arr], freq=[0])
+            return np.asarray(point[0], dtype=float)[:horizon]
+        except Exception as exc:
+            log.warning("TimesFM forecast failed (%s) — naive fallback", exc)
+            return naive_forecast(series, horizon)
+
+
+class ChronosModel:
+    """Amazon Chronos-2 wrapper (falls back to a Chronos-Bolt pipeline)."""
+
+    def __init__(self) -> None:
+        self.available = False
+        self.label = "Chronos-2"
+        self._impl = None
+        self._mode: Optional[str] = None
+        self._load()
+
+    def _load(self) -> None:
+        # --- Preferred: Chronos-2 -----------------------------------------
+        try:
+            from chronos import Chronos2Pipeline  # type: ignore
+
+            self._impl = Chronos2Pipeline.from_pretrained(
+                "amazon/chronos-2", device_map="cpu"
+            )
+            self._mode = "chronos2"
+            self.label = "Chronos-2"
+            self.available = True
+            log.info("Loaded Chronos-2 (amazon/chronos-2)")
+            return
+        except Exception as exc:
+            log.warning("Chronos-2 load failed (%s) — trying Chronos-Bolt", exc)
+
+        # --- Fallback: Chronos-Bolt (older base pipeline) -----------------
+        try:
+            from chronos import BaseChronosPipeline  # type: ignore
+
+            self._impl = BaseChronosPipeline.from_pretrained(
+                "amazon/chronos-bolt-base", device_map="cpu"
+            )
+            self._mode = "bolt"
+            self.label = "Chronos-Bolt"
+            self.available = True
+            log.info("Loaded Chronos-Bolt (amazon/chronos-bolt-base)")
+        except Exception as exc:
+            log.warning("Chronos-Bolt load failed (%s) — using naive fallback", exc)
+
+    def forecast(
+        self,
+        series: Sequence[float],
+        horizon: int,
+        covariates: Optional[Sequence[Sequence[float]]] = None,
+    ) -> np.ndarray:
+        if not self.available or self._impl is None:
+            return naive_forecast(series, horizon)
+
+        arr = np.asarray(series, dtype=float)
+        try:
+            if self._mode == "chronos2":
+                return self._forecast_chronos2(arr, horizon, covariates)
+            return self._forecast_bolt(arr, horizon)
+        except Exception as exc:
+            log.warning("Chronos forecast failed (%s) — naive fallback", exc)
+            return naive_forecast(series, horizon)
+
+    def _forecast_chronos2(
+        self,
+        arr: np.ndarray,
+        horizon: int,
+        covariates: Optional[Sequence[Sequence[float]]],
+    ) -> np.ndarray:
+        import pandas as pd
+
+        ts = pd.date_range("2000-01-01", periods=arr.size, freq="D")
+        frame = {"id": ["series"] * arr.size, "timestamp": ts, "target": arr}
+
+        # Chronos-2 natively supports past covariates; attach any we were given.
+        if covariates:
+            for idx, cov in enumerate(covariates):
+                cov_arr = np.asarray(cov, dtype=float)
+                if cov_arr.size == arr.size:
+                    frame[f"cov_{idx}"] = cov_arr
+
+        context_df = pd.DataFrame(frame)
+        pred_df = self._impl.predict_df(
+            context_df,
+            prediction_length=int(horizon),
+            quantile_levels=[0.1, 0.5, 0.9],
+            id_column="id",
+            timestamp_column="timestamp",
+            target="target",
+        )
+        return self._extract_median(pred_df, horizon)
+
+    @staticmethod
+    def _extract_median(pred_df, horizon: int) -> np.ndarray:
+        import pandas as pd  # noqa: F401
+
+        # predict_df output column naming varies; prefer the 0.5 quantile/mean.
+        for candidate in ("0.5", 0.5, "median", "mean", "predictions", "target"):
+            if candidate in pred_df.columns:
+                return np.asarray(pred_df[candidate].to_numpy(), dtype=float)[:horizon]
+        # Last resort: first purely-numeric column that is not an id/time column.
+        for col in pred_df.columns:
+            if col in ("id", "timestamp", "item_id"):
+                continue
+            try:
+                return np.asarray(pred_df[col].to_numpy(), dtype=float)[:horizon]
+            except Exception:
+                continue
+        raise ValueError("could not locate a forecast column in predict_df output")
+
+    def _forecast_bolt(self, arr: np.ndarray, horizon: int) -> np.ndarray:
+        import torch  # type: ignore
+
+        context = torch.tensor(arr, dtype=torch.float32)
+        # BaseChronosPipeline exposes predict_quantiles(...) -> (quantiles, mean)
+        quantiles, mean = self._impl.predict_quantiles(
+            context=context,
+            prediction_length=int(horizon),
+            quantile_levels=[0.1, 0.5, 0.9],
+        )
+        median = quantiles[0, :, 1]  # the 0.5 quantile
+        return np.asarray(median, dtype=float)[:horizon]

pipeline.py

@@ -0,0 +1,599 @@
+"""FutureQuery research pipeline — LangGraph SCAN -> TRIAGE -> SYNTHESISE.
+
+Runs the full loop for a single question across Polymarket, Kalshi, Metaculus
+and Manifold, calls the ensemble ``/forecast`` endpoint, fuses
+everything into one probability, pauses at a human checkpoint, and writes the
+result to ``db.sqlite`` for calibration tracking.
+
+Usage::
+
+    python pipeline.py "Will UK interest rates fall below 4% by Jan 2027?"
+    python pipeline.py --numeric "US CPI year-over-year in Dec 2026"
+
+At the checkpoint, press **Enter** to accept the proposed probability or type
+``override 0.42`` to set your own.
+
+Graph::
+
+    scan_markets -> triage -> fetch_history -> run_ensemble
+                 -> synthesise -> human_checkpoint -> output
+
+The source fetchers and the ``/forecast`` call are module-level functions on
+purpose, so they can be monkeypatched in tests and so a single failing source
+never takes down the run.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import os
+import re
+import sys
+import urllib.parse
+import urllib.request
+import uuid
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Dict, List, Optional, TypedDict
+
+FORECAST_URL = os.environ.get("FORECAST_URL", "http://127.0.0.1:8008")
+HORIZON = int(os.environ.get("PIPELINE_HORIZON", "5"))
+ARB_SPREAD_THRESHOLD = 0.10
+
+# Platform credibility weights used in triage scoring.
+CREDIBILITY = {
+    "Polymarket": 1.0,
+    "Kalshi": 1.0,
+    "Metaculus": 0.9,
+    "Manifold": 0.6,  # play-money — lower weight
+}
+
+STOP_WORDS = {
+    "will", "the", "a", "an", "of", "to", "in", "on", "by", "be", "is", "are",
+    "before", "after", "than", "this", "that", "and", "or", "for", "with",
+}
+
+
+# --- shared state ---------------------------------------------------------
+
+class PipelineState(TypedDict, total=False):
+    question: str
+    question_type: str
+    horizon: int
+    hits: List[Dict]
+    scored: List[Dict]
+    top: List[Dict]
+    arbitrage: Optional[Dict]
+    history: List[float]
+    history_source: Optional[str]
+    ensemble: Dict
+    use_forecast: bool
+    ensemble_trend: Optional[str]
+    brier_estimate: Optional[float]
+    market_consensus: Optional[float]
+    final_probability: Optional[float]
+    rationale: str
+    decision_note: str
+    forecast_id: Optional[str]
+
+
+# --- tiny HTTP helpers (stdlib only) -------------------------------------
+
+def _get_json(url: str, timeout: int = 10):
+    req = urllib.request.Request(url, headers={"User-Agent": "futurequery-pipeline/1.0"})
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return json.loads(resp.read().decode("utf-8"))
+
+
+def _post_json(url: str, payload: dict, timeout: int = 20):
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        url, data=data, headers={"Content-Type": "application/json"}
+    )
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return json.loads(resp.read().decode("utf-8"))
+
+
+def _keywords(question: str) -> str:
+    words = [
+        w for w in re.sub(r"[^a-z0-9\s-]", " ", question.lower()).split()
+        if len(w) > 2 and w not in STOP_WORDS
+    ]
+    return " ".join(dict.fromkeys(words)) or question
+
+
+def _as_float(value, default=None):
+    try:
+        f = float(value)
+        return f if math.isfinite(f) else default
+    except (TypeError, ValueError):
+        return default
+
+
+# --- source fetchers (each returns a list of hit dicts) ------------------
+# Hit = {source, question, probability, volume, url, clob_token_id?, end_date?}
+
+def fetch_polymarket(question: str) -> List[Dict]:
+    try:
+        q = urllib.parse.urlencode(
+            {"active": "true", "closed": "false", "limit": "8", "search": _keywords(question)}
+        )
+        data = _get_json(f"https://gamma-api.polymarket.com/markets?{q}")
+        hits = []
+        for m in data if isinstance(data, list) else []:
+            outcomes = _json_array(m.get("outcomes"))
+            prices = [_as_float(p) for p in _json_array(m.get("outcomePrices"))]
+            tokens = _json_array(m.get("clobTokenIds"))
+            yes_idx = next((i for i, o in enumerate(outcomes) if str(o).lower() == "yes"), 0)
+            prob = prices[yes_idx] if yes_idx < len(prices) else None
+            if prob is None:
+                continue
+            hits.append({
+                "source": "Polymarket",
+                "question": m.get("question") or m.get("title") or "",
+                "probability": max(0.0, min(1.0, prob)),
+                "volume": _as_float(m.get("volumeNum") or m.get("volume"), 0.0) or 0.0,
+                "url": f"https://polymarket.com/event/{m.get('slug')}" if m.get("slug") else None,
+                "clob_token_id": tokens[yes_idx] if yes_idx < len(tokens) else (tokens[0] if tokens else None),
+                "end_date": m.get("endDate"),
+            })
+        return hits
+    except Exception:
+        return []
+
+
+def fetch_kalshi(question: str) -> List[Dict]:
+    api_key = os.environ.get("KALSHI_API_KEY")
+    if not api_key:
+        return []  # Kalshi search needs auth — skip gracefully
+    try:
+        q = urllib.parse.urlencode({"limit": "8", "status": "open", "search": _keywords(question)})
+        req = urllib.request.Request(
+            f"https://api.elections.kalshi.com/trade-api/v2/markets?{q}",
+            headers={"Authorization": f"Bearer {api_key}", "User-Agent": "futurequery/1.0"},
+        )
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            data = json.loads(resp.read().decode("utf-8"))
+        hits = []
+        for m in data.get("markets", []) or []:
+            bid, ask = _as_float(m.get("yes_bid")), _as_float(m.get("yes_ask"))
+            cents = (bid + ask) / 2 if bid is not None and ask is not None else _as_float(m.get("last_price"))
+            if cents is None or not m.get("title"):
+                continue
+            hits.append({
+                "source": "Kalshi",
+                "question": m["title"],
+                "probability": max(0.0, min(1.0, cents / 100.0)),
+                "volume": _as_float(m.get("volume"), 0.0) or 0.0,
+                "url": f"https://kalshi.com/markets/{m.get('ticker')}" if m.get("ticker") else None,
+                "end_date": m.get("close_time"),
+            })
+        return hits
+    except Exception:
+        return []
+
+
+def fetch_metaculus(question: str) -> List[Dict]:
+    try:
+        q = urllib.parse.urlencode({"search": _keywords(question), "limit": "5", "type": "forecast"})
+        data = _get_json(f"https://www.metaculus.com/api2/questions/?{q}")
+        hits = []
+        for item in data.get("results", []) or []:
+            prob = _metaculus_probability(item)
+            title = item.get("title") or item.get("question", {}).get("title")
+            if prob is None or not title:
+                continue
+            hits.append({
+                "source": "Metaculus",
+                "question": title,
+                "probability": max(0.0, min(1.0, prob)),
+                "volume": float(item.get("number_of_forecasters", 0) or 0),
+                "url": f"https://www.metaculus.com{item.get('page_url', '')}" if item.get("page_url") else None,
+                "end_date": item.get("resolve_time") or item.get("scheduled_resolve_time"),
+            })
+        return hits
+    except Exception:
+        return []
+
+
+def fetch_manifold(question: str) -> List[Dict]:
+    try:
+        q = urllib.parse.urlencode({"term": _keywords(question), "limit": "6"})
+        data = _get_json(f"https://api.manifold.markets/v0/search-markets?{q}")
+        hits = []
+        for m in data if isinstance(data, list) else []:
+            prob = _as_float(m.get("probability"))
+            if prob is None or m.get("outcomeType") not in (None, "BINARY"):
+                continue
+            hits.append({
+                "source": "Manifold",
+                "question": m.get("question", ""),
+                "probability": max(0.0, min(1.0, prob)),
+                "volume": _as_float(m.get("volume"), 0.0) or 0.0,
+                "url": m.get("url"),
+                "end_date": _ms_to_iso(m.get("closeTime")),
+            })
+        return hits
+    except Exception:
+        return []
+
+
+def fetch_polymarket_history(token_id: str, days: int = 30) -> List[float]:
+    """30-day yes-price history from the Polymarket CLOB prices-history API."""
+    try:
+        import time as _time
+
+        end = int(_time.time())
+        start = end - days * 86400
+        q = urllib.parse.urlencode(
+            {"market": token_id, "startTs": start, "endTs": end, "fidelity": 360}
+        )
+        data = _get_json(f"https://clob.polymarket.com/prices-history?{q}")
+        points = data.get("history", []) if isinstance(data, dict) else []
+        series = [_as_float(p.get("p")) for p in points if _as_float(p.get("p")) is not None]
+        return series
+    except Exception:
+        return []
+
+
+def call_forecast(prices: List[float], question_type: str, horizon: int) -> Dict:
+    """POST to the ensemble service; degrade gracefully if it is offline."""
+    try:
+        return _post_json(
+            f"{FORECAST_URL}/forecast",
+            {"prices": prices, "question_type": question_type, "horizon": horizon},
+        )
+    except Exception as exc:
+        return {
+            "timesfm": {"forecast": [], "trend": "flat"},
+            "chronos2": {"forecast": [], "trend": "flat"},
+            "ensemble": {"forecast": [], "trend": "flat", "brier_estimate": None},
+            "use_forecast": False,
+            "warning": f"forecast-service unreachable at {FORECAST_URL} ({exc})",
+        }
+
+
+# --- graph nodes ----------------------------------------------------------
+
+def scan_markets(state: PipelineState) -> Dict:
+    question = state["question"]
+    sources = {
+        "Polymarket": fetch_polymarket,
+        "Kalshi": fetch_kalshi,
+        "Metaculus": fetch_metaculus,
+        "Manifold": fetch_manifold,
+    }
+    hits: List[Dict] = []
+    with ThreadPoolExecutor(max_workers=len(sources)) as ex:
+        futures = {ex.submit(fn, question): name for name, fn in sources.items()}
+        for fut in as_completed(futures):
+            try:
+                hits.extend(fut.result() or [])
+            except Exception:
+                pass
+    print(f"[scan] {len(hits)} hits across {len(sources)} sources")
+    return {"hits": hits}
+
+
+def triage(state: PipelineState) -> Dict:
+    hits = state.get("hits", [])
+    scored = []
+    for h in hits:
+        cred = CREDIBILITY.get(h["source"], 0.5)
+        volume = max(h.get("volume", 0.0) or 0.0, 0.0)
+        recency = _recency_factor(h.get("end_date"))
+        score = cred * math.log1p(volume + 1.0) * recency
+        scored.append({**h, "score": round(score, 4), "credibility": cred, "recency": round(recency, 3)})
+
+    scored.sort(key=lambda x: x["score"], reverse=True)
+    top = scored[:3]
+
+    arbitrage = None
+    probs = [h["probability"] for h in scored]
+    if len(probs) >= 2:
+        hi = max(scored, key=lambda x: x["probability"])
+        lo = min(scored, key=lambda x: x["probability"])
+        spread = hi["probability"] - lo["probability"]
+        if spread > ARB_SPREAD_THRESHOLD:
+            arbitrage = {
+                "spread": round(spread, 4),
+                "high": {"source": hi["source"], "probability": hi["probability"]},
+                "low": {"source": lo["source"], "probability": lo["probability"]},
+            }
+            print(f"[triage] ⚠ arbitrage: {spread:.2f} spread "
+                  f"({lo['source']} {lo['probability']:.2f} → {hi['source']} {hi['probability']:.2f})")
+    print(f"[triage] top {len(top)}: " + ", ".join(f"{h['source']}({h['score']})" for h in top))
+    return {"scored": scored, "top": top, "arbitrage": arbitrage}
+
+
+def fetch_history(state: PipelineState) -> Dict:
+    for h in state.get("top", []):
+        if h["source"] == "Polymarket" and h.get("clob_token_id"):
+            series = fetch_polymarket_history(h["clob_token_id"], days=30)
+            if len(series) >= 2:
+                print(f"[history] {len(series)} points from Polymarket "
+                      f"({h['question'][:48]})")
+                return {"history": series, "history_source": h.get("url") or h["source"]}
+    print("[history] no Polymarket price history available")
+    return {"history": [], "history_source": None}
+
+
+def run_ensemble(state: PipelineState) -> Dict:
+    prices = state.get("history", [])
+    qtype = state.get("question_type", "event")
+    horizon = state.get("horizon", HORIZON)
+    result = call_forecast(prices, qtype, horizon)
+    ens = result.get("ensemble", {})
+    print(f"[ensemble] use_forecast={result.get('use_forecast')} "
+          f"trend={ens.get('trend')} warning={result.get('warning')}")
+    return {
+        "ensemble": result,
+        "use_forecast": bool(result.get("use_forecast", False)),
+        "ensemble_trend": ens.get("trend"),
+        "brier_estimate": ens.get("brier_estimate"),
+    }
+
+
+def synthesise(state: PipelineState) -> Dict:
+    top = state.get("top", []) or state.get("scored", [])
+    consensus = _volume_weighted_mean(top)
+    ensemble = state.get("ensemble", {})
+    ens_forecast = ensemble.get("ensemble", {}).get("forecast", [])
+    ens_p = float(ens_forecast[-1]) if (state.get("use_forecast") and ens_forecast) else None
+
+    if consensus is None and ens_p is None:
+        final = 0.5
+    elif ens_p is None:
+        final = consensus
+    elif consensus is None:
+        final = ens_p
+    else:
+        # Markets are the strong prior; the ensemble nudges it.
+        final = 0.6 * consensus + 0.4 * ens_p
+
+    # base-rate prior: regress genuinely-uncertain calls toward 0.5
+    if final is not None and abs(final - 0.5) <= 0.1:
+        final = final + 0.15 * (0.5 - final)
+    final = max(0.0, min(1.0, final)) if final is not None else None
+
+    rationale = _rationale(state, consensus, ens_p, final)
+    print(f"[synthesise] consensus={_fmt(consensus)} ensemble={_fmt(ens_p)} "
+          f"→ proposed={_fmt(final)}")
+    return {"market_consensus": consensus, "final_probability": final, "rationale": rationale}
+
+
+def human_checkpoint(state: PipelineState) -> Dict:
+    from langgraph.types import interrupt
+
+    summary = {
+        "question": state["question"],
+        "market_consensus": state.get("market_consensus"),
+        "ensemble_trend": state.get("ensemble_trend"),
+        "use_forecast": state.get("use_forecast"),
+        "arbitrage": state.get("arbitrage"),
+        "proposed_probability": state.get("final_probability"),
+        "rationale": state.get("rationale"),
+        "instructions": "Press Enter to accept, or type 'override 0.XX' to set your own probability.",
+    }
+    # Pauses the graph; the value returned is whatever the driver resumes with.
+    user_input = interrupt(summary)
+
+    final = state.get("final_probability")
+    note = "accepted"
+    if isinstance(user_input, str):
+        text = user_input.strip()
+        if text.lower().startswith("override"):
+            parts = text.split()
+            if len(parts) >= 2:
+                v = _as_float(parts[1])
+                if v is not None:
+                    if v > 1.0:
+                        v = v / 100.0
+                    final = max(0.0, min(1.0, v))
+                    note = f"overridden to {final:.2f}"
+    return {"final_probability": final, "decision_note": note}
+
+
+def output(state: PipelineState) -> Dict:
+    fid = None
+    try:
+        import calibration
+
+        fid = calibration.record_forecast(
+            question=state["question"],
+            question_type=state.get("question_type", "event"),
+            horizon=state.get("horizon", HORIZON),
+            p_forecast=state.get("final_probability"),
+            p_market=state.get("market_consensus"),
+            trend=state.get("ensemble_trend"),
+            brier_estimate=state.get("brier_estimate"),
+            use_forecast=state.get("use_forecast", False),
+        )
+        print(f"[output] stored forecast {fid} in db.sqlite")
+    except Exception as exc:
+        print(f"[output] could not store forecast: {exc}")
+    return {"forecast_id": fid}
+
+
+# --- graph assembly -------------------------------------------------------
+
+def build_graph():
+    from langgraph.graph import END, START, StateGraph
+
+    try:
+        from langgraph.checkpoint.memory import InMemorySaver as _Saver
+    except Exception:  # pragma: no cover - older langgraph
+        from langgraph.checkpoint.memory import MemorySaver as _Saver
+
+    builder = StateGraph(PipelineState)
+    builder.add_node("scan_markets", scan_markets)
+    builder.add_node("triage", triage)
+    builder.add_node("fetch_history", fetch_history)
+    builder.add_node("run_ensemble", run_ensemble)
+    builder.add_node("synthesise", synthesise)
+    builder.add_node("human_checkpoint", human_checkpoint)
+    builder.add_node("output", output)
+
+    builder.add_edge(START, "scan_markets")
+    builder.add_edge("scan_markets", "triage")
+    builder.add_edge("triage", "fetch_history")
+    builder.add_edge("fetch_history", "run_ensemble")
+    builder.add_edge("run_ensemble", "synthesise")
+    builder.add_edge("synthesise", "human_checkpoint")
+    builder.add_edge("human_checkpoint", "output")
+    builder.add_edge("output", END)
+
+    return builder.compile(checkpointer=_Saver())
+
+
+# --- helpers --------------------------------------------------------------
+
+def _json_array(value) -> List:
+    if isinstance(value, list):
+        return value
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+            return parsed if isinstance(parsed, list) else []
+        except Exception:
+            return []
+    return []
+
+
+def _metaculus_probability(item: Dict) -> Optional[float]:
+    cp = item.get("community_prediction") or {}
+    if isinstance(cp, dict):
+        full = cp.get("full") or {}
+        for key in ("q2", "median", "mean"):
+            if _as_float(full.get(key)) is not None:
+                return _as_float(full.get(key))
+    q = item.get("question") or {}
+    agg = (q.get("aggregations") or {}).get("recency_weighted") or {}
+    latest = agg.get("latest") or {}
+    centers = latest.get("centers")
+    if isinstance(centers, list) and centers:
+        return _as_float(centers[0])
+    return None
+
+
+def _recency_factor(end_date: Optional[str]) -> float:
+    """Closer-to-resolution markets are more actionable → mild score boost."""
+    if not end_date:
+        return 1.0
+    try:
+        from datetime import datetime, timezone
+
+        dt = datetime.fromisoformat(str(end_date).replace("Z", "+00:00"))
+        days = (dt - datetime.now(timezone.utc)).total_seconds() / 86400.0
+        if days <= 0:
+            return 0.6  # already past resolution — stale
+        return max(0.7, min(1.2, 1.2 - days / 365.0))
+    except Exception:
+        return 1.0
+
+
+def _ms_to_iso(ms) -> Optional[str]:
+    if not ms:
+        return None
+    try:
+        from datetime import datetime, timezone
+
+        return datetime.fromtimestamp(float(ms) / 1000.0, tz=timezone.utc).isoformat()
+    except Exception:
+        return None
+
+
+def _volume_weighted_mean(hits: List[Dict]) -> Optional[float]:
+    if not hits:
+        return None
+    total = sum(max(h.get("volume", 0.0) or 0.0, 0.0) for h in hits)
+    if total <= 0:
+        return sum(h["probability"] for h in hits) / len(hits)
+    return sum(h["probability"] * max(h.get("volume", 0.0) or 0.0, 0.0) for h in hits) / total
+
+
+def _rationale(state, consensus, ens_p, final) -> str:
+    bits = []
+    n = len(state.get("scored", []))
+    if consensus is not None:
+        bits.append(f"{n} market hit(s), volume-weighted consensus {consensus:.0%}")
+    if ens_p is not None:
+        bits.append(f"ensemble forecast {ens_p:.0%} ({state.get('ensemble_trend')})")
+    elif not state.get("use_forecast"):
+        bits.append("ensemble withheld (insufficient price history)")
+    if state.get("arbitrage"):
+        a = state["arbitrage"]
+        bits.append(f"cross-platform arbitrage spread {a['spread']:.0%}")
+    bits.append(f"synthesised probability {final:.0%}" if final is not None else "no estimate")
+    return "; ".join(bits)
+
+
+def _fmt(value) -> str:
+    return "—" if value is None else f"{value:.2%}"
+
+
+# --- CLI driver -----------------------------------------------------------
+
+def _print_checkpoint(payload: Dict) -> None:
+    print("\n" + "=" * 64)
+    print("HUMAN CHECKPOINT")
+    print("=" * 64)
+    print(f"Question : {payload.get('question')}")
+    print(f"Consensus: {_fmt(payload.get('market_consensus'))}")
+    print(f"Ensemble : trend={payload.get('ensemble_trend')} "
+          f"use_forecast={payload.get('use_forecast')}")
+    if payload.get("arbitrage"):
+        a = payload["arbitrage"]
+        print(f"Arbitrage: {a['spread']:.2%} ({a['low']['source']} → {a['high']['source']})")
+    print(f"Rationale: {payload.get('rationale')}")
+    print("-" * 64)
+    print(f"PROPOSED PROBABILITY: {_fmt(payload.get('proposed_probability'))}")
+    print(payload.get("instructions", ""))
+
+
+def main(argv: List[str]) -> int:
+    args = [a for a in argv[1:] if a]
+    question_type = "event"
+    if "--numeric" in args:
+        question_type = "numeric"
+        args = [a for a in args if a != "--numeric"]
+    if not args:
+        print('Usage: python pipeline.py [--numeric] "Your forecast question?"')
+        return 2
+
+    question = " ".join(args)
+    from langgraph.types import Command
+
+    graph = build_graph()
+    config = {"configurable": {"thread_id": f"pipeline-{uuid.uuid4().hex[:8]}"}}
+    initial: PipelineState = {
+        "question": question,
+        "question_type": question_type,
+        "horizon": HORIZON,
+    }
+
+    print(f"\n▶ FutureQuery pipeline — {question!r} ({question_type})\n")
+    state = graph.invoke(initial, config=config)
+
+    # Drive through any human-checkpoint interrupts from the terminal.
+    while isinstance(state, dict) and state.get("__interrupt__"):
+        payload = state["__interrupt__"][0].value
+        _print_checkpoint(payload)
+        try:
+            user_input = input("> ")
+        except EOFError:
+            user_input = ""
+        state = graph.invoke(Command(resume=user_input), config=config)
+
+    print("\n" + "=" * 64)
+    print("FINAL")
+    print("=" * 64)
+    print(f"Question   : {state.get('question')}")
+    print(f"Probability: {_fmt(state.get('final_probability'))}  ({state.get('decision_note')})")
+    print(f"Rationale  : {state.get('rationale')}")
+    print(f"Stored as  : {state.get('forecast_id')}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv))

pipeline_api.py

@@ -0,0 +1,161 @@
+"""HTTP wrapper around the LangGraph pipeline (pipeline.py).
+
+Drives the full SCAN -> TRIAGE -> ENSEMBLE -> SYNTHESISE -> HUMAN CHECKPOINT loop
+from the browser:
+
+    POST /pipeline/start   {question, question_type} -> runs to the checkpoint
+    POST /pipeline/resume  {thread_id, resume}       -> "" accepts, "override 0.XX"
+
+Robustness:
+* the ensemble is computed IN-PROCESS (no HTTP self-call), avoiding
+  localhost/IPv6 and nested-request issues.
+* any failure is returned as a readable {"status":"error","error":...} with
+  HTTP 200, so the UI shows the message instead of a raw 500.
+"""
+
+from __future__ import annotations
+
+import traceback
+import uuid
+from typing import Literal
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+
+import pipeline
+
+router = APIRouter(prefix="/pipeline", tags=["pipeline"])
+
+_SNAPSHOT_KEYS = [
+    "question", "question_type", "horizon", "hits", "scored", "top", "arbitrage",
+    "ensemble", "use_forecast", "ensemble_trend", "brier_estimate",
+    "market_consensus", "final_probability", "rationale", "history_source",
+    "decision_note", "forecast_id",
+]
+
+_GRAPH = None
+_PATCHED = False
+
+
+def _graph():
+    global _GRAPH
+    if _GRAPH is None:
+        _GRAPH = pipeline.build_graph()
+    return _GRAPH
+
+
+def _use_inprocess_forecast():
+    """Replace the pipeline's HTTP forecast call with a direct in-process call."""
+    global _PATCHED
+    if _PATCHED:
+        return
+
+    def _inprocess(prices, question_type, horizon):
+        import main  # function-level import avoids a circular import at load
+        req = main.ForecastRequest(
+            prices=list(prices or []),
+            question_type=question_type,
+            horizon=horizon,
+        )
+        return main.forecast(req)
+
+    pipeline.call_forecast = _inprocess
+    _PATCHED = True
+
+
+def _snapshot(graph, config) -> dict:
+    try:
+        values = graph.get_state(config).values or {}
+    except Exception:
+        values = {}
+    return {k: values.get(k) for k in _SNAPSHOT_KEYS if k in values}
+
+
+def _interrupt_payload(result):
+    if isinstance(result, dict) and result.get("__interrupt__"):
+        try:
+            return result["__interrupt__"][0].value
+        except Exception:
+            return None
+    return None
+
+
+def _error(thread_id, graph, config, exc) -> dict:
+    return {
+        "thread_id": thread_id,
+        "status": "error",
+        "state": _snapshot(graph, config) if graph is not None else {},
+        "checkpoint": None,
+        "error": f"{type(exc).__name__}: {exc}",
+        "trace": traceback.format_exc()[-1500:],
+    }
+
+
+class StartRequest(BaseModel):
+    question: str = Field(min_length=3, max_length=500)
+    question_type: Literal["numeric", "event"] = "event"
+    horizon: int = Field(default=5, ge=1, le=64)
+
+
+class ResumeRequest(BaseModel):
+    thread_id: str
+    resume: str = ""
+
+
+@router.post("/start")
+def start(req: StartRequest) -> dict:
+    _use_inprocess_forecast()
+    thread_id = f"web-{uuid.uuid4().hex[:10]}"
+    try:
+        graph = _graph()
+    except Exception as exc:
+        return _error(thread_id, None, None, exc)
+
+    config = {"configurable": {"thread_id": thread_id}}
+    initial = {
+        "question": req.question,
+        "question_type": req.question_type,
+        "horizon": req.horizon,
+    }
+    try:
+        result = graph.invoke(initial, config=config)
+    except Exception as exc:
+        return _error(thread_id, graph, config, exc)
+
+    payload = _interrupt_payload(result)
+    return {
+        "thread_id": thread_id,
+        "status": "checkpoint" if payload is not None else "done",
+        "state": _snapshot(graph, config),
+        "checkpoint": payload,
+    }
+
+
+@router.post("/resume")
+def resume(req: ResumeRequest) -> dict:
+    from langgraph.types import Command
+
+    try:
+        graph = _graph()
+    except Exception as exc:
+        return _error(req.thread_id, None, None, exc)
+
+    config = {"configurable": {"thread_id": req.thread_id}}
+    try:
+        existing = graph.get_state(config)
+        if existing is None or not existing.values:
+            return {
+                "thread_id": req.thread_id, "status": "error", "state": {},
+                "checkpoint": None, "error": "unknown or expired thread_id",
+            }
+        result = graph.invoke(Command(resume=req.resume), config=config)
+    except Exception as exc:
+        return _error(req.thread_id, graph, config, exc)
+
+    payload = _interrupt_payload(result)
+    return {
+        "thread_id": req.thread_id,
+        "status": "checkpoint" if payload is not None else "done",
+        "state": _snapshot(graph, config),
+        "checkpoint": payload,
+    }

requirements-lite.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn
+numpy
+pandas
+langgraph

requirements.txt

@@ -0,0 +1,15 @@
+# FutureQuery forecast-service — ensemble forecasting + LangGraph pipeline
+# Python 3.10–3.12 recommended. First `python main.py` downloads model
+# weights from HuggingFace (~2GB) — this is expected and only happens once.
+
+timesfm[torch]            # Google TimesFM 2.x time-series foundation model
+chronos-forecasting>=2.0  # Amazon Chronos-2 (Chronos2Pipeline requires >=2.0)
+langgraph                 # SCAN -> TRIAGE -> SYNTHESISE orchestration (Phase 3)
+openai>=1.0.0             # OpenRouter (OpenAI-compatible) LLM synthesis
+fastapi                   # /forecast + /calibration API
+uvicorn                   # ASGI server (started from main.py on :8008)
+scikit-learn              # calibration_curve / reliability diagram helpers
+properscoring             # Brier score (graceful fallback to numpy if missing)
+statsmodels               # baseline statistical checks / fallbacks
+numpy
+pandas