Cookbook

Ready-to-run snippets for common Pulse tasks.

Each recipe below is a self-contained snippet you can drop straight into a notebook or script. Swap in your own symbols, dates, and parameters as needed. All examples assume you have the SDK installed and an API key set — see Quickstart if not.

Simulations

Run, wait, and download in one blockPro

Submit a job, block until it finishes, then pull down the first simulation’s data — the most common Pro-tier pattern.

python

import time
from simudyne import PulseABM

client = PulseABM()  # reads SIMUDYNE_API_KEY from environment

result = client.simulation.run(
    symbol="700.HK",
    cal_date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    n_runs=5,
)
job_id  = result["job_id"]
sim_ids = result["sim_ids"]
print(f"Submitted {job_id} — {len(sim_ids)} runs queued")

while True:
    status = client.simulation.get_job_status(job_id)
    done   = status["status_summary"].get("complete", 0)
    print(f"\r  {done}/{len(sim_ids)} complete", end="", flush=True)
    if status["is_complete"] or status.get("has_errors"):
        break
    time.sleep(30)

df = client.simulation.get_sim_data(sim_ids[0])
print(df.head())

Scenarios

Compare normal vs flash crash on the same datePro

Submit both scenarios concurrently and collect their job IDs, then poll them together before downloading results. The shared cal_date ensures the background market microstructure is identical across scenarios.

python

import time

scenarios = ["normal", "flash_crash"]
jobs: dict[str, str] = {}

for scenario in scenarios:
    result = client.simulation.run(
        symbol="700.HK",
        cal_date="2025-09-02",
        provider="omd",
        exchange="hkex_securities",
        n_runs=5,
        scenario=scenario,
    )
    jobs[scenario] = result["job_id"]
    print(f"Submitted {scenario}: {result['job_id']}")

# Wait for all jobs
for scenario, job_id in jobs.items():
    while True:
        status = client.simulation.get_job_status(job_id)
        if status["is_complete"] or status.get("has_errors"):
            print(f"{scenario} done")
            break
        time.sleep(30)

Execution algorithms

Build a CSS execution schedulePro

css takes an explicit pd.Series indexed by a full datetime — date and time combined. Sizes are in lots, the same unit as order_size for TWAP and VWAP, not shares: 100 lots of 700.HK (lot size 100) is 10,000 shares. Spread the remainder across the first orders so the total is exact. Orders falling outside HKEX continuous trading hours (09:30–12:00, 13:00–16:00) are silently dropped.

python

import numpy as np
import pandas as pd
from datetime import date, datetime

def make_schedule(total_lots, start_t, end_t, cal_date, target_clip=1):
    """Distribute total_lots evenly across a datetime window, target_clip lots per order."""
    start_dt = datetime.combine(cal_date, start_t)
    end_dt   = datetime.combine(cal_date, end_t)

    n_orders = int(np.ceil(total_lots / target_clip))
    idx = pd.date_range(start=start_dt, end=end_dt, periods=n_orders)

    # Spread remainder across first orders so total is exact
    q, r = divmod(total_lots, n_orders)
    sizes = np.full(n_orders, q, dtype=int)
    sizes[:r] += 1

    return pd.Series(sizes, index=idx)

cal_date = date(2025, 9, 2)
schedule = make_schedule(
    total_lots=100,  # 10,000 shares of 700.HK
    start_t=datetime.strptime("09:30", "%H:%M").time(),
    end_t=datetime.strptime("12:00", "%H:%M").time(),
    cal_date=cal_date,
)

print(schedule)
print(f"Orders: {len(schedule)}, Total lots: {int(schedule.sum())}")

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
2025-09-02 09:34:32.727272727    1
2025-09-02 09:36:03.636363636    1
                                ..
2025-09-02 11:53:56.363636363    1
2025-09-02 11:55:27.272727272    1
2025-09-02 11:56:58.181818181    1
2025-09-02 11:58:29.090909090    1
2025-09-02 12:00:00.000000000    1
Length: 100, dtype: int64
Orders: 100, Total lots: 100

Compare TWAP across multiple execution windowsPro

Pass multiple algo configs in a single exec_algos list. Each strategy runs against the same shared baseline, so cost comparisons are apples-to-apples.

python

import time
from datetime import datetime

windows = [
    ("09:30:00", "11:30:00"),  # morning only
    ("09:30:00", "16:00:00"),  # full day
    ("13:00:00", "16:00:00"),  # afternoon only
]

def window_seconds(start, end):
    """Wall-clock length of a window; time outside trading hours is skipped."""
    fmt = "%H:%M:%S"
    return int((datetime.strptime(end, fmt) - datetime.strptime(start, fmt)).total_seconds())

exec_algos = [
    {
        "type": "twap",
        "order_size": 50,                       # lots
        "horizon": window_seconds(start, end),  # this window's length in seconds
        "frequency": 30,                        # seconds between child orders
        "start_time": start,
    }
    for start, end in windows
]

result = client.simulation.run(
    symbol="700.HK",
    cal_date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    n_runs=5,
    exec_algos=exec_algos,
)
job_id = result["job_id"]

# sim_ids layout: baseline[0:n_runs], strategy_0, strategy_1, ...
sim_ids = result["sim_ids"]
baseline_ids = sim_ids[:5]
strategy_ids  = [sim_ids[5 + i * 5 : 10 + i * 5] for i in range(len(windows))]

while True:
    s = client.simulation.get_job_status(result["job_id"])
    if s["is_complete"] or s.get("has_errors"):
        break
    time.sleep(30)

# Flat list of all algo sim IDs — used by the metrics recipe below
algo_sim_ids = [sid for group in strategy_ids for sid in group]

# Lookup map and metadata — used by the threadpool download recipe below
strategies = [{"start": start, "end": end} for start, end in windows]
algo_sim_map = {
    (strat_idx, run_idx): sid
    for strat_idx, group in enumerate(strategy_ids)
    for run_idx, sid in enumerate(group)
}

Compare CSS schedules against one baselinePro

Pass several css configs in a single exec_algos list and every strategy is measured against the same shared baseline. Each gets its own set of Monte Carlo runs inside the one job, so the comparison is like-for-like.

python

import numpy as np
import pandas as pd
from datetime import date, datetime, time

def make_schedule(total_lots, start_t, end_t, cal_date, target_clip=1):
    """Distribute total_lots evenly across a datetime window, target_clip lots per order."""
    start_dt = datetime.combine(cal_date, start_t)
    end_dt   = datetime.combine(cal_date, end_t)
    n_orders = int(np.ceil(total_lots / target_clip))
    idx = pd.date_range(start=start_dt, end=end_dt, periods=n_orders)
    q, r = divmod(total_lots, n_orders)
    sizes = np.full(n_orders, q, dtype=int)
    sizes[:r] += 1
    return pd.Series(sizes, index=idx)

cal_date = date(2025, 9, 2)

# Build a grid of execution windows to compare
windows = [
    (time(9, 30),  time(12, 0)),   # full morning
    (time(9, 30),  time(16, 0)),   # full day
    (time(13, 0),  time(16, 0)),   # full afternoon
]

exec_algos = [
    {"type": "css", "orders": make_schedule(100, s, e, cal_date)}  # 100 lots each
    for s, e in windows
]

result = client.simulation.run(
    symbol="700.HK",
    cal_date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    n_runs=5,
    exec_algos=exec_algos,
)
job_id = result["job_id"]

# sim_ids are ordered: baseline[0:n_runs], strategy_0[n_runs:2*n_runs], ...
sim_ids = result["sim_ids"]

Retrieve execution quality metricsPro

Once a job with execution algorithms is complete, exec_results.parquet contains per-run slippage and impact metrics. Aggregate across runs for a summary.

python

import time
import polars as pl

from datetime import datetime

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

# Submit a multi-window TWAP job to produce algo runs to analyse
windows = [("09:30:00", "11:30:00"), ("13:00:00", "16:00:00")]
result = client.simulation.run(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities", n_runs=5,
    exec_algos=[{"type": "twap", "order_size": 50,  # lots
                 "horizon": window_seconds(s, e),  # this window's length
                 "frequency": 30, "start_time": s}
                for s, e in windows],
)
while True:
    status = client.simulation.get_job_status(result["job_id"])
    if status["is_complete"] or status.get("has_errors"):
        break
    time.sleep(30)

sim_ids = result["sim_ids"]
strategy_ids = [sim_ids[5 + i * 5 : 10 + i * 5] for i in range(len(windows))]
algo_sim_ids = [sid for group in strategy_ids for sid in group]

rows = []
for sim_id in algo_sim_ids:
    df = client.simulation.get_sim_data(sim_id, "exec_results.parquet")
    rows.append(df.to_pandas().iloc[0].to_dict() | {"sim_id": sim_id})

results_df = pl.DataFrame(rows)
print(results_df.select(["sim_id", "market_slippage", "market_impact"]).describe())

Working with results

List cached baselines and plot Monte Carlo paths by scenarioFree

Use list_cached() to discover available pre-computed baselines, then download and overlay every run coloured by scenario. No job submission required.

python

import matplotlib.pyplot as plt
from simudyne import PulseABM

client = PulseABM()  # reads SIMUDYNE_API_KEY from environment

COLORS  = {"normal": "steelblue", "flash_crash": "crimson"}
LABELS  = {"normal": "Normal",    "flash_crash": "Flash Crash"}

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

# one entry per scenario — pick the variant with the most runs
by_scenario = {}
for s in cached["simulations"]:
    sc = s["scenario"]
    if sc not in by_scenario or s["n_runs"] > by_scenario[sc]["n_runs"]:
        by_scenario[sc] = s

fig, ax = plt.subplots(figsize=(14, 5))

for sim in by_scenario.values():
    scenario = sim["scenario"]
    if scenario not in LABELS:
        continue
    base  = sim["example_sim_id"].rsplit(":", 1)[0]
    first = True
    n_runs = min(sim["n_runs"], 5)
    for i in range(n_runs):
        print(f"\r  {LABELS[scenario]}  {i + 1}/{n_runs}", end="", flush=True)
        try:
            df = client.simulation.get_sim_data(f"{base}:{i:04d}", "mid_price_by_min.parquet")
        except Exception:
            continue
        ax.plot(df["time"].to_list(), df["mid_price"].to_list(),
                color=COLORS[scenario], linewidth=0.6, alpha=0.35,
                label=LABELS[scenario] if first else "_nolegend_")
        first = False
    print()

ax.set_title("700.HK  2025-09-02 — Monte Carlo paths (mid price by minute)")
ax.set_xlabel("Time")
ax.set_ylabel("Mid Price")
ax.legend(title="Scenario")
fig.autofmt_xdate()
plt.tight_layout()
plt.savefig("mc_paths_700HK_20250902.png", dpi=150)
plt.show()

Bulk download and read without saving to diskFree

Pull a set of simulations as a ZIP archive and read every parquet file directly into Polars — no intermediate files written.

python

import io, zipfile, polars as pl

cached = client.simulation.list_cached(symbol="700.HK", scenario="normal")
sim = cached["simulations"][0]

# example_sim_id is only run 0000 — expand it to the first 5 Monte Carlo
# runs (the run index is the final field). get_bulk_data accepts up to 100 ids.
base = sim["example_sim_id"].rsplit(":", 1)[0]
sim_ids = [f"{base}:{i:04d}" for i in range(min(sim["n_runs"], 5))]

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

frames: dict[str, pl.DataFrame] = {}
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}")

Triage a failed jobPro

has_errors tells you a simulation failed; the per-simulation error_message usually tells you why, and the engine log is for when it does not. This is also the output to attach if you send the problem to us.

python

job_id = client.simulation.get_jobs(limit=1)["jobs"][0]["job_id"]
status = client.simulation.get_job_status(job_id)

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

The log is written once a job reaches a worker, so a job rejected before that has none and get_job_logs() raises PulseAPIError with status 404.

Threadpooled download of execution resultsPro

For faster download across a large set of simulation runs, use a thread pool to fetch files in parallel. This is especially useful when collecting exec_results.parquet across many strategy variants.

python

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from datetime import datetime

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

# Submit a multi-window TWAP job to produce algo runs to download
windows = [("09:30:00", "11:30:00"), ("13:00:00", "16:00:00")]
result = client.simulation.run(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities", n_runs=5,
    exec_algos=[{"type": "twap", "order_size": 50,  # lots
                 "horizon": window_seconds(s, e),  # this window's length
                 "frequency": 30, "start_time": s}
                for s, e in windows],
)
while True:
    status = client.simulation.get_job_status(result["job_id"])
    if status["is_complete"] or status.get("has_errors"):
        break
    time.sleep(30)

sim_ids = result["sim_ids"]
strategy_ids = [sim_ids[5 + i * 5 : 10 + i * 5] for i in range(len(windows))]
strategies = [{"start": s, "end": e} for s, e in windows]
algo_sim_map = {(si, ri): sid
                for si, group in enumerate(strategy_ids)
                for ri, sid in enumerate(group)}

FILE_TO_DOWNLOAD = "exec_results.parquet"
DOWNLOAD_WORKERS = 8

def fetch_file(sim_id):
    """Download exec_results.parquet for a single sim_id."""
    try:
        df = client.simulation.get_sim_data(sim_id, FILE_TO_DOWNLOAD)
        return df.to_pandas().iloc[0].to_dict()
    except Exception as exc:
        return {"error": f"download_failed: {exc}"}

print(f"Downloading {len(algo_sim_map)} algo sims...")
all_rows = []
done_count = 0

with ThreadPoolExecutor(max_workers=DOWNLOAD_WORKERS) as pool:
    futures = {}
    for (strat_idx, run_idx), sid in algo_sim_map.items():
        futures[pool.submit(fetch_file, sid)] = (strat_idx, run_idx, sid)

    for future in as_completed(futures):
        strat_idx, run_idx, sid = futures[future]
        row = future.result()
        row["sim_id"] = sid
        row["strategy_index"] = strat_idx
        row["run_index"] = run_idx
        row["start"] = str(strategies[strat_idx]["start"])
        row["end"] = str(strategies[strat_idx]["end"])
        all_rows.append(row)

        done_count += 1
        print(f"\r  {done_count}/{len(algo_sim_map)}", end="", flush=True)