PyTorch深度學習框架60天進階學習計劃-第57天：因果推理模型（一）

第57天：因果推理模型（一）- 揭開因果關系的神秘面紗

🎯 學習目標概覽

今天我們要踏入一個既古老又前沿的領域——因果推理！如果說傳統的機器學習是在找"相關性"，那因果推理就是在挖掘"因果性"。想象一下，你不僅要知道"下雨時人們會帶傘"，更要理解"是因為要下雨所以人們帶傘，還是因為人們帶傘所以下雨"。聽起來有點繞？別擔心，我會用最通俗易懂的方式帶你進入這個fascinating的世界！

第一部分：因果推理的理論基礎與反事實推理框架

🧠 1. 因果推理：從相關到因果的華麗轉身

在開始我們的因果推理之旅之前，讓我先給你講個小故事。假設你是一個電商平臺的數據科學家，發現了一個有趣的現象：購買紅色商品的用戶轉化率比購買藍色商品的用戶高30%。傳統的機器學習會告訴你"紅色和高轉化率相關"，但因果推理會問：“是紅色導致了高轉化率，還是喜歡紅色的用戶本身就更容易轉化？”

這就是**相關性（Correlation）與因果性（Causation）**的根本區別！

1.1 因果推理的核心概念

讓我們用一個經典的框架來理解因果推理的三個層次：

層次	名稱	問題類型	典型問題	所需信息	PyTorch應用場景
第一層	關聯/觀測	看到了什么？	癥狀X和疾病Y的相關性是多少？	觀測數據P(Y\|X)	傳統預測模型、分類器
第二層	介入/干預	如果我們這樣做會怎樣？	如果給病人藥物X，康復率會提高多少？	實驗數據P(Y\|do(X))	A/B測試分析、策略優化
第三層	反事實	如果當時不這樣做會怎樣？	如果病人沒有服用藥物X，他還會康復嗎？	因果模型P(Y_x\|X’,Y’)	個性化推薦、決策解釋

關鍵區別說明：

P(Y|X)：在觀測到X的條件下Y的概率（被動觀測）
P(Y|do(X))：主動設置X后Y的概率（主動干預）
P(Y_x|X’,Y’)：在觀測到X’和Y’的情況下，如果X被設置為x時Y的概率（反事實推理）

1.2 反事實推理：時光機器般的思維實驗

反事實推理可以說是因果推理的"終極形態"。它不僅要求我們理解"如果做X會發生Y"，還要能回答"如果當時沒做X，現在會是什么樣？"

想象你是一個推薦系統的工程師，用戶小明點擊了你推薦的電影A并給了好評。反事實推理會問：“如果我當時推薦的是電影B，小明還會給好評嗎？” 這種思維方式能幫我們更好地理解推薦策略的真實效果。

🔍 2. 介入分布 vs 觀測分布：數據背后的兩個世界

這是因果推理中最容易讓人困惑，但也是最重要的概念之一。讓我用一個生動的例子來解釋：

2.1 觀測分布：被動的旁觀者

觀測分布 P(Y|X) 就像是一個被動的攝像頭，只能記錄發生的事情。比如：

觀測到：下雨天80%的人會帶雞
數學表達：P(帶傘=是|下雨=是) = 0.8

但這個分布包含了所有的"混淆因素"：

天氣預報的影響
個人習慣的差異
季節性因素
地域文化等等

2.2 介入分布：主動的實驗者

介入分布 P(Y|do(X)) 就像是一個主動的實驗者，能夠人為地設置某個變量的值。比如：

實驗：強制讓一部分人在晴天也帶傘，觀察他們的行為變化
數學表達：P(心情=好|do(帶傘=是)) = ?

介入分布"切斷"了X的所有上游因素，只關注X對Y的直接影響。### 🎯 3. 因果圖：讓因果關系可視化

因果圖（Causal Graph）是理解復雜因果關系的神器！它就像是一張"關系地圖"，能清晰地顯示變量之間的因果流向。

觀測分布 vs 介入分布：深度對比

核心差異表

維度	觀測分布 P(Y\|X)	介入分布 P(Y\|do(X))
定義	在觀測到X條件下Y的概率	人為設置X后Y的概率
數據來源	自然觀測數據	實驗/隨機化數據
混淆因素	包含所有混淆	消除了X的混淆
因果解釋	不能確定因果關系	可以確定因果關系
計算復雜度	簡單統計	需要因果模型

經典案例：Simpson悖論

考慮一個醫院的治療效果研究：

觀測數據顯示：

治療A總體成功率：78% (78/100)
治療B總體成功率：83% (83/100)
結論：治療B更好

按病情嚴重程度分層后：

輕癥患者：

治療A成功率：93% (93/100)
治療B成功率：87% (87/100)

重癥患者：

治療A成功率：73% (73/100)
治療B成功率：69% (69/100)

真實結論：治療A在每個分層中都更好！

為什么會出現這種悖論？

因為病情嚴重程度是一個混淆變量：

重癥患者更多選擇治療A（最后的希望）
輕癥患者更多選擇治療B（保守治療）

這就是觀測分布的陷阱！

3.1 因果圖的基本元素

在這里插入圖片描述

3.2 因果圖中的關鍵路徑

在上面的因果圖中，我們可以識別出幾種重要的路徑類型：

直接因果路徑：X → Y（工作技能直接影響收入）
混淆路徑：X ← Z1 → Y 和 X ← Z2 → Y（教育水平和家庭背景同時影響技能和收入）
后門路徑：需要被"阻斷"的非因果路徑

🔧 4. PyTorch實現：從理論到實踐

現在讓我們用PyTorch來實現一個簡單而實用的因果推理模型！我們將構建一個能夠區分觀測分布和介入分布的框架。

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import matplotlib.pyplot as plt
import pandas as pd
from sklearn.model_selection import train_test_split
import warnings
warnings.filterwarnings('ignore')# 設置隨機種子確保結果可重復
torch.manual_seed(42)
np.random.seed(42)class CausalDataGenerator:"""因果數據生成器：模擬具有已知因果結構的數據結構化因果模型：Z (混淆變量) -> X (處理變量)Z (混淆變量) -> Y (結果變量)  X (處理變量) -> Y (結果變量)"""def __init__(self, n_samples=1000, causal_effect=2.0, confounding_strength=1.5):self.n_samples = n_samplesself.causal_effect = causal_effect  # X對Y的真實因果效應self.confounding_strength = confounding_strength  # 混淆變量的影響強度def generate_data(self):"""生成具有已知因果結構的數據"""# 1. 生成混淆變量Z（比如：社會經濟地位）Z = np.random.normal(0, 1, self.n_samples)# 2. 生成處理變量X（比如：是否接受培訓）# X受到Z的影響（有錢人更容易接受培訓）prob_treatment = torch.sigmoid(torch.tensor(0.5 + self.confounding_strength * Z)).numpy()X = np.random.binomial(1, prob_treatment, self.n_samples)# 3. 生成結果變量Y（比如：收入水平）# Y同時受到Z和X的影響noise = np.random.normal(0, 0.5, self.n_samples)Y = (self.confounding_strength * Z +          # 混淆變量的影響self.causal_effect * X +                 # 真實的因果效應noise)                                   # 隨機噪聲# 創建數據框data = pd.DataFrame({'Z': Z,           # 混淆變量'X': X,           # 處理變量'Y': Y,           # 結果變量'prob_treatment': prob_treatment  # 處理概率（用于分析）})return datadef calculate_true_effects(self):"""計算真實的因果效應（已知答案）"""return {'ATE': self.causal_effect,  # 平均處理效應'observational_bias': self.confounding_strength**2 / 2  # 觀測偏差的近似值}class ObservationalModel(nn.Module):"""觀測模型：直接從X預測Y，不考慮混淆變量這個模型會高估因果效應！"""def __init__(self, input_dim=1):super(ObservationalModel, self).__init__()self.linear = nn.Linear(input_dim, 1)def forward(self, x):return self.linear(x)class AdjustedModel(nn.Module):"""調整模型：同時考慮X和Z，試圖消除混淆偏差這個模型應該能給出更準確的因果效應估計"""def __init__(self, input_dim=2):super(AdjustedModel, self).__init__()self.layers = nn.Sequential(nn.Linear(input_dim, 16),nn.ReLU(),nn.Linear(16, 8),nn.ReLU(), nn.Linear(8, 1))def forward(self, x):return self.layers(x)class CausalInferenceFramework:"""因果推理框架：對比觀測估計和調整估計"""def __init__(self):self.observational_model = ObservationalModel()self.adjusted_model = AdjustedModel()# 優化器self.obs_optimizer = optim.Adam(self.observational_model.parameters(), lr=0.01)self.adj_optimizer = optim.Adam(self.adjusted_model.parameters(), lr=0.01)# 損失函數self.criterion = nn.MSELoss()def train_models(self, data, epochs=500):"""訓練兩個模型"""# 準備數據X = torch.tensor(data['X'].values.reshape(-1, 1), dtype=torch.float32)Z = torch.tensor(data['Z'].values.reshape(-1, 1), dtype=torch.float32)XZ = torch.cat([X, Z], dim=1)  # X和Z的組合Y = torch.tensor(data['Y'].values.reshape(-1, 1), dtype=torch.float32)# 訓練歷史obs_losses = []adj_losses = []for epoch in range(epochs):# 訓練觀測模型（只用X）self.obs_optimizer.zero_grad()obs_pred = self.observational_model(X)obs_loss = self.criterion(obs_pred, Y)obs_loss.backward()self.obs_optimizer.step()obs_losses.append(obs_loss.item())# 訓練調整模型（用X和Z）self.adj_optimizer.zero_grad()adj_pred = self.adjusted_model(XZ)adj_loss = self.criterion(adj_pred, Y)adj_loss.backward()self.adj_optimizer.step()adj_losses.append(adj_loss.item())if epoch % 100 == 0:print(f'Epoch {epoch}: Obs Loss = {obs_loss:.4f}, Adj Loss = {adj_loss:.4f}')return obs_losses, adj_lossesdef estimate_causal_effects(self, data):"""估計因果效應"""# 準備數據X0 = torch.zeros(len(data), 1)  # 不接受處理X1 = torch.ones(len(data), 1)   # 接受處理Z = torch.tensor(data['Z'].values.reshape(-1, 1), dtype=torch.float32)# 觀測模型的估計（有偏差）with torch.no_grad():obs_y0 = self.observational_model(X0)obs_y1 = self.observational_model(X1)obs_ate = (obs_y1 - obs_y0).mean().item()# 調整模型的估計（去除偏差）with torch.no_grad():X0Z = torch.cat([X0, Z], dim=1)X1Z = torch.cat([X1, Z], dim=1)adj_y0 = self.adjusted_model(X0Z)adj_y1 = self.adjusted_model(X1Z)adj_ate = (adj_y1 - adj_y0).mean().item()return obs_ate, adj_atedef analyze_confounding_bias(self, data):"""分析混淆偏差"""# 觀測分布中的關聯treated_group = data[data['X'] == 1]['Y'].mean()control_group = data[data['X'] == 0]['Y'].mean()naive_estimate = treated_group - control_groupreturn naive_estimatedef run_causal_analysis():"""運行完整的因果分析流程"""print("=" * 60)print("🚀 PyTorch因果推理實驗開始！")print("=" * 60)# 1. 生成數據print("\n📊 步驟1：生成具有已知因果結構的數據...")generator = CausalDataGenerator(n_samples=2000, causal_effect=2.0, confounding_strength=1.5)data = generator.generate_data()true_effects = generator.calculate_true_effects()print(f"? 生成了 {len(data)} 個樣本")print(f"📈 數據統計:")print(data.describe())# 2. 訓練模型print("\n🧠 步驟2：訓練觀測模型和調整模型...")framework = CausalInferenceFramework()obs_losses, adj_losses = framework.train_models(data, epochs=500)# 3. 估計因果效應print("\n🔍 步驟3：估計因果效應...")obs_ate, adj_ate = framework.estimate_causal_effects(data)naive_estimate = framework.analyze_confounding_bias(data)# 4. 結果對比print("\n📊 因果效應估計結果對比:")print("-" * 50)print(f"真實因果效應（ATE）:        {true_effects['ATE']:.3f}")print(f"樸素估計（觀測差異）:      {naive_estimate:.3f}")print(f"觀測模型估計:             {obs_ate:.3f}")print(f"調整模型估計:             {adj_ate:.3f}")print("-" * 50)print(f"樸素估計偏差:             {abs(naive_estimate - true_effects['ATE']):.3f}")print(f"觀測模型偏差:             {abs(obs_ate - true_effects['ATE']):.3f}")print(f"調整模型偏差:             {abs(adj_ate - true_effects['ATE']):.3f}")# 5. 可視化結果print("\n📈 步驟4：可視化訓練過程...")plt.figure(figsize=(15, 5))# 訓練損失plt.subplot(1, 3, 1)plt.plot(obs_losses, label='觀測模型', alpha=0.7)plt.plot(adj_losses, label='調整模型', alpha=0.7)plt.xlabel('訓練輪次')plt.ylabel('損失值')plt.title('模型訓練過程')plt.legend()plt.grid(True, alpha=0.3)# 因果效應對比plt.subplot(1, 3, 2)methods = ['真實值', '樸素估計', '觀測模型', '調整模型']estimates = [true_effects['ATE'], naive_estimate, obs_ate, adj_ate]colors = ['green', 'red', 'orange', 'blue']bars = plt.bar(methods, estimates, color=colors, alpha=0.7)plt.ylabel('因果效應估計')plt.title('不同方法的因果效應估計')plt.axhline(y=true_effects['ATE'], color='green', linestyle='--', alpha=0.8, label='真實值')plt.xticks(rotation=45)plt.grid(True, alpha=0.3)# 數據分布plt.subplot(1, 3, 3)treated = data[data['X'] == 1]['Y']control = data[data['X'] == 0]['Y']plt.hist(control, alpha=0.6, label='對照組 (X=0)', bins=30, density=True)plt.hist(treated, alpha=0.6, label='處理組 (X=1)', bins=30, density=True)plt.xlabel('結果變量 Y')plt.ylabel('密度')plt.title('處理組vs對照組分布')plt.legend()plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()return data, framework, true_effects# 運行實驗
if __name__ == "__main__":data, framework, true_effects = run_causal_analysis()print("\n🎉 實驗完成！我們成功展示了觀測分布和介入分布的差異！")

🎪 5. 混淆變量：因果推理的"隱形殺手"

混淆變量就像是因果推理中的"隱形殺手"——它們悄悄地影響著我們的結論，讓我們誤以為找到了因果關系，實際上只是發現了虛假的相關性。

5.1 混淆變量的識別與處理

在我們的PyTorch代碼中，變量Z就是一個經典的混淆變量。它同時影響了處理變量X（是否接受培訓）和結果變量Y（收入水平）。這創造了一個"后門路徑"：X ← Z → Y，使得X和Y之間產生了非因果的關聯。

混淆變量處理方法詳細對比

方法分類表

方法類別	具體方法	適用場景	優勢	劣勢	PyTorch實現難度
觀測數據方法	后門調整	已知所有混淆變量	理論基礎扎實	需要強假設	??
	前門調整	存在中介變量	不需要觀測所有混淆變量	需要滿足前門準則	???
	工具變量	存在合適的工具變量	可處理未觀測混淆	工具變量難找	???
實驗方法	隨機對照試驗	可進行隨機化	金標準	成本高、倫理限制	?
	自然實驗	存在外生變異	接近隨機化	機會稀少	??
機器學習方法	雙重機器學習	高維數據	處理復雜非線性關系	理論要求高	????
	因果森林	異質性處理效應	個性化效應估計	計算復雜	?????

混淆偏差的數學表達

假設真實的因果效應為 τ，觀測到的關聯為 β，那么：

β = τ + 偏差其中偏差 = Cov(X,U) × Effect(U,Y) / Var(X)

X: 處理變量
Y: 結果變量
U: 未觀測混淆變量
τ: 真實因果效應
β: 觀測關聯

我們的PyTorch實驗中的偏差分析

在代碼中，我們設置了：

真實因果效應：2.0
混淆強度：1.5
理論偏差：≈ 1.52 / 2 = 1.125

這解釋了為什么樸素估計會系統性地高估因果效應！

🔬 6. 因果發現算法：從數據中挖掘因果結構

因果發現算法就像是一個"偵探"，它試圖從觀測數據中推斷出變量之間的因果關系結構。這是因果推理中最具挑戰性的任務之一！

6.1 因果發現的基本思路

想象你是一個數據偵探，面前有一堆變量：X?, X?, X?, …, X?。你的任務是回答：

哪些變量之間存在直接的因果關系？
因果關系的方向是什么？
是否存在潛在的混淆變量？

6.2 主流因果發現算法分類

基于約束的方法（Constraint-Based）

PC算法：通過條件獨立性測試來識別因果結構
FCI算法：能處理潛在混淆變量的情況

基于分數的方法（Score-Based）

GES算法：通過最大化某個分數函數來搜索最優因果圖
LINGAM：假設線性非高斯噪聲模型

基于函數因果模型的方法

ANM（Additive Noise Models）：假設因果關系為加性噪聲模型
PNL（Post-Nonlinear Models）：允許更復雜的函數關系

import torch
import torch.nn as nn
import numpy as np
import pandas as pd
from scipy import stats
from itertools import combinations, permutations
import matplotlib.pyplot as plt
import seaborn as snsclass SimpleCausalDiscovery:"""簡化版因果發現算法基于條件獨立性測試和非線性回歸的混合方法"""def __init__(self, significance_level=0.05):self.significance_level = significance_levelself.adjacency_matrix = Noneself.causal_strengths = Nonedef conditional_independence_test(self, X, Y, Z=None, method='partial_correlation'):"""條件獨立性測試：檢驗X和Y在給定Z的條件下是否獨立Args:X, Y: 要測試的變量Z: 條件變量（可選）method: 測試方法Returns:p_value: p值is_independent: 是否獨立"""if Z is None:# 無條件獨立性測試if method == 'correlation':corr, p_value = stats.pearsonr(X, Y)is_independent = p_value > self.significance_levelelse:# 使用互信息p_value = self._mutual_information_test(X, Y)is_independent = p_value > self.significance_levelelse:# 條件獨立性測試p_value = self._partial_correlation_test(X, Y, Z)is_independent = p_value > self.significance_levelreturn p_value, is_independentdef _partial_correlation_test(self, X, Y, Z):"""偏相關測試"""try:# 創建數據矩陣if Z.ndim == 1:Z = Z.reshape(-1, 1)data = np.column_stack([X, Y, Z])# 計算相關矩陣corr_matrix = np.corrcoef(data.T)if corr_matrix.shape[0] < 3:return 1.0  # 如果維度不夠，返回高p值# 計算偏相關系數# partial_corr(X,Y|Z) = (corr(X,Y) - corr(X,Z)*corr(Y,Z)) / sqrt((1-corr(X,Z)^2)*(1-corr(Y,Z)^2))r_xy = corr_matrix[0, 1]r_xz = corr_matrix[0, 2:].mean()  # 簡化處理：取平均r_yz = corr_matrix[1, 2:].mean()denom = np.sqrt((1 - r_xz**2) * (1 - r_yz**2))if abs(denom) < 1e-8:return 1.0partial_corr = (r_xy - r_xz * r_yz) / denom# 計算t統計量和p值n = len(X)t_stat = partial_corr * np.sqrt((n - 3) / (1 - partial_corr**2))p_value = 2 * (1 - stats.t.cdf(abs(t_stat), n - 3))return p_valueexcept:return 1.0  # 出錯時返回高p值def _mutual_information_test(self, X, Y):"""互信息測試（簡化版）"""try:# 離散化變量X_discrete = pd.cut(X, bins=10, labels=False)Y_discrete = pd.cut(Y, bins=10, labels=False)# 計算互信息mi = self._calculate_mutual_information(X_discrete, Y_discrete)# 簡化的p值計算（實際應該用更嚴格的統計測試）# 這里我們用一個啟發式方法p_value = np.exp(-mi * len(X) / 100)  # 簡化公式return min(p_value, 1.0)except:return 1.0def _calculate_mutual_information(self, X, Y):"""計算互信息"""try:# 計算聯合分布和邊際分布xy_counts = pd.crosstab(X, Y)x_counts = pd.Series(X).value_counts()y_counts = pd.Series(Y).value_counts()mi = 0n = len(X)for x in xy_counts.index:for y in xy_counts.columns:if xy_counts.loc[x, y] > 0:p_xy = xy_counts.loc[x, y] / np_x = x_counts[x] / np_y = y_counts[y] / nmi += p_xy * np.log(p_xy / (p_x * p_y))return max(mi, 0)except:return 0def discover_causal_structure(self, data, variable_names=None):"""發現因果結構Args:data: numpy數組或DataFramevariable_names: 變量名列表Returns:adjacency_matrix: 鄰接矩陣causal_strengths: 因果強度矩陣"""if isinstance(data, pd.DataFrame):variable_names = data.columns.tolist()data = data.valueselif variable_names is None:variable_names = [f'X{i}' for i in range(data.shape[1])]n_vars = data.shape[1]self.variable_names = variable_names# 初始化矩陣self.adjacency_matrix = np.zeros((n_vars, n_vars))self.causal_strengths = np.zeros((n_vars, n_vars))print(f"🔍 開始因果發現，分析 {n_vars} 個變量...")# 第一步：識別所有可能的邊possible_edges = []for i in range(n_vars):for j in range(n_vars):if i != j:# 測試i和j之間是否有直接關系X_i = data[:, i]X_j = data[:, j]# 收集其他所有變量作為潛在的條件變量other_vars = [k for k in range(n_vars) if k != i and k != j]if len(other_vars) > 0:# 測試條件獨立性（使用所有其他變量）Z = data[:, other_vars]p_value, is_independent = self.conditional_independence_test(X_i, X_j, Z)else:# 無條件獨立性測試p_value, is_independent = self.conditional_independence_test(X_i, X_j)if not is_independent:possible_edges.append((i, j, p_value))print(f"📊 發現 {len(possible_edges)} 個可能的因果邊")# 第二步：確定邊的方向for i, j, p_value in possible_edges:# 使用非線性回歸來估計因果方向strength_i_to_j = self._estimate_causal_strength(data[:, i], data[:, j])strength_j_to_i = self._estimate_causal_strength(data[:, j], data[:, i])# 選擇更強的方向if strength_i_to_j > strength_j_to_i:self.adjacency_matrix[i, j] = 1self.causal_strengths[i, j] = strength_i_to_jprint(f"? 發現因果關系: {variable_names[i]} -> {variable_names[j]} (強度: {strength_i_to_j:.3f})")else:self.adjacency_matrix[j, i] = 1self.causal_strengths[j, i] = strength_j_to_iprint(f"? 發現因果關系: {variable_names[j]} -> {variable_names[i]} (強度: {strength_j_to_i:.3f})")return self.adjacency_matrix, self.causal_strengthsdef _estimate_causal_strength(self, cause, effect):"""估計因果強度使用非線性回歸的R2作為強度指標"""try:# 簡單的非線性特征X_features = np.column_stack([cause,cause**2,np.sin(cause),np.cos(cause)])# 使用PyTorch進行非線性回歸X_tensor = torch.tensor(X_features, dtype=torch.float32)y_tensor = torch.tensor(effect, dtype=torch.float32).reshape(-1, 1)# 簡單的神經網絡model = nn.Sequential(nn.Linear(4, 8),nn.ReLU(),nn.Linear(8, 1))optimizer = optim.Adam(model.parameters(), lr=0.01)criterion = nn.MSELoss()# 快速訓練for _ in range(100):optimizer.zero_grad()pred = model(X_tensor)loss = criterion(pred, y_tensor)loss.backward()optimizer.step()# 計算R2with torch.no_grad():pred = model(X_tensor)ss_res = torch.sum((y_tensor - pred) ** 2)ss_tot = torch.sum((y_tensor - torch.mean(y_tensor)) ** 2)r_squared = 1 - ss_res / ss_totreturn max(r_squared.item(), 0)except:return 0def visualize_causal_graph(self):"""可視化因果圖"""if self.adjacency_matrix is None:print("? 請先運行因果發現算法！")returnplt.figure(figsize=(12, 8))# 創建子圖fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))# 鄰接矩陣熱圖sns.heatmap(self.adjacency_matrix, annot=True, xticklabels=self.variable_names,yticklabels=self.variable_names,cmap='Reds',ax=ax1)ax1.set_title('因果鄰接矩陣\n(行->列表示因果方向)')# 因果強度熱圖sns.heatmap(self.causal_strengths,annot=True,fmt='.3f',xticklabels=self.variable_names,yticklabels=self.variable_names,cmap='Blues',ax=ax2)ax2.set_title('因果強度矩陣\n(數值表示因果強度)')plt.tight_layout()plt.show()# 測試因果發現算法
def test_causal_discovery():"""測試因果發現算法"""print("🚀 測試因果發現算法")print("=" * 50)# 生成測試數據（已知因果結構）np.random.seed(42)n_samples = 1000# 真實因果結構: Z -> X -> Y, Z -> YZ = np.random.normal(0, 1, n_samples)X = 0.8 * Z + np.random.normal(0, 0.5, n_samples)Y = 0.6 * Z + 1.2 * X + np.random.normal(0, 0.3, n_samples)# 添加一個無關變量W = np.random.normal(0, 1, n_samples)data = pd.DataFrame({'Z': Z,'X': X, 'Y': Y,'W': W})print("📊 測試數據生成完成")print("🎯 真實因果結構: Z -> X -> Y, Z -> Y")print(f"📈 數據統計:\n{data.describe()}")# 運行因果發現discovery = SimpleCausalDiscovery(significance_level=0.05)adj_matrix, strengths = discovery.discover_causal_structure(data)# 可視化結果discovery.visualize_causal_graph()return discovery# 運行測試
if __name__ == "__main__":discovery = test_causal_discovery()

📝 總結：

因果推理的本質：從"相關性"到"因果性"的思維躍遷，理解了觀測、干預、反事實三個層次的遞進關系。
分布差異的深度理解：掌握了P(Y|X)與P(Y|do(X))的根本區別，明白了為什么Simpson悖論會發生，以及如何通過正確的因果框架來避免錯誤結論。
混淆變量的識別與處理：學會了如何識別"隱形殺手"般的混淆變量，并掌握了多種處理方法，從簡單的分層調整到復雜的后門調整。
因果發現的算法思維：了解了如何從純觀測數據中推斷因果結構，掌握了基于條件獨立性測試和非線性回歸的混合方法。
PyTorch實戰能力：通過具體的代碼實現，我們不僅理解了理論，更重要的是學會了如何用PyTorch將這些抽象概念轉化為可執行的程序。