data-anonymization

name	data-anonymization
description	Implement data anonymization and pseudonymization techniques for GDPR compliance, safe analytics, and test data generation. Outputs anonymization pipelines, k-anonymity validation, differential privacy, and de-identification strategies.
argument-hint	["data types","compliance requirements","analytics needs","re-identification risk tolerance"]
allowed-tools	Read, Write, Bash

Data Anonymization

Anonymization removes personal identifiers so that individuals cannot be re-identified. Pseudonymization replaces identifiers with reversible tokens. True anonymization is irreversible — once done, GDPR no longer applies to that data.

Techniques

Technique	Reversible	GDPR Applies	Use Case
Pseudonymization	Yes (with key)	Yes	Data sharing where re-linking may be needed
Generalization	No	No	Analytics (age group vs. exact age)
Suppression	No	No	Remove outliers or rare categories
Data masking	No	No	Test data generation
K-anonymity	No	No	Dataset release (each record looks like k others)
Differential privacy	No	No	Aggregate statistics with math guarantees

Output Format

Pseudonymization Pipeline

# anonymization/pseudonymizer.py
import hashlib
import hmac
import secrets
import os
from functools import lru_cache

class Pseudonymizer:
    """
    Reversible pseudonymization using HMAC.
    Same input + same key = same output (deterministic).
    Needed for joining pseudonymized datasets.
    """
    
    def __init__(self, secret_key: bytes = None):
        self.key = secret_key or os.environ.get("PSEUDONYMIZATION_KEY", "").encode()
        if not self.key:
            raise ValueError("PSEUDONYMIZATION_KEY environment variable required")
    
    def pseudonymize(self, value: str, domain: str = "default") -> str:
        """
        Deterministically hash a value with domain separation.
        Different domains produce different outputs for the same input.
        """
        msg = f"{domain}:{value}".encode()
        return hmac.new(self.key, msg, hashlib.sha256).hexdigest()[:16]
    
    def pseudonymize_email(self, email: str) -> str:
        """Replace email while preserving domain for analytics."""
        local, domain = email.split(, )
        pseudonym = .pseudonymize(email, )
         
    
     () -> :
        
           ip:   
            parts = ip.split()
             .join(parts[:]) + 
        :           
            parts = ip.split()
             .join(parts[:]) + 
    
     () -> :
        
         pandas  pd
        result = df.copy()
        
         col, col_type  columns.items():
             col   result.columns:
                
            
             col_type == :
                result[col] = result[col].astype().apply(
                     x: .pseudonymize(x, )
                )
             col_type == :
                result[col] = result[col].apply(.pseudonymize_email)
             col_type == :
                result[col] = result[col].apply(.pseudonymize_ip)
             col_type == :
                result[col] = result[col].apply(
                     x: .pseudonymize((x), )[:].title()
                )
             col_type == :
                result[col] = result[col].apply(
                     x:  + .pseudonymize((x), )[:]
                )
             col_type == :
                result = result.drop(columns=[col])
        
         result

Data Masking for Test Data

# anonymization/masker.py
from faker import Faker
import pandas as pd
import random

fake = Faker()
Faker.seed(42)   # Deterministic for reproducible test data

class DataMasker:
    """Replace real PII with realistic fake data for development/testing."""
    
    MASKING_RULES = {
        "name":         lambda _: fake.name(),
        "first_name":   lambda _: fake.first_name(),
        "last_name":    lambda _: fake.last_name(),
        "email":        lambda _: fake.email(),
        "phone":        lambda _: fake.phone_number(),
        "address":      lambda _: fake.address().replace('\n', ', '),
        "city":         lambda _: fake.city(),
        "country":      lambda _: fake.country_code(),  # Keep country for analytics
        "zip_code":     lambda _: fake.zipcode(),
        "ssn":          lambda _: fake.ssn(),
        "credit_card":  lambda _: fake.credit_card_number(),
        "dob":          lambda dob: fake.date_of_birth(
                            minimum_age=18, maximum_age=90
                        ).strftime("%Y-%m-%d"),
        "ip_address":   lambda _: fake.ipv4(),
        :    _: fake.user_agent(),
        :         _: fake.sentence(),
    }
    
     () -> pd.DataFrame:
        result = df.copy()
        
         col, col_type  column_types.items():
             col   result.columns:
                
            
             col_type  .MASKING_RULES:
                rule = .MASKING_RULES[col_type]
                result[col] = result[col].apply(rule)
             col_type == :
                
                result[col] = result[col].apply(._partial_mask_email)
             col_type == :
                result[col] = result[col].apply(
                     age:   
                )
             col_type == :
                result = result.drop(columns=[col])
        
         result
    
     () -> :
          (email, )     email:
             
        local, domain = email.split(, )
        masked_local = local[] +  * ((local) - )
         
    
     () -> pd.DataFrame:
        
        masked = .mask_dataframe(real_df, schema)
        
         output_path:
            masked.to_parquet(output_path, index=)
            ()
        
         masked

K-Anonymity Validation

# anonymization/k_anonymity.py
import pandas as pd

def check_k_anonymity(
    df: pd.DataFrame,
    quasi_identifiers: list[str],
    k: int = 5
) -> dict:
    """
    Verify k-anonymity: every combination of quasi-identifiers
    appears at least k times in the dataset.
    k=5 means each record is indistinguishable from at least 4 others.
    """
    groups = df.groupby(quasi_identifiers).size().reset_index(name='count')
    
    violations = groups[groups['count'] < k]
    
    return {
        "k": k,
        "total_groups": len(groups),
        "violating_groups": len(violations),
        "min_group_size": groups['count'].min(),
        "is_k_anonymous": len(violations) == 0,
        "sample_violations": violations.head(5).to_dict('records'),
    }

def generalize_to_achieve_k_anonymity(
    df: pd.DataFrame,
    quasi_identifiers: list[str],
    k: int = 5
) -> pd.DataFrame:
    """Suppress records in groups smaller than k."""
    groups = df.groupby(quasi_identifiers).size().reset_index(name='_count')
    df_with_count = df.merge(groups, on=quasi_identifiers)
    
    # Suppress rows in groups below k
    result = df_with_count[df_with_count['_count'] >= k].drop(columns=['_count'])
    
    suppressed = (df) - (result)
     suppressed > :
        ()
    
     result


df = pd.read_parquet()


result = check_k_anonymity(
    df,
    quasi_identifiers=[, , ],
    k=
)

()
()
  result[]:
    ()

Production Data Pipeline

# pipelines/anonymize_for_analytics.py
"""
Nightly job: anonymize production data for analytics use.
Output: anonymized dataset safe for BI tools and data scientists.
"""

import pandas as pd
from anonymization.pseudonymizer import Pseudonymizer
from anonymization.masker import DataMasker

pseudonymizer = Pseudonymizer()
masker = DataMasker()

def anonymize_orders_for_analytics():
    # Load raw data
    df = pd.read_parquet("s3://prod-data/orders/today/")
    
    print(f"Anonymizing {len(df):,} orders...")
    
    # Step 1: Pseudonymize joinable IDs (keep analytics cross-dataset linking)
    df['user_id'] = df['user_id'].apply(
        lambda x: pseudonymizer.pseudonymize(str(x), "user_id")
    )
    
    # Step 2: Generalize quasi-identifiers
    df['age_group'] = pd.cut(
        df['age'],
        bins=[0, 18, 25, 35, 45, 55, 65, 120],
        labels=['<18', '18-24', '25-34', '35-44', '45-54', '55-64', '65+']
    )
    
    # Step 3: Drop direct identifiers
    columns_to_drop = [
        'shipping_name', 'shipping_address', 'billing_name',
        , , ,   
        ,              
    ]
    df = df.drop(columns=[c  c  columns_to_drop  c  df.columns])
    
    
    
    
    
    qi = [, , ]
    result = check_k_anonymity(df, qi, k=)
      result[]:
        df = generalize_to_achieve_k_anonymity(df, qi, k=)
    
    
    df.to_parquet(, index=)
    ()
    
     {: (df), : result[]}

Rules

• Anonymization is context-dependent — a dataset anonymous alone may be re-identifiable when joined with others.
• Test re-identification risk — don't assume anonymization works; attempt to re-identify with auxiliary data.
• k-anonymity minimum k=5 — below 5, records are too easily singled out; aim for k=10+ for sensitive data.
• Pseudonymization ≠ anonymization — if you keep the key, GDPR still applies.
• Never anonymize in place on production — anonymize to a separate dataset; never modify the source.
• Validate statistically — after anonymization, verify distributions are preserved (utility) and re-identification is hard (privacy).
• Generalize, don't mask at random — fake data should be plausible and statistically similar to real data.
• Separate keys from data — pseudonymization keys must be stored separately from the pseudonymized dataset.
• Document what was done — anonymization transformations must be documented for audit and reproducibility.
• Re-evaluate when schema changes — new columns may introduce re-identification risk; audit anonymization on every schema change.

Worked Example and Anti-Patterns

Anti-Patterns to Avoid

Anti-pattern	Problem	Fix
No runbook	On-call engineer has no guidance during incident	Write runbook before going to production
Single point of failure	One component down takes everything with it	Design for redundancy at every layer
No monitoring	Problems discovered by users, not engineers	Instrument before launch
Manual toil	Repeated manual steps slow down and introduce errors	Automate anything done more than twice
Undocumented decisions	Next engineer repeats the same mistakes	Use Architecture Decision Records (ADRs)

Rules

• Start with the simplest thing that works -- complexity should be earned, not assumed.
• Make it observable before making it complex -- logs, metrics, and traces first.
• Automate toil -- anything done manually more than twice should be scripted.
• Document decisions -- use ADRs; future engineers will thank you.
• Test failure modes -- chaos engineering starts small; break one thing at a time.
• Prefer reversible decisions -- irreversible architecture decisions need the most careful thought.
• Own your runbooks -- every service needs a runbook before it goes to production.
• Measure before optimizing -- do not optimize what you have not profiled.
• Design for the 99th percentile user -- the average case is not the hard case.
• Keep it boring -- stable, predictable, well-understood technology over cutting-edge.