alpinedataworks 0.1.0

Python SDK for the Alpine DataWorks Intelligence Object API

Alpine DataWorks Python SDK

Official Python SDK for the Alpine DataWorks Intelligence Object API. Serves all 14 ADW Intelligence Object Manifests (IOMs) — crypto sentiment, macro health, DeFi TVL, real-estate affordability, and more — to any Python application or AI agent.

Install

pip install -e .          # editable local install (development)
pip install alpinedataworks   # once published to PyPI (future)

Requires Python 3.9+ and requests.

Demo keys (evaluation)

Key	Tier	Rate limit	Field depth
adw_free_demo	Free	100/hr	score, trend, freshness, methodology_version
adw_silver_demo	Silver	10,000/hr	+ confidence, top_drivers, coverage, recommended_use
adw_gold_demo	Gold	100,000/hr	+ source_lineage, allowed_use, /history endpoint
adw_platinum_demo	Platinum	1,000,000/hr	everything + larger batch

Quick start

from alpinedataworks import Client

# Use demo key; switch base_url for local dev
client = Client(
    api_key="adw_gold_demo",
    base_url="http://localhost:8787/v1",   # default: https://api.alpinedataworks.ai/v1
)

# Public — no auth required
print(client.health())           # {"status": "ok", ...}
schema = client.schemas()        # IOM JSON Schema
result = client.validate(obj)    # {"valid": bool, "errors": [...]}

# Authenticated — list the catalog
products = client.list_products(domain="Crypto")
for p in products:
    print(p.product_id, p.tier, p.name)

# Fetch a single IOM (fields gated by your tier)
obj = client.get_product("ADW-001")
print(obj.score, obj.trend, obj.confidence)   # confidence=None on Free tier

# Alias for the canonical intelligence path
obj = client.get_intelligence("ADW-001")

# Batch (up to 20 IDs; up to 100 for Platinum)
batch = client.batch(["ADW-001", "ADW-002", "ADW-003"])

# Driver decomposition (Silver+ required)
drivers = client.drivers("ADW-001")
for d in drivers:
    print(d.factor, d.contribution)

# 30-day history (Gold+ required)
pts = client.history("ADW-001")   # list of {"date": ..., "score": ...}

# Rate-limit info from the last response
print(client.rate_limit)   # {"limit": "100000", "remaining": "99997", "reset": "..."}

Data types

IntelligenceObject

A @dataclass mapping the full IOM contract.

Field	Type	Tier
product_id	str	Free+
entity	str	Free+
score	float | str	Free+
trend	str | None	Free+
freshness	str	Free+
methodology_version	str	Free+
confidence	float | None	Silver+
top_drivers	list[Driver] | None	Silver+
prediction_horizon	str | None	Silver+
recommended_use	str | None	Silver+
coverage	str | None	Silver+
source_lineage	list[str] | None	Gold+
allowed_use	str | None	Gold+
raw	dict	always — full API payload

Driver

@dataclass
class Driver:
    factor: str           # e.g. "ETF Inflows"
    contribution: float | str   # weight or label

Product

Catalog summary returned by list_products(). Key fields: product_id, name, domain, type, tier, refresh.

Error handling

All errors inherit from AlpineAPIError.

from alpinedataworks import (
    AlpineAPIError,
    AuthError,           # HTTP 401 — bad or missing key
    TierForbiddenError,  # HTTP 403 — tier too low; .required_tier has the minimum
    NotFoundError,       # HTTP 404
    RateLimitError,      # HTTP 429; .retry_after has seconds to wait
    BadRequestError,     # HTTP 400
)

try:
    client.history("ADW-001")
except TierForbiddenError as e:
    print(e.status_code, e.code, e.required_tier)
except RateLimitError as e:
    import time
    time.sleep(e.retry_after or 60)
except AlpineAPIError as e:
    print(e.code, e.message, e.request_id)

Agent / LangChain integration

client.as_tool(product_id) returns a zero-argument callable that fetches the IOM and returns the raw dict. No LangChain dependency is added to the SDK — you just wrap it:

from langchain.tools import Tool
from alpinedataworks import Client

client = Client(api_key="adw_gold_demo")

adw_crypto = Tool.from_function(
    func=client.as_tool("ADW-001"),
    name="adw_crypto_sentiment",
    description=(
        "Returns the Alpine DataWorks Crypto Market Sentiment & Volatility Index. "
        "Use when you need a current crypto market signal."
    ),
)

# Pass to your agent:
# agent = initialize_agent(tools=[adw_crypto], ...)

For frameworks that accept plain callables (AutoGen, CrewAI, bare function-calling loops):

# Just use the callable directly
fn = client.as_tool("ADW-002")
macro_signal = fn()   # -> {"product_id": "ADW-002", "score": 68, ...}

Running tests locally

The test suite runs against a live local wrangler dev worker:

# 1. Install SDK in dev mode
cd ~/Documents/GitHub/adw-sdk-python
pip install -e ".[dev]"

# 2. Run the harness (starts wrangler, runs pytest, stops wrangler)
bash scripts/run_tests.sh

# Or manually:
cd ~/Documents/GitHub/adw-api && npx wrangler dev --local --port 8787 &
cd ~/Documents/GitHub/adw-sdk-python
ADW_BASE_URL=http://localhost:8787/v1 pytest tests/ -v

Project layout

adw-sdk-python/
├── src/
│   └── alpinedataworks/
│       ├── __init__.py    # public surface: Client, models, errors
│       ├── client.py      # Client class — all API methods
│       ├── models.py      # IntelligenceObject, Driver, Product dataclasses
│       └── errors.py      # error hierarchy + raise_for_status factory
├── tests/
│   ├── conftest.py        # fixtures (base_url, free/silver/gold/platinum clients)
│   └── test_sdk.py        # integration tests — all endpoints + error paths
├── examples/
│   └── quickstart.py      # runnable end-to-end walkthrough
├── scripts/
│   └── run_tests.sh       # test harness (start worker → pytest → stop)
├── pyproject.toml
└── README.md

License

MIT