本文目录导读：
针对监控数据匿名化效果的评估,我提供一个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）
- 实际应用中应结合具体业务需求调整评估指标权重
