Composer Python SDK

A thin, dependency-light Python client for the Composer AI basket advisor. It wraps the same endpoints documented in the REST API reference: build a draft risk-sized basket from a theme + risk profile, fetch a saved draft, and find historical analogs. Composer returns draft recommendations only — no live trading, nothing bought or sold. Not investment advice.

Install

pip install requests

The SDK is a single file you can drop into your project — no package install required. It talks to the public API over HTTPS.

Quickstart

from composer import ComposerClient

client = ComposerClient(
    base_url="https://cymetica.com",
    email="you@example.com",
    password="…",
)

# Build a draft, risk-sized basket from a theme + risk profile (1-5 Likert)
draft = client.compose(
    theme="data center electricity",
    time_horizon_years=5,
    loss_capacity=3,
    loss_tolerance=3,
    knowledge=3,
    liquidity_need=3,
)
print("draft:", draft["public_ref"], "risk:", draft["risk_score"])
for leg in draft["assets"]:
    print(f"  {leg['symbol']:6} {leg['asset_class']:8} {leg['weight']:.0%}")
print(draft["legal_disclaimer"])

Fetch a Saved Draft

same = client.get_draft(draft["public_ref"])
print(same["rationale"])

Historical Analogs

# How did similar past events perform? (semantic headline-embedding match)
analogs = client.historical_analogs(draft["public_ref"], window_days=30, limit=8)
print(analogs)

Hand Off to MacroMarket (backtest the draft)

Composer is advisory; to backtest a draft as an AI Index Basket, post it to MacroMarket — see the MacroMarket SDK:

import requests
r = requests.post(
    f"https://cymetica.com/api/v1/macromarket/from-composer/{draft['public_ref']}",
    headers={"Authorization": f"Bearer {client.token}"},
    params={"period_days": 365}, timeout=60,
)
print(r.json())   # hypothetical backtest metrics — paper only

SDK Source (paste this file)

"""composer.py — minimal client for the EventTrader Composer basket advisor."""
import requests


class ComposerClient:
    def __init__(self, base_url: str, email: str | None = None, password: str | None = None):
        self.base = base_url.rstrip("/")
        self.s = requests.Session()
        self.token = None
        if email and password:
            r = self.s.post(f"{self.base}/auth/login",
                            json={"email": email, "password": password}, timeout=15)
            r.raise_for_status()
            self.token = r.json()["access_token"]
            self.s.headers.update({"Authorization": f"Bearer {self.token}"})

    def compose(self, theme: str, time_horizon_years: int, loss_capacity: int,
                loss_tolerance: int, knowledge: int, liquidity_need: int):
        r = self.s.post(f"{self.base}/api/v1/composer/instruments", json={
            "theme": theme,
            "time_horizon_years": time_horizon_years,
            "loss_capacity": loss_capacity,
            "loss_tolerance": loss_tolerance,
            "knowledge": knowledge,
            "liquidity_need": liquidity_need,
        }, timeout=60)
        r.raise_for_status()
        return r.json()

    def get_draft(self, public_ref: str):
        r = self.s.get(f"{self.base}/api/v1/composer/instruments/{public_ref}", timeout=15)
        r.raise_for_status()
        return r.json()

    def historical_analogs(self, public_ref: str, window_days: int = 30, limit: int = 8):
        r = self.s.post(
            f"{self.base}/api/v1/composer/instruments/{public_ref}/historical-analogs",
            json={"window_days": window_days, "limit": limit}, timeout=30)
        r.raise_for_status()
        return r.json()

Example: scan several themes by risk score

themes = ["data center electricity", "cures for cancer", "space industry"]
for t in themes:
    d = client.compose(t, 5, 3, 3, 3, 3)
    print(f"{t:28} risk={d['risk_score']} vol={d['projected_vol']} legs={len(d['assets'])}")