🧩 一、因果图建模实战

使用DoWhy构建金融风控因果图：

from dowhy import CausalModel
import networkx as nx

# 构建信贷风险评估因果图

model = CausalModel(
    data=df,
    treatment='loan_amount',
    outcome='default_rate',
    graph="""
    digraph {
        income -> loan_amount;
        credit_score -> loan_amount;
        loan_amount -> default_rate;
        employment_length -> default_rate;
        credit_score -> default_rate;
        age -> credit_score;
        education_level -> income;
    }
    """
)

# 可视化因果图

nx.drawing.nx_pydot.to_pydot(model._graph._graph).write_png('causal_graph.png')

⚖️ 二、因果效应估计

基于双重机器学习（Double ML）的效应估计：

from sklearn.ensemble import GradientBoostingRegressor
from econml.dml import LinearDML

# 定义估计器

est = LinearDML(
    model_y=GradientBoostingRegressor(),
    model_t=GradientBoostingRegressor(),
    discrete_treatment=False,
    cv=5
)

# 拟合模型

est.fit(
    y=df['default_rate'], 
    T=df['loan_amount'],
    X=df[['income', 'credit_score', 'age']],
    W=df[['employment_length', 'education_level']]
)

# 输出平均处理效应
print(est.ate_)

🔄 三、反事实预测系统

使用Pyro实现贝叶斯反事实推理：

import pyro
import pyro.distributions as dist
from pyro.infer import Predictive

def causal_model(data):
    
    # 先验分布
    
    income_coef = pyro.sample("income_coef", dist.Normal(0, 1))
    credit_coef = pyro.sample("credit_coef", dist.Normal(0, 1))
    
    # 因果关系
    
    with pyro.plate("data", len(data)):
        loan_amount = pyro.sample("loan_amount", 
            dist.Normal(
                income_coef * data['income'] + credit_coef * data['credit_score'],
                0.5
            )
        )
        default_rate = pyro.sample("default_rate",
            dist.Beta(
                1 + 0.1 * loan_amount,
                1 + 0.2 * data['credit_score']
            )
        )
    return default_rate

# 生成反事实样本

predictive = Predictive(causal_model, num_samples=1000)
samples = predictive(data)

📈 四、金融场景应用

1. 贷款额度优化

# 最优贷款额度计算

def optimize_loan(client_data):
    from scipy.optimize import minimize_scalar
    
    def expected_loss(loan_amount):
        client_data['loan_amount'] = loan_amount
        samples = predictive(client_data)
        return samples['default_rate'].mean()
    
    result = minimize_scalar(expected_loss, bounds=(1000, 50000))
    return result.x
    

2. 策略干预评估

# 评估利率调整政策效果

def policy_effect(data, rate_change):
    model = CausalModel(
        data=data.assign(interest_rate=lambda x: x['interest_rate'] + rate_change),
        treatment='interest_rate',
        outcome='default_rate',
        graph="interest_rate -> default_rate; loan_amount -> default_rate;"
    )
    
    identified_estimand = model.identify_effect()
    estimate = model.estimate_effect(
        identified_estimand,
        method_name="backdoor.linear_regression"
    )
    return estimate.value
    

🔧 五、系统部署方案

生产环境部署架构：

• 实时推理服务：FastAPI封装Pyro模型

• 特征工程管道：使用Feast特征存储

• 监控系统：Prometheus收集指标

# FastAPI服务示例

from fastapi import FastAPI
import pandas as pd

app = FastAPI()

@app.post("/predict")
async def predict(data: dict):
    df = pd.DataFrame([data])
    optimal_loan = optimize_loan(df)
    return {"optimal_loan_amount": optimal_loan}
    

💡 六、前沿技术展望

因果推理最新研究方向：

• 动态因果发现与时间序列分析

• 非参数因果效应估计

• 因果强化学习框架