别再只懂k-anonymity了：用Python实战带你搞懂隐私模型三剑客（附代码）

在数据驱动的时代，隐私保护已成为每个数据从业者的必修课。你可能听说过k-anonymity，但真正要构建可靠的隐私保护方案，还需要掌握l-diversity和t-closeness这两个进阶模型。本文将用Python代码带你完整实现这三个模型，解决实际项目中常见的隐私泄露陷阱。

1. 环境准备与数据模拟

首先我们需要创建一个包含敏感信息的模拟数据集。这里使用pandas构建一个包含邮编、年龄、性别和疾病信息的医疗数据集：

import pandas as pd
import numpy as np

# 生成模拟数据
np.random.seed(42)
data = {
    'zipcode': ['10001', '10002', '10003', '10001', '10002', '10003']*50,
    'age': np.random.randint(20, 70, 300),
    'gender': np.random.choice(['M', 'F'], 300),
    'disease': np.random.choice(['Diabetes', 'Hypertension', 'Asthma', 'Cancer', 'Depression'], 300, 
                              p=[0.3, 0.25, 0.2, 0.15, 0.1])
}

df = pd.DataFrame(data)

注意：实际项目中应使用脱敏后的真实数据，这里仅用于演示目的

数据集中的准标识符(QI)是zipcode和age，敏感属性是disease。我们先对数据进行基本分析：

print(f"数据集大小: {df.shape}")
print("\n疾病分布:")
print(df['disease'].value_counts(normalize=True))

2. 实现k-anonymity的泛化与抑制

k-anonymity要求每组准标识符至少对应k条记录。我们先实现数据泛化：

def generalize_data(df, k=3):
    # 年龄泛化为10岁区间
    df['age_group'] = (df['age'] // 10 * 10).astype(str) + '-' + (df['age'] // 10 * 10 + 9).astype(str)
    
    # 邮编保留前3位
    df['zipcode_generalized'] = df['zipcode'].str[:3] + 'XX'
    
    return df

df = generalize_data(df)

检查k-anonymity满足情况：

def check_k_anonymity(df, quasi_identifiers, k=3):
    groups = df.groupby(quasi_identifiers).size()
    return groups.min() >= k

quasi_ids = ['zipcode_generalized', 'age_group']
print(f"满足3-anonymity: {check_k_anonymity(df, quasi_ids)}")

对于不满足k-anonymity的组，我们需要实施抑制：

def enforce_k_anonymity(df, quasi_identifiers, k=3):
    group_sizes = df.groupby(quasi_identifiers).size()
    small_groups = group_sizes[group_sizes < k].index
    
    # 抑制小群体记录
    suppressed = df[~df.set_index(quasi_identifiers).index.isin(small_groups)]
    return suppressed

df_anon = enforce_k_anonymity(df, quasi_ids)
print(f"处理后记录数: {len(df_anon)}")

3. 实现l-diversity检查与增强

k-anonymity存在同质化攻击风险，我们需要实现l-diversity检查：

from collections import Counter
import math

def calculate_entropy(group):
    counts = Counter(group)
    total = len(group)
    entropy = 0
    for count in counts.values():
        p = count / total
        entropy -= p * math.log2(p)
    return entropy

def check_l_diversity(df, quasi_identifiers, sensitive_attr, l=2):
    groups = df.groupby(quasi_identifiers)[sensitive_attr]
    
    # 检查可区分l-diversity
    distinct_check = groups.nunique() >= l
    
    # 检查熵l-diversity
    entropy_check = groups.apply(calculate_entropy) >= math.log2(l)
    
    return all(distinct_check) & all(entropy_check)

print(f"满足2-diversity: {check_l_diversity(df_anon, quasi_ids, 'disease')}")

对于不满足l-diversity的组，我们可以通过进一步泛化或敏感值抑制来处理：

def enhance_l_diversity(df, quasi_identifiers, sensitive_attr, l=2):
    groups = df.groupby(quasi_identifiers)
    
    # 识别不满足l-diversity的组
    problematic = []
    for name, group in groups:
        if len(group[sensitive_attr].unique()) < l:
            problematic.append(name)
    
    # 进一步泛化年龄
    enhanced = df.copy()
    enhanced['age_group'] = (enhanced['age'] // 20 * 20).astype(str) + '-' + (enhanced['age'] // 20 * 20 + 19).astype(str)
    
    return enhanced

df_ldiv = enhance_l_diversity(df_anon, quasi_ids, 'disease')
print(f"增强后满足2-diversity: {check_l_diversity(df_ldiv, quasi_ids, 'disease')}")

4. 实现t-closeness距离度量

t-closeness要求组内敏感属性分布与整体分布的距离不超过阈值t。我们实现EMD(地球移动距离)计算：

from scipy.stats import wasserstein_distance

def calculate_t_closeness(df, quasi_identifiers, sensitive_attr, t=0.2):
    # 获取全局分布
    global_dist = df[sensitive_attr].value_counts(normalize=True).sort_index()
    
    groups = df.groupby(quasi_identifiers)[sensitive_attr]
    results = []
    
    for name, group in groups:
        # 计算组内分布
        local_dist = group.value_counts(normalize=True).reindex(global_dist.index, fill_value=0)
        
        # 计算EMD距离
        emd = wasserstein_distance(global_dist.values, local_dist.values)
        results.append(emd <= t)
    
    return all(results)

# 计算整体分布
print("疾病全局分布:")
print(df['disease'].value_counts(normalize=True))

print(f"\n满足t-closeness(t=0.2): {calculate_t_closeness(df_ldiv, quasi_ids, 'disease')}")

对于不满足t-closeness的组，我们可以调整泛化策略：

def improve_t_closeness(df, quasi_identifiers, sensitive_attr, t=0.2):
    # 获取全局分布
    global_dist = df[sensitive_attr].value_counts(normalize=True).sort_index()
    
    improved = df.copy()
    groups = improved.groupby(quasi_identifiers)
    
    for name, group in groups:
        local_dist = group[sensitive_attr].value_counts(normalize=True).reindex(global_dist.index, fill_value=0)
        emd = wasserstein_distance(global_dist.values, local_dist.values)
        
        if emd > t:
            # 合并相邻年龄组
            improved.loc[improved.set_index(quasi_identifiers).index == name, 'age_group'] = \
                improved.loc[improved.set_index(quasi_identifiers).index == name, 'age'].apply(
                    lambda x: f"{(x//30)*30}-{(x//30)*30+29}")
    
    return improved

df_tclose = improve_t_closeness(df_ldiv, quasi_ids, 'disease')
print(f"改进后满足t-closeness: {calculate_t_closeness(df_tclose, quasi_ids, 'disease')}")

5. 三模型联合应用实战

现在我们将三个模型组合应用到一个完整的数据发布流程中：

def full_anonymization_pipeline(df, quasi_identifiers, sensitive_attr, k=3, l=2, t=0.2):
    # 初始泛化
    df = generalize_data(df)
    
    # 实施k-anonymity
    df = enforce_k_anonymity(df, quasi_identifiers, k)
    
    # 检查并增强l-diversity
    if not check_l_diversity(df, quasi_identifiers, sensitive_attr, l):
        df = enhance_l_diversity(df, quasi_identifiers, sensitive_attr, l)
    
    # 检查并改进t-closeness
    if not calculate_t_closeness(df, quasi_identifiers, sensitive_attr, t):
        df = improve_t_closeness(df, quasi_identifiers, sensitive_attr, t)
    
    return df

final_df = full_anonymization_pipeline(df, ['zipcode', 'age'], 'disease')

评估最终结果：

print("\n最终数据评估:")
print(f"记录保留率: {len(final_df)/len(df):.1%}")
print(f"满足k-anonymity(k=3): {check_k_anonymity(final_df, ['zipcode_generalized', 'age_group'])}")
print(f"满足l-diversity(l=2): {check_l_diversity(final_df, ['zipcode_generalized', 'age_group'], 'disease')}")
print(f"满足t-closeness(t=0.2): {calculate_t_closeness(final_df, ['zipcode_generalized', 'age_group'], 'disease')}")

print("\n示例记录:")
print(final_df.sample(5))

在实际项目中，还需要考虑以下优化点：

• 性能优化 ：对于大数据集，可以使用采样方法估算分布
• 参数调优 ：通过实验确定最佳的k、l、t参数组合
• 可视化监控 ：绘制信息损失与隐私保护的权衡曲线

# 可视化信息损失
import matplotlib.pyplot as plt

original_entropy = calculate_entropy(df['disease'])
final_entropy = calculate_entropy(final_df['disease'])

plt.bar(['原始数据', '匿名化数据'], [original_entropy, final_entropy])
plt.ylabel('疾病信息熵')
plt.title('匿名化前后的信息保留情况')
plt.show()

隐私保护不是一次性工作，而是一个需要持续优化的过程。在实际项目中，我发现最常犯的错误是过度追求k值而忽视l-diversity，结果导致同质化攻击风险。建议先确保l-diversity，再调整k值平衡可用性与隐私性。