docssdkspython-sdk

Python

Python SDKpip install deltadaemonRepository →

Production-quality async Python SDK for the DeltaDaemon API. Install with pip install deltadaemon. The client covers health, station metadata, accuracy (summary, exceedance, by-bucket, by-regime), bias correction, and data endpoints. Source, examples, and full API reference are in the repository linked below.

Python SDK repositorySource, README, examples (compare_stations, trading_bot_workflow, poll_accuracy), and full API reference.https://github.com/Delta-Daemon/python-sdk →
How to interpret accuracy statsLearn what count, mean_error, std_dev, MAE, and RMSE mean in practice, with real station data.Learn more →
← Observations
TypeScript →
Python
# Install: pip install deltadaemon
# Repository: https://github.com/Delta-Daemon/python-sdk


import asyncio
import os
from deltadaemon import DeltaDaemonClient
from deltadaemon.errors import AuthenticationError, PaymentRequiredError


# --- Quick start ---
async def quick_start():
    async with DeltaDaemonClient(api_key=os.environ.get("DELTADAEMON_API_KEY", "your-api-key")) as client:
        health = await client.get_health()
        print(health.status)
        stations = await client.get_station_metadata()
        for s in stations[:5]:
            print(f"{s.station_id}: {s.city_name}")
        summary = await client.get_accuracy_summary(city="Los Angeles", days=90)
        print(summary)


# --- Risk and markets (summary + exceedance + temperature buckets) ---
async def risk_and_markets(station_id="KLAX", days=90):
    async with DeltaDaemonClient(api_key=os.environ.get("DELTADAEMON_API_KEY")) as client:
        summary = await client.get_accuracy_summary(station_id=station_id, days=days)
        if isinstance(summary, dict) and summary.get("success"):
            d = summary["data"]
            print(f"MAE: {d['mae']}°F  bias: {d['mean_error']}°F")
        exc = await client.get_exceedance(station_id=station_id, thresholds="2,3,5", days=days)
        for t in exc.get("data", {}).get("thresholds", []):
            print(f"  Miss >{t['threshold_f']}°F: {t['exceedance_rate']:.0%}")
        buckets = await client.get_bucket_accuracy(station_id, days=days)
        for b in buckets.get("data", [])[:5]:
            print(f"  {b['temperature_range']}: hit {b.get('accuracy_when_forecast_in_range', 0):.0%}")


# --- Compare stations (which cities to trade) ---
async def compare_stations():
    async with DeltaDaemonClient(api_key=os.environ.get("DELTADAEMON_API_KEY")) as client:
        acc = await client.get_city_accuracy(days=90, min_samples=100, sort_by="mae")
        data = acc.get("data", acc) if isinstance(acc, dict) else acc
        for row in (data if isinstance(data, list) else [])[:10]:
            stats = row.get("stats") or {}
            print(f"{row.get('city_name')}: MAE={stats.get('mae')}  n={stats.get('verification_pairs')}")


# --- Trading workflow (run stats, bias correction, exceedance) ---
async def trading_workflow():
    async with DeltaDaemonClient(api_key=os.environ.get("DELTADAEMON_API_KEY")) as client:
        try:
            run_stats = await client.get_forecast_run_stats(station_id="KLAX")
            bias = await client.get_bias_correction(station_id="KLAX", forecast_temp=75, days=90)
            exc = await client.get_exceedance(station_id="KLAX", thresholds="1,2,3,5", days=90)
            print("Run stats:", run_stats, "Bias:", bias, "Exceedance:", exc)
        except AuthenticationError:
            print("Invalid API key")
        except PaymentRequiredError:
            print("Subscription past due or cancelled")


asyncio.run(quick_start())