简介：Python-Causality 是一套面向数据分析与机器学习从业者的数据集因果分析工具包，适合具备一定 Python 基础、希望从相关性分析进阶到因果推断的研究人员与数据科学家。它围绕结构方程模型、贝叶斯网络、格兰杰因果检验、倾向得分匹配、工具变量法与 Do-Calculus 等算法，帮助用户识别变量间的潜在因果结构，可应用于社会科学、经济学、生物医学与市场营销等场景。资源包共 29 个文件，以 18 个 py 源码文件为核心，辅以 2 个 csv 数据样例、2 个 md 说明文档、3 张 png 示意图及 xml、txt、in 等配置与依赖文件，压缩包约 74KB，目录涵盖 inference、estimation、util 等模块，结构清晰便于按功能查阅。目前已有 3293 人学习下载。通过阅读源码与示例，读者可掌握数据预处理、模型选择、训练评估、因果发现及可视化呈现的完整流程，并借助 README 与测试用例快速上手，为药物疗效评估、政策效果分析等实际决策提供因果层面的支撑。
1. 因果推断落地：从一份数据集到可解释的因果图
拿到一份业务数据集，老板问“哪个指标真正影响了转化”，多数人第一反应是跑相关性、画热力图。但相关性会把“冰淇淋销量”和“溺水人数”绑在一起，因为它们背后都站着“气温”这个混杂因子。Python-Causality 就是冲着这个痛点来的：它把因果结构学习、干预计算、反事实推断打包成一套可复用的 Python 工具，让你在 Jupyter 里就能从原始数据走到一张有向无环图，再算出“如果我把价格降 5%，销量会变成多少”。它适合做数据分析、增长实验、风控归因的开发者，也适合刚接触因果推断、不想一上来就啃理论公式的人。下面我按“能干什么 → 怎么跑 → 怎么改”的顺序，把这份源码拆开讲。
2. 项目定位与技术选型：为什么不是又一个相关性工具
2.1 它到底解决哪类问题
普通数据分析工具回答的是“X 和 Y 一起动了吗”，Python-Causality 回答的是“我动了 X，Y 会怎么变”。前者是观察，后者是干预。项目把这两件事分得很清楚：数据进来先做结构学习，得到候选因果图；再基于图做后门调整或前门调整，估计平均处理效应；最后支持 do-演算，模拟“强制把某个变量设成固定值”后的分布。整套流程不依赖外部服务，纯 Python 实现，核心算法用 numpy 和 scipy 打底，图结构用 networkx 管理，可视化走 matplotlib。常见做法是把它嵌进现有的 pandas 工作流里，读 CSV、清洗、丢进 CausalityAnalyzer，拿回图对象和效应值。
2.2 技术栈与依赖边界
项目没有把深度学习框架拉进来，这是有意为之。因果结构学习在小样本、高噪声场景下，基于约束的方法（PC 算法）和基于评分的方法（BIC 评分 + 爬山搜索）往往比神经网络更稳，也更容易解释。依赖清单里主要是 numpy、pandas、scipy、networkx、matplotlib、scikit-learn（只用到互信息、卡方检验这类基础统计量）。Python 版本建议 3.8 以上，3.10 也能跑。如果你打算做大规模数据，可以自己接 Dask 或 Polars，但项目默认走 pandas，因为因果推断通常发生在特征工程之后，数据量已经被压到几千到几万行。
2.3 目录结构一眼看懂
源码包解压后，顶层是causality/包，下面分core/、algorithms/、utils/、viz/四个子目录。core/放图对象和因果效应估计器；algorithms/放 PC、GES、后门调整这些具体算法；utils/是数据校验、条件独立检验、bootstrap 重采样；viz/负责画图。根目录还有examples/和tests/，examples 里给了三个 notebook：一个做结构学习，一个做效应估计，一个做反事实。tests 覆盖了条件独立检验和 PC 算法的边界情况。这种布局的好处是算法和数据结构解耦，你想换评分函数，只动algorithms/scoring.py，不用碰图对象。
3. 跑通第一个因果分析：从 CSV 到效应值
3.1 环境准备与安装
先把仓库拉到本地，建议用虚拟环境，避免和系统里的 numpy 版本打架。安装命令分两种：如果你只想跑 examples，直接pip install -e .走 editable 模式；如果要改源码，同样用 editable，改完不用重装。依赖会自动装，但 networkx 和 matplotlib 的版本最好锁一下，networkx 2.8 以上、matplotlib 3.5 以上，否则画图时节点标签会错位。
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -e . pip install jupyter # 跑 examples 需要装完可以跑pytest tests/验证环境，正常情况全绿。如果卡在条件独立检验的测试上，多半是 scipy 版本太老，升到 1.9 以上。
3.2 数据加载与预处理
项目对输入数据的要求不复杂：pandas DataFrame，列名是变量名，不能有缺失值，类别变量要么转成数值，要么在初始化时声明。下面这段代码读一个模拟的营销数据集，包含价格、广告投入、销量、季节四个字段。注意CausalityAnalyzer初始化时要传treatment和outcome，这两个是后续效应估计的目标变量。
import pandas as pd from causality.core import CausalityAnalyzer # 读数据，假设 CSV 里没有索引列 df = pd.read_csv("examples/data/marketing_sim.csv") # 声明哪些是离散变量，PC 算法会据此选检验方法 analyzer = CausalityAnalyzer( data=df, treatment="price", # 干预变量 outcome="sales", # 结果变量 discrete=["season"] # 季节是类别变量 ) # 先做基础统计，确认没有常量列和极端偏态 print(analyzer.summary())summary()会输出每列的均值、方差、缺失计数和唯一值数量。如果某列唯一值只有 1 个，PC 算法会直接报错，因为条件独立检验没法算。遇到这种情况，要么删列，要么在初始化时用exclude参数排除。
3.3 结构学习：PC 算法跑出因果图
结构学习是整条链路里最耗时的一步，因为 PC 算法要做大量条件独立检验。项目默认用卡方检验处理离散变量，用 Fisher-z 处理连续变量，混合类型时自动切换。下面这行代码触发结构学习，alpha是显著性水平，默认 0.05，调小会让图更稀疏，调大则保留更多边。
# 运行 PC 算法，返回一个 DAG 对象 dag = analyzer.learn_structure( method="pc", alpha=0.05, max_cond_vars=3 # 条件集最大变量数，控制计算量 ) # 打印边列表和每个节点的父节点 print(dag.edges()) print(dag.parents("sales"))max_cond_vars是关键参数。设成 3 意味着条件集最多包含 3 个变量，再大计算量会指数上升。小数据集（几百行）可以设到 4 或 5，上万行建议降到 2。跑完先别急着信，用dag.to_dot()导出 DOT 格式，丢进 Graphviz 看一眼有没有反向边。PC 算法对噪声敏感，偶尔会把“销量 → 广告”学成“广告 → 销量”，这时候需要人工加先验约束。
3.4 效应估计与反事实
图有了，接下来算平均处理效应。项目提供两种估计器：后门调整和逆概率加权。后门调整更直观，它把混杂因子的分布加权掉，再算 treatment 对 outcome 的回归系数。下面代码用后门调整估计价格对销量的效应，bootstrap参数控制重采样次数，用来算置信区间。
# 后门调整估计平均处理效应 ate = analyzer.estimate_effect( method="backdoor", dag=dag, bootstrap=200 ) print(f"ATE: {ate.value:.3f}, 95% CI: {ate.ci}")ate.value是效应值，负值表示价格上升销量下降，符合预期。ate.ci是 bootstrap 得到的置信区间，如果区间跨 0，说明效应不显著。反事实推断用counterfactual()方法，传一个干预后的数据框，返回预测结果。注意反事实依赖模型假设，图错了结果就错了，所以结构学习那步不能省。
4. 核心模块拆解：图对象、条件独立检验与评分函数
4.1 DAG 对象的设计
core/dag.py里的DAG类是整个项目的骨架。它内部用 networkx 的 DiGraph 存边，但对外暴露的是因果语义的方法：parents()、children()、do()、is_valid()。do()方法模拟干预，把指向 treatment 的边全部切断，这在后门调整里反复用到。is_valid()检查图是否无环，PC 算法输出的是 CPDAG，可能含无向边，需要先定向再校验。
class DAG: def __init__(self, edges=None): self.graph = nx.DiGraph() if edges: self.graph.add_edges_from(edges) def do(self, node): """返回干预后的图副本，切断所有指向 node 的边""" new_graph = self.graph.copy() for parent in list(new_graph.predecessors(node)): new_graph.remove_edge(parent, node) return DAG(new_graph.edges()) def is_valid(self): """检查是否是有向无环图""" return nx.is_directed_acyclic_graph(self.graph)这段代码里do()返回的是新对象，不是原地修改，避免污染原始图。is_valid()直接复用 networkx 的检查，省去手写拓扑排序。实际用的时候，如果 PC 输出含无向边，is_valid()会返回 False，需要先跑orient_edges()做定向。
4.2 条件独立检验的三种实现
utils/ci_tests.py里实现了三种检验：Fisher-z、卡方、G-test。Fisher-z 用于连续变量，卡方和 G-test 用于离散变量。选择逻辑在CausalityAnalyzer初始化时根据discrete参数自动判断。下面以 Fisher-z 为例，它把相关系数转成 z 统计量，再算 p 值。
def fisher_z_test(data, x, y, cond_set): """条件独立检验：x 和 y 在 cond_set 下是否独立""" if cond_set: # 对 cond_set 做线性回归，取残差 resid_x = residualize(data[x], data[cond_set]) resid_y = residualize(data[y], data[cond_set]) else: resid_x, resid_y = data[x], data[y] r = np.corrcoef(resid_x, resid_y)[0, 1] n = len(data) z = 0.5 * np.log((1 + r) / (1 - r)) * np.sqrt(n - len(cond_set) - 3) p_value = 2 * (1 - norm.cdf(abs(z))) return p_valueresidualize()用最小二乘把 cond_set 的影响去掉，剩下的残差再做相关。样本量 n 要减去条件集大小再减 3，这是 Fisher-z 的自由度修正。p 值小于 alpha 就拒绝独立假设，认为 x 和 y 之间还有边。注意这个检验假设线性关系，如果变量间是非线性，得换互信息或核方法，项目里留了ci_test参数让你自己传函数。
4.3 评分函数与爬山搜索
除了 PC，项目还实现了基于评分的 GES 算法。评分函数在algorithms/scoring.py，默认用 BIC。BIC 由两部分组成：拟合优度减去复杂度惩罚。拟合优度用对数似然，复杂度用边数乘 log(n)。下面代码展示 BIC 评分的核心计算。
def bic_score(data, dag): """计算 DAG 的 BIC 分数，越高越好""" n = len(data) score = 0 for node in dag.nodes(): parents = dag.parents(node) # 对每个节点，用父节点做回归，算残差平方和 if parents: X = data[list(parents)] y = data[node] model = LinearRegression().fit(X, y) rss = ((y - model.predict(X)) ** 2).sum() else: rss = ((data[node] - data[node].mean()) ** 2).sum() # 对数似然（高斯假设） ll = -0.5 * n * np.log(rss / n) # 复杂度惩罚 penalty = 0.5 * len(parents) * np.log(n) score += ll - penalty return scorebic_score对每个节点单独算局部分数再求和，这是 DAG 评分函数的可分解性，也是 GES 能高效搜索的前提。爬山搜索从空图开始，每步尝试加边、删边、反向边，选 BIC 提升最大的操作，直到没有提升为止。max_iter参数控制最大迭代次数，默认 100，一般几十步就收敛。
5. 避坑与排查：跑因果分析时最容易翻车的五件事
5.1 现象：PC 算法跑得极慢，半小时没输出
原因通常是max_cond_vars设太大，或者变量数超过 20 个。PC 算法的条件独立检验次数随变量数指数增长，20 个变量、条件集大小 3 时，检验次数轻松上百万。解决方法是先做特征筛选，把变量降到 10 个以内，或者把max_cond_vars降到 2。如果必须保留多变量，改用 GES，它基于评分搜索，通常比 PC 快一个数量级。
5.2 现象：学出来的图全是无向边，is_valid()返回 False
PC 算法输出的是 CPDAG（完全部分有向无环图），里面有无向边是正常的，因为数据没法区分边的方向。is_valid()要求全有向，所以会 False。解决方法是跑orient_edges()做定向，它用 Meek 规则把能定向的边定下来。如果还有无向边，说明数据信息不足，需要加先验知识，比如手动指定“价格不可能被销量影响”，用add_constraint()强制方向。
5.3 现象：ATE 估计值符号和业务常识相反
先检查图的方向对不对。如果图里“销量 → 价格”而不是“价格 → 销量”，后门调整会算出完全相反的效应。PC 算法在样本量小的时候容易搞反方向，尤其是两个变量高度相关时。解决办法是加先验约束，或者换 GES 加 BIC 评分，BIC 对边方向的惩罚更合理。另一个可能是混杂因子没被观测到，后门调整假设所有混杂因子都在数据里，如果漏了关键变量，效应估计有偏。
5.4 现象：bootstrap 置信区间宽得离谱，跨了好几个数量级
这通常意味着数据里有极端值，或者样本量太小。bootstrap 重采样时，极端值会被反复抽到，导致方差爆炸。先做异常值处理，把超过 3 倍标准差的点去掉，或者用稳健回归替代最小二乘。如果样本量低于 200，bootstrap 本身就不稳，建议改用解析方法算置信区间，或者直接报告效应值和 p 值，不报区间。
5.5 现象：反事实预测结果全是 NaN
检查干预后的数据框列名是否和训练数据一致。counterfactual()内部用训练时的列顺序做对齐，如果干预数据缺列或多列，会直接返回 NaN。另外，如果干预值超出训练数据的取值范围，模型外推可能产生 NaN 或 inf。解决办法是干预值设在训练数据的 5% 到 95% 分位数之间，别做极端外推。
6. 二次开发与改造：换算法、接大数据、做自动化报告
6.1 替换结构学习算法
项目默认的 PC 和 GES 不一定适合你的场景。如果你想接 NOTEARS 这类基于连续优化的方法，只需要在algorithms/下新建notears.py，实现一个learn(data, **kwargs)函数，返回边列表，然后在CausalityAnalyzer.learn_structure()里加一个分支。接口约定很简单：输入 DataFrame，输出(edges, score)，edges 是(parent, child)元组列表。这样不用改图对象和效应估计器，链路照常跑。
6.2 接 Polars 或 Dask 处理大数据
项目默认走 pandas，但因果推断的前置步骤（特征工程）可能产生百万行数据。改造思路是把utils/ci_tests.py里的 numpy 操作换成 Polars 表达式，或者用 Dask 做分块条件独立检验。注意 bootstrap 在分布式环境下要改成按块重采样再合并，不能直接调np.random.choice。我一般会保留 pandas 接口做小数据验证，大数据走 Polars 后端，用环境变量切换。
6.3 自动化报告与验证习惯
最后分享一个我常用的验证套路：每次改完算法，先跑tests/里的合成数据测试，确认边恢复率没下降；再用examples/里的营销数据跑一遍，对比 ATE 的符号和量级；最后把图导出成 PNG，肉眼扫一遍有没有明显反向边。这套流程走下来，基本能拦住大部分低级错误。从那以后我每次动结构学习模块，都强制走一遍合成数据加真实数据的双验证，省得后面效应估计翻车了再回头查图。希望帮到你。
本文还有配套的精品资源，点击获取
