Cookbook

Complete workflows you can copy and run top to bottom.

Each recipe runs end to end on its own: paste it into a notebook or script with your API key set and it works as written. Every parameter is documented in the API reference; each recipe links the entries it uses. New to the SDK? Start with the Quickstart.

Find a calibrated day, run a simulation and load the mid price

The basic Pro-tier loop: check which days are calibrated, submit a simulation for one, wait for it, and read the minute-by-minute mid price.

python
import time
from simudyne import PulseABM

client = PulseABM()  # reads SIMUDYNE_API_KEY from the environment

# 1. Which days of 700.HK are calibrated and ready to simulate?
rows = client.data.calibrated_data(symbol="700.HK", provider="omd")["symbols"]
dates = sorted(d["date"] for row in rows for d in row["available_dates"])
print(f"{len(dates)} calibrated days, latest {dates[-1]}")

cal_date = "2025-09-02"  # any date from the list works
assert cal_date in dates

# 2. Submit two Monte Carlo runs
job = client.simulation.run(
    symbol="700.HK",
    cal_date=cal_date,
    provider="omd",
    exchange="hkex_securities",
    n_runs=2,
)
job_id = job["job_id"]
print(f"Submitted {job_id}: {len(job['sim_ids'])} runs")

# 3. Wait for the job
while True:
    status = client.simulation.get_job_status(job_id)
    print(f"  {status['status_summary']}")
    if status["is_complete"]:
        break
    time.sleep(30)

if status["has_errors"]:
    raise RuntimeError(client.simulation.get_job_logs(job_id)[:2000])

# 4. Load the mid price of every run
for sim_id in job["sim_ids"]:
    mid = client.simulation.get_sim_data(sim_id, "mid_price_by_min.parquet")
    print(f"{sim_id.rsplit(':', 1)[-1]}: {len(mid)} minutes, "
          f"open {mid['mid_price'][0]:.2f}, close {mid['mid_price'][-1]:.2f}")

Prints the number of calibrated days, the job's progress every 30 seconds, then one line per run with its opening and closing mid price. Swap mid_price_by_min.parquet for sim_data.parquet to get the full tick-level book (see Output format).

Reference: data.calibrated_data, simulation.run, simulation.get_job_status, simulation.get_job_logs, simulation.get_sim_data

Download cached baselines and plot them (free tier)

No job to submit: use the shared library of pre-computed baselines, which every tier can read, and overlay their Monte Carlo paths by scenario.

python
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt

COLORS = {"normal": "steelblue", "flash_crash": "crimson"}

cached = client.simulation.list_cached(symbol="700.HK", date="2025-09-02")

# One group per scenario; keep the one with the most runs
groups = {}
for g in cached["simulations"]:
    if g["scenario"] in COLORS and g["n_runs"] > groups.get(g["scenario"], {}).get("n_runs", 0):
        groups[g["scenario"]] = g

fig, ax = plt.subplots(figsize=(14, 5))
for scenario, g in groups.items():
    # example_sim_id is run 0000; swap the last field for the other runs
    base = g["example_sim_id"].rsplit(":", 1)[0]
    for i in range(min(g["n_runs"], 3)):
        df = client.simulation.get_sim_data(f"{base}:{i:04d}", "mid_price_by_min.parquet")
        ax.plot(df["time"].to_list(), df["mid_price"].to_list(),
                color=COLORS[scenario], linewidth=0.7, alpha=0.5,
                label=scenario if i == 0 else "_nolegend_")
    print(f"{scenario}: plotted {min(g['n_runs'], 3)} of {g['n_runs']} runs")

ax.set_title("700.HK 2025-09-02: mid price by minute")
ax.set_ylabel("Mid price")
ax.legend(title="Scenario")
fig.autofmt_xdate()
fig.tight_layout()
fig.savefig("mc_paths_700HK_20250902.png", dpi=150)
print("saved mc_paths_700HK_20250902.png")

Writes mc_paths_700HK_20250902.png. On the free tier each scenario group you download counts once against your daily allowance; a group you have downloaded before stays free (see Account).

Reference: simulation.list_cached, simulation.get_sim_data, profile.downloads

Run a scenario with an execution algorithm and compare to normal

Measure what a stressed market does to an execution: run the same TWAP under normal and flash_crash on the same calibrated day, then compare the price path and the algorithm's costs.

python
import time

twap = {
    "type": "twap",
    "order_size": 50,        # lots (700.HK lot size is 100 shares); positive = sell
    "horizon": 7200,         # seconds
    "start_time": "10:00:00",
    "frequency": 30,         # seconds between child orders
}

jobs = {}
for scenario in ["normal", "flash_crash"]:
    jobs[scenario] = client.simulation.run(
        symbol="700.HK",
        cal_date="2025-09-02",
        provider="omd",
        exchange="hkex_securities",
        n_runs=2,
        scenario=scenario,
        scenario_params={"start_time": "11:00:00"} if scenario == "flash_crash" else None,
        exec_algos=[twap],
    )
    print(f"{scenario}: {jobs[scenario]['job_id']}")

for scenario, job in jobs.items():
    while not client.simulation.get_job_status(job["job_id"])["is_complete"]:
        time.sleep(30)

for scenario, job in jobs.items():
    # The second-to-last sim_id field names the arm: "baseline" or the algo
    algo_runs = [s for s in job["sim_ids"] if s.split(":")[-2] != "baseline"]
    lows, slippage, impact = [], [], []
    for sim_id in algo_runs:
        mid = client.simulation.get_sim_data(sim_id, "mid_price_by_min.parquet")
        lows.append(mid["mid_price"].min())
        costs = client.simulation.get_sim_data(sim_id, "exec_results.parquet").row(0, named=True)
        slippage.append(costs["market_slippage"])
        impact.append(costs["market_impact"])
    n = len(algo_runs)
    print(f"{scenario:12s} low {sum(lows) / n:8.2f}  "
          f"slippage {sum(slippage) / n:.4f}  impact {sum(impact) / n:.4f}")

Prints both job ids, then one line per scenario: the average intraday low of the mid price and the average slippage and impact of the TWAP. Every key in scenario_params and exec_algos is listed in the reference.

Reference: simulation.run, simulation.get_job_status, simulation.get_sim_data

Compare TWAP execution windows against one baseline

Several algorithms in one exec_algos list share a single baseline, so their costs are compared like for like. Here: the same TWAP over three windows, with the results downloaded in parallel.

python
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime

windows = [("09:30:00", "11:30:00"), ("13:00:00", "16:00:00"), ("09:30:00", "16:00:00")]

def seconds(start, end):
    fmt = "%H:%M:%S"
    return int((datetime.strptime(end, fmt) - datetime.strptime(start, fmt)).total_seconds())

job = client.simulation.run(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities",
    n_runs=2,
    exec_algos=[
        {"type": "twap", "order_size": 50, "horizon": seconds(s, e),
         "frequency": 30, "start_time": s}
        for s, e in windows
    ],
)
job_id = job["job_id"]

while not client.simulation.get_job_status(job_id)["is_complete"]:
    time.sleep(30)

# Group the algo runs by arm (second-to-last sim_id field)
arms = defaultdict(list)
for sim_id in job["sim_ids"]:
    if sim_id.split(":")[-2] != "baseline":
        arms[sim_id.split(":")[-2]].append(sim_id)

def costs(sim_id):
    return client.simulation.get_sim_data(sim_id, "exec_results.parquet").row(0, named=True)

with ThreadPoolExecutor(max_workers=8) as pool:
    for arm, sim_ids in arms.items():
        rows = list(pool.map(costs, sim_ids))
        print(f"{arm}: slippage {sum(r['market_slippage'] for r in rows) / len(rows):.4f}, "
              f"impact {sum(r['market_impact'] for r in rows) / len(rows):.4f}")

Prints one line per window with its mean slippage and impact. The runs of one job come back baseline first, then each algorithm in the order given, but grouping by the arm field of the sim_id does not depend on that order.

Reference: simulation.run, simulation.get_job_status, simulation.get_sim_data

Build CSS schedules and compare them

When TWAP and VWAP are not the shape you want, a CSS (custom static schedule) algorithm sends exactly the orders you list. Build the schedule as a pd.Series indexed by full datetimes, then compare several against one baseline.

python
import time
from datetime import date, datetime
from datetime import time as clock

import numpy as np
import pandas as pd

def make_schedule(total_lots, start_t, end_t, cal_date, clip=1):
    """Spread total_lots evenly across a window, clip lots per order."""
    n_orders = int(np.ceil(total_lots / clip))
    idx = pd.date_range(datetime.combine(cal_date, start_t),
                        datetime.combine(cal_date, end_t), periods=n_orders)
    q, r = divmod(total_lots, n_orders)
    sizes = np.full(n_orders, q, dtype=int)
    sizes[:r] += 1  # spread the remainder so the total is exact
    return pd.Series(sizes, index=idx)

cal_date = date(2025, 9, 2)
windows = [(clock(9, 30), clock(12, 0)), (clock(13, 0), clock(16, 0))]
schedules = [make_schedule(100, s, e, cal_date) for s, e in windows]  # 100 lots each
print(schedules[0].head(3))
print(f"orders per schedule: {[len(s) for s in schedules]}, lots: {[int(s.sum()) for s in schedules]}")

job = client.simulation.run(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities",
    n_runs=2,
    exec_algos=[{"type": "css", "orders": s} for s in schedules],
)
job_id = job["job_id"]

while not client.simulation.get_job_status(job_id)["is_complete"]:
    time.sleep(30)

for sim_id in job["sim_ids"]:
    arm = sim_id.split(":")[-2]
    if arm != "baseline":
        row = client.simulation.get_sim_data(sim_id, "exec_results.parquet").row(0, named=True)
        print(f"{arm} run {sim_id[-4:]}: slippage {row['market_slippage']:.4f}")
Output
2025-09-02 09:30:00.000000000    1
2025-09-02 09:31:30.909090909    1
2025-09-02 09:33:01.818181818    1
dtype: int64
orders per schedule: [100, 100], lots: [100, 100]
...

Sizes are in lots, as for TWAP and VWAP. Orders outside continuous trading (HKEX 09:30–12:00 and 13:00–16:00) are dropped.

Reference: simulation.run, simulation.get_job_status, simulation.get_sim_data

Validate simulations against the real day

Score simulated order books against the real trading day with Pulse-Check: distributional distances, stylised facts, impact response and the inception distances. This uses cached baseline runs, so there is nothing to simulate first; pass your own job's sim_ids the same way.

python
import time

group = client.simulation.list_cached(symbol="700.HK", date="2025-09-02", scenario="normal")["simulations"][0]
base = group["example_sim_id"].rsplit(":", 1)[0]
sim_ids = [f"{base}:{i:04d}" for i in range(min(group["n_runs"], 3))]

job_id = client.validation.run(
    symbol="700.HK",
    date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    sim_ids=sim_ids,
)["job_id"]
print(f"validation job {job_id}")

while not client.validation.get_job_status(job_id)["is_complete"]:
    time.sleep(15)

result = client.validation.get_job(job_id)
print(f"status: {result['status']}")
if result["status"] == "failed":
    raise RuntimeError(result["error"])
if result.get("errors"):
    print(f"areas that failed: {result['errors']}")

# distances: {metric: {"l1": [per run], "w": [per run]}}; null where a run failed
for metric, d in sorted((result.get("distances") or {}).items()):
    w = [x for x in d.get("w") or [] if x is not None]
    if w:
        print(f"  {metric:30s} mean Wasserstein {sum(w) / len(w):.4g}")
print(f"FID: {result.get('fid_scores')}")
print(f"MIND: {result.get('mind_scores')}")

Prints the validation job id, its final status, then one line per metric and the FID / MIND scores per run. Lower is closer to the real day. What each metric means is on the Pulse-Check page.

Reference: simulation.list_cached, validation.run, validation.get_job_status, validation.get_job

Validate against a stretch of the day (match_window)

Use this when your runs cover only part of the day, or when you care about one stretch of it. By default (match_window=False) every run is scored against the whole historical day; match_window=True cuts the historical day to the runs' own span, and ["10:00", "10:30"] cuts both sides to that interval.

python
import time

group = client.simulation.list_cached(symbol="700.HK", date="2025-09-02", scenario="normal")["simulations"][0]
base = group["example_sim_id"].rsplit(":", 1)[0]
sim_ids = [f"{base}:{i:04d}" for i in range(min(group["n_runs"], 2))]

windows = {
    "runs' own span": True,               # historical day cut to the runs' span
    "10:00-10:30": ["10:00", "10:30"],    # both sides cut to this interval
}

jobs = {}
for label, match_window in windows.items():
    jobs[label] = client.validation.run(
        symbol="700.HK",
        date="2025-09-02",
        provider="omd",
        exchange="hkex_securities",
        sim_ids=sim_ids,
        match_window=match_window,
        impact=False,  # the slowest area; skip it while iterating
    )["job_id"]

for label, job_id in jobs.items():
    while not client.validation.get_job_status(job_id)["is_complete"]:
        time.sleep(15)
    result = client.validation.get_job(job_id)
    window = (result.get("metadata") or {}).get("historical_window") or {}
    print(f"{label}: {result['status']}")
    print(f"  mode={window.get('mode')} {window.get('start')}-{window.get('end')} "
          f"historical rows {window.get('hist_rows_before')} -> {window.get('hist_rows_after')}")

metadata["historical_window"] reports what was applied: mode ("off", "simulation" or "interval"), the start and end used, and the row counts before and after the cut. An interval holding no rows of the day or of a run fails the job with the reason in error; an unreadable one is rejected at submission with a 422.

Reference: validation.run, validation.get_job_status, validation.get_job

Compare two populations in one validation

Pass a mapping of named groups rather than a list and one job compares the populations, against the real day and each other, instead of pooling every run together. Here: normal against flash crash.

python
import time

def runs(scenario, n=2):
    group = client.simulation.list_cached(symbol="700.HK", date="2025-09-02", scenario=scenario)["simulations"][0]
    base = group["example_sim_id"].rsplit(":", 1)[0]
    return [f"{base}:{i:04d}" for i in range(min(group["n_runs"], n))]

job_id = client.validation.run(
    symbol="700.HK",
    date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    sim_ids={"normal": runs("normal"), "flash_crash": runs("flash_crash")},
    impact=False,
    plots=["statistical.radar", "stylised_facts.overall"],
)["job_id"]

while not client.validation.get_job_status(job_id)["is_complete"]:
    time.sleep(15)

result = client.validation.get_job(job_id, plot_dir="figures")
print(f"status: {result['status']}")
print(f"FID by population:  {result.get('fid_scores')}")
print(f"MIND by population: {result.get('mind_scores')}")
print(f"figures: {result.get('plot_paths')}")

Scores come back keyed by population name, and the radar and verdict figures draw one series per population, written to figures/. Up to 25 runs in total per job.

Reference: simulation.list_cached, validation.run, validation.get_job

Validate your own parquet files

Score a model of your own: upload parquet files in pulse format with sim_files, optionally beside platform runs so both land on one set of figures. The historical day is still fetched for you.

python
import time
from pathlib import Path

# Your model's output, one pulse-format parquet per run
files = sorted(Path("my_sims").glob("run_*.parquet"))

# Platform runs to compare against
group = client.simulation.list_cached(symbol="700.HK", date="2025-09-02", scenario="normal")["simulations"][0]
base = group["example_sim_id"].rsplit(":", 1)[0]

job_id = client.validation.run(
    symbol="700.HK",
    date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    sim_files={"mine": files},
    sim_ids={"platform": [f"{base}:0000", f"{base}:0001"]},
)["job_id"]

while not client.validation.get_job_status(job_id)["is_complete"]:
    time.sleep(15)

result = client.validation.get_job(job_id)
print(result["status"], result.get("fid_scores"))

sim_files also takes polars or pandas DataFrames, or (filename, bytes) pairs. If your frames are L2 snapshots rather than an event stream, add lob=True; areas that need the message stream are then skipped and listed in metadata["areas_skipped"].

Reference: validation.run, validation.get_job

Foundation-model rollout, then score it

Generate a short stretch of order-book activity with a foundation model, then score it against the same minutes of the real day. Without match_window=True a five-minute rollout would be compared with the whole day, opening and close included.

python
import time

# Models differ by environment, so take one this key can run rather than
# naming it. Prompt it with the busiest instrument in its market, on that
# instrument's latest calibrated day: a quiet name can have too few rows to start.
model = client.fm.models()["models"][0]
market = (model["supported_data"] or [{}])[0]
inst = max(
    client.data.calibrated_data(provider=market.get("provider"), exchange=market.get("exchange"))["symbols"],
    key=lambda s: s["avg_number_of_msgs"],
)
day = {"symbol": inst["symbol"], "provider": inst["provider"], "exchange": inst["exchange"],
       "dates": [d["date"] for d in inst["available_dates"]]}

job = client.simulation.run(
    engine="fm",
    model_id=model["model_id"],
    symbol=day["symbol"],
    cal_date=day["dates"][-1],
    provider=day["provider"],
    exchange=day["exchange"],
    duration_minutes=5,
    n_runs=2,
)
while not (status := client.simulation.get_job_status(job["job_id"]))["is_complete"]:
    print(f"{status.get('status', 'queued')}: {status.get('message', '')}")
    time.sleep(30)

val_id = client.validation.run(
    symbol=day["symbol"],
    date=day["dates"][-1],
    provider=day["provider"],
    exchange=day["exchange"],
    sim_ids=job["sim_ids"],
    match_window=True,  # score against the same minutes of the real day
)["job_id"]

while not client.validation.get_job_status(val_id)["is_complete"]:
    time.sleep(15)

result = client.validation.get_job(val_id)
meta = result.get("metadata") or {}
print(result["status"], meta.get("historical_window"))
print(f"prompt rows dropped: {meta.get('historical_rows_dropped')}")
print(f"skipped: {meta.get('areas_skipped')}")

Validation drops the prompt context rows automatically, so the scores describe generated frames only. Foundation-model output has no order flow, so the areas that need it are skipped, not failed. See Foundation models.

Foundation-model inference runs on GPU and costs real compute; keep duration_minutes and n_runs small while iterating.

Reference: fm.models, simulation.run, simulation.get_job_status, validation.run, validation.get_job

Liquidity risk across a ladder of order sizes

Estimate market impact as a function of order size: one call runs an execution algorithm per size against a single shared baseline.

python
import time
from collections import defaultdict

job = client.simulation.run_lrm(
    symbol="700.HK",
    cal_date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    order_sizes=[100, 200, 400],  # lots, one arm each
    strategy="twap",
    side="buy",
    horizon_mins=30,
    n_runs=20,                    # 20 * (1 + 3) = 80 simulations
)
job_id = job["job_id"]  # run_lrm returns no sim_ids; read them from the results

while not client.simulation.get_job_status(job_id)["is_complete"]:
    time.sleep(30)

impact = defaultdict(list)
for sim in client.simulation.get_job_results(job_id)["simulations"]:
    arm = sim["sim_id"].split(":")[-2]
    if arm != "baseline" and "exec_results.parquet" in sim["available_files"]:
        row = client.simulation.get_sim_data(sim["sim_id"], "exec_results.parquet").row(0, named=True)
        impact[arm].append(row["market_impact"])

for arm, values in impact.items():
    print(f"{arm}: mean impact {sum(values) / len(values):.4f} over {len(values)} runs")

Prints one line per order size. The cost is n_runs × (1 + len(order_sizes)) simulations, so a ladder is far cheaper than one baseline per size, but still large.

Reference: simulation.run_lrm, simulation.get_job_status, simulation.get_job_results, simulation.get_sim_data

Bulk download several runs as a ZIP

Fetch many runs in one request (up to 100 ids) and read every parquet straight from the ZIP, without writing anything to disk.

python
import io
import zipfile

import polars as pl

group = client.simulation.list_cached(symbol="700.HK", date="2025-09-02", scenario="normal")["simulations"][0]
base = group["example_sim_id"].rsplit(":", 1)[0]
sim_ids = [f"{base}:{i:04d}" for i in range(min(group["n_runs"], 3))]

zip_bytes = client.simulation.get_bulk_data(
    sim_ids=sim_ids,
    include_sim_data=False,
    include_mid_price=True,
)

frames = {}
with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
    for name in zf.namelist():
        if name.endswith(".parquet"):
            frames[name] = pl.read_parquet(io.BytesIO(zf.read(name)))

print(f"Loaded {len(frames)} files")
for name, df in frames.items():
    print(f"  {name}: {df.shape}")

Prints one line per file, laid out one directory per simulation. Set include_sim_data=True for the full books, or include_l2_by_second=True for the per-second L2 book. To keep the archive, write zip_bytes to a file instead.

Reference: simulation.list_cached, simulation.get_bulk_data

Triage a failed job

has_errors says a simulation failed; each run's error_message usually says why, and the engine log covers the rest. This is also the output to attach when you report a problem.

python
jobs = client.simulation.get_jobs(limit=1)["jobs"]
if not jobs:
    raise RuntimeError("no jobs yet")

latest = jobs[0]["job_id"]
status = client.simulation.get_job_status(latest)

if status["has_errors"]:
    for sim in status["simulations"]:
        if sim["status"] in ("error", "failed"):
            print(f"{sim['sim_id']}: {sim['error_message']}")
    print(client.simulation.get_job_logs(latest)[:2000])
else:
    print(f"{latest}: {status['status_summary']}")

Prints the failing runs and the first 2,000 characters of the log, or the status summary for a healthy job.

A job rejected before reaching a worker has no log, and get_job_logs() raises PulseAPIError with status 404.

Reference: simulation.get_jobs, simulation.get_job_status, simulation.get_job_logs

Handle API errors

Every non-2xx response raises PulseAPIError with the status, the server's detail and, for a 422, the field-level errors. Separate “fix the request” from “wrong tier” from “try later”; network failures and 429/502/503/504 are already retried for you.

python
from simudyne import PulseAPIError

def submit(**kwargs):
    try:
        return client.simulation.run(**kwargs)
    except PulseAPIError as exc:
        if exc.status_code == 422:      # the request is wrong: fix and resubmit
            print(f"rejected: {exc.detail}")
            for err in exc.errors or []:
                print(f"  {err}")
        elif exc.status_code == 403:    # your tier cannot do this
            print(f"needs a higher tier: {exc.detail}")
        elif exc.status_code >= 500:    # still failing after the SDK's retries
            print(f"service unavailable, try later: {exc.detail}")
        else:
            raise
        return None

# A scenario_params key that does not exist is a 422 naming the field
result = submit(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities",
    n_runs=2, scenario="flash_crash",
    scenario_params={"not_a_param": 1},
)
print(f"submitted: {result is not None}")

# A sim_id that does not exist is a 404
try:
    client.simulation.get_sim_params("not-a-real-sim-id")
except PulseAPIError as exc:
    print(f"{exc.status_code}: {exc.detail}")

Prints the 422 detail naming not_a_param, submitted: False, then the 404. The full status-code table and retry settings are on Errors and retries.

A validation job that cannot load its data does not raise: it ends with status: "failed" and the reason in error. Check the status after polling, as the validation recipes above do.

Reference: simulation.run, simulation.get_sim_params