1ctgan/
 2├── __init__.py              # 入口：匯出 CTGAN, TVAE, load_demo
 3├── data_transformer.py      # DataTransformer：Bayesian GMM + OneHot
 4├── data_sampler.py           # DataSampler：條件向量取樣邏輯
 5├── synthesizers/
 6│   ├── base.py               # BaseSynthesizer：random state 管理、序列化
 7│   ├── ctgan.py              # CTGAN：Generator + Discriminator + 訓練迴圈
 8│   └── tvae.py               # TVAE：Encoder + Decoder + ELBO loss
 9├── data.py                   # CLI 資料讀取（CSV/TSV）
10├── demo.py                   # 內建 demo 資料集（Adult Census）
11└── errors.py                 # 自訂例外

1loss_d = -(torch.mean(y_real) - torch.mean(y_fake))  # Wasserstein distance
2pen = discriminator.calc_gradient_penalty(...)         # λ=10 gradient penalty

1Input: z (128) + cond (C)
2→ Residual(128+C, 256) → dim = 128+C+256
3→ Residual(128+C+256, 256) → dim = 128+C+256+256
4→ Linear(dim, data_dim) → output

1Input: data_dim × pac (10)
2→ Linear → LeakyReLU(0.2) → Dropout(0.5)
3→ Linear → LeakyReLU(0.2) → Dropout(0.5)
4→ Linear → 1 (real/fake score)

1pip install ctgan

1conda install -c pytorch -c conda-forge ctgan

1pip install sdv

1git clone https://github.com/sdv-dev/CTGAN.git
2cd CTGAN
3
4# 使用 uv（推薦）
5uv venv .venv && source .venv/bin/activate
6uv pip install -e ".[dev]"
7
8# 或使用 pip
9pip install -e ".[dev]"

1# 強制使用 CPU
2ctgan = CTGAN(enable_gpu=False)
3
4# 使用 CUDA GPU（預設行為）
5ctgan = CTGAN(enable_gpu=True)

 1from ctgan import CTGAN, load_demo
 2
 3# 載入 Adult Census 資料集（32,561 筆、15 欄位）
 4real_data = load_demo()
 5
 6# 定義離散欄位名稱（CTGAN 必須明確知道哪些欄位是離散的）
 7discrete_columns = [
 8    'workclass',
 9    'education',
10    'marital-status',
11    'occupation',
12    'relationship',
13    'race',
14    'sex',
15    'native-country',
16    'income',
17]
18
19# 初始化並訓練 CTGAN
20ctgan = CTGAN(
21    embedding_dim=128,       # 噪聲向量維度
22    generator_dim=(256, 256),  # Generator 殘差層大小
23    discriminator_dim=(256, 256),  # Discriminator 層大小
24    batch_size=500,
25    epochs=50,               # 生產環境建議 300+
26    verbose=True,            # 顯示訓練進度
27)
28
29ctgan.fit(real_data, discrete_columns)
30
31# 生成 1000 筆合成資料
32synthetic_data = ctgan.sample(1000)
33
34# 驗證基本統計特性
35print("=== 原始資料 ===")
36print(real_data.describe())
37print("\n=== 合成資料 ===")
38print(synthetic_data.describe())
39
40# 檢查離散欄位分佈是否保留
41for col in discrete_columns:
42    print(f"\n--- {col} ---")
43    print("原始:", real_data[col].value_counts(normalize=True).head(3).to_dict())
44    print("合成:", synthetic_data[col].value_counts(normalize=True).head(3).to_dict())

 1import pandas as pd
 2import numpy as np
 3from ctgan import CTGAN
 4
 5# 模擬一個藥物反應資料集（類似臨床試驗 CRF 資料）
 6np.random.seed(42)
 7n_patients = 500
 8
 9clinical_data = pd.DataFrame({
10    'age': np.random.normal(55, 12, n_patients).clip(18, 85).astype(int),
11    'weight_kg': np.random.normal(70, 15, n_patients).clip(40, 150),
12    'bmi': np.random.normal(25, 5, n_patients).clip(15, 45),
13    'biomarker_level': np.random.lognormal(2, 0.8, n_patients),
14    'treatment_arm': np.random.choice(
15        ['Drug_A', 'Drug_B', 'Placebo'],
16        n_patients,
17        p=[0.4, 0.4, 0.2]
18    ),
19    'response': np.random.choice(
20        ['CR', 'PR', 'SD', 'PD'],  # Complete/Partial Response, Stable/Progressive Disease
21        n_patients,
22        p=[0.05, 0.15, 0.50, 0.30]  # CR 是少數類別
23    ),
24    'adverse_event': np.random.choice(
25        ['None', 'Mild', 'Moderate', 'Severe'],
26        n_patients,
27        p=[0.5, 0.3, 0.15, 0.05]
28    ),
29    'egfr': np.random.normal(80, 20, n_patients).clip(15, 120),
30})
31
32discrete_columns = ['treatment_arm', 'response', 'adverse_event']
33
34# 訓練 CTGAN
35ctgan = CTGAN(
36    epochs=100,
37    batch_size=100,        # 較小資料集用較小 batch_size
38    generator_dim=(128, 128),
39    discriminator_dim=(128, 128),
40    pac=10,
41    verbose=True,
42)
43ctgan.fit(clinical_data, discrete_columns)
44
45# === 條件式生成：補足少數類別 ===
46# 生成更多 Complete Response (CR) 的患者資料
47synthetic_cr = ctgan.sample(
48    200,
49    condition_column='response',
50    condition_value='CR'
51)
52
53print(f"原始 CR 樣本數: {(clinical_data['response'] == 'CR').sum()}")
54print(f"合成 CR 樣本數: {(synthetic_cr['response'] == 'CR').sum()}")
55print(f"合成 CR 比例: {(synthetic_cr['response'] == 'CR').mean():.2%}")
56
57# === 儲存與載入模型 ===
58import pickle
59
60with open('ctgan_clinical_model.pkl', 'wb') as f:
61    pickle.dump(ctgan, f)
62
63# 後續載入
64with open('ctgan_clinical_model.pkl', 'rb') as f:
65    loaded_ctgan = pickle.load(f)
66    new_synthetic = loaded_ctgan.sample(500)

 1import pandas as pd
 2import numpy as np
 3from ctgan import CTGAN, load_demo
 4
 5# 載入資料
 6real_data = load_demo()
 7discrete_columns = [
 8    'workclass', 'education', 'marital-status', 'occupation',
 9    'relationship', 'race', 'sex', 'native-country', 'income',
10]
11
12# --- 方法 A：CTGAN ---
13ctgan = CTGAN(epochs=50, verbose=True)
14ctgan.fit(real_data, discrete_columns)
15synthetic_ctgan = ctgan.sample(len(real_data))
16
17# 查看訓練 loss 曲線
18loss_df = ctgan.loss_values
19print("CTGAN 最終 Generator Loss:", loss_df['Generator Loss'].iloc[-1])
20print("CTGAN 最終 Discriminator Loss:", loss_df['Discriminator Loss'].iloc[-1])
21
22# --- 方法 B：TVAE ---
23from ctgan import TVAE
24
25tvae = TVAE(
26    embedding_dim=128,
27    compress_dims=(128, 128),
28    decompress_dims=(128, 128),
29    epochs=50,
30    batch_size=500,
31    loss_factor=2,
32    verbose=True,
33)
34tvae.fit(real_data, discrete_columns)
35synthetic_tvae = tvae.sample(len(real_data))
36
37# --- 手動評估（不依賴 SDMetrics 也可做基本比較） ---
38continuous_cols = [c for c in real_data.columns if c not in discrete_columns]
39
40print("\n=== 連續欄位統計比較 ===")
41for col in continuous_cols:
42    real_mean = real_data[col].mean()
43    ctgan_mean = synthetic_ctgan[col].mean()
44    tvae_mean = synthetic_tvae[col].mean()
45    print(f"{col:>20s}: 真實={real_mean:.2f} | CTGAN={ctgan_mean:.2f} | TVAE={tvae_mean:.2f}")
46
47print("\n=== 離散欄位 Jensen-Shannon Divergence ===")
48from scipy.spatial.distance import jensenshannon
49
50for col in discrete_columns[:3]:  # 取前三個欄位示範
51    real_dist = real_data[col].value_counts(normalize=True).sort_index()
52    ctgan_dist = synthetic_ctgan[col].value_counts(normalize=True).reindex(real_dist.index, fill_value=0)
53    tvae_dist = synthetic_tvae[col].value_counts(normalize=True).reindex(real_dist.index, fill_value=0)
54
55    jsd_ctgan = jensenshannon(real_dist.values, ctgan_dist.values)
56    jsd_tvae = jensenshannon(real_dist.values, tvae_dist.values)
57    print(f"{col:>20s}: CTGAN JSD={jsd_ctgan:.4f} | TVAE JSD={jsd_tvae:.4f}")
58
59# --- 使用 SDMetrics 做完整品質報告（如已安裝） ---
60try:
61    from sdmetrics.reports.single_table import QualityReport
62
63    metadata = {
64        'columns': {col: {'sdtype': 'categorical'} if col in discrete_columns
65                    else {'sdtype': 'numerical'}
66                    for col in real_data.columns}
67    }
68
69    report = QualityReport()
70    report.generate(real_data, synthetic_ctgan, metadata)
71    print(f"\nSDMetrics 品質分數: {report.get_score():.4f}")
72except ImportError:
73    print("\n（未安裝 sdmetrics，跳過完整品質報告）")

 1第一世代：GAN-based（2019-2022 主流）
 2├── CTGAN ← 你在這裡
 3├── dp_cgans（CTGAN + Differential Privacy）
 4├── PATE-GAN
 5└── TableGAN
 6
 7第二世代：Diffusion-based（2022-2025 興起）
 8├── TabDDPM（Denoising Diffusion）
 9├── TabDiff（Score-based Diffusion）
10├── ForestDiffusion（XGBoost + Diffusion）
11└── TabCSDI（Conditional Score-based Diffusion）
12
13第三世代：LLM-based（2024- 新興）
14├── GReaT（GPT-2 fine-tuned on tabular）
15├── distilabel（LLM-as-judge 合成）
16└── REaLTabFormer（Transformer-based）

 1# CTGAN 生成 → SDMetrics 評估 → 迭代改善
 2from sdmetrics.single_table import KSComplement, TVComplement
 3
 4# 連續欄位品質
 5ks_score = KSComplement.compute(real_data, synthetic_data,
 6                                 column_name='age')
 7
 8# 離散欄位品質
 9tv_score = TVComplement.compute(real_data, synthetic_data,
10                                 column_name='sex')

1from dpctgan import DPCTGAN
2
3dp_model = DPCTGAN(
4    epochs=100,
5    epsilon=1.0,   # 隱私預算（越小越嚴格）
6    delta=1e-5,    # 隱私鬆弛參數
7)
8dp_model.fit(clinical_data, discrete_columns)
9private_synthetic = dp_model.sample(1000)

 1# 統一評估介面
 2results = {}
 3
 4# CTGAN baseline
 5from ctgan import CTGAN
 6ctgan = CTGAN(epochs=300)
 7ctgan.fit(train_data, discrete_columns)
 8results['CTGAN'] = ctgan.sample(len(train_data))
 9
10# TabDDPM（如已安裝）
11# from tab_ddpm import TabDDPM
12# ddpm = TabDDPM(...)
13# ddpm.fit(train_data)
14# results['TabDDPM'] = ddpm.sample(len(train_data))
15
16# 統一用 SDMetrics 比較
17from sdmetrics.reports.single_table import QualityReport
18for name, syn in results.items():
19    report = QualityReport()
20    report.generate(train_data, syn, metadata)
21    print(f"{name}: Quality Score = {report.get_score():.4f}")

1品質排名（由高到低，視資料集而異）：
2TabDDPM / ForestDiffusion ≈ TabDiff
3  > TVAE ≈ CTGAN
4  > GaussianCopula
5  > 簡單統計取樣