Example Scripts

The following example scripts demonstrate common CorePlexML SDK workflows. Each script is self-contained, uses argparse for configuration, and can be run directly against a CorePlexML instance.

All examples are located in the docs-site/examples/ directory.

01 – Quick Start

End-to-end ML workflow: create a project, upload data, train a classifier, deploy to staging, and make predictions.

python 01_quickstart.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --csv data.csv --target churned

  1"""CorePlexML Quick Start -- End-to-end ML workflow.
  2
  3Creates a project, uploads a dataset, trains a classification model,
  4deploys it to staging, makes predictions, and cleans up.
  5
  6Usage:
  7    python 01_quickstart.py --base-url https://your-domain.com --api-key YOUR_KEY --csv data.csv
  8"""
  9
 10import argparse
 11import sys
 12
 13from coreplexml import CorePlexMLClient, CorePlexMLError
 14
 15
 16def main():
 17    parser = argparse.ArgumentParser(description="CorePlexML Quick Start")
 18    parser.add_argument(
 19        "--base-url", default="http://localhost:8888", help="CorePlexML server URL"
 20    )
 21    parser.add_argument("--api-key", required=True, help="API key for authentication")
 22    parser.add_argument("--csv", required=True, help="Path to the training CSV file")
 23    parser.add_argument(
 24        "--target", default="target", help="Target column name (default: target)"
 25    )
 26    args = parser.parse_args()
 27
 28    client = CorePlexMLClient(base_url=args.base_url, api_key=args.api_key, timeout=120)
 29    print("Connected to CorePlexML")
 30
 31    try:
 32        # Step 1: Create a project
 33        project = client.projects.create(
 34            "Quick Start Project", description="SDK quick start demo"
 35        )
 36        project_id = project["id"]
 37        print(f"[1/8] Created project: {project_id}")
 38
 39        # Step 2: Upload the dataset
 40        ds = client.datasets.upload(project_id, args.csv, "Training Data")
 41        dataset_id = ds["id"]
 42        version_id = ds["version_id"]
 43        print(f"[2/8] Uploaded dataset: {dataset_id} (version {version_id})")
 44
 45        # Step 3: Inspect columns
 46        cols = client.datasets.columns(dataset_id)
 47        col_names = [c["name"] for c in cols["columns"]]
 48        print(f"[3/8] Detected {len(col_names)} columns: {', '.join(col_names[:5])}...")
 49
 50        # Step 4: Create an experiment
 51        exp = client.experiments.create(
 52            project_id=project_id,
 53            dataset_version_id=version_id,
 54            target_column=args.target,
 55            name="Quick Start Classifier",
 56            problem_type="classification",
 57            config={"max_models": 5, "max_runtime_secs": 120},
 58        )
 59        experiment_id = exp["id"]
 60        print(f"[4/8] Started experiment: {experiment_id}")
 61
 62        # Step 5: Wait for training
 63        print("       Waiting for training to complete...")
 64        result = client.experiments.wait(experiment_id, interval=5.0, timeout=1800.0)
 65        print(f"[5/8] Experiment status: {result['status']}")
 66
 67        if result["status"] != "succeeded":
 68            print(f"Training did not succeed: {result.get('error', 'unknown')}")
 69            sys.exit(1)
 70
 71        # Step 6: Get the best model
 72        models = client.models.list(experiment_id=experiment_id)
 73        best = models["items"][0]
 74        model_id = best["id"]
 75        print(f"[6/8] Best model: {best.get('algorithm', 'N/A')} ({model_id})")
 76
 77        # Step 7: Deploy to staging
 78        dep = client.deployments.create(
 79            project_id=project_id,
 80            model_id=model_id,
 81            name="Quick Start Deployment",
 82            stage="staging",
 83        )
 84        deployment_id = dep["id"]
 85        print(f"[7/8] Deployed to staging: {deployment_id}")
 86
 87        # Step 8: Make a prediction
 88        sample_input = {name: 0 for name in col_names if name != args.target}
 89        pred = client.deployments.predict(deployment_id, inputs=sample_input)
 90        print(f"[8/8] Prediction: {pred.get('prediction', pred)}")
 91
 92        print("\nQuick start complete!")
 93
 94        # Cleanup
 95        client.deployments.deactivate(deployment_id)
 96        client.projects.delete(project_id)
 97        print("Cleaned up resources.")
 98
 99    except CorePlexMLError as e:
100        print(f"Error ({e.status_code}): {e.message}")
101        sys.exit(1)
102
103
104if __name__ == "__main__":
105    main()

02 – Batch Predictions

Read rows from a CSV, send them in configurable batches to a deployment endpoint, and write predictions to an output file.

python 02_batch_predictions.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --deployment-id DEPLOY_UUID \
    --input new_data.csv --output predictions.csv --batch-size 500

 1"""Batch predictions with deployed models.
 2
 3Reads rows from a CSV file, sends them in batches to a deployment endpoint,
 4and writes predictions to an output CSV.
 5
 6Usage:
 7    python 02_batch_predictions.py --base-url https://your-domain.com \
 8        --api-key YOUR_KEY --deployment-id DEPLOY_UUID --input data.csv --output predictions.csv
 9"""
10
11import argparse
12import csv
13import sys
14
15from coreplexml import CorePlexMLClient, CorePlexMLError
16
17
18def chunked(items, size):
19    """Yield successive chunks of the given size."""
20    for i in range(0, len(items), size):
21        yield items[i : i + size]
22
23
24def main():
25    parser = argparse.ArgumentParser(description="Batch predictions")
26    parser.add_argument("--base-url", default="http://localhost:8888")
27    parser.add_argument("--api-key", required=True)
28    parser.add_argument("--deployment-id", required=True, help="UUID of the deployment")
29    parser.add_argument("--input", required=True, help="Path to input CSV")
30    parser.add_argument(
31        "--output", default="predictions.csv", help="Path to output CSV"
32    )
33    parser.add_argument(
34        "--batch-size", type=int, default=500, help="Rows per batch (default: 500)"
35    )
36    args = parser.parse_args()
37
38    client = CorePlexMLClient(base_url=args.base_url, api_key=args.api_key, timeout=120)
39
40    # Verify the deployment exists
41    try:
42        dep = client.deployments.get(args.deployment_id)
43        print(
44            f"Deployment: {dep.get('name', args.deployment_id)} (stage={dep.get('stage', 'unknown')})"
45        )
46    except CorePlexMLError as e:
47        print(f"Cannot access deployment: {e.message}")
48        sys.exit(1)
49
50    # Read input CSV
51    with open(args.input, newline="") as f:
52        reader = csv.DictReader(f)
53        rows = list(reader)
54    print(f"Loaded {len(rows)} rows from {args.input}")
55
56    if not rows:
57        print("No rows to predict.")
58        sys.exit(0)
59
60    # Run batch predictions
61    all_predictions = []
62    for i, batch in enumerate(chunked(rows, args.batch_size)):
63        try:
64            result = client.deployments.predict(args.deployment_id, inputs=batch)
65            preds = result.get("predictions", [])
66            all_predictions.extend(preds)
67            processed = min((i + 1) * args.batch_size, len(rows))
68            print(f"  Batch {i + 1}: {processed}/{len(rows)} rows processed")
69        except CorePlexMLError as e:
70            print(f"  Batch {i + 1} failed: {e.message}")
71            # Fill with error markers so row count stays aligned
72            all_predictions.extend([{"prediction": "ERROR"}] * len(batch))
73
74    # Write output CSV
75    fieldnames = list(rows[0].keys()) + ["prediction", "probability"]
76    with open(args.output, "w", newline="") as f:
77        writer = csv.DictWriter(f, fieldnames=fieldnames)
78        writer.writeheader()
79        for row, pred in zip(rows, all_predictions, strict=False):
80            row["prediction"] = pred.get("prediction", "")
81            row["probability"] = pred.get("probability", pred.get("probabilities", ""))
82            writer.writerow(row)
83
84    print(f"\nWrote {len(all_predictions)} predictions to {args.output}")
85
86
87if __name__ == "__main__":
88    main()

03 – Experiment Comparison

Run multiple experiments with different AutoML configurations on the same dataset, wait for all to complete, and print a ranked comparison.

python 03_experiment_comparison.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --project-id PROJ_UUID --version-id VER_UUID \
    --target churned --problem-type classification

  1"""Compare multiple experiments and find the best model.
  2
  3Runs several experiments with different configurations on the same dataset,
  4waits for all to complete, and prints a ranked comparison of the best model
  5from each experiment.
  6
  7Usage:
  8    python 03_experiment_comparison.py --base-url https://your-domain.com \
  9        --api-key YOUR_KEY --project-id PROJ_UUID --version-id VER_UUID --target churned
 10"""
 11
 12import argparse
 13import sys
 14
 15from coreplexml import CorePlexMLClient, CorePlexMLError
 16
 17EXPERIMENT_CONFIGS = [
 18    {"name": "Fast (5 models, 60s)", "max_models": 5, "max_runtime_secs": 60},
 19    {"name": "Standard (10 models, 180s)", "max_models": 10, "max_runtime_secs": 180},
 20    {"name": "Thorough (20 models, 300s)", "max_models": 20, "max_runtime_secs": 300},
 21]
 22
 23
 24def main():
 25    parser = argparse.ArgumentParser(description="Experiment comparison")
 26    parser.add_argument("--base-url", default="http://localhost:8888")
 27    parser.add_argument("--api-key", required=True)
 28    parser.add_argument("--project-id", required=True, help="UUID of the project")
 29    parser.add_argument(
 30        "--version-id", required=True, help="UUID of the dataset version"
 31    )
 32    parser.add_argument("--target", required=True, help="Target column name")
 33    parser.add_argument(
 34        "--problem-type",
 35        default="classification",
 36        choices=["classification", "regression"],
 37    )
 38    args = parser.parse_args()
 39
 40    client = CorePlexMLClient(base_url=args.base_url, api_key=args.api_key, timeout=60)
 41
 42    # Launch all experiments
 43    experiments = []
 44    for cfg in EXPERIMENT_CONFIGS:
 45        try:
 46            exp = client.experiments.create(
 47                project_id=args.project_id,
 48                dataset_version_id=args.version_id,
 49                target_column=args.target,
 50                name=cfg["name"],
 51                problem_type=args.problem_type,
 52                config={
 53                    "max_models": cfg["max_models"],
 54                    "max_runtime_secs": cfg["max_runtime_secs"],
 55                },
 56            )
 57            experiments.append({"id": exp["id"], "name": cfg["name"]})
 58            print(f"Started: {cfg['name']} ({exp['id']})")
 59        except CorePlexMLError as e:
 60            print(f"Failed to start '{cfg['name']}': {e.message}")
 61
 62    if not experiments:
 63        print("No experiments started.")
 64        sys.exit(1)
 65
 66    # Wait for all experiments
 67    print("\nWaiting for all experiments to complete...")
 68    for exp in experiments:
 69        try:
 70            status = client.experiments.wait(exp["id"], interval=10.0, timeout=3600.0)
 71            exp["status"] = status.get("status", "unknown")
 72            print(f"  {exp['name']}: {exp['status']}")
 73        except CorePlexMLError as e:
 74            exp["status"] = "timeout"
 75            print(f"  {exp['name']}: timed out ({e.message})")
 76
 77    # Collect best model from each successful experiment
 78    print("\n--- Results ---")
 79    best_overall = None
 80    metric_key = "auc" if args.problem_type == "classification" else "rmse"
 81
 82    for exp in experiments:
 83        if exp["status"] != "succeeded":
 84            print(f"  {exp['name']}: SKIPPED ({exp['status']})")
 85            continue
 86        models = client.models.list(experiment_id=exp["id"])
 87        if not models.get("items"):
 88            print(f"  {exp['name']}: No models produced")
 89            continue
 90        top = models["items"][0]
 91        metric_val = top.get("metrics", {}).get(metric_key, "N/A")
 92        print(f"  {exp['name']}: {top.get('algorithm', '?')} {metric_key}={metric_val}")
 93
 94        if best_overall is None:
 95            best_overall = (exp, top, metric_val)
 96        else:
 97            try:
 98                current_best = float(best_overall[2])
 99                candidate = float(metric_val)
100                # For AUC, higher is better; for RMSE, lower is better
101                if (
102                    args.problem_type == "classification"
103                    and candidate > current_best
104                    or args.problem_type == "regression"
105                    and candidate < current_best
106                ):
107                    best_overall = (exp, top, metric_val)
108            except (ValueError, TypeError):
109                pass
110
111    if best_overall:
112        exp_info, model_info, metric = best_overall
113        print(
114            f"\nBest overall: {model_info.get('algorithm')} from '{exp_info['name']}'"
115        )
116        print(f"  Model ID: {model_info['id']}")
117        print(f"  {metric_key}: {metric}")
118    else:
119        print("\nNo successful experiments to compare.")
120
121
122if __name__ == "__main__":
123    main()

04 – Privacy Workflow

Create a HIPAA compliance policy, scan a dataset for PII, apply privacy transformations (masking, hashing, redaction), and retrieve results.

python 04_privacy_workflow.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --project-id PROJ_UUID --dataset-id DS_UUID \
    --profile hipaa

05 – Synthetic Data Generation

Train a SynthGen model (CTGAN/CopulaGAN/TVAE/Gaussian Copula) on a dataset version, wait for training, and generate synthetic rows with optional reproducibility seed.

python 05_synthetic_data.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --project-id PROJ_UUID --version-id VER_UUID \
    --model-type ctgan --epochs 300 --num-rows 5000 --seed 42

06 – Model Monitoring

Deploy a model, promote to production, and check for data drift.

python 06_model_monitoring.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --project-id PROJ_UUID --model-id MODEL_UUID

07 – What-If Analysis

Create a studio session with baseline inputs, define counterfactual scenarios, run predictions, and compare results side by side.

python 07_what_if_analysis.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --project-id PROJ_UUID --deployment-id DEP_UUID

08 – Report Generation

Generate experiment and deployment reports with AI insights, wait for PDF generation, and download the results.

python 08_reports.py --base-url https://your-domain.com \
    --api-key YOUR_KEY --project-id PROJ_UUID \
    --experiment-id EXP_UUID --output-dir ./reports

Running the Examples

All examples follow the same pattern:

  1. Install the SDK:

    pip install coreplexml

  2. Set your API key and server URL. You can pass them as arguments or export them as environment variables and modify the scripts to read from os.environ.

  3. Run the script:

    python 01_quickstart.py --base-url https://ml.example.com --api-key cp_ab12cd34.your-secret-key --csv train.csv

  4. Each script prints progress to stdout and exits with code 0 on success or 1 on error.