Run Synthesis on Your Data

import os
from radmah_sdk import RadMahClient

client = RadMahClient(
    api_key=os.environ["RADMAH_API_KEY"],
    base_url=os.environ.get("RADMAH_API_URL", "https://api.radmah.ai"),
)

dataset = client.upload_dataset(
    "customers_2026.csv",
    description="Owned training dataset for Synthesis",
)
print(dataset.id, dataset.rows, dataset.columns_json)

train_quote = client.estimate_cost(
    kind="train",
    dataset_id=str(dataset.id),
    mode="quality",
    compute="auto",
    epochs=2_400,
    batch_size=512,
    learning_rate=0.001,
)

training_job = client.submit_job_with_budget(
    kind="train",
    dataset_id=str(dataset.id),
    seed=42,
    mode="quality",
    compute="auto",
    options={
        "epochs": 2_400,
        "batch_size": 512,
        "learning_rate": 0.001,
    },
    max_credits=int(train_quote["credits_required"]),
).wait(timeout=1_800)

model_sha256 = (training_job.result_summary or {}).get(
    "checkpoint_artifact_sha256"
)
if training_job.status != "succeeded" or not model_sha256:
    raise RuntimeError("Reusable Synthesis model was not produced")

from radmah_sdk import (
    SynthesisNumericConstraint,
    SynthesisReleasePolicy,
)

release_policy = SynthesisReleasePolicy(
    minimum_overall_fidelity=0.75,
    minimum_column_fidelity=0.75,
    minimum_bivariate_fidelity=0.60,
    maximum_membership_advantage=0.20,
    maximum_linkage_risk=0.05,
    require_zero_exact_copies=True,
    # Optional, off by default: refuse delivery if a delivered text value
    # is shaped like a live identifier (see "Live-identifier control").
    block_live_identifiers=False,
)

quote = client.estimate_cost(
    kind="synthesize",
    dataset_id=str(dataset.id),
    rows=10_000,
    mode="quality",
    compute="auto",
    epochs=2_400,
)

job = client.submit_job_with_budget(
    kind="synthesize",
    dataset_id=str(dataset.id),
    rows=10_000,
    seed=42,
    mode="quality",
    compute="auto",
    options={
        "epochs": 2_400,
        "batch_size": 512,
        "learning_rate": 0.001,
    },
    numeric_constraints=[
        SynthesisNumericConstraint(
            column="customer_age",
            minimum=18.0,
            maximum=90.0,
        ),
    ],
    release_policy=release_policy,
    # Optional: a successful owned training-job id from the exact dataset.
    # checkpoint_source_job_id=str(training_job.id),
    max_credits=int(quote["credits_required"]),
)
print(job.id, job.status, job.mode)

# Quote, submit, wait, and download the primary customer artifact
rady synthesize \
  --dataset "$DATASET_ID" \
  --rows 10000 \
  --mode quality \
  --epochs 2400 \
  --batch-size 512 \
  --learning-rate 0.001 \
  --seed 42 \
  --compute auto \
  --minimum-overall-fidelity 75 \
  --minimum-relationship-fidelity 60 \
  --maximum-membership-advantage 20 \
  --maximum-linkage-indicator 5 \
  --output synthesis.csv

# Operational lifecycle
rady jobs get "$JOB_ID"
rady jobs cancel "$JOB_ID"
rady jobs rerun "$JOB_ID"

# Any mode. Counts training-epoch time only (not data preparation,
# calibration or generation) and is checked after each completed epoch, so
# every model trains at least one epoch and may run past the cap by up to one
# epoch of each training phase. Not valid with checkpoint_source_job_id. A
# capped run's stop point depends on host speed; its saved model is the
# replay authority.
job = client.submit_job_with_budget(
    kind="synthesize",
    dataset_id=str(dataset.id),
    rows=10_000,
    seed=42,
    mode="automatic",
    compute="auto",
    options={"max_training_minutes": 30},
    release_policy=release_policy,
    max_credits=int(quote["credits_required"]),
).wait(timeout=3_600)

print(job.training_stop_reason)   # e.g. "training_time_cap_reached"
print(job.training_time_cap)      # requested, reached, measured seconds
print(job.training_resume)        # resumed after a worker loss, and from where

import json
from pathlib import Path
from radmah_sdk.verify import verify_evidence_bundle

artifacts = client.list_artifacts(job.id)
primary = [a for a in artifacts if a.artifact_type == "primary"]
if len(primary) != 1:
    raise RuntimeError(f"expected one primary artifact, received {len(primary)}")

csv_bytes = client.download_artifact_bytes(job.id, primary[0].id)
Path("synthesis.csv").write_bytes(csv_bytes)

quality = client.get_utility_report(job.id)
privacy = client.get_privacy_report(job.id)
bundle_bytes = client.download_evidence_bundle(job.id)
Path("evidence_bundle.json").write_bytes(bundle_bytes)

verification = verify_evidence_bundle(json.loads(bundle_bytes))
if not verification.ok:
    raise RuntimeError(verification.reason)

print(quality)
print(privacy)
print(verification.kind.value, verification.integrity)

listing = client.list_table_exports(job.id)
for table in listing["tables"]:
    print(table["name"], table["rows"], table["formats"]["xlsx"])

client.export_table(job.id, listing["tables"][0]["name"], "table.csv")
workbook = client.export_table(job.id, listing["tables"][0]["name"], "table.xlsx", format="xlsx")
print(workbook.source_sha256)  # the sealed CSV this workbook came from

client.export_tables_zip(job.id, "tables.zip", format="csv")