Forecast VIX with forward-dated market data

pip install requests

curl -s -X POST https://api.polybridge.ai/v1/forecast \
  -H "Content-Type: application/json" \
  -d '{"question": "Will the VIX close above 30 at any point in the next 90 days?"}' \
  | python3 -m json.tool

{
  "probability": 0.10,
  "confidence": 0.34,
  "confidence_interval": {
    "lower": 0.05,
    "upper": 0.15
  },
  "markets_used": [
    {
      "id": "example-market-id",
      "question": "Example source market question",
      "platform": "polymarket",
      "platform_url": "https://example.com"
    }
  ]
}

import time
import requests

API_URL = "https://api.polybridge.ai/v1/forecast"
TIMEOUT = 75
MAX_RETRIES = 4

HEADLINE = "Will the VIX close above 30 at any point in the next 90 days?"

DRIVERS = [
    (
        "Crude oil > $90 (next quarter)",
        "Will crude oil settle above $90 at any point in the next 90 days?",
    ),
    (
        "SPX drawdown > 10% (next quarter)",
        "Will SPX draw down more than 10% at any point in the next 90 days?",
    ),
    (
        "Gold +10% (next quarter)",
        "Will gold rise more than 10% from its current price at any point in the next 90 days?",
    ),
    (
        "Hormuz regular traffic (next quarter)",
        "Will the Strait of Hormuz be open to regular traffic at the end of the next 90 days?",
    ),
]

def forecast(session, question):
    headers = {"Content-Type": "application/json"}

    for attempt in range(MAX_RETRIES + 1):
        try:
            r = session.post(
                API_URL,
                headers=headers,
                json={"question": question},
                timeout=TIMEOUT,
            )
        except requests.RequestException:
            if attempt >= MAX_RETRIES:
                raise
            time.sleep(min(20.0, 2.0 ** attempt))
            continue

        if r.status_code in (429, 503):
            if attempt >= MAX_RETRIES:
                r.raise_for_status()
            try:
                wait = float(r.headers.get("Retry-After", ""))
            except (TypeError, ValueError):
                wait = min(20.0, 2.0 ** attempt)
            time.sleep(wait)
            continue

        r.raise_for_status()
        return r.json()

    raise RuntimeError(f"Failed after {MAX_RETRIES} retries: {question}")

def bar(p):
    filled = round(p * 20)
    return "█" * filled + "░" * (20 - filled)

with requests.Session() as session:
    result = forecast(session, HEADLINE)
    p = float(result.get("probability", 0))
    print("SIGNAL")
    print(f"{p * 100:>5.1f}%  {bar(p)}  VIX > 30 in 90d")
    print()
    print("HIGHLIGHTED DRIVERS")
    for label, question in DRIVERS:
        result = forecast(session, question)
        p = float(result.get("probability", 0))
        print(f"{p * 100:>5.1f}%  {bar(p)}  {label}")