┌─────────────────────────────────────────────────────────────┐
│                    应用层：决策仿真与建议生成                  │
│  ┌──────────────┐  ┌──────────────┐  ┌────────────────────┐  │
│  │ 反事实干预   │  │ 效应分解归因 │  │ 决策路径优化推荐   │  │
│  │ 模拟器       │  │ 看板         │  │ 引擎               │  │
│  └──────────────┘  └──────────────┘  └────────────────────┘  │
├─────────────────────────────────────────────────────────────┤
│                    推理层：因果效应估计核心                    │
│  ┌──────────────┐  ┌──────────────┐  ┌────────────────────┐  │
│  │ 反事实生成   │  │ 去偏估计器   │  │ 敏感性分析模块     │  │
│  │ 引擎 (CFE)   │  │ (IPW/DR/TMLE)│  │ (E-value/置信区间) │  │
│  └──────────────┘  └──────────────┘  └────────────────────┘  │
├─────────────────────────────────────────────────────────────┤
│                    因果图层：结构化知识表达                    │
│  ┌──────────────┐  ┌──────────────┐  ┌────────────────────┐  │
│  │ 领域知识图谱 │  │ 因果发现算法 │  │ 图校验与冲突消解   │  │
│  │ 构建 (KG)    │  │ (NOTEARS/PC) │  │ 机制               │  │
│  └──────────────┘  └──────────────┘  └────────────────────┘  │
├─────────────────────────────────────────────────────────────┤
│                    数据层：多源异构会议数据治理                │
│  语音转文本 → 实体关系抽取 → 会议结构化日志 → 特征工程管线      │
└─────────────────────────────────────────────────────────────┘

# 伪代码：CFE 核心前向过程
class CounterfactualGenerator(nn.Module):
    def __init__(self, causal_graph: nx.DiGraph, dim_latent: int = 256):
        super().__init__()
        self.causal_graph = causal_graph
        self.encoder = TransformerEncoder(d_model=dim_latent)
        self.diffusion = ConditionalDiffusion(
            condition_dim=dim_latent + len(T),  # X + 目标干预 T'
            target_dim=dim_latent
        )
        self.decoder = TransformerDecoder(d_model=dim_latent)
        # 结构约束：邻接矩阵掩码，强制生成符合因果拓扑
        self.structural_mask = self._build_structural_mask(causal_graph)

    def forward(self, X_obs: Tensor, T_target: Tensor) -> Tensor:
        # 1. 编码观测协变量
        z_obs = self.encoder(X_obs)  # [B, D]
        
        # 2. 条件扩散生成反事实潜变量
        # 条件 = [z_obs, T_target]，目标 = z_cf
        z_cf = self.diffusion.sample(condition=torch.cat([z_obs, T_target], dim=-1))
        
        # 3. 结构约束投影：确保生成的 z_cf 满足因果图拓扑约束
        z_cf = self._project_to_causal_manifold(z_cf, self.structural_mask)
        
        # 4. 解码为完整反事实样本 (X_cf, Y_cf)
        X_cf, Y_cf = self.decoder(z_cf)
        return X_cf, Y_cf

def identify_adjustment_set(causal_graph: nx.DiGraph, treatment: str, outcome: str) -> List[str]:
    """
    返回满足后门准则的最小调整集列表（可能有多个等价集合）
    """
    # 1. 枚举所有后门路径
    backdoor_paths = find_all_backdoor_paths(causal_graph, treatment, outcome)
    
    # 2. 构建 hitting set 问题：寻找最小节点集覆盖所有路径
    # 使用 ILP 求解器或贪心近似
    adjustment_sets = solve_min_hitting_set(backdoor_paths, 
                                            forbidden_nodes=descendants(causal_graph, treatment))
    
    # 3. 可行性校验：剔除不可观测/高缺失率变量
    feasible_sets = [s for s in adjustment_sets if check_feasibility(s, data_catalog)]
    
    return feasible_sets

def select_estimator(data_meta: DataMeta, causal_graph: nx.DiGraph) -> EstimatorType:
    # 启发式规则 + 元学习模型
    if data_meta.treatment_type == 'continuous':
        return 'DR-Continuous'  # 广义倾向分 + 结果模型
    if data_meta.missing_rate > 0.3 and data_meta.mnar_evidence:
        return 'TMLE'  # 对 MNAR 鲁棒性更强
    if data_meta.sample_size < 5000:
        return 'IPW-Stabilized'  # 小样本下方差可控
    return 'DR-CrossFit'  # 默认推荐：双重稳健 + 交叉拟合

## 敏感性分析报告示例

**因果效应估计**：引入技术评审环节 → 决策执行偏差率降低 12.3% (95% CI: [8.1%, 16.5%])

**E-value (点估计)**：2.41
> 解释：未观测混杂因子需同时与“技术评审”和“执行偏差”呈 2.41 倍风险比关联，才能将效应解释为零。

**E-value (CI 下限)**：1.67
> 解释：即使在置信区间下限，未观测混杂也需 1.67 倍关联强度才能使效应不显著。

**Rosenbaum Γ 边界**：Γ = 1.8 时 p-value > 0.05
> 解释：若隐性偏倚导致同协变量组内处理概率比达 1.8 倍，结论将不再显著。

**业务判读**：当前组织文化、隐性政治联盟等未观测因子极难达到 2.4 倍关联强度，结论具备工程鲁棒性。

class ContinuousTimeSCM(nn.Module):
    """
    连续时间结构因果模型
    dZ/dt = f_θ(Z(t), T(t), t) + g_φ(Z(t)) * dW_t  (随机微分方程可选)
    """
    def __init__(self, dim_latent: int, causal_graph: nx.DiGraph):
        super().__init__()
        self.causal_graph = causal_graph
        # 邻接矩阵掩码，强制稀疏连接符合因果拓扑
        self.adj_mask = torch.tensor(nx.adjacency_matrix(causal_graph).todense(), dtype=torch.float32)
        
        # 向量场网络：输入 [Z, T, t] -> 输出 dZ/dt
        self.vector_field = nn.Sequential(
            nn.Linear(dim_latent + dim_treatment + 1, 256),
            nn.Softplus(),  # 保证流形平滑性
            nn.Linear(256, dim_latent)
        )
        
        # 干预响应网络：单独建模 T(t) 对 Z 的直接冲击
        self.intervention_encoder = nn.Linear(dim_treatment, dim_latent)

    def forward(self, z0: Tensor, t_span: Tensor, T_traj: Callable[[float], Tensor]) -> Tensor:
        """
        z0: 初始潜状态 [B, D]
        t_span: 积分时间点 [T_steps]
        T_traj: 处理轨迹函数 t -> [B, dim_T]
        返回: 轨迹 [B, T_steps, D]
        """
        def ode_func(t, z):
            # 1. 获取当前时刻处理
            T_curr = T_traj(t)  # [B, dim_T]
            
            # 2. 结构约束：仅允许父节点影响子节点
            # 利用邻接矩阵掩码实现消息传递
            z_structured = z @ self.adj_mask.T  # [B, D] 仅保留因果父节点信息
            
            # 3. 向量场计算
            inp = torch.cat([z_structured, T_curr, t.expand(z.shape[0], 1)], dim=-1)
            dz_dt = self.vector_field(inp)
            
            # 4. 干预直接效应注入 (类似残差连接)
            dz_dt = dz_dt + self.intervention_encoder(T_curr)
            return dz_dt
        
        # 使用自适应步长 ODE 求解器
        z_traj = odeint(ode_func, z0, t_span, method='dopri5', rtol=1e-4, atol=1e-6)
        return z_traj.permute(1, 0, 2)  # [B, T_steps, D]

class ContinuousDRFEstimator:
    def __init__(self, treatment_type: str = 'continuous'):
        # 1. GPS 模型: f(T | X) 通常假设正态分布或混合高斯
        self.gps_model = ConditionalNormalizingFlow(
            condition_dim=dim_X, 
            target_dim=1, 
            n_flows=8
        )
        # 2. 结果模型: μ(t, X) 使用 B-Spline 或 Gaussian Process 捕捉非线性
        self.outcome_model = BayesianAdditiveRegressionTrees(
            n_trees=200, 
            n_chains=4
        )
        # 3. 核加权回归带宽自适应选择
        self.bandwidth_selector = CrossValidatedBandwidth()
    
    def estimate_drf(self, X: Tensor, T: Tensor, Y: Tensor, t_grid: Tensor) -> Dict:
        """
        返回: {
            'drf': DRF(t_grid),           # 点估计
            'drf_ci': (lower, upper),     # 95% 置信带
            'mer': MER(t_grid),           # 边际效应
            'optimal_t': t_optimal,       # 论最大化收益/最小化成本的最优剂量
            'gps_diagnosis': {...}        # GPS 重叠度诊断
        }
        """
        # Step 1: 估计 GPS 密度
        log_gps = self.gps_model.log_prob(T, condition=X)  # [N]
        gps = torch.exp(log_gps)
        
        # Step 2: 交叉拟合 DR 估计
        drf_estimates = []
        for train_idx, val_idx in KFold(n_splits=5).split(X):
            # 训练结果模型
            self.outcome_model.fit(X[train_idx], T[train_idx], Y[train_idx])
            mu_hat = self.outcome_model.predict(X[val_idx], t_grid)  # [N_val, len(t_grid)]
            
            # 计算 GPS 权重 (稳定化)
            gps_val = gps[val_idx]
            gps_marginal = self.gps_model.marginal_prob(t_grid)  # [len(t_grid)]
            weights = (gps_marginal / gps_val.unsqueeze(1)).clamp(max=100)  # 裁剪极端权重
            
            # DR 校正项
            residuals = Y[val_idx].unsqueeze(1) - mu_hat  # [N_val, len(t_grid)]
            dr_correction = (weights * residuals).mean(dim=0)
            
            drf_fold = mu_hat.mean(dim=0) + dr_correction
            drf_estimates.append(drf_fold)
        
        drf = torch.stack(drf_estimates).mean(dim=0)
        
        # Step 3: Bootstrap 置信带 (并行化)
        ci_lower, ci_upper = self._bootstrap_ci(X, T, Y, t_grid, n_boot=500)
        
        # Step 4: 边际效应数值微分 + 平滑
        mer = torch.gradient(drf, spacing=t_grid)[0]
        mer = gaussian_filter1d(mer.numpy(), sigma=1.5)
        
        return {
            'drf': drf, 'drf_ci': (ci_lower, ci_upper), 
            'mer': torch.tensor(mer),
            'optimal_t': t_grid[torch.argmax(drf)]  # 简化：假设单峰最大化
        }

class HTEEngine:
    def __init__(self, meta_learner: str = 'auto'):
        self.meta_learner = self._select_meta_learner(meta_learner)
        self.base_learners = {
            'propensity': GradientBoostingRegressor(n_estimators=300, max_depth=4),
            'outcome': HistGradientBoostingRegressor(max_iter=500, max_depth=5),
            'cate': CausalForest(n_estimators=2000, min_samples_leaf=20, honest=True)
        }
    
    def _select_meta_learner(self, mode: str) -> BaseMetaLearner:
        if mode != 'auto': return REGISTRY[mode]
        # 启发式选择逻辑
        if self.data_meta.n_samples > 50000 and self.data_meta.treatment_cardinality > 2:
            return DR_Learner()  # 多重处理/连续处理首选
        elif self.data_meta.n_samples < 5000:
            return X_Learner()   # 小样本利用倾向分分层
        else:
            return R_Learner()   # 通用稳健选择
    
    def fit(self, X, T, Y):
        # 1. 交叉拟合生成伪结果
        pseudo_outcomes = self.meta_learner.compute_pseudo_outcomes(X, T, Y, self.base_learners)
        
        # 2. 训练 CATE 模型 (因果森林/变换树/神经网络)
        self.cate_model = self.base_learners['cate']
        self.cate_model.fit(X, pseudo_outcomes)
        
        # 3. 校准：确保 CATE 平均回归到 ATE
        self.calibrator = IsotonicRegression()
        self.calibrator.fit(self.cate_model.predict(X), Y - self.base_learners['outcome'].predict(X, T=0))
        
        return self
    
    def predict_cate(self, X_new: Tensor) -> Tensor:
        raw_cate = self.cate_model.predict(X_new)
        return self.calibrator.predict(raw_cate)
    
    def policy_learning(self, X: Tensor, cost_model: Callable) -> Policy:
        """
        学习最优个性化干预策略 π(X) -> T
        目标: max E[ CATE(X) * π(X) - Cost(π(X)) ]
        """
        cate = self.predict_cate(X)
        # 离散动作空间: 论证每个动作的净收益
        if self.action_space.is_discrete:
            net_benefit = cate.unsqueeze(-1) * self.action_effects - cost_model(self.actions)
            optimal_actions = self.actions[net_benefit.argmax(dim=-1)]
        # 连续动作空间: 梯度上升求解
        else:
            optimal_actions = self._continuous_policy_optimization(cate, cost_model)
        return Policy(optimal_actions)

## 子群体画像：技术评审高受益群 (CATE = +0.42 SD, 占比 18%)

**特征规则**：
- 议题标签 ∈ {架构重构, 核心库升级, 安全合规}
- 历史技术债务指数 > 75 分位
- 参会架构师数量 ≥ 2
- 会议时长预算 > 60 分钟

**干预建议**：
- ✅ 强制嵌入 30 分钟技术评审环节
- ✅ 邀请 ≥ 1 名外部领域专家
- ❌ 避免压缩讨论时间低于 45 分钟

**边际收益曲线**：评审时长 20-40 分钟边际收益递增，40 分钟达峰，之后递减。

graph LR
    A[原始语料: 会议纪要/访谈/制度文档] --> B(阶段1: 因果三元组抽取)
    B --> C{阶段2: 结构冲突消解}
    C --> D[阶段3: 图神经网络精排]
    D --> E[可审计因果图 v1.0]
    
    subgraph 阶段1: LLM抽取
    B1[Prompt: 识别 变量/处理/结果/混杂/中介] --> B2[输出: (原因, 结果, 机制文本, 置信度)]
    B2 --> B3[实体链接到业务数据字典]
    end
    
    subgraph 阶段2: 逻辑约束求解
    C1[约束1: 无环性] --> C2[约束2: 时间先后序]
    C2 --> C3[约束3: 业务必知边(如: 决策->执行)]
    C3 --> C4[ILP求解最大一致子图]
    end
    
    subgraph 阶段3: 数据驱动精排
    D1[NOTEARS/GOLEM 评分] --> D2[置换检验剔除伪相关]
    D2 --> D3[专家复核界面: 高亮低置信度边]
    end

CAUSAL_EXTRACTION_PROMPT = """
角色：资深组织行为学因果分析师
任务：从会议记录/访谈中抽取结构化因果知识三元组

【输入文本】
{raw_text}

【业务变量字典】(必须映射到标准变量名)
{variable_catalog}

【抽取规则】
1. 仅抽取**直接因果**关系，非相关性
2. 必须标注变量角色：Treatment/Outcome/Confounder/Mediator/Collider
3. 必须给出**机制文本证据** (原文片段)
4. 置信度打分 0-1：基于证据明确性、时间先后、排除混杂合理性
5. 识别**效应修饰变量** (Moderator)：影响因果强度的条件

【输出格式】JSON Lines:
{{"cause": "技术评审环节", "effect": "决策执行偏差率", "role": "Treatment->Outcome", 
  "mechanism": "评审暴露隐性风险，迫使方案修正", "evidence": "会议纪要p3: 经评审发现接口兼容性风险...", 
  "confidence": 0.87, "moderators": ["议题复杂度", "评审专家资深度"]}}
"""

def generate_semi_synthetic_benchmark(real_data: DataFrame, scm: NeuralSCM) -> BenchmarkSuite:
    """
    保留真实协变量分布 P(X)，注入已知因果机制
    """
    # 1. 拟合真实 X 分布 (CTGAN/TVAE)
    x_sampler = fit_tabular_gan(real_data[X_cols])
    
    # 2. 定义地面真理 SCM (可控非线性/交互/异质性)
    true_scm = ConfigurableSCM(
        treatment_effect_fn=lambda x: 0.5 + 0.3*x['complexity'] - 0.2*x['urgency'], # CATE
        confounding_strength=0.7,
        selection_bias_fn=lambda x: x['political_sensitivity'] > 0.8
    )
    
    # 3. 生成反事实完整数据集
    synth_data = true_scm.simulate(n=50000, x_sampler=x_sampler)
    
    # 4. 构建评估任务：ATE / CATE / DRF / Policy Value
    return BenchmarkSuite(
        ate_true=true_scm.ate(),
        cate_true=true_scm.cate_fn,
        drf_true=true_scm.drf_fn,
        optimal_policy_value=true_scm.max_policy_value(),
        data=synth_data
    )

# 服务器端
class FedCausalServer:
    def __init__(self, global_cate_model: nn.Module):
        self.global_model = global_cate_model
        self.optimizer = torch.optim.Adam(self.global_model.parameters(), lr=1e-3)
    
    def aggregate(self, client_updates: List[Dict]) -> Dict:
        """
        client_updates: [
            {'pseudo_outcome_grad': Tensor, 'hessian_diag': Tensor, 
             'n_samples': int, 'local_ate': float, 'metrics': {...}}, ...
        ]
        """
        # FedAvg 风格聚合伪结果梯度 (或使用 FedProx/SCAFFOLD 修正漂移)
        total_samples = sum(u['n_samples'] for u in client_updates)
        global_grad = sum(u['pseudo_outcome_grad'] * u['n_samples'] for u in client_updates) / total_samples
        
        # 服务器端更新全局 CATE 模型
        self.optimizer.zero_grad()
        for p, g in zip(self.global_model.parameters(), global_grad):
            p.grad = g
        self.optimizer.step()
        
        # 下发新模型权重
        return {'global_weights': self.global_model.state_dict(), 'round': self.round}

# 客户端 (各部门/公司)
class FedCausalClient:
    def __init__(self, local_data: DataFrame, global_model: nn.Module):
        self.data = local_data
        self.local_model = copy.deepcopy(global_model)
        # 本地估计器 (DR-Learner)
        self.estimator = DRLearner(base_models=...)
    
    def local_update(self, global_weights: Dict) -> Dict:
        self.local_model.load_state_dict(global_weights)
        
        # 1. 本地估计倾向分 e(X) 和结果模型 μ(T,X) (仅用本地数据)
        self.estimator.fit(self.data)
        
        # 2. 计算伪结果 (R-learner 风格)
        # pseudo = (Y - μ_0(X)) / e(X) * T - (Y - μ_1(X)) / (1-e(X)) * (1-T)  [二元简化]
        pseudo_outcomes = self.estimator.compute_pseudo_outcomes(self.data)
        
        # 3. 本地计算 CATE 模型梯度 (冻结表示层，仅训练头？或全量)
        # 目标: min || CATE_model(X) - pseudo_outcomes ||^2
        loss = F.mse_loss(self.local_model(self.data.X), pseudo_outcomes)
        loss.backward()
        
        # 4. 可选：计算海森对角用于二阶聚合 (FedNewton)
        hess_diag = self._compute_hessian_diag()
        
        return {
            'pseudo_outcome_grad': [p.grad.clone() for p in self.local_model.parameters()],
            'hessian_diag': hess_diag,
            'n_samples': len(self.data),
            'local_ate': self.estimator.ate_,
            'metrics': self.estimator.diagnostics_
        }

# Prometheus 规则示例
groups:
- name: causal_engine_alerts
  rules:
  # 数据漂移
  - alert: CovariateShiftDetected
    expr: ks_test_pvalue{covariate=~".*"} < 0.01
    for: 1h
    labels: {severity: warning}
    annotations: {summary: "协变量分布漂移，需重训练倾向分模型"}
  
  # 正交性破坏 (混杂残留)
  - alert: OrthogonalityViolation
    expr: abs(corr(residual, treatment | X)) > 0.1
    for: 30m
    labels: {severity: critical}
    annotations: {summary: "去偏失效，疑似新增未观测混杂"}
  
  # 反事实生成质量下降
  - alert: CFGenerationQualityDrop
    expr: structural_consistency_rate < 0.9
    for: 15m
    labels: {severity: warning}
  
  # 业务指标回测偏离
  - alert: PolicyRegretHigh
    expr: (predicted_policy_value - realized_value) / predicted_policy_value > 0.2
    for: 1d
    labels: {severity: critical}

def logic_regularization_loss(model: NeuralSCM, logic_rules: List[FOLRule]) -> Tensor:
    """
    使用 T-Norm 模糊逻辑软化硬约束
    规则: A -> B  转化为: 1 - t_norm(A, 1-B)  (Gödel/Łukasiewicz/Product T-Norm)
    """
    loss = 0.0
    for rule in logic_rules:
        # 采样反事实场景验证规则满足度
        X_cf = sample_counterfactual_contexts(rule.variables)
        preds = model.predict_proba(X_cf)  # P(B|A, X_cf)
        
        # 模糊蕴涵满足度
        satisfaction = fuzzy_implication(rule.antecedent(X_cf), preds, t_norm='product')
        loss += (1 - satisfaction).mean()
    return loss * LAMBDA_LOGIC

class CausalDecisionAgent:
    def __init__(self, engine: CausalEngine, env: MeetingEnv):
        self.engine = engine
        self.env = env
        self.planner = LLMPlanner(tools=[
            engine.estimate_ate, engine.optimize_drf, 
            engine.discover_heterogeneity, env.deploy_policy, env.monitor_kpi
        ])
    
    def run_optimization_cycle(self, business_goal: str):
        # 1. 理解目标 -> 因果查询
        query = self.planner.think(f"目标: {business_goal}. 需要回答哪些因果问题?")
        
        # 2. 执行因果分析流水线
        evidence = self.engine.execute_query_plan(query)
        
        # 3. 生成干预策略组合
        policies = self.engine.learn_optimal_policies(evidence, constraints=BUSINESS_RULES)
        
        # 4. 设计最小化遗憾实验
        exp_design = self.planner.design_experiment(policies, budget=EXPERIMENT_BUDGET)
        
        # 5. 部署与监控
        task_id = self.env.deploy(exp_design)
        monitor = self.env.monitor(task_id, callback=self._on_anomaly)
        
        return OptimizationReport(task_id, policies, exp_design, monitor)