In‑Depth Guide: Building a High‑Fidelity Synthetic Data Pipeline with CTGAN and SDV

1️⃣ Installation & Environment Setup

Begin by creating an isolated Python environment. The following pip command pulls the exact versions required for reproducibility:

pip install "ctgan" "sdv" "sdmetrics" "scikit-learn" "pandas" "numpy" "matplotlib"

UBOS’s Workflow automation studio can orchestrate this step across cloud VMs, guaranteeing consistent runtime across teams.

2️⃣ Data Preparation

Load your raw dataset, clean column names, and separate categorical from numerical features. A typical snippet looks like this:

import pandas as pd
real = pd.read_csv("customer_data.csv")
real.columns = [c.strip().replace(" ", "_") for c in real.columns]
categorical_cols = real.select_dtypes(include=["object"]).columns.tolist()
numerical_cols   = [c for c in real.columns if c not in categorical_cols]

For startups that need rapid onboarding, the UBOS for startups page offers a one‑click data‑prep template.

3️⃣ Training a Standalone CTGAN Model

Instantiate CTGAN with a modest number of epochs for a quick proof‑of‑concept, then fit it on the cleaned dataframe:

from ctgan import CTGAN
ctgan = CTGAN(epochs=30, batch_size=500, verbose=True)
ctgan.fit(real, discrete_columns=categorical_cols)

After training, generate a synthetic sample to sanity‑check column distributions:

synthetic = ctgan.sample(5000)
print(synthetic.head())

The AI marketing agents can automatically draft a data‑quality summary from this output.

4️⃣ Enforcing Business Rules with SDV Constraints

SDV lets you declare semantic metadata and structural constraints. For example, enforce that age >= 18 and that country and currency appear as valid pairs:

from sdv.metadata import SingleTableMetadata
from sdv.cag import Inequality, FixedCombinations

metadata = SingleTableMetadata()
metadata.detect_from_dataframe(real)
metadata.update_column(column_name="age", sdtype="numerical")
metadata.update_column(column_name="country", sdtype="categorical")
metadata.update_column(column_name="currency", sdtype="categorical")

constraints = [
    Inequality(low_column_name="age", high_column_name="age", low=18),
    FixedCombinations(column_names=["country", "currency"])
]

These constraints are attached to the synthesiser before training:

from sdv.single_table import CTGANSynthesizer
synth = CTGANSynthesizer(metadata=metadata, epochs=30, batch_size=500)
synth.add_constraints(constraints)
synth.fit(real)

SMBs can explore the UBOS solutions for SMBs to manage constraint libraries centrally.

5️⃣ Visualising Generator & Discriminator Losses

SDV exposes a loss dataframe that can be plotted with matplotlib. The following code automatically picks the appropriate column names, making the script robust to library updates:

import matplotlib.pyplot as plt
loss_df = synth.get_loss_values()
xcol = next((c for c in ["epoch","step","iteration"] if c in loss_df.columns), None)
gcol = next((c for c in ["generator_loss","gen_loss"] if c in loss_df.columns), None)
dcol = next((c for c in ["discriminator_loss","disc_loss"] if c in loss_df.columns), None)

plt.figure(figsize=(10,4))
x = loss_df[xcol] if xcol else range(len(loss_df))
if gcol: plt.plot(x, loss_df[gcol], label="Generator")
if dcol: plt.plot(x, loss_df[dcol], label="Discriminator")
plt.xlabel(xcol or "step")
plt.ylabel("Loss")
plt.legend()
plt.title("CTGAN Training Dynamics (SDV Wrapper)")
plt.show()

Embedding this chart in a Web app editor on UBOS lets data scientists share live training dashboards with stakeholders.

6️⃣ Conditional Sampling for Targeted Scenarios

Suppose you need synthetic records where region = "EMEA". SDV’s Condition API makes this trivial:

from sdv.sampling import Condition
condition = Condition({"region": "EMEA"}, num_rows=2000)
synthetic_cond = synth.sample_from_conditions([condition])
print(synthetic_cond["region"].value_counts())

Such focused generation is ideal for privacy‑preserving A/B tests, and the ChatGPT and Telegram integration can push the sampled slice directly to a secure Slack‑like channel for rapid review.

7️⃣ Rigorous Evaluation Using SDMetrics

SDMetrics provides two complementary reports: DiagnosticReport (distributional fidelity) and QualityReport (utility for downstream models). Example usage:

from sdmetrics.reports.single_table import DiagnosticReport, QualityReport
diagnostic = DiagnosticReport()
diagnostic.generate(real_data=real, synthetic_data=synthetic_cond, metadata=metadata.to_dict())
print("Diagnostic score:", diagnostic.get_score())

quality = QualityReport()
quality.generate(real_data=real, synthetic_data=synthetic_cond, metadata=metadata.to_dict())
print("Quality score:", quality.get_score())

High scores (>0.8) indicate that the synthetic data preserves both statistical properties and predictive power, a prerequisite for compliance‑driven data sharing.

8️⃣ Testing Downstream ML Models on Synthetic Data

Train a simple classifier on synthetic data and evaluate it on a real hold‑out set. This “train‑synthetic → test‑real” experiment quantifies utility loss:

from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score

train_real, test_real = train_test_split(real, test_size=0.25, random_state=42, stratify=real["target"])

preprocess = ColumnTransformer([
    ("cat", OneHotEncoder(handle_unknown="ignore"), categorical_cols),
    ("num", "passthrough", numerical_cols)
])

clf = Pipeline([("prep", preprocess), ("logreg", LogisticRegression(max_iter=200))])
clf.fit(synthetic_cond.drop(columns=["target"]), synthetic_cond["target"])
pred_syn = clf.predict_proba(test_real.drop(columns=["target"]))[:,1]
auc_syn = roc_auc_score(test_real["target"], pred_syn)
print("Synthetic‑train → Real‑test AUC:", auc_syn)

If the AUC gap is minimal, you have a production‑ready synthetic dataset ready for model‑as‑a‑service deployments.

9️⃣ Persisting the Synthesiser for Future Use

Save the trained synthesiser to disk and reload it later without retraining:

model_path = "ctgan_sdv.pkl"
synth.save(model_path)

from sdv.utils import load_synthesizer
loaded = load_synthesizer(model_path)
sample = loaded.sample(1000)
print(sample.head())

The UBOS partner program offers managed model‑registry services that store these artefacts securely and expose them via REST APIs.