监控数据匿名化效果的脚本如何写

本文目录导读：

1. 基础评估脚本
2. 关键说明

针对监控数据匿名化效果的评估,我提供一个Python脚本框架，包含常见的评估指标：

基础评估脚本

import pandas as pd
import numpy as np
from scipy import stats
from sklearn.metrics import pairwise_distances
from collections import Counter
import warnings
warnings.filterwarnings('ignore')
class AnonymizationEvaluator:
    def __init__(self, original_data, anonymized_data, identifier_cols, quasi_identifiers):
        """
        初始化评估器
        Parameters:
        - original_data: DataFrame, 原始数据
        - anonymized_data: DataFrame, 匿名化后的数据
        - identifier_cols: list, 标识符列名（如姓名、ID等）
        - quasi_identifiers: list, 准标识符列名（如年龄、性别、邮编等）
        """
        self.original = original_data.copy()
        self.anonymized = anonymized_data.copy()
        self.identifier_cols = [col for col in identifier_cols if col in original_data.columns]
        self.qi_cols = [col for col in quasi_identifiers if col in original_data.columns]
    def check_record_count(self):
        """检查记录数量变化"""
        orig_count = len(self.original)
        anon_count = len(self.anonymized)
        return {
            'original_records': orig_count,
            'anonymized_records': anon_count,
            'difference': abs(orig_count - anon_count)
        }
    def compute_k_anonymity(self):
        """计算k-匿名性"""
        if not self.qi_cols:
            return {'k': len(self.anonymized), 'min_k': len(self.anonymized)}
        grouped = self.anonymized[self.qi_cols].groupby(self.qi_cols).size()
        k_values = grouped.values
        return {
            'k': grouped.min(),
            'min_k': grouped.min(),
            'max_k': grouped.max(),
            'avg_k': grouped.mean(),
            'k_distribution': Counter(k_values)
        }
    def compute_l_diversity(self, sensitive_col):
        """计算l-多样性"""
        if not self.qi_cols or sensitive_col not in self.anonymized.columns:
            return {'l': 0}
        # 对每个等价类计算敏感值多样性
        diversity_values = []
        for _, group in self.anonymized.groupby(self.qi_cols):
            l = group[sensitive_col].nunique()
            diversity_values.append(l)
        return {
            'l': min(diversity_values) if diversity_values else 0,
            'min_l': min(diversity_values) if diversity_values else 0,
            'avg_l': np.mean(diversity_values) if diversity_values else 0
        }
    def compute_t_closeness(self, sensitive_col):
        """
        计算t-近邻性
        比较每个等价类中敏感值分布与全局分布的距离
        """
        if not self.qi_cols or sensitive_col not in self.anonymized.columns:
            return {'t': 1.0}
        global_dist = self.anonymized[sensitive_col].value_counts(normalize=True)
        max_distance = 0
        for _, group in self.anonymized.groupby(self.qi_cols):
            eq_dist = group[sensitive_col].value_counts(normalize=True)
            # 使用EMD（Earth Mover's Distance）或KL散度
            all_categories = list(set(global_dist.index) | set(eq_dist.index))
            global_probs = [global_dist.get(cat, 0) for cat in all_categories]
            eq_probs = [eq_dist.get(cat, 0) for cat in all_categories]
            distance = sum(abs(p1 - p2) for p1, p2 in zip(global_probs, eq_probs)) / 2
            max_distance = max(max_distance, distance)
        return {'t': max_distance}
    def compute_data_utility(self):
        """
        计算数据效用
        使用统计相似度方法
        """
        utility_metrics = {}
        for col in self.anonymized.columns:
            if col in self.qi_cols or col in self.identifier_cols:
                continue
            if self.original[col].dtype in ['int64', 'float64']:
                # 数值型：计算相关性
                common_idx = self.original.index.intersection(self.anonymized.index)
                if len(common_idx) > 0:
                    corr = stats.pearsonr(
                        self.original.loc[common_idx, col],
                        self.anonymized.loc[common_idx, col]
                    )[0]
                    utility_metrics[f'{col}_correlation'] = corr
                # 均值差异
                mean_diff = abs(self.original[col].mean() - self.anonymized[col].mean())
                utility_metrics[f'{col}_mean_diff'] = mean_diff
            else:
                # 分类型：计算分布一致性
                orig_dist = self.original[col].value_counts(normalize=True)
                anon_dist = self.anonymized[col].value_counts(normalize=True)
                all_categories = list(set(orig_dist.index) | set(anon_dist.index))
                dist_diff = sum(abs(
                    orig_dist.get(cat, 0) - anon_dist.get(cat, 0)
                ) for cat in all_categories) / 2
                utility_metrics[f'{col}_distribution_diff'] = dist_diff
        return utility_metrics
    def compute_re_identification_risk(self, sample_size=1000):
        """
        计算重识别风险
        基于记录链接方法
        """
        if len(self.original) < 2 or len(self.anonymized) < 2:
            return {'re_identification_risk': 0.0}
        sample_size = min(sample_size, len(self.original), len(self.anonymized))
        orig_sample = self.original[self.qi_cols].sample(n=sample_size, random_state=42)
        anon_sample = self.anonymized[self.qi_cols].sample(n=sample_size, random_state=42)
        # 计算距离矩阵
        distances = pairwise_distances(orig_sample, anon_sample, metric='euclidean')
        # 对每个原始记录，找最近的匿名记录
        min_distances = distances.min(axis=1)
        # 定义匹配阈值（基于数据标准差）
        thresholds = np.percentile(min_distances, [25, 50, 75])
        risk_scores = {
            'mean_distance': min_distances.mean(),
            'std_distance': min_distances.std(),
            'risk_low_threshold': thresholds[0],
            'risk_medium_threshold': thresholds[1],
            'risk_high_threshold': thresholds[2],
        }
        # 计算可能的匹配数
        matches = (min_distances < thresholds[1]).sum()
        risk_scores['estimated_risk'] = matches / sample_size
        return risk_scores
    def full_evaluation(self, sensitive_columns=None):
        """
        执行完整评估
        """
        results = {}
        # 1. 基本信息
        results['record_count'] = self.check_record_count()
        # 2. k-匿名性
        results['k_anonymity'] = self.compute_k_anonymity()
        # 3. 准标识符统计
        results['qi_statistics'] = {
            col: {
                'original_unique': self.original[col].nunique(),
                'anonymized_unique': self.anonymized[col].nunique(),
                'unique_change': self.original[col].nunique() - self.anonymized[col].nunique()
            }
            for col in self.qi_cols
        }
        # 4. l-多样性（如果指定了敏感属性）
        if sensitive_columns:
            l_diversity_results = {}
            for sens_col in sensitive_columns:
                if sens_col in self.anonymized.columns:
                    l_diversity_results[sens_col] = self.compute_l_diversity(sens_col)
            results['l_diversity'] = l_diversity_results
            # 5. t-近邻性
            t_closeness_results = {}
            for sens_col in sensitive_columns:
                if sens_col in self.anonymized.columns:
                    t_closeness_results[sens_col] = self.compute_t_closeness(sens_col)
            results['t_closeness'] = t_closeness_results
        # 6. 数据效用
        results['data_utility'] = self.compute_data_utility()
        # 7. 重识别风险
        results['re_identification_risk'] = self.compute_re_identification_risk()
        return results
# 使用示例
if __name__ == "__main__":
    # 创建示例数据
    original_df = pd.DataFrame({
        'name': ['Alice', 'Bob', 'Charlie', 'David'],
        'age': [25, 30, 35, 40],
        'gender': ['F', 'M', 'M', 'M'],
        'zipcode': ['10001', '10002', '10003', '10001'],
        'salary': [50000, 60000, 70000, 80000],
        'disease': ['flu', 'cancer', 'flu', 'diabetes']
    })
    # 模拟匿名化数据（例如泛化年龄和删除姓名）
    anonymized_df = original_df.copy()
    anonymized_df['name'] = ['P1', 'P2', 'P3', 'P4']  # 替换为伪标识符
    anonymized_df['age'] = [25, 30, 35, 40]  # 实际应用中应进行泛化
    # 配置标识符和准标识符
    identifiers = ['name']
    quasi_ids = ['age', 'gender', 'zipcode']  # 至少2个准标识符
    # 评估
    evaluator = AnonymizationEvaluator(original_df, anonymized_df, identifiers, quasi_ids)
    results = evaluator.full_evaluation(sensitive_columns=['salary', 'disease'])
    # 输出结果
    print("=== 匿名化效果评估报告 ===")
    print(f"\n记录数: {results['record_count']}")
    print(f"\nK-匿名性: min_k={results['k_anonymity']['min_k']}, avg_k={results['k_anonymity']['avg_k']:.2f}")
    print("\n准标识符统计:")
    for col, stats in results['qi_statistics'].items():
        print(f"  {col}: 原始唯一值{stats['original_unique']} -> 匿名后{stats['anonymized_unique']}")
    if 'l_diversity' in results:
        print("\nl-多样性:")
        for sens_col, l_results in results['l_diversity'].items():
            print(f"  {sens_col}: min_l={l_results['min_l']}, avg_l={l_results['avg_l']:.2f}")
    if 't_closeness' in results:
        print("\nt-近邻性:")
        for sens_col, t_results in results['t_closeness'].items():
            print(f"  {sens_col}: t={t_results['t']:.3f}")
    print("\n数据效用: (部分指标)")
    for metric, value in list(results['data_utility'].items())[:5]:
        print(f"  {metric}: {value:.3f}")
    print(f"\n重识别风险: {results['re_identification_risk']['estimated_risk']:.2%}")

关键说明

评估指标解释

• k-匿名性：每个等价类至少包含k条记录，k越大隐私保护越好
• l-多样性：每个等价类中敏感值至少有l种不同值
• t-近邻性：等价类中敏感值分布与全局分布的距离
• 数据效用：匿名化后数据统计分析结果的保留程度
• 重识别风险：通过记录链接方法评估攻击者识别个体的难度

使用建议

# 1. 至少配置2个准标识符
identifiers = ['name', 'id']
quasi_ids = ['age', 'gender', 'zipcode', 'birthdate']
# 2. 如果有敏感属性，指定进行评估
sensitive_cols = ['salary', 'disease', 'diagnosis']
# 3. 对大型数据集进行采样
results = evaluator.compute_re_identification_risk(sample_size=5000)

注意事项

• 准标识符的选择对评估结果影响很大
• 匿名化效果是隐私保护与数据效用之间的平衡
• 不同行业可能有不同的阈值标准（如医疗领域要求k≥5）
• 实际应用中应结合具体业务需求调整评估指标权重