数据清洗对模型性能的真实影响：安全视角下的防投毒实践

"""
安全数据清洗脚本，包含去除重复数据、处理缺失值、纠正错误数据和检测异常值
"""

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SecurityDataCleaner:
    """安全数据清洗器"""
    
    def __init__(self, contamination=0.05):
        """初始化数据清洗器
        
        Args:
            contamination: 异常值比例
        """
        self.contamination = contamination
        self.scaler = StandardScaler()
        self.outlier_detector = IsolationForest(contamination=contamination, random_state=42)
        logger.info(f"初始化安全数据清洗器，异常值比例：{contamination}")
    
    def remove_duplicates(self, data):
        """去除重复数据
        
        Args:
            data: 原始数据，DataFrame格式
            
        Returns:
            去除重复数据后的DataFrame
        """
        logger.info(f"开始去除重复数据，原始样本数：{len(data)}")
        data_cleaned = data.drop_duplicates()
        duplicate_count = len(data) - len(data_cleaned)
        logger.info(f"去除重复数据完成，去除样本数：{duplicate_count}，剩余样本数：{len(data_cleaned)}")
        return data_cleaned
    
    def handle_missing_values(self, data, method="mean"):
        """处理缺失值
        
        Args:
            data: 数据，DataFrame格式
            method: 处理方法，可选：mean, median, mode, drop
            
        Returns:
            处理缺失值后的DataFrame
        """
        logger.info(f"开始处理缺失值，方法：{method}")
        
        # 统计缺失值情况
        missing_info = data.isnull().sum()
        logger.info(f"缺失值统计：\n{missing_info}")
        
        if method == "drop":
            data_cleaned = data.dropna()
        elif method == "mean":
            data_cleaned = data.fillna(data.mean())
        elif method == "median":
            data_cleaned = data.fillna(data.median())
        elif method == "mode":
            data_cleaned = data.fillna(data.mode().iloc[0])
        else:
            raise ValueError(f"不支持的缺失值处理方法：{method}")
        
        logger.info(f"处理缺失值完成，剩余样本数：{len(data_cleaned)}")
        return data_cleaned
    
    def correct_error_data(self, data, rules):
        """根据规则纠正错误数据
        
        Args:
            data: 数据，DataFrame格式
            rules: 纠正规则字典，格式：{"column": {"condition": "value"}}
            
        Returns:
            纠正错误数据后的DataFrame
        """
        logger.info(f"开始纠正错误数据，规则数量：{len(rules)}")
        
        data_cleaned = data.copy()
        for column, rule in rules.items():
            for condition, value in rule.items():
                # 解析条件，这里简化处理，仅支持基本比较
                if ">" in condition:
                    threshold = float(condition.split(">").strip())
                    data_cleaned.loc[data_cleaned[column] > threshold, column] = value
                elif "<" in condition:
                    threshold = float(condition.split("<").strip())
                    data_cleaned.loc[data_cleaned[column] < threshold, column] = value
                elif "==" in condition:
                    target = condition.split("==").strip()
                    data_cleaned.loc[data_cleaned[column] == target, column] = value
        
        logger.info("纠正错误数据完成")
        return data_cleaned
    
    def detect_outliers(self, data, features):
        """检测异常值
        
        Args:
            data: 数据，DataFrame格式
            features: 用于检测异常值的特征列表
            
        Returns:
            检测异常值后的DataFrame，添加is_outlier列
        """
        logger.info(f"开始检测异常值，使用特征：{features}")
        
        # 提取特征数据
        X = data[features].values
        
        # 特征缩放
        X_scaled = self.scaler.fit_transform(X)
        
        # 训练异常检测模型
        self.outlier_detector.fit(X_scaled)
        
        # 检测异常值
        outliers = self.outlier_detector.predict(X_scaled)
        
        # 添加异常值标记列
        data_cleaned = data.copy()
        data_cleaned["is_outlier"] = outliers
        
        # 统计异常值数量
        outlier_count = np.sum(outliers == -1)
        logger.info(f"检测异常值完成，异常值数量：{outlier_count}/{len(data_cleaned)}")
        
        return data_cleaned
    
    def filter_outliers(self, data):
        """过滤异常值
        
        Args:
            data: 带有is_outlier列的DataFrame
            
        Returns:
            过滤异常值后的DataFrame
        """
        logger.info(f"开始过滤异常值，原始样本数：{len(data)}")
        
        data_cleaned = data[data["is_outlier"] == 1].drop(columns=["is_outlier"])
        outlier_count = len(data) - len(data_cleaned)
        
        logger.info(f"过滤异常值完成，过滤样本数：{outlier_count}，剩余样本数：{len(data_cleaned)}")
        return data_cleaned
    
    def full_clean(self, data, features, rules=None, missing_method="mean"):
        """完整的数据清洗流程
        
        Args:
            data: 原始数据，DataFrame格式
            features: 用于检测异常值的特征列表
            rules: 纠正规则字典
            missing_method: 缺失值处理方法
            
        Returns:
            完全清洗后的数据
        """
        logger.info("开始完整数据清洗流程")
        
        # 1. 去除重复数据
        data_cleaned = self.remove_duplicates(data)
        
        # 2. 处理缺失值
        data_cleaned = self.handle_missing_values(data_cleaned, method=missing_method)
        
        # 3. 纠正错误数据
        if rules:
            data_cleaned = self.correct_error_data(data_cleaned, rules)
        
        # 4. 检测异常值
        data_cleaned = self.detect_outliers(data_cleaned, features)
        
        # 5. 过滤异常值
        data_cleaned = self.filter_outliers(data_cleaned)
        
        logger.info("完整数据清洗流程完成")
        return data_cleaned

# 示例用法
if __name__ == "__main__":
    # 生成示例安全数据
    np.random.seed(42)
    data = pd.DataFrame({
        'timestamp': pd.date_range('2026-01-01', periods=1000, freq='H'),
        'source_ip': [f'192.168.1.{i%100}' for i in range(1000)],
        'destination_ip': [f'10.0.0.{i%50}' for i in range(1000)],
        'bytes_sent': np.random.randint(100, 10000, 1000),
        'bytes_received': np.random.randint(100, 10000, 1000),
        'is_attack': np.random.choice([0, 1], 1000, p=[0.95, 0.05])
    })
    
    # 添加一些噪声和错误数据
    # 添加重复数据
    data = pd.concat([data, data.sample(50)], ignore_index=True)
    # 添加缺失值
    data.loc[::10, 'bytes_sent'] = np.nan
    # 添加错误数据
    data.loc[::20, 'bytes_received'] = 999999  # 异常大的值
    # 添加投毒数据（模拟）
    poison_data = pd.DataFrame({
        'timestamp': pd.date_range('2026-01-01', periods=100, freq='H'),
        'source_ip': ['1.1.1.1' for _ in range(100)],  # 可疑IP
        'destination_ip': ['2.2.2.2' for _ in range(100)],  # 可疑IP
        'bytes_sent': [1000000 for _ in range(100)],  # 异常大的值
        'bytes_received': [1000000 for _ in range(100)],  # 异常大的值
        'is_attack': [0 for _ in range(100)]  # 错误标签，实际是攻击
    })
    data = pd.concat([data, poison_data], ignore_index=True)
    
    # 初始化数据清洗器
    cleaner = SecurityDataCleaner(contamination=0.1)
    
    # 定义纠正规则
    rules = {
        'bytes_sent': {'<0': 0, '>100000': 100000},
        'bytes_received': {'<0': 0, '>100000': 100000}
    }
    
    # 执行完整的数据清洗
    features = ['bytes_sent', 'bytes_received']
    cleaned_data = cleaner.full_clean(data, features, rules, missing_method="mean")
    
    # 输出清洗前后的对比
    print(f"清洗前样本数：{len(data)}")
    print(f"清洗后样本数：{len(cleaned_data)}")
    print(f"清洗前攻击样本比例：{data['is_attack'].mean():.4f}")
    print(f"清洗后攻击样本比例：{cleaned_data['is_attack'].mean():.4f}")

"""
投毒数据检测脚本
使用机器学习算法检测安全数据中的投毒数据
"""

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.preprocessing import StandardScaler
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PoisonDetectionModel:
    """投毒数据检测模型"""
    
    def __init__(self):
        """初始化投毒检测模型"""
        self.scaler = StandardScaler()
        self.model = RandomForestClassifier(n_estimators=100, random_state=42, class_weight='balanced')
        self.is_trained = False
        logger.info("初始化投毒检测模型")
    
    def prepare_features(self, data):
        """准备用于检测的特征
        
        Args:
            data: 原始数据，DataFrame格式
            
        Returns:
            特征数据X
        """
        logger.info("开始准备特征")
        
        # 这里简化处理，仅使用数值特征
        # 实际应用中应根据具体情况设计特征
        numerical_features = data.select_dtypes(include=[np.number]).columns.tolist()
        logger.info(f"使用的数值特征：{numerical_features}")
        
        X = data[numerical_features].values
        logger.info(f"特征准备完成，特征数量：{X.shape[1]}")
        return X
    
    def generate_poison_labels(self, data):
        """生成投毒数据标签（模拟）
        
        Args:
            data: 原始数据，DataFrame格式
            
        Returns:
            标签数据y，1表示正常数据，0表示投毒数据
        """
        logger.info("开始生成投毒标签")
        
        y = np.ones(len(data), dtype=int)  # 默认都是正常数据
        
        # 模拟投毒数据标签生成
        # 1. 统计异常：bytes_sent或bytes_received异常大
        y[(data['bytes_sent'] > 100000) | (data['bytes_received'] > 100000)] = 0
        
        # 2. 时序异常：同一IP短时间内发送大量数据
        # 这里简化处理，仅检测同一source_ip的bytes_sent总和
        ip_stats = data.groupby('source_ip')['bytes_sent'].transform('sum')
        y[ip_stats > 500000] = 0
        
        # 3. 标签异常：is_attack与实际特征不符
        # 这里简化处理，仅检测bytes_sent大但is_attack为0的数据
        y[(data['bytes_sent'] > 50000) & (data['is_attack'] == 0)] = 0
        
        logger.info(f"生成投毒标签完成，投毒数据比例：{1 - y.mean():.4f}")
        return y
    
    def train(self, data):
        """训练投毒检测模型
        
        Args:
            data: 训练数据，DataFrame格式
        """
        logger.info("开始训练投毒检测模型")
        
        # 准备特征
        X = self.prepare_features(data)
        
        # 生成投毒标签
        y = self.generate_poison_labels(data)
        
        # 划分训练集和验证集
        X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)
        logger.info(f"训练集样本数：{len(X_train)}，验证集样本数：{len(X_val)}")
        
        # 特征缩放
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_val_scaled = self.scaler.transform(X_val)
        
        # 训练模型
        self.model.fit(X_train_scaled, y_train)
        
        # 在验证集上评估模型
        y_pred = self.model.predict(X_val_scaled)
        logger.info("模型训练完成，验证集评估结果：")
        logger.info(f"分类报告：\n{classification_report(y_val, y_pred)}")
        logger.info(f"混淆矩阵：\n{confusion_matrix(y_val, y_pred)}")
        
        self.is_trained = True
    
    def detect(self, data):
        """检测投毒数据
        
        Args:
            data: 待检测数据，DataFrame格式
            
        Returns:
            检测结果，1表示正常数据，0表示投毒数据
        """
        if not self.is_trained:
            raise ValueError("模型未训练，无法进行检测")
        
        logger.info(f"开始检测投毒数据，样本数：{len(data)}")
        
        # 准备特征
        X = self.prepare_features(data)
        
        # 特征缩放
        X_scaled = self.scaler.transform(X)
        
        # 进行检测
        y_pred = self.model.predict(X_scaled)
        
        # 统计检测结果
        poison_count = np.sum(y_pred == 0)
        logger.info(f"检测完成，投毒数据数量：{poison_count}/{len(data)}")
        
        return y_pred
    
    def filter_poison_data(self, data):
        """过滤投毒数据
        
        Args:
            data: 原始数据，DataFrame格式
            
        Returns:
            过滤投毒数据后的DataFrame
        """
        logger.info("开始过滤投毒数据")
        
        # 检测投毒数据
        y_pred = self.detect(data)
        
        # 过滤投毒数据
        filtered_data = data[y_pred == 1].copy()
        
        # 记录投毒数据
        poison_data = data[y_pred == 0].copy()
        logger.info(f"投毒数据记录：\n{poison_data.head()}")
        
        logger.info(f"过滤投毒数据完成，剩余样本数：{len(filtered_data)}")
        return filtered_data

# 示例用法
if __name__ == "__main__":
    # 生成示例安全数据
    np.random.seed(42)
    data = pd.DataFrame({
        'timestamp': pd.date_range('2026-01-01', periods=1000, freq='H'),
        'source_ip': [f'192.168.1.{i%100}' for i in range(1000)],
        'destination_ip': [f'10.0.0.{i%50}' for i in range(1000)],
        'bytes_sent': np.random.randint(100, 10000, 1000),
        'bytes_received': np.random.randint(100, 10000, 1000),
        'is_attack': np.random.choice([0, 1], 1000, p=[0.95, 0.05])
    })
    
    # 添加投毒数据
    poison_data = pd.DataFrame({
        'timestamp': pd.date_range('2026-01-01', periods=200, freq='H'),
        'source_ip': ['1.1.1.1' for _ in range(200)],
        'destination_ip': ['2.2.2.2' for _ in range(200)],
        'bytes_sent': [1000000 for _ in range(200)],
        'bytes_received': [1000000 for _ in range(200)],
        'is_attack': [0 for _ in range(200)]
    })
    data = pd.concat([data, poison_data], ignore_index=True)
    
    # 初始化投毒检测模型
    detector = PoisonDetectionModel()
    
    # 训练模型
    detector.train(data)
    
    # 检测并过滤投毒数据
    filtered_data = detector.filter_poison_data(data)
    
    # 输出过滤前后的对比
    print(f"过滤前样本数：{len(data)}")
    print(f"过滤后样本数：{len(filtered_data)}")
    print(f"过滤前攻击样本比例：{data['is_attack'].mean():.4f}")
    print(f"过滤后攻击样本比例：{filtered_data['is_attack'].mean():.4f}")

"""
数据清洗效果评估脚本
量化数据清洗对模型性能的影响
"""

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.preprocessing import StandardScaler
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CleaningEffectEvaluator:
    """数据清洗效果评估器"""
    
    def __init__(self):
        """初始化评估器"""
        self.model = RandomForestClassifier(n_estimators=100, random_state=42, class_weight='balanced')
        logger.info("初始化数据清洗效果评估器")
    
    def evaluate_model(self, X_train, y_train, X_test, y_test, name):
        """评估模型性能
        
        Args:
            X_train: 训练特征
            y_train: 训练标签
            X_test: 测试特征
            y_test: 测试标签
            name: 评估名称
            
        Returns:
            评估结果字典
        """
        logger.info(f"开始评估模型：{name}")
        
        # 训练模型
        self.model.fit(X_train, y_train)
        
        # 预测
        y_pred = self.model.predict(X_test)
        
        # 生成评估报告
        report = classification_report(y_test, y_pred, output_dict=True)
        conf_matrix = confusion_matrix(y_test, y_pred)
        
        # 提取关键指标
        accuracy = report['accuracy']
        precision = report['1']['precision']  # 攻击样本的精确率
        recall = report['1']['recall']  # 攻击样本的召回率
        f1_score = report['1']['f1-score']  # 攻击样本的F1分数
        
        logger.info(f"{name} 评估结果：")
        logger.info(f"准确率：{accuracy:.4f}")
        logger.info(f"精确率：{precision:.4f}")
        logger.info(f"召回率：{recall:.4f}")
        logger.info(f"F1分数：{f1_score:.4f}")
        logger.info(f"混淆矩阵：\n{conf_matrix}")
        
        return {
            'name': name,
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall,
            'f1_score': f1_score,
            'confusion_matrix': conf_matrix.tolist()
        }
    
    def evaluate_cleaning_effect(self, raw_data, cleaned_data, label_column='is_attack'):
        """评估数据清洗效果
        
        Args:
            raw_data: 原始数据，DataFrame格式
            cleaned_data: 清洗后的数据，DataFrame格式
            label_column: 标签列名
            
        Returns:
            清洗效果评估结果字典
        """
        logger.info("开始评估数据清洗效果")
        
        # 提取特征和标签
        features = raw_data.select_dtypes(include=[np.number]).columns.tolist()
        features.remove(label_column)
        
        # 原始数据
        X_raw = raw_data[features].values
        y_raw = raw_data[label_column].values
        
        # 清洗后数据
        X_cleaned = cleaned_data[features].values
        y_cleaned = cleaned_data[label_column].values
        
        # 划分训练集和测试集
        X_raw_train, X_raw_test, y_raw_train, y_raw_test = train_test_split(
            X_raw, y_raw, test_size=0.2, random_state=42, stratify=y_raw
        )
        
        X_cleaned_train, X_cleaned_test, y_cleaned_train, y_cleaned_test = train_test_split(
            X_cleaned, y_cleaned, test_size=0.2, random_state=42, stratify=y_cleaned
        )
        
        # 特征缩放
        scaler = StandardScaler()
        X_raw_train_scaled = scaler.fit_transform(X_raw_train)
        X_raw_test_scaled = scaler.transform(X_raw_test)
        
        X_cleaned_train_scaled = scaler.fit_transform(X_cleaned_train)
        X_cleaned_test_scaled = scaler.transform(X_cleaned_test)
        
        # 评估原始数据上的模型性能
        raw_result = self.evaluate_model(X_raw_train_scaled, y_raw_train, X_raw_test_scaled, y_raw_test, "原始数据")
        
        # 评估清洗后数据上的模型性能
        cleaned_result = self.evaluate_model(X_cleaned_train_scaled, y_cleaned_train, X_cleaned_test_scaled, y_cleaned_test, "清洗后数据")
        
        # 计算清洗带来的性能提升
        improvement = {
            'accuracy_improvement': cleaned_result['accuracy'] - raw_result['accuracy'],
            'precision_improvement': cleaned_result['precision'] - raw_result['precision'],
            'recall_improvement': cleaned_result['recall'] - raw_result['recall'],
            'f1_score_improvement': cleaned_result['f1_score'] - raw_result['f1_score']
        }
        
        logger.info("数据清洗效果评估完成")
        logger.info(f"性能提升：\n{improvement}")
        
        return {
            'raw_result': raw_result,
            'cleaned_result': cleaned_result,
            'improvement': improvement
        }

# 示例用法
if __name__ == "__main__":
    # 生成示例安全数据
    np.random.seed(42)
    data = pd.DataFrame({
        'timestamp': pd.date_range('2026-01-01', periods=1000, freq='H'),
        'source_ip': [f'192.168.1.{i%100}' for i in range(1000)],
        'destination_ip': [f'10.0.0.{i%50}' for i in range(1000)],
        'bytes_sent': np.random.randint(100, 10000, 1000),
        'bytes_received': np.random.randint(100, 10000, 1000),
        'is_attack': np.random.choice([0, 1], 1000, p=[0.95, 0.05])
    })
    
    # 添加噪声和错误数据
    data = pd.concat([data, data.sample(50)], ignore_index=True)  # 重复数据
    data.loc[::10, 'bytes_sent'] = np.nan  # 缺失值
    data.loc[::20, 'bytes_received'] = 999999  # 错误数据
    
    # 添加投毒数据
    poison_data = pd.DataFrame({
        'timestamp': pd.date_range('2026-01-01', periods=200, freq='H'),
        'source_ip': ['1.1.1.1' for _ in range(200)],
        'destination_ip': ['2.2.2.2' for _ in range(200)],
        'bytes_sent': [1000000 for _ in range(200)],
        'bytes_received': [1000000 for _ in range(200)],
        'is_attack': [0 for _ in range(200)]
    })
    data = pd.concat([data, poison_data], ignore_index=True)
    
    # 导入之前定义的数据清洗器和投毒检测器
    from G052_data_cleaner import SecurityDataCleaner
    from G052_poison_detector import PoisonDetectionModel
    
    # 执行数据清洗
    cleaner = SecurityDataCleaner(contamination=0.1)
    rules = {
        'bytes_sent': {'<0': 0, '>100000': 100000},
        'bytes_received': {'<0': 0, '>100000': 100000}
    }
    features = ['bytes_sent', 'bytes_received']
    cleaned_data = cleaner.full_clean(data, features, rules, missing_method="mean")
    
    # 执行投毒检测
    detector = PoisonDetectionModel()
    detector.train(cleaned_data)
    cleaned_poison_filtered = detector.filter_poison_data(cleaned_data)
    
    # 评估数据清洗效果
    evaluator = CleaningEffectEvaluator()
    
    # 评估原始数据 vs 仅清洗数据
    result1 = evaluator.evaluate_cleaning_effect(data, cleaned_data)
    
    # 评估原始数据 vs 清洗+防投毒数据
    result2 = evaluator.evaluate_cleaning_effect(data, cleaned_poison_filtered)
    
    # 输出对比结果
    print("=== 原始数据 vs 仅清洗数据 ===")
    print(f"原始数据准确率：{result1['raw_result']['accuracy']:.4f}")
    print(f"仅清洗数据准确率：{result1['cleaned_result']['accuracy']:.4f}")
    print(f"准确率提升：{result1['improvement']['accuracy_improvement']:.4f}")
    print(f"F1分数提升：{result1['improvement']['f1_score_improvement']:.4f}")
    
    print("\n=== 原始数据 vs 清洗+防投毒数据 ===")
    print(f"原始数据准确率：{result2['raw_result']['accuracy']:.4f}")
    print(f"清洗+防投毒数据准确率：{result2['cleaned_result']['accuracy']:.4f}")
    print(f"准确率提升：{result2['improvement']['accuracy_improvement']:.4f}")
    print(f"F1分数提升：{result2['improvement']['f1_score_improvement']:.4f}")
    
    print("\n=== 仅清洗 vs 清洗+防投毒 ===")
    print(f"仅清洗数据准确率：{result1['cleaned_result']['accuracy']:.4f}")
    print(f"清洗+防投毒数据准确率：{result2['cleaned_result']['accuracy']:.4f}")
    print(f"准确率提升：{(result2['cleaned_result']['accuracy'] - result1['cleaned_result']['accuracy']):.4f}")

# 安装必要的库
pip install pandas numpy scikit-learn matplotlib seaborn pyspark

# 安装PySpark（如果需要分布式数据清洗）
# 注意：PySpark需要Java环境
pip install pyspark