1# 建議使用 uv 建立隔離環境
 2uv venv smartnoise-env
 3source smartnoise-env/bin/activate
 4
 5# 安裝 SQL 套件（僅查詢功能）
 6uv pip install smartnoise-sql
 7
 8# 安裝合成器套件（含 PyTorch 相依）
 9uv pip install smartnoise-synth
10
11# 同時安裝兩者
12uv pip install smartnoise-sql smartnoise-synth
13
14# 安裝評估套件（可選）
15uv pip install smartnoise-eval

1# 安裝 PyTorch 系列合成器（DP-CTGAN, PATE-CTGAN, PATE-GAN）
2uv pip install "smartnoise-synth[pytorch]"
3
4# 安裝 MST 所需的 OpenDP 支援
5uv pip install "smartnoise-synth[mst]"

 1import snsql
 2from snsql import Privacy
 3print(f"smartnoise-sql version: {snsql.__version__}")
 4
 5import snsynth
 6print(f"smartnoise-synth version: {snsynth.__version__}")
 7
 8# 檢查可用合成器
 9from snsynth import MWEMSynthesizer
10print("MWEM available: OK")

 1# metadata.yaml - 模擬患者資料表描述
 2Hospital:
 3  Patients:
 4    Patients:
 5      row_privacy: True      # 每列代表一個獨立個體
 6      rows: 5000              # 近似列數（可略大於實際）
 7      age:
 8        type: int
 9        lower: 0              # 數值下界
10        upper: 120            # 數值上界
11      sex:
12        type: string          # 類別型（不需 upper/lower）
13      diagnosis_code:
14        type: string
15      lab_value:
16        type: float
17        lower: 0.0
18        upper: 1000.0
19      treatment_cost:
20        type: int
21        lower: 0
22        upper: 5000000

 1"""
 2範例 1：差分隱私 SQL 查詢
 3場景：在不暴露個別患者資訊的情況下，統計臨床試驗受試者的年齡與性別分佈
 4"""
 5import pandas as pd
 6import snsql
 7from snsql import Privacy
 8
 9# === 步驟 1：準備資料 ===
10# 模擬臨床試驗患者資料
11data = pd.DataFrame({
12    'patient_id': range(1, 501),
13    'age': [45, 52, 38, 61, 29, 55, 43, 67, 34, 58] * 50,
14    'sex': ['M', 'F', 'F', 'M', 'F', 'M', 'M', 'F', 'M', 'F'] * 50,
15    'treatment_arm': ['Drug_A', 'Drug_A', 'Placebo', 'Drug_A', 'Placebo'] * 100,
16    'response': [1, 0, 0, 1, 0, 1, 1, 0, 1, 0] * 50,
17})
18
19# === 步驟 2：定義 Metadata（也可用 YAML 檔案） ===
20metadata = {
21    'ClinicalTrial': {
22        'Patients': {
23            'Patients': {
24                'row_privacy': True,
25                'rows': 500,
26                'age': {'type': 'int', 'lower': 18, 'upper': 85},
27                'sex': {'type': 'string'},
28                'treatment_arm': {'type': 'string'},
29                'response': {'type': 'int', 'lower': 0, 'upper': 1},
30            }
31        }
32    }
33}
34
35# === 步驟 3：建立隱私保護連線 ===
36# epsilon=1.0 是常見的中等隱私保護水準
37# delta=1e-5 (遠小於 1/n) 是標準做法
38privacy = Privacy(epsilon=1.0, delta=1e-5)
39reader = snsql.from_df(data, privacy=privacy, metadata=metadata)
40
41# === 步驟 4：執行 DP SQL 查詢 ===
42
43# 查詢 1：各治療組的平均年齡
44result1 = reader.execute(
45    'SELECT treatment_arm, AVG(age) AS avg_age, COUNT(*) AS n '
46    'FROM ClinicalTrial.Patients GROUP BY treatment_arm'
47)
48print("各治療組統計（差分隱私）：")
49for row in result1:
50    print(f"  {row}")
51
52# 查詢 2：各性別的反應率
53result2 = reader.execute(
54    'SELECT sex, SUM(response) AS responders, COUNT(*) AS total '
55    'FROM ClinicalTrial.Patients GROUP BY sex'
56)
57print("\n各性別反應統計（差分隱私）：")
58for row in result2:
59    print(f"  {row}")
60
61# === 步驟 5：檢查隱私預算消耗 ===
62eps_spent, delta_spent = reader.odometer.spent
63print(f"\n累計隱私花費：epsilon={eps_spent:.4f}, delta={delta_spent:.2e}")
64
65# 預估下一次查詢的隱私成本（不實際執行）
66cost = reader.get_privacy_cost(
67    'SELECT AVG(age), AVG(response) FROM ClinicalTrial.Patients'
68)
69print(f"預估下次查詢成本：epsilon={cost[0]:.4f}, delta={cost[1]:.2e}")

 1"""
 2範例 2：MST 合成器生成合成 EHR 資料
 3場景：從真實 EHR 資料產生可公開分享的合成版本，用於跨機構研究合作
 4"""
 5import pandas as pd
 6import numpy as np
 7from snsynth import Synthesizer
 8
 9# === 步驟 1：準備模擬 EHR 資料 ===
10np.random.seed(42)
11n_patients = 2000
12
13ehr_data = pd.DataFrame({
14    'age_group': np.random.choice(
15        ['18-30', '31-45', '46-60', '61-75', '76+'],
16        n_patients, p=[0.15, 0.25, 0.30, 0.20, 0.10]
17    ),
18    'sex': np.random.choice(['M', 'F'], n_patients, p=[0.48, 0.52]),
19    'diagnosis': np.random.choice(
20        ['Hypertension', 'Diabetes_T2', 'COPD', 'CKD', 'Healthy'],
21        n_patients, p=[0.25, 0.20, 0.15, 0.10, 0.30]
22    ),
23    'medication': np.random.choice(
24        ['ACE_inhibitor', 'Metformin', 'Bronchodilator', 'None'],
25        n_patients, p=[0.25, 0.20, 0.15, 0.40]
26    ),
27    'smoking_status': np.random.choice(
28        ['Never', 'Former', 'Current'],
29        n_patients, p=[0.50, 0.30, 0.20]
30    ),
31    'readmission_30d': np.random.choice(
32        ['Yes', 'No'], n_patients, p=[0.12, 0.88]
33    ),
34})
35
36print(f"原始資料形狀：{ehr_data.shape}")
37print(f"原始診斷分佈：\n{ehr_data['diagnosis'].value_counts(normalize=True)}\n")
38
39# === 步驟 2：使用 MST 合成器 ===
40# MST 要求所有欄位為類別型（已滿足）
41# epsilon=1.0 提供合理的隱私-效用平衡
42synth = Synthesizer.create(
43    "mst",
44    epsilon=1.0,
45    verbose=True
46)
47
48# 指定所有欄位為類別型
49categorical_columns = ehr_data.columns.tolist()
50
51# 訓練合成器
52synth.fit(
53    ehr_data,
54    categorical_columns=categorical_columns
55)
56
57# === 步驟 3：生成合成資料 ===
58synthetic_ehr = synth.sample(2000)
59synthetic_df = pd.DataFrame(synthetic_ehr, columns=ehr_data.columns)
60
61print(f"合成資料形狀：{synthetic_df.shape}")
62print(f"合成診斷分佈：\n{synthetic_df['diagnosis'].value_counts(normalize=True)}\n")
63
64# === 步驟 4：簡單品質檢查 ===
65print("=== 邊際分佈比較 ===")
66for col in ehr_data.columns:
67    orig_dist = ehr_data[col].value_counts(normalize=True).sort_index()
68    synth_dist = synthetic_df[col].value_counts(normalize=True).sort_index()
69    merged = pd.DataFrame({
70        'Original': orig_dist,
71        'Synthetic': synth_dist
72    }).fillna(0)
73    max_diff = (merged['Original'] - merged['Synthetic']).abs().max()
74    print(f"  {col}: max 邊際偏差 = {max_diff:.4f}")
75
76# === 步驟 5：匯出合成資料（可安全分享） ===
77synthetic_df.to_csv('synthetic_ehr_dp.csv', index=False)
78print("\n合成資料已儲存至 synthetic_ehr_dp.csv（可安全分享）")

 1"""
 2範例 3：PATE-CTGAN 合成混合型臨床試驗資料
 3場景：臨床試驗資料含連續型實驗室數值與類別型診斷，需要保留複雜的
 4      非線性關聯（如年齡 vs. 肌酸酐 vs. CKD 分期的交互作用）
 5"""
 6import pandas as pd
 7import numpy as np
 8from snsynth.pytorch.nn import PATECTGAN
 9from snsynth.pytorch import PytorchDPSynthesizer
10
11# === 步驟 1：模擬含連續值的臨床資料 ===
12np.random.seed(42)
13n = 1000
14
15clinical_data = pd.DataFrame({
16    'age': np.random.normal(55, 15, n).clip(18, 90).astype(int),
17    'bmi': np.random.normal(26, 5, n).clip(15, 50).round(1),
18    'creatinine': np.random.lognormal(0.2, 0.4, n).clip(0.3, 8.0).round(2),
19    'hba1c': np.random.normal(6.5, 1.5, n).clip(4.0, 14.0).round(1),
20    'sex': np.random.choice(['M', 'F'], n),
21    'ckd_stage': np.random.choice(
22        ['Stage_1', 'Stage_2', 'Stage_3a', 'Stage_3b', 'Stage_4', 'Stage_5'],
23        n, p=[0.30, 0.25, 0.20, 0.12, 0.08, 0.05]
24    ),
25    'diabetes': np.random.choice(['Yes', 'No'], n, p=[0.35, 0.65]),
26})
27
28print(f"原始資料：{clinical_data.shape}")
29print(clinical_data.describe())
30
31# === 步驟 2：設定 PATE-CTGAN ===
32# 標示類別型欄位
33categorical_columns = ['sex', 'ckd_stage', 'diabetes']
34
35# 建立合成器
36# epsilon=3.0：適中的隱私保護，適合不含直接識別子的臨床資料
37synth = PytorchDPSynthesizer(
38    epsilon=3.0,
39    gan=PATECTGAN(
40        regularization='dragan',  # DRAGAN 正則化穩定訓練
41        batch_size=64,
42        epochs=100,
43    ),
44    preprocessor=None,  # 使用內建前處理
45)
46
47# === 步驟 3：訓練與生成 ===
48synth.fit(
49    clinical_data,
50    categorical_columns=categorical_columns,
51)
52
53synthetic_clinical = synth.sample(1000)
54print(f"\n合成資料：{synthetic_clinical.shape}")
55print(synthetic_clinical.describe())
56
57# === 步驟 4：交叉驗證 -- 連續欄位分佈比較 ===
58print("\n=== 連續欄位 KS 統計量 ===")
59from scipy import stats
60
61for col in ['age', 'bmi', 'creatinine', 'hba1c']:
62    ks_stat, p_value = stats.ks_2samp(
63        clinical_data[col].astype(float),
64        synthetic_clinical[col].astype(float)
65    )
66    quality = "Good" if ks_stat < 0.1 else "Fair" if ks_stat < 0.2 else "Poor"
67    print(f"  {col}: KS={ks_stat:.4f}, p={p_value:.4f} ({quality})")

 1# OpenDP 底層（原子操作，彈性高但複雜）
 2import opendp.prelude as dp
 3dp.enable_features("contrib", "honest-but-curious")
 4
 5space = dp.atom_domain(T=int), dp.absolute_distance(T=int)
 6base_lap = space >> dp.then_laplace(scale=1.0)
 7
 8# SmartNoise 高階（SQL 操作，易用但彈性較低）
 9import snsql
10reader = snsql.from_df(df, privacy=Privacy(epsilon=1.0), metadata=meta)
11result = reader.execute("SELECT AVG(age) FROM table")  # 內部自動呼叫 OpenDP

 1# 步驟 1：用 SmartNoise MST 生成 DP 合成資料
 2from snsynth import Synthesizer
 3synth = Synthesizer.create("mst", epsilon=1.0)
 4synth.fit(sensitive_data, categorical_columns=cat_cols)
 5dp_synthetic = synth.sample(len(sensitive_data))
 6
 7# 步驟 2：用 synthcity 評估合成資料品質
 8from synthcity.metrics import Metrics
 9score = Metrics.evaluate(
10    X_gt=sensitive_data,      # 僅在安全環境中比較
11    X_syn=dp_synthetic,
12    metrics={"stats": ["jensenshannon_dist", "ks_test"]},
13)
14print(score)

 1# 情境：Synthea 產生的模擬資料仍可能過度擬合真實人口分佈
 2# 加上 DP 層可消除這類風險
 3
 4# 步驟 1：讀取 Synthea 輸出的 CSV
 5synthea_patients = pd.read_csv("synthea_output/patients.csv")
 6
 7# 步驟 2：透過 DP SQL 查詢統計數據
 8privacy = Privacy(epsilon=0.5, delta=1e-6)
 9reader = snsql.from_df(synthea_patients, privacy=privacy, metadata=synthea_meta)
10stats = reader.execute("SELECT RACE, AVG(age) FROM Patients GROUP BY RACE")