Pulse-Check

Validate simulation quality against historical market data.

Pulse-Check is the validation API for Pulse. It scores how closely simulated order book data matches real historical market data for the same symbol and date, using distributional distance metrics (L1 and Wasserstein) and impact response analysis. It is a separate service from the native Pulse simulation API: validation jobs run under /validation and are accessed via client.validation in the Python SDK.

Pulse-Check is available on the Pro tier only. All validation endpoints require a Pro API key — see Plans.

How it works

Submit a validation job with run() — returns a job_id immediately
Historical data for the symbol and date is fetched automatically; simulation data comes from your sim_ids (or uploaded parquet files)
Poll with get_job(job_id), or use run_pipeline() to block until done
Read results: per-metric distances

Validate simulation runs

The most common flow: run simulations with the native Pulse API, then pass the resulting sim_ids to Pulse-Check. Validation compares each run's sim_data.parquet against historical data for the same symbol and date.

python

import time

# 1. Run simulations with the native Pulse API
sim = client.simulation.run(
    symbol="700.HK",
    cal_date="2025-09-02",
    provider="omd",
    exchange="hkex_securities",
    n_runs=5
)
while True:
    status = client.simulation.get_job_status(sim["job_id"])
    if status["is_complete"] or status.get("has_errors"):
        break
    time.sleep(30)

# 2. Submit a Pulse-Check validation job for those runs
validation_job = client.validation.run(
    symbol="700.HK",
    date="2025-09-02",
    sim_ids=sim["queued_sim_ids"],
)
validation_job_id = validation_job["job_id"]
print(f"Validation job submitted: {validation_job_id}")

Parameters

Parameter	Type	Default	Description
symbol	str	required	Trading symbol (e.g. 700.HK)
date	str	required	Date in YYYY-MM-DD format — historical data for this date is the benchmark
sim_ids	list[str]	required	Simulation IDs to validate (1–25)
ticksize	float	1.0	Tick size for the symbol
run_metrics	bool	True	Compute L1/Wasserstein distributional distances
run_impact	bool	False	Compute impact response curves
n_levels	int	10	Number of L2 book levels to use
rescale_volumes	bool	True	Multiply simulated L2 size columns by lot_size
lot_size	int	1	Lot size multiplier for volume rescaling

The date must have completed historical data processing — use the same calibration dates that work for simulations.

run_impact=True is significantly more expensive than the default metrics — for liquid symbols it can take well over the default 600s timeout. Raise timeout substantially (or poll manually with get_job()) if you enable it.

Run and wait in one call

run_pipeline() submits the job and polls until it completes, returning the full result. It accepts the same parameters as run() plus poll_interval (default 3s) and timeout (default 600s).

python

import time

# Run simulations first, then validate those runs
sim = client.simulation.run(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities", n_runs=5,
)
sim_ids = sim["queued_sim_ids"]
while True:
    status = client.simulation.get_job_status(sim["job_id"])
    if status["is_complete"] or status.get("has_errors"):
        break
    time.sleep(30)

result = client.validation.run_pipeline(
    symbol="700.HK",
    date="2025-09-02",
    sim_ids=sim_ids,
)

print(result["status"])  # "completed"

Check job status

Validation jobs move through: pending → running → completed (or failed).

python

import time

# Submit simulations and an async validation job to poll
sim = client.simulation.run(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities", n_runs=5,
)
while True:
    status = client.simulation.get_job_status(sim["job_id"])
    if status["is_complete"] or status.get("has_errors"):
        break
    time.sleep(30)
validation_job_id = client.validation.run(
    symbol="700.HK", date="2025-09-02", sim_ids=sim["queued_sim_ids"],
)["job_id"]

while True:
    result = client.validation.get_job(validation_job_id)
    print(f"Status: {result['status']}")
    if result["status"] in ("completed", "failed"):
        break
    time.sleep(5)

if result["status"] == "failed":
    print(f"Error: {result['error']}")

Result fields

Field	Type	Description
status	str	pending, running, completed, or failed
distances	dict	Per-metric distances: {metric: {"l1": [...], "w": [...]}} — one value per simulation run
metadata	dict	Run parameters used for the job
error	str	Error message (when failed)
created_at / completed_at	str	Job timestamps

Validation metrics

With run_metrics=True, Pulse-Check computes stylised-fact distributions on both the historical and simulated order books, then scores the gap between them with L1 and Wasserstein distances. Lower is better. This distributional approach to scoring generative order book models follows the methodology of LOB-Bench (Nagy et al.). Metrics include:

Group	Metrics
Spread & touch	spread, touch_bid_volume, touch_ask_volume
Order flow	obi, ofi (plus up/down/stay and rolling variants), log_interarrivaltime
Volume	total_bid_volume, total_ask_volume, volume_per_minute
Order placement	ask/bid_limit_depth, ask/bid_limit_level
Cancellations	time_to_cancel, ask/bid_cancel_depth, ask/bid_cancel_level

python

import time

# Run simulations and validate them to get a completed result
sim = client.simulation.run(
    symbol="700.HK", cal_date="2025-09-02",
    provider="omd", exchange="hkex_securities", n_runs=5,
)
while True:
    status = client.simulation.get_job_status(sim["job_id"])
    if status["is_complete"] or status.get("has_errors"):
        break
    time.sleep(30)
result = client.validation.run_pipeline(
    symbol="700.HK", date="2025-09-02",
    sim_ids=sim["queued_sim_ids"],
)

# Mean distance per metric across all runs
import numpy as np

for metric, scores in result["distances"].items():
    w = np.mean(scores["w"])
    l1 = np.mean(scores["l1"])
    print(f"{metric:24s}  wasserstein={w:.4f}  l1={l1:.4f}")

View past jobs

list_jobs() returns your validation jobs (newest first, without the full results payload). Use get_job() to fetch results for a specific job.

python

jobs = client.validation.list_jobs(limit=50)  # max 200

print(f"You have {jobs['total']} validation jobs")
for job in jobs["jobs"]:
    params = job["params"]
    print(f"{job['job_id']}  {job['status']}")
    print(f"  {params['symbol']} {params['date']} — {len(params.get('sim_ids') or [])} sims")
    print(f"  Created: {job['created_at']}")

Validate uploaded files

You can also validate simulation output that isn't stored in Pulse — for example parquets downloaded from another environment — by uploading sim_data.parquet files directly to the /validation/run/upload endpoint. The SDK session handles authentication:

python

from pathlib import Path

# Stage files to upload. Here we download 5 runs of a cached simulation —
# in practice these are sim_data parquets you produced elsewhere.
Path("my_sims").mkdir(exist_ok=True)
cached = client.simulation.list_cached(symbol="700.HK", scenario="normal")["simulations"][0]
base = cached["example_sim_id"].rsplit(":", 1)[0]
for i in range(min(cached["n_runs"], 5)):
    # list_cached only guarantees sim_data.parquet for example_sim_id itself —
    # sibling runs in the same batch may still be processing or missing files.
    try:
        df = client.simulation.get_sim_data(f"{base}:{i:04d}")
    except Exception:
        continue
    df.write_parquet(f"my_sims/sim_{i:04d}.parquet")

parquet_paths = sorted(Path("my_sims").glob("sim_*.parquet"))

files = [
    ("sim_files", (p.name, p.read_bytes(), "application/octet-stream"))
    for p in parquet_paths
]
data = {
    # provider/exchange identify the historical data the sims are validated
    # against (pulse-format/{provider}_{exchange}/...) — required for uploads.
    "provider": "omd",
    "exchange": "hkex_securities",
    "symbol": "700.HK",
    "date": "2025-09-02",
    "ticksize": "1.0",
    # Optional: JSON string of config overrides
    # "config": '{"run_impact": false, "n_levels": 10}',
}

response = client.session.post(
    f"{client.base_url}/validation/run/upload",
    files=files,
    data=data,
)
response.raise_for_status()
validation_job_id = response.json()["job_id"]

# Poll as usual
result = client.validation.get_job(validation_job_id)

Maximum 25 simulation files (or sim_ids) per validation job.

REST endpoints

For non-Python integrations, Pulse-Check is plain HTTP. Authenticate with your Pro API key in the X-API-Key header.

Method	Path	Description
POST	/validation/run	Submit a job from sim_ids (JSON body: symbol, date, sim_ids, ticksize, config)
POST	/validation/run/upload	Submit a job from uploaded parquet files (multipart form)
GET	/validation/jobs	List your validation jobs (?limit=, default 50, max 200)
GET	/validation/jobs/{job_id}	Get job status and results

MCP integration→