可解释性：不是为了合规，而是调试

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import shap
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

# 定义简单的神经网络模型
class SimpleNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(SimpleNN, self).__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(hidden_size, output_size)
        self.sigmoid = nn.Sigmoid()
    
    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        x = self.sigmoid(x)
        return x

# 加载数据集
data = load_breast_cancer()
X = data.data
y = data.target
feature_names = data.feature_names

# 数据预处理
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2, random_state=42)

# 转换为PyTorch张量
X_train_tensor = torch.FloatTensor(X_train)
y_train_tensor = torch.FloatTensor(y_train).unsqueeze(1)
X_test_tensor = torch.FloatTensor(X_test)
y_test_tensor = torch.FloatTensor(y_test).unsqueeze(1)

# 初始化模型
input_size = X_train.shape[1]
hidden_size = 32
output_size = 1
model = SimpleNN(input_size, hidden_size, output_size)

# 定义损失函数和优化器
criterion = nn.BCELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

# 训练模型
epochs = 100
train_losses = []
test_losses = []

for epoch in range(epochs):
    # 训练模式
    model.train()
    optimizer.zero_grad()
    outputs = model(X_train_tensor)
    loss = criterion(outputs, y_train_tensor)
    loss.backward()
    optimizer.step()
    train_losses.append(loss.item())
    
    # 评估模式
    model.eval()
    with torch.no_grad():
        test_outputs = model(X_test_tensor)
        test_loss = criterion(test_outputs, y_test_tensor)
        test_losses.append(test_loss.item())
    
    if (epoch+1) % 10 == 0:
        print(f'Epoch [{epoch+1}/{epochs}], Train Loss: {loss.item():.4f}, Test Loss: {test_loss.item():.4f}')

# 计算模型准确率
model.eval()
with torch.no_grad():
    train_outputs = model(X_train_tensor)
    train_pred = (train_outputs >= 0.5).float()
    train_acc = (train_pred == y_train_tensor).sum().item() / y_train_tensor.shape[0]
    
    test_outputs = model(X_test_tensor)
    test_pred = (test_outputs >= 0.5).float()
    test_acc = (test_pred == y_test_tensor).sum().item() / y_test_tensor.shape[0]
    
print(f'Train Accuracy: {train_acc:.4f}, Test Accuracy: {test_acc:.4f}')

# 使用SHAP解释模型
# 创建SHAP解释器
explainer = shap.DeepExplainer(model, X_train_tensor)

# 计算SHAP值
shap_values = explainer.shap_values(X_test_tensor)

# 将SHAP值转换为numpy数组
shap_values = np.array(shap_values)[0]  # 取第一个输出的SHAP值

# 绘制汇总图
plt.figure(figsize=(12, 8))
shap.summary_plot(shap_values, X_test, feature_names=feature_names, plot_type="bar")
plt.title("特征重要性SHAP汇总图")
plt.tight_layout()
plt.savefig('shap_summary.png')
print("SHAP汇总图已保存为 shap_summary.png")

# 绘制单个样本的SHAP值图
plt.figure(figsize=(12, 8))
shap.initjs()
shap.force_plot(
    explainer.expected_value[0],
    shap_values[0],
    X_test[0],
    feature_names=feature_names,
    matplotlib=True,
    show=False
)
plt.title("单个样本的SHAP值图")
plt.tight_layout()
plt.savefig('shap_force.png')
print("单个样本的SHAP力导向图已保存为 shap_force.png")

# 绘制部分依赖图
plt.figure(figsize=(12, 8))
shap.dependence_plot(0, shap_values, X_test, feature_names=feature_names, show=False)
plt.title("第一个特征的部分依赖图")
plt.tight_layout()
plt.savefig('shap_dependence.png')
print("部分依赖图已保存为 shap_dependence.png")

# 分析攻击路径（模拟）
# 假设我们有一个恶意样本，分析模型如何被欺骗
malicious_sample = X_test[0].copy()

# 修改样本，使其成为恶意样本
malicious_sample[0] += 5  # 增加第一个特征的值
malicious_sample[1] += 5  # 增加第二个特征的值

# 转换为张量
malicious_sample_tensor = torch.FloatTensor(malicious_sample.reshape(1, -1))

# 模型预测
with torch.no_grad():
    original_pred = model(torch.FloatTensor(X_test[0].reshape(1, -1))).item()
    malicious_pred = model(malicious_sample_tensor).item()
    
print(f"原始样本预测概率: {original_pred:.4f}")
print(f"恶意样本预测概率: {malicious_pred:.4f}")

# 计算恶意样本的SHAP值
malicious_shap_values = explainer.shap_values(malicious_sample_tensor)[0]

# 绘制恶意样本的SHAP值图
plt.figure(figsize=(12, 8))
shap.force_plot(
    explainer.expected_value[0],
    malicious_shap_values,
    malicious_sample,
    feature_names=feature_names,
    matplotlib=True,
    show=False
)
plt.title("恶意样本的SHAP力导向图")
plt.tight_layout()
plt.savefig('malicious_shap_force.png')
print("恶意样本的SHAP力导向图已保存为 malicious_shap_force.png")

# 比较原始样本和恶意样本的SHAP值差异
diff_shap = malicious_shap_values - shap_values[0]

# 创建差异数据框
diff_df = pd.DataFrame({
    'Feature': feature_names,
    'Original_SHAP': shap_values[0],
    'Malicious_SHAP': malicious_shap_values,
    'SHAP_Diff': diff_shap
})

# 按SHAP差异排序
diff_df = diff_df.sort_values(by='SHAP_Diff', ascending=False)

print("\n原始样本与恶意样本的SHAP值差异（按影响从大到小排序）：")
print(diff_df.head(10))

# 绘制差异图
plt.figure(figsize=(12, 8))
plt.barh(diff_df['Feature'][:10], diff_df['SHAP_Diff'][:10])
plt.xlabel('SHAP值差异')
plt.ylabel('特征')
plt.title('原始样本与恶意样本的SHAP值差异（Top 10）')
plt.tight_layout()
plt.savefig('shap_diff.png')
print("SHAP值差异图已保存为 shap_diff.png")

import numpy as np
import pandas as pd
import shap
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

# 自适应解释框架类
class AdaptiveExplainer:
    def __init__(self, model, X_train, feature_names=None):
        self.model = model
        self.X_train = X_train
        self.feature_names = feature_names
        self.explainers = {}
        
        # 初始化不同类型的解释器
        self._init_explainers()
    
    def _init_explainers(self):
        """初始化不同类型的解释器"""
        # SHAP解释器
        self.explainers['shap'] = shap.TreeExplainer(self.model, self.X_train)
        
    def _select_explanation_strategy(self, X, task_type='classification', user_level='expert'):
        """根据数据、任务和用户水平选择解释策略"""
        # 简单策略选择，实际应用中可以更复杂
        if task_type == 'classification':
            if user_level == 'expert':
                return 'shap_force'  # 专家用户需要详细的力导向图
            else:
                return 'shap_summary'  # 普通用户需要直观的汇总图
        elif task_type == 'regression':
            return 'shap_dependence'  # 回归任务需要依赖图
        else:
            return 'shap_summary'  # 默认使用汇总图
    
    def explain(self, X, task_type='classification', user_level='expert'):
        """生成自适应解释"""
        # 选择解释策略
        strategy = self._select_explanation_strategy(X, task_type, user_level)
        
        # 生成SHAP值
        shap_values = self.explainers['shap'].shap_values(X)
        
        # 根据策略生成解释
        if strategy == 'shap_summary':
            return self._generate_summary_explanation(shap_values, X)
        elif strategy == 'shap_force':
            return self._generate_force_explanation(shap_values, X)
        elif strategy == 'shap_dependence':
            return self._generate_dependence_explanation(shap_values, X)
        else:
            return self._generate_summary_explanation(shap_values, X)
    
    def _generate_summary_explanation(self, shap_values, X):
        """生成汇总解释"""
        print("生成SHAP汇总解释...")
        
        # 绘制汇总图
        import matplotlib.pyplot as plt
        plt.figure(figsize=(12, 8))
        shap.summary_plot(shap_values, X, feature_names=self.feature_names, plot_type="bar")
        plt.title("特征重要性SHAP汇总图")
        plt.tight_layout()
        plt.savefig('adaptive_shap_summary.png')
        print("自适应SHAP汇总图已保存为 adaptive_shap_summary.png")
        
        return {
            'type': 'summary',
            'plot_path': 'adaptive_shap_summary.png',
            'explanation': '此图显示了每个特征对模型决策的平均影响，帮助理解模型的整体行为。'
        }
    
    def _generate_force_explanation(self, shap_values, X):
        """生成力导向解释"""
        print("生成SHAP力导向解释...")
        
        # 绘制第一个样本的力导向图
        import matplotlib.pyplot as plt
        plt.figure(figsize=(12, 8))
        shap.initjs()
        shap.force_plot(
            self.explainers['shap'].expected_value[0],
            shap_values[0],
            X[0],
            feature_names=self.feature_names,
            matplotlib=True,
            show=False
        )
        plt.title("单个样本的SHAP力导向图")
        plt.tight_layout()
        plt.savefig('adaptive_shap_force.png')
        print("自适应SHAP力导向图已保存为 adaptive_shap_force.png")
        
        return {
            'type': 'force',
            'plot_path': 'adaptive_shap_force.png',
            'explanation': '此图显示了每个特征对单个样本决策的具体影响，帮助理解模型的局部行为。'
        }
    
    def _generate_dependence_explanation(self, shap_values, X):
        """生成依赖图解释"""
        print("生成SHAP依赖图解释...")
        
        # 绘制第一个特征的依赖图
        import matplotlib.pyplot as plt
        plt.figure(figsize=(12, 8))
        shap.dependence_plot(0, shap_values, X, feature_names=self.feature_names, show=False)
        plt.title("第一个特征的部分依赖图")
        plt.tight_layout()
        plt.savefig('adaptive_shap_dependence.png')
        print("自适应SHAP依赖图已保存为 adaptive_shap_dependence.png")
        
        return {
            'type': 'dependence',
            'plot_path': 'adaptive_shap_dependence.png',
            'explanation': '此图显示了单个特征与模型输出之间的关系，帮助理解特征的非线性影响。'
        }

# 示例用法
if __name__ == "__main__":
    # 加载数据集
    data = load_iris()
    X = data.data
    y = data.target
    feature_names = data.feature_names
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    
    # 训练模型
    model = RandomForestClassifier(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)
    
    # 创建自适应解释器
    adaptive_explainer = AdaptiveExplainer(model, X_train, feature_names=feature_names)
    
    # 生成不同用户水平的解释
    print("\n=== 为专家用户生成解释 ===")
    expert_explanation = adaptive_explainer.explain(X_test[:1], task_type='classification', user_level='expert')
    
    print("\n=== 为普通用户生成解释 ===")
    novice_explanation = adaptive_explainer.explain(X_test[:1], task_type='classification', user_level='novice')
    
    # 生成回归任务的解释（模拟）
    print("\n=== 为回归任务生成解释 ===")
    regression_explanation = adaptive_explainer.explain(X_test[:1], task_type='regression', user_level='expert')
    
    # 评估模型准确率
    accuracy = model.score(X_test, y_test)
    print(f"\n模型准确率: {accuracy:.4f}")

import numpy as np
import pandas as pd
import dowhy
from dowhy import CausalModel
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error

# 生成模拟数据
def generate_causal_data(n_samples=1000, seed=42):
    np.random.seed(seed)
    
    # 生成特征
    X1 = np.random.normal(0, 1, n_samples)  # 外生变量1
    X2 = np.random.normal(0, 1, n_samples)  # 外生变量2
    U = np.random.normal(0, 1, n_samples)   # 未观察到的混杂因子
    
    # 生成处理变量T（受X1和U影响）
    T = 0.5 * X1 + 0.3 * U + np.random.normal(0, 0.1, n_samples)
    
    # 生成结果变量Y（受T、X2和U影响）
    Y = 2.0 * T + 0.5 * X2 + 0.2 * U + np.random.normal(0, 0.1, n_samples)
    
    # 创建数据框
    data = pd.DataFrame({
        'X1': X1,
        'X2': X2,
        'T': T,
        'Y': Y
    })
    
    return data

# 生成数据
data = generate_causal_data()
print("生成的数据示例：")
print(data.head())

# 步骤1：构建因果模型
model = CausalModel(
    data=data,
    treatment='T',
    outcome='Y',
    common_causes=['X1', 'X2'],  # 观察到的混杂因子
    instruments=None  # 工具变量
)

# 步骤2：识别因果效应
identified_estimand = model.identify_effect()
print("\n识别的因果效应：")
print(identified_estimand)

# 步骤3：估计因果效应
# 使用倾向评分匹配方法
estimate = model.estimate_effect(
    identified_estimand,
    method_name="backdoor.propensity_score_matching"
)
print("\n估计的平均因果效应（ATE）：")
print(estimate)

# 使用多种方法估计，验证结果
methods = [
    "backdoor.propensity_score_matching",
    "backdoor.propensity_score_weighting",
    "backdoor.linear_regression"
]

print("\n使用多种方法估计因果效应：")
for method in methods:
    estimate = model.estimate_effect(
        identified_estimand,
        method_name=method
    )
    print(f"{method}: {estimate.value:.4f}")

# 步骤4：反驳估计结果
print("\n=== 反驳测试 ===")

# 1. 随机化处理变量
refute1_results = model.refute_estimate(
    identified_estimand, estimate, method_name="random_common_cause"
)
print(f"随机化混杂因子: {refute1_results}")

# 2. 安慰剂处理
refute2_results = model.refute_estimate(
    identified_estimand, estimate, method_name="placebo_treatment_refuter"
)
print(f"安慰剂处理: {refute2_results}")

# 3. 数据子集验证
refute3_results = model.refute_estimate(
    identified_estimand, estimate, method_name="data_subset_refuter", subset_fraction=0.8
)
print(f"数据子集验证: {refute3_results}")

# 使用因果特征选择
print("\n=== 因果特征选择 ===")

# 生成包含无关特征的数据
np.random.seed(42)
data_with_noise = data.copy()
for i in range(5):
    data_with_noise[f'Noise_{i}'] = np.random.normal(0, 1, len(data))

# 定义因果图
causal_graph = """
digraph {
    X1 -> T;
    X2 -> Y;
    T -> Y;
    U -> T;
    U -> Y;
    Noise_0 -> Y [arrowhead="none"];  # 无关特征，没有因果关系
    Noise_1 -> Y [arrowhead="none"];
    Noise_2 -> Y [arrowhead="none"];
    Noise_3 -> Y [arrowhead="none"];
    Noise_4 -> Y [arrowhead="none"];
}
"""

# 使用包含无关特征的数据构建模型
model_with_noise = CausalModel(
    data=data_with_noise,
    treatment='T',
    outcome='Y',
    graph=causal_graph
)

# 识别因果效应
identified_estimand_with_noise = model_with_noise.identify_effect()

# 估计因果效应（使用线性回归，包含所有特征）
estimate_with_noise = model_with_noise.estimate_effect(
    identified_estimand_with_noise,
    method_name="backdoor.linear_regression",
    control_value=0,
    treatment_value=1
)

print(f"包含无关特征的线性回归ATE估计: {estimate_with_noise.value:.4f}")

# 比较基于相关性的特征重要性和基于因果性的特征重要性
print("\n=== 相关性 vs 因果性特征重要性 ===")

# 基于相关性的特征重要性
corr_importance = data_with_noise.corr()['Y'].abs().sort_values(ascending=False)
print("基于相关性的特征重要性：")
print(corr_importance)

# 基于因果性的特征重要性（使用因果图分析）
causal_features = ['T', 'X2']  # 根据因果图，只有T和X2直接影响Y
print("\n基于因果图的特征重要性：")
print(f"直接因果特征: {causal_features}")
print(f"无关特征: {[col for col in data_with_noise.columns if col not in causal_features + ['Y', 'X1']]}")

# 反事实解释示例
print("\n=== 反事实解释示例 ===")

# 选择一个样本
sample = data.iloc[0]
print(f"原始样本: {sample.to_dict()}")

# 训练预测模型
from sklearn.linear_model import LinearRegression

X = data[['T', 'X1', 'X2']]
y = data['Y']

predict_model = LinearRegression()
predict_model.fit(X, y)

# 原始预测
original_pred = predict_model.predict(sample[['T', 'X1', 'X2']].values.reshape(1, -1))[0]
print(f"原始预测: {original_pred:.4f}")

# 反事实：如果T增加1，预测结果会怎样
counterfactual_sample = sample.copy()
counterfactual_sample['T'] += 1

counterfactual_pred = predict_model.predict(counterfactual_sample[['T', 'X1', 'X2']].values.reshape(1, -1))[0]
print(f"反事实预测（T增加1）: {counterfactual_pred:.4f}")
print(f"预测变化: {counterfactual_pred - original_pred:.4f}")

# 基于因果估计的反事实预测
causal_effect = estimate.value  # 使用之前估计的因果效应
causal_counterfactual_pred = original_pred + causal_effect
print(f"基于因果效应的反事实预测: {causal_counterfactual_pred:.4f}")
print(f"因果效应: {causal_effect:.4f}")

# 可视化因果关系
import networkx as nx
import matplotlib.pyplot as plt

# 构建因果图
G = nx.DiGraph()
G.add_edges_from([
    ('X1', 'T'),
    ('U', 'T'),
    ('U', 'Y'),
    ('T', 'Y'),
    ('X2', 'Y')
])

# 绘制因果图
plt.figure(figsize=(10, 6))
pos = nx.spring_layout(G, seed=42)
nx.draw(G, pos, with_labels=True, node_size=2000, node_color='lightblue', font_size=12, font_weight='bold', arrows=True, arrowsize=20)
nx.draw_networkx_edge_labels(G, pos, edge_labels={('X1', 'T'): '0.5', ('U', 'T'): '0.3', ('U', 'Y'): '0.2', ('T', 'Y'): '2.0', ('X2', 'Y'): '0.5'})
plt.title('因果图可视化')
plt.tight_layout()
plt.savefig('causal_graph.png')
print("\n因果图已保存为 causal_graph.png")

# 安装基本依赖
pip install numpy pandas scikit-learn matplotlib seaborn

# 安装深度学习框架
pip install torch torchvision tensorflow

# 安装可解释性库
pip install shap lime dowhy interpret captum alibi

# 安装因果推断库
pip install causality==0.0.1 causalnex==0.11.0

# 安装可视化库
pip install networkx pydot graphviz

# 安装安全相关库
pip install adversarial-robustness-toolbox