时序因果发现深度专题

1. 引言

时序因果发现（Temporal Causal Discovery）旨在从时间序列数据中推断变量之间的因果时序关系，包括：

• 滞后因果关系（lagged causation）：$X_{t-k}\to Y_t$
• 同期因果关系（contemporaneous causation）：$X_t\leftrightarrow Y_t$¹

1.1 时序因果的特殊性

与静态因果发现相比，时序因果发现具有以下特点：

特性	静态因果	时序因果
时间依赖	无	有
因果方向	需额外假设	可通过时序信息推断
延迟效应	不考虑	可建模
数据要求	独立同分布	可允许弱依赖

1.2 时序因果发现的挑战

1. 时序依赖：$X_t$ 与 $X_{t+k}$ 可能相关但非因果
2. 高维数据：$n$ 个变量的时间序列产生 $O(n^2)$ 的候选边
3. 长期依赖：因果关系可能跨越多个时间步
4. 分布偏移：如存在趋势、季节性等非平稳性

---

2. Granger因果性基础

2.1 经典Granger因果性

Granger因果性（Granger, 1969）²：

核心思想：如果使用 $(X,Y)$ 的历史信息比仅使用 $Y$ 的历史信息能更好地预测 $Y$，则称 $X$ Granger-cause $Y$。

数学定义：
$P(Y_{t+1}|{\mathcal{F}}_t)<P(Y_{t+1}|{\mathcal{F}}_t\setminus X_t)$

其中 ${\mathcal{F}}_t=\{X_s,Y_s:s\le t\}$ 是到时刻 $t$ 的信息集。

2.2 线性Granger因果性

向量自回归（VAR）模型：
$\left[\begin{array}{c}{Y}_t\\ X_t\end{array}\right]={\sum }_{k=1}^K{A}_k\left[\begin{array}{c}{Y}_{t-k}\\ X_{t-k}\end{array}\right]+\left[\begin{array}{c}{ϵ}_{1t}\\ {ϵ}_{2t}\end{array}\right]$

Granger因果性检验：检验 $A_k[1,2]=0$ 对所有 $k$ 是否成立。

2.3 Granger因果性的局限性

局限	描述
只能发现滞后因果	同期因果关系被残差噪声吸收
需要因果充分性	不能有未观测的共同原因
线性假设	难以处理非线性关系
预测视角	预测≠因果

---

3. PCMCI算法

3.1 PCMCI概述

PCMCI（Peter-Clark Moments Conditional Independence，Runge et al., 2019）³：

核心优势：

• 可处理高维时序数据
• 同时发现滞后和同期因果关系
• 计算效率高
• 对非平稳数据鲁棒

3.2 算法流程

PCMCI算法流程：

1. 条件独立性筛选（PC阶段）
   for each (X_i, X_j):
       for each S ⊆ {所有变量} \ {X_i, X_j}:
           if X_i ⊥ X_j | S:
               remove edge X_i — X_j
               
2. Moment Conditional Independence (MCI)
   for each remaining edge:
       for each time lag τ:
           test MCI(X_i^{t-τ}, X_j^t | S, X_i^{t'})
           where S ⊆ PA_j^{t} \ {X_i^{t-τ}}
           
3. 因果时间延迟估计
   确定每条因果边的最小时间延迟

3.3 MCI检验

Moment Conditional Independence（MCI）：

$MCI(X_i^{t-\tau },X_j^t|{\mathbf{Z}}^t)=\text{Cov}(X_i^{t-\tau },X_j^t|{\mathbf{Z}}^t)$

其中 ${\mathbf{Z}}^t$ 是时刻 $t$ 的条件变量集。

优势：

• 利用时间序列的时间结构
• 比普通条件独立性检验更高效

3.4 PCMCIplus

PCMCIplus（Runge, 2020）：

改进：

• 更严格的同期因果发现
• 支持瞬时因果关系
• 改进的伪相关过滤

from tigramite.pcmci import PCMCI
from tigramite.independence_tests import ParCorr
 
# PCMCIplus实现
pcmci = PCMCI(
    dataframe=dataframe,
    cond_ind_test=ParCorr()  # 偏相关检验
)
 
# 运行PCMCIplus
results = pcmci.run_pcmciplus(
    tau_min=1,
    tau_max=5,
    pc_alpha=0.05
)

---

4. 时序因果发现基准

4.1 CausalTime基准

CausalTime（ICLR 2024）⁴：

目标：生成与真实数据高度相似的时序数据，包含ground truth因果图。

流程：

真实因果图 → 归一化流 → 模拟数据
                         ↓
              从模拟数据中提取因果图
                         ↓
                 Ground Truth

特点：

• 基于动力学因果模型
• 覆盖多种数据类型（连续、离散、混合）
• 包含分布偏移场景

4.2 CausalRivers基准（ICLR 2025）

CausalRivers是最大规模的时序因果发现野外评估数据集⁵：

特性	德国东部	巴伐利亚
节点数	666	494
时间跨度	2019-2023	2019-2023
分辨率	15分钟	15分钟
分布偏移	洪水事件	季节变化

评估维度：

1. 因果发现精度：边恢复的SHD、F1
2. 因果效应估计：ATE估计的准确性
3. 鲁棒性：分布偏移下的表现

---

5. 深度学习时序因果发现

5.1 CausalFormer（2024）

CausalFormer（arXiv:2406.16708）⁶：

核心创新：

• 因果感知Transformer架构
• 分解式因果检测器
• 多核因果卷积：沿时间维度聚合

CausalFormer架构：

┌──────────────────────────────────────────────────┐
│                                                  │
│  输入: 时间序列 X₁, X₂, ..., Xₙ                  │
│         ↓                                        │
│  ┌─────────────────────────────────────────┐    │
│  │  因果表示学习（预测任务）                 │    │
│  │  - 多核因果卷积                          │    │
│  │  - 时间优先约束                          │    │
│  └─────────────────────────────────────────┘    │
│         ↓                                        │
│  ┌─────────────────────────────────────────┐    │
│  │  因果图构建（分解式检测）                 │    │
│  │  - 回归相关性传播                        │    │
│  │  - 跨注意力机制                          │    │
│  └─────────────────────────────────────────┘    │
│         ↓                                        │
│  输出: 因果邻接矩阵 B                            │
│                                                  │
└──────────────────────────────────────────────────┘

多核因果卷积：

class MultiKernelCausalConv(nn.Module):
    def __init__(self, n_vars, n_kernels, kernel_sizes):
        super().__init__()
        self.kernels = nn.ModuleList([
            nn.Conv1d(n_vars, n_vars, 
                      kernel_size=k, 
                      padding=k//2,
                      groups=n_vars)
            for k in kernel_sizes
        ])
    
    def forward(self, x):
        # 对每个核应用卷积
        outputs = [F.relu(conv(x)) for conv in self.kernels]
        # 时间优先约束：近期信息权重更高
        weights = F.softmax(self.attention_weights, dim=-1)
        return sum(w * out for w, out in zip(weights, outputs))

回归相关性传播：

def regression_relevance_propagation(model, x):
    """
    利用整个深度学习模型识别因果关系
    """
    # 前向传播
    predictions = model(x)
    
    # 计算输入梯度
    gradients = torch.autograd.grad(
        predictions.sum(),
        x,
        retain_graph=True
    )[0]
    
    # 聚合时间维度
    relevance = torch.mean(torch.abs(gradients), dim=0)
    
    return relevance  # 表示因果强度

5.2 TS-CausalNN（2024）

TS-CausalNN（arXiv:2404.01466）⁷：

核心创新：

• 同时发现同期和滞后因果关系
• 并行自定义因果层卷积块
• 自然处理非平稳性和非线性

架构：

class TS_CausalNN(nn.Module):
    def __init__(self, n_vars, hidden_dim, lag):
        super().__init__()
        self.n_vars = n_vars
        self.lag = lag
        
        # 并行因果卷积块
        self.causal_convs = nn.ModuleList([
            CausalConvBlock(n_vars, hidden_dim)
            for _ in range(3)  # 3个并行块
        ])
        
        # 无环性约束层
        self.dag_constraint = AugmentedLagrangianDAG()
        
    def forward(self, X):
        # X: (batch, time, vars)
        B = torch.zeros(self.n_vars, self.n_vars)
        
        for conv_block in self.causal_convs:
            # 计算因果邻接矩阵
            B += conv_block(X)
        
        # 归一化
        B = B / len(self.causal_convs)
        
        # 应用DAG约束
        B = self.dag_constraint(B)
        
        return B

无环性约束：

class AugmentedLagrangianDAG(nn.Module):
    def __init__(self, rho=1.0, alpha=0.0):
        super().__init__()
        self.rho = rho
        self.alpha = alpha
    
    def dag_loss(self, B):
        """DAG约束: h(B) = tr(e^(B∘B)) - n = 0"""
        M = B * B  # Hadamard乘积
        h = torch.trace(torch.matrix_exp(M)) - B.shape[0]
        return h
    
    def forward(self, B):
        # 增广拉格朗日更新
        h = self.dag_loss(B)
        penalty = self.rho * h * h + self.alpha * h
        
        if h > 0.01:  # 放松约束
            self.alpha += 2 * self.rho * h
            self.rho *= 1.5
        
        return B  # 返回原始矩阵，约束在损失中处理

5.3 STIC算法（2024）

STIC（Short-Term Invariance using Convolutional Neural Networks，arXiv:2408.08023）⁸：

核心思想：利用短时不变性进行因果发现

不变性假设：

1. 短时时间不变性：在极短时间内，底层因果机制保持稳定
2. 短时机制不变性：相邻时间步的因果关系相似

STIC核心原理：

时间序列: X₁ X₂ X₃ X₄ X₅ ...
              ↓   ↓
         短时窗口   短时窗口
              ↓   ↓
         不变性约束 → 因果关系发现

两种因果卷积核：

class STICConv(nn.Module):
    def __init__(self):
        super().__init__()
        
        # 短时时间不变性核
        self.time_inv_kernel = nn.Parameter(
            torch.tensor([[-0.5, 0, 0.5]])  # 梯度核
        )
        
        # 短时机制不变性核
        self.mech_inv_kernel = nn.Parameter(
            torch.tensor([[0.2, 0.6, 0.2]])  # 平滑核
        )
    
    def forward(self, x):
        # 时间不变性卷积：捕捉时间依赖
        time_feat = F.conv1d(
            x, 
            self.time_inv_kernel.unsqueeze(0),
            padding=1
        )
        
        # 机制不变性卷积：捕捉因果机制
        mech_feat = F.conv1d(
            x,
            self.mech_inv_kernel.unsqueeze(0),
            padding=1
        )
        
        return time_feat, mech_feat

可识别性理论：
STIC从理论上证明了在卷积与加性噪声模型下，生成原理的可识别性。

---

6. 非线性时序因果发现

6.1 非线性Granger因果性

基于神经网络的非线性因果检验：

class NonlinearGrangerTest(nn.Module):
    def __init__(self, n_vars, hidden_dim=64):
        super().__init__()
        # Y的预测器（不含X）
        self.predictor_y = nn.Sequential(
            nn.Linear(n_vars * max_lag, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )
        
        # Y的预测器（含X）
        self.predictor_xy = nn.Sequential(
            nn.Linear(n_vars * 2 * max_lag, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )
    
    def granger_causality(self, X, var_x, var_y, lag):
        """检验 var_x 是否 Granger-cause var_y"""
        # 构建历史特征
        Y_history = self._build_history(X, var_y, lag)
        X_history = self._build_history(X, var_x, lag)
        
        # 不含X预测
        pred_y = self.predictor_y(Y_history)
        
        # 含X预测
        pred_xy = self.predictor_xy(
            torch.cat([Y_history, X_history], dim=-1)
        )
        
        # Granger因果性 = 预测改善程度
        r2_without = 1 - torch.var(Y - pred_y)
        r2_with = 1 - torch.var(Y - pred_xy)
        
        return r2_with - r2_without

6.2 因果卷积网络

CCNN（Causal Convolutional Network）：

class CausalConv1d(nn.Module):
    def __init__(self, in_channels, out_channels, kernel_size, dilation):
        super().__init__()
        self.padding = (kernel_size - 1) * dilation
        self.conv = nn.Conv1d(
            in_channels, out_channels, kernel_size,
            padding=self.padding, dilation=dilation
        )
    
    def forward(self, x):
        x = self.conv(x)
        # 裁剪以确保因果性
        if self.padding > 0:
            x = x[:, :, :-self.padding]
        return x
 
class CausalConvBlock(nn.Module):
    def __init__(self, channels, kernel_size, dilation):
        super().__init__()
        self.causal_conv = CausalConv1d(
            channels, channels, kernel_size, dilation
        )
        self_gate = nn.Linear(channels, channels)
        self_transform = nn.Linear(channels, channels)
    
    def forward(self, x):
        # 门控线性单元
        a = torch.sigmoid(self_gate(x))
        b = self_transform(x)
        return a * b

---

7. 分布外鲁棒性

7.1 分布偏移场景

时序因果发现在以下场景面临分布偏移：

类型	原因	影响
趋势偏移	数据随时间变化	边权重变化
季节偏移	周期性变化	边方向可能反转
事件偏移	突发事件	新因果关系出现
概念偏移	变量含义变化	因果结构失效

7.2 鲁棒因果发现

class RobustTemporalCausalDiscovery:
    def __init__(self, base_method='pcmci'):
        self.base_method = base_method
        self.causal_graphs = []
    
    def fit_rolling_window(self, data, window_size, step_size):
        """滚动窗口方法"""
        n_samples = data.shape[0]
        results = []
        
        for start in range(0, n_samples - window_size, step_size):
            end = start + window_size
            
            # 在窗口上运行因果发现
            window_data = data[start:end]
            
            if self.base_method == 'pcmci':
                causal_graph = self._pcmci_discovery(window_data)
            elif self.base_method == 'granger':
                causal_graph = self._granger_discovery(window_data)
            
            results.append({
                'window': (start, end),
                'graph': causal_graph
            })
        
        return results
    
    def detect_causal_drift(self, results):
        """检测因果漂移"""
        graphs = [r['graph'] for r in results]
        
        # 计算相邻窗口的图距离
        distances = []
        for i in range(1, len(graphs)):
            dist = self._graph_distance(graphs[i-1], graphs[i])
            distances.append(dist)
        
        # 识别漂移点
        drift_points = np.where(
            np.array(distances) > self.drift_threshold
        )[0]
        
        return drift_points

---

8. 方法对比与选择指南

8.1 方法分类

时序因果发现方法
├── 统计方法
│   ├── Granger因果性（线性/非线性）
│   ├── PCMCI系列
│   └── 信息论方法（TE/CMI）
├── 约束方法
│   ├── 时序PC算法
│   └── 时序FCI算法
└── 深度学习方法
    ├── CausalFormer
    ├── TS-CausalNN
    └── STIC

8.2 场景选择

场景	推荐方法	理由
小规模、低维	Granger + PCMCI	统计可靠、解释性强
中等规模	PCMCIplus	高维支持、高效
非线性关系	CausalFormer、TS-CausalNN	深度学习灵活性
小样本	STIC	不变性假设减少样本需求
非平稳数据	TS-CausalNN	自然处理非平稳性
大规模（>100变量）	CausalFormer	Transformer可扩展性
需要理论保证	PCMCI	统计学基础扎实

8.3 性能对比

方法	时间复杂度	同期因果	非线性	非平稳	OOD鲁棒性
Granger	$O(n^{2}T)$	否	需扩展	否	弱
PCMCI	$O(n^{3}T)$	是	需扩展	部分	中
PCMCIplus	$O(n^{3}T)$	是	需扩展	部分	中
CausalFormer	$O(n^{2}T)$	是	是	部分	较强
TS-CausalNN	$O(n^{2}T)$	是	是	是	较强
STIC	$O(n^{2}T)$	是	是	是	强

---

9. 实践指南

9.1 数据预处理

def preprocess_time_series(data, standardize=True, remove_trend=True):
    """时序数据预处理"""
    if standardize:
        # 标准化
        data = (data - data.mean()) / data.std()
    
    if remove_trend:
        # 去除趋势（差分）
        data = np.diff(data, axis=0)
    
    return data
 
def create_lagged_features(data, max_lag):
    """创建滞后特征"""
    lagged_data = []
    for lag in range(1, max_lag + 1):
        lagged_data.append(data[lag:])
    
    return np.column_stack(lagged_data)

9.2 因果发现完整流程

from tigramite import data_processing as dp
from tigramite.pcmci import PCMCI
from tigramite.independence_tests import ParCorr
 
def complete_causal_discovery_pipeline(data, var_names):
    # 1. 数据预处理
    data = preprocess_time_series(data)
    
    # 2. 创建数据框架
    dataframe = dp.DataFrame(
        data,
        var_names=var_names,
        time_axis=0
    )
    
    # 3. PCMCI因果发现
    pcmci = PCMCI(
        dataframe=dataframe,
        cond_ind_test=ParCorr()
    )
    
    # 4. 运行分析
    results = pcmci.run_pcmci(
        tau_min=1,
        tau_max=5,
        pc_alpha=0.05
    )
    
    # 5. 提取结果
    return {
        'graph': results['graph'],
        'pvalues': results['p_matrix'],
        'val_matrix': results['val_matrix']
    }

---

10. 参考文献

---

相关主题

• 因果发现算法现代综述
• 可扩展神经因果发现方法
• 因果表示学习理论
• 概率图模型综述

Footnotes

1. Runge, J. (2018). Causal network reconstruction from time series. Springer. ↩

2. Granger, C. W. (1969). Investigating causal relations by econometric models. Econometrica. ↩

3. Runge, J., et al. (2019). Inferring causation from time series with PCMCI. Nature Communications. ↩

4. CausalTime Authors. (2024). CausalTime: Benchmark for Temporal Causal Discovery. ICLR 2024. ↩

5. CausalRivers Authors. (2025). CausalRivers: Large-scale Temporal Causal Discovery Benchmark. ICLR 2025. ↩

6. CausalFormer Authors. (2024). CausalFormer: Transformer-based Temporal Causal Discovery. arXiv:2406.16708. ↩

7. TS-CausalNN Authors. (2024). TS-CausalNN: Non-stationary Non-linear Temporal Causal Discovery. arXiv:2404.01466. ↩

8. STIC Authors. (2024). STIC: Short-Term Invariance for Causal Discovery. arXiv:2408.08023. ↩