Using Synthetic Data

• 1: Python SDK Installation
• 2: Python SDK Reference

1 - Python SDK Installation

Use this section to install, configure, and validate the Python SDK for Protegrity Synthetic Data.

Prerequisites

Ensure one of the following is met:

• pip is available in your active Python virtual environment.
• A conda environment is activated.

Overview

The Synthetic Data service can be accessed programmatically using the Python SDK.

1. Install the Synthetic Data Python SDK

Install the SDK from PyPI or conda using the version that matches the deployed Synthetic Data service.

Using pip:

pip install protegrity-synthetic-data-sdk==2.1.0

Using conda:

conda install -c protegrity protegrity-synthetic-data-sdk==2.1.0

2. Configure the SDK

Create a client configuration pointing to your PPC gateway. The gateway URL can be obtained using:

export GATEWAY_URL="https://$(kubectl get configmap/nfa-config -n default -o jsonpath='{.data.FQDN}')"

Then configure the client in Python:

from synthetic_data_sdk.config import ClientConfig

config = ClientConfig(
    base_url="<GATEWAY_URL>/pty/syntheticdata/v2",
    username="<USERNAME>",
    password="<PASSWORD>",
    verify_ssl=True,
)

Note: Replace default credentials and URLs for production environments. For example, for <USERNAME>, enter syntheticdata_admin and for <PASSWORD>, enter StrongPassword123!.

3. Test the Synthetic Data Python SDK

from synthetic_data_sdk import SyntheticDataClient
from synthetic_data_sdk.config import ClientConfig

config = ClientConfig(
    base_url="<GATEWAY_URL>/pty/syntheticdata/v2",
    username="<USERNAME>",
    password="<PASSWORD>",
    verify_ssl=True,
)

with SyntheticDataClient(config=config) as client:
    print("Connection established successfully.")

The connection is established successfully. If the connection fails, an error message is displayed.

2 - Python SDK Reference

Protegrity AI (Developer or TEAM) Editions - This section documents all classes and methods available in Protegrity Synthetic Data TEAMS and later. Methods marked as (TEAMS) are available only in the AI TEAM edition and above.

ClientConfig

Configuration dataclass for all Synthetic Data SDK clients.

from synthetic_data_sdk import ClientConfig

Attributes:

• endpoint (str) - Base URL of the Synthetic Data API (e.g. "<GATEWAY_URL>/pty/syntheticdata/v2").
• timeout (int, default 300) - Request timeout in seconds.
• max_retries (int, default 3) - Retry attempts on connection failure.
• verify_ssl (bool, default True) - Set to False for self-signed certificates.
• api_key (str | None, default None) - API key for authentication if required.
• headers (dict, default {}) - Additional HTTP headers for every request.

Single-Table Synthesizers

All single-table synthesizers share the same fit / transform / fit_transform / evaluate / summary interface, inherited from a shared base class.

RemoteVineCopula

Single-table vine copula synthesizer. Available in all tiers.

from synthetic_data_sdk import RemoteVineCopula

synth = RemoteVineCopula(
    endpoint="<GATEWAY_URL>/pty/syntheticdata/v2",
    categorical_cols=["city", "product"],
)

Constructor parameters:

• endpoint (str) - API endpoint URL. Not required when config is provided.
• model_version (str, optional) - Version identifier for model persistence. Reuse to avoid refitting.
• config (ClientConfig, optional) - Advanced client configuration.
• mlops_config (dict, optional) - Per-request MLOps tracking override.
• **parameters - Model-specific hyper-parameters (e.g. categorical_cols, vine_type).

RemoteSMOTE

SMOTE-based oversampling synthesizer. Useful for augmenting minority classes in imbalanced datasets. Available in all tiers.

from synthetic_data_sdk import RemoteSMOTE

synth = RemoteSMOTE(
    endpoint="<GATEWAY_URL>/pty/syntheticdata/v2",
    categorical_cols=["class"],
    k=5,
)

Constructor parameters: Same as RemoteVineCopula. Model-specific parameters: categorical_cols, k, noise_scale.

RemoteTabDiff (TEAMS)

Diffusion-based single-table synthesizer. Ideal for GPU-intensive synthesis without local GPU resources. Requires TEAMS.

from synthetic_data_sdk import RemoteTabDiff

synth = RemoteTabDiff(
    endpoint="<GATEWAY_URL>/pty/syntheticdata/v2",
    categorical_cols=["city", "product"],
    epochs=1000,
)

Constructor parameters: Same as RemoteVineCopula. Model-specific parameters: categorical_cols, epochs.

RemoteTabularGAN (TEAMS)

TabularGAN (CTABGAN architecture) synthesizer with mode-specific normalization for mixed continuous/categorical columns. Requires TEAMS.

from synthetic_data_sdk import RemoteTabularGAN

synth = RemoteTabularGAN(
    endpoint="<GATEWAY_URL>/pty/syntheticdata/v2",
    categorical_cols=["city", "product"],
    epochs=300,
)

Constructor parameters: Same as RemoteVineCopula. Model-specific parameters: categorical_cols, epochs.

Shared Methods (all single-table synthesizers)

fit

def fit(df: pd.DataFrame | str | Path) -> Self

Fit the model on training data. The fitted model is stored on the server using the configured model_version.

Parameters:

• df (DataFrame | str | Path) - Training data as a DataFrame, local file path, or cloud URI (s3://, gs://, azure://, minio://). Cloud URIs (s3://, gs://, azure://) require TEAMS.

Returns: Self (for method chaining).

Raises:

• SynthesisAPIError - If fitting fails.

transform

def transform(n: int) -> pd.DataFrame

Generate synthetic data using a fitted model.

Parameters:

• n (int) - Number of synthetic samples to generate.

Returns: DataFrame - Synthetic data with the same schema as the training data.

Raises:

• RuntimeError - If the model has not been fitted and no model_version exists on the server.
• SynthesisAPIError - If generation fails.

fit_transform

def fit_transform(df: pd.DataFrame, n: int) -> pd.DataFrame

Fit the model and generate synthetic data in a single call.

Parameters:

• df (DataFrame) - Training data.
• n (int) - Number of synthetic samples to generate.

Returns: DataFrame - Synthetic data.

evaluate

def evaluate(
    real_data: pd.DataFrame | str,
    synthetic_data: pd.DataFrame | str,
    categorical_cols: list[str] | None = None,
    target_col: str | None = None,
    task_type: str | None = None,
    eval_params: dict[str, Any] | None = None,
) -> dict[str, Any]

Evaluate synthetic data quality against the real data.

Parameters:

• real_data (DataFrame | str) - Real training data.
• synthetic_data (DataFrame | str) - Synthetic data to evaluate.
• categorical_cols (list[str], optional) - Categorical column names.
• target_col (str, optional) - Target column for TSTR/TRTR evaluation.
• task_type (str, optional) - "classification" or "regression" for predictive evaluation.
• eval_params (dict, optional) - Additional FidelityEvaluator configuration.

Returns: dict - Evaluation metrics. Basic metrics (distributional, column-level utility, memorization) are available in all tiers. Advanced metrics (privacy attacks, causal fidelity, certification, rare-population) require TEAMS.

summary

def summary() -> dict[str, Any]

Get summary statistics from a fitted model.

Returns: dict - Model summary with statistics and metadata.

Raises:

• RuntimeError - If the model has not been fitted.

transform_conditional (TEAMS - RemoteVineCopula only)

def transform_conditional(
    df: pd.DataFrame,
    n: int,
    conditions: dict[str, Any] | None = None,
    amplify_patterns: float | None = None,
    inject_drift: dict[str, float] | None = None,
    random_state: int | None = None,
) -> pd.DataFrame

Generate synthetic data matching specific conditional scenarios. Requires TEAMS (generation:conditional).

Parameters:

• df (DataFrame) - Training data to fit on.
• n (int) - Number of synthetic samples to generate.
• conditions (dict, optional) - Column filter conditions:
  • Exact match: {"fraud": 1, "status": "active"}
  • Comparison: {"age": ">65", "income": "<=50000"}
  • Range: {"age": "between(30,50)"}
  • Membership: {"city": "in(NYC,LA,Chicago)"}
• amplify_patterns (float, optional) - Multiplier for conditional pattern amplification (e.g. 1.5 for 50% increase).
• inject_drift (dict, optional) - Column drift shifts (e.g. {"income": -20000} for recession scenario).
• random_state (int, optional) - Random seed for reproducibility.

Returns: DataFrame - Synthetic data matching the specified conditions.

Multi-Table Synthesizer

RemoteMultiTableVineCopula

Multi-table vine copula synthesizer. Preserves foreign-key relationships across tables. Available in all tiers.

from synthetic_data_sdk import RemoteMultiTableVineCopula

synth = RemoteMultiTableVineCopula(
    endpoint="<GATEWAY_URL>/pty/syntheticdata/v2",
    relationships=[
        ("customers", "customer_id", "orders", "customer_id"),
        ("orders", "order_id", "items", "order_id"),
    ],
)
tables = {"customers": customers_df, "orders": orders_df, "items": items_df}
synth.fit(tables)
synthetic = synth.transform(n=500)

Exposes the same fit, transform, fit_transform, evaluate, summary, and validate_relationships workflow as RemoteVineCopula, but accepts and returns dict[str, DataFrame] instead of DataFrame.

Evaluation Clients (TEAMS)

The following evaluation clients require TEAMS and provide advanced quality and privacy assessments beyond the metrics returned by evaluate().

PrivacyEvaluator (TEAMS)

Evaluates privacy risks in synthetic data using membership inference attacks, sensitive attribute reconstruction, and linkage attack risk analysis.

from synthetic_data_sdk import PrivacyEvaluator

evaluator = PrivacyEvaluator(endpoint="<GATEWAY_URL>/pty/syntheticdata/v2")
results = evaluator.evaluate(
    train_real_data=train_df,
    test_real_data=test_df,
    synthetic_data=synthetic_df,
    sensitive_columns=["ssn", "salary", "diagnosis"],
)
print(f"Overall Risk: {results['overall_risk']}")

evaluate(train_real_data, test_real_data, synthetic_data, sensitive_columns, k_values, config) → dict

• train_real_data - Real training data used to generate synthetic data.
• test_real_data - Held-out real data (not seen during training).
• synthetic_data - Synthetic data to evaluate.
• sensitive_columns (list[str], optional) - Columns to test for attribute inference.
• k_values (list[int], optional) - K values for linkage attack evaluation.
• config (dict, optional) - Attack configuration (e.g. {"shadow_models": 10, "attack_model": "xgboost"}).

Returns: dict - Contains overall_risk, attacks list, and summary with successful_attacks.

CertificationClient (TEAMS)

Produces a comprehensive certification score (0–100) with letter grade (A+ to F) by aggregating fidelity, privacy, utility, and completeness metrics.

Score components: Fidelity 40% · Privacy 30% · Utility 20% · Completeness 10%.

from synthetic_data_sdk import CertificationClient

cert = CertificationClient(endpoint="<GATEWAY_URL>/pty/syntheticdata/v2")
result = cert.certify(
    real_data=real_df,
    synthetic_data=synthetic_df,
    categorical_cols=["city", "gender"],
    target_col="income",
    task_type="regression",
)
print(f"Grade: {result['grade']}  Score: {result['overall_score']:.1f}/100")

certify(real_data, synthetic_data, categorical_cols, target_col, task_type, include_privacy_attacks, ...) → dict

Returns: dict - Contains grade (A+ to F), overall_score (0–100), risk_level, summary, and recommendations.

CausalEvaluator (TEAMS)

Evaluates whether synthetic data preserves causal relationships, decision boundaries, and fairness properties from the original real data.

from synthetic_data_sdk import CausalEvaluator

evaluator = CausalEvaluator(endpoint="<GATEWAY_URL>/pty/syntheticdata/v2")
results = evaluator.evaluate(
    real_data=real_df,
    synthetic_data=synthetic_df,
    treatment_col="received_treatment",
    outcome_col="recovery_time",
    covariates=["age", "severity"],
)
print(f"Preservation Rate: {results['summary']['preservation_rate']:.1%}")

evaluate(real_data, synthetic_data, treatment_col, outcome_col, target_col, feature_cols, task_type, sensitive_attr, covariates) → dict

Returns: dict - Contains overall_preserved, evaluations list, and summary.

Low-Level Client

SynthesisClient

Low-level client for direct API interaction. Most users should prefer the high-level synthesizer classes above.

synthesize

def synthesize(
    model_name: str,
    action: str,
    training_data: str | None = None,
    training_data_path: str | None = None,
    training_data_tables: dict[str, str] | None = None,
    n_samples: int | None = None,
    model_version: str | None = None,
    parameters: dict[str, Any] | None = None,
    output_uri: str | None = None,
    mlops_config: dict[str, Any] | None = None,
) -> dict[str, Any]

Send a synthesis request (synchronous - waits for completion).

Parameters:

• model_name (str) - Model type. "vine", "vine_multitable", "smote" are available in all tiers. "tabdiff" and "tabulargan" require TEAMS.
• action (str) - Action: "fit", "transform", or "fit_transform".
• training_data (str, optional) - Base64-encoded CSV for single-table inline data.
• training_data_path (str, optional) - Cloud URI or local path. s3://, azure://, gcs:// require TEAMS; minio:// is available in all tiers.
• training_data_tables (dict[str, str], optional) - Table name → path/URI mapping for multi-table synthesis.
• n_samples (int, optional) - Number of synthetic samples to generate.
• model_version (str, optional) - Version identifier for model persistence.
• parameters (dict, optional) - Model-specific hyper-parameters.
• output_uri (str, optional) - Cloud URI for output. Same tier rules as training_data_path.
• mlops_config (dict, optional) - Per-request MLOps tracking configuration.

Returns: dict - API response with status, data, and metadata.

Raises:

• SynthesisAPIError - If the request fails.

synthesize_async

Same parameters as synthesize(). Returns immediately with {"job_id": "...", "status": "queued"}.

generate_conditional (TEAMS)

def generate_conditional(
    real_data: str | pd.DataFrame,
    model_name: str,
    n_samples: int,
    conditions: dict[str, Any] | None = None,
    amplify_patterns: float | None = None,
    inject_drift: dict[str, float] | None = None,
    categorical_cols: list[str] | None = None,
    random_state: int | None = None,
) -> dict[str, Any]

Generate conditional synthetic data via the low-level client. Requires TEAMS (generation:conditional).

Parameters:

• real_data (str | DataFrame) - Training data.
• model_name (str) - One of "vine", "smote", "tabdiff" (TEAMS), "tabulargan" (TEAMS).
• n_samples (int) - Samples to generate.
• conditions (dict, optional) - Column filter conditions (see transform_conditional).
• amplify_patterns (float, optional) - Pattern amplification multiplier.
• inject_drift (dict, optional) - Column drift shifts.
• categorical_cols (list[str], optional) - Categorical columns.
• random_state (int, optional) - Random seed.

Returns: dict - Contains success, n_samples, synthetic_data (base64 CSV), conditions_applied, drift_applied, warnings, and metadata.

Job Management

get_job_status

def get_job_status(job_id: str) -> dict[str, Any]

Get the current status of an asynchronous job.

Parameters:

• job_id (str) - Job identifier returned by synthesize_async().

Returns: dict - Contains job_id, status (pending, running, completed, failed, cancelled), progress, step, message, error, synth_data_uri, and timestamps.

wait_for_job

def wait_for_job(
    job_id: str,
    poll_interval: float = 2.0,
    timeout: float = 600.0,
    callback: Any | None = None,
) -> dict[str, Any]

Block until a job reaches a terminal state.

Parameters:

• job_id (str) - Job identifier.
• poll_interval (float, default 2.0) - Seconds between status polls.
• timeout (float, default 600.0) - Maximum seconds to wait.
• callback (callable, optional) - Called with the current status dict after each poll.

Returns: dict - Final job status.

Raises:

• TimeoutError - If the job does not complete within timeout.

list_jobs

def list_jobs(
    status: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> dict[str, Any]

List jobs with optional status filter and pagination.

Parameters:

• status (str, optional) - Filter by status: "pending", "running", "completed", "failed", "cancelled".
• limit (int, default 100) - Page size (1–1000).
• offset (int, default 0) - Page offset.

Returns: dict - Contains jobs list, total, limit, and offset.

get_job_history

def get_job_history(job_id: str) -> list[dict[str, Any]]

Get the full state-transition audit trail for a job.

Parameters:

• job_id (str) - Job identifier.

Returns: list[dict] - Each entry contains sequence, status, progress, step, and changed_at.

delete_job

def delete_job(job_id: str) -> None

Delete a job record, or cancel it if still running.

Parameters:

• job_id (str) - Job identifier.

Exceptions (SynthesisAPIError) are importable from synthetic_data_sdk.exceptions.