Le Règlement Californien sur la Protection de la Vie Privée des Consommateurs (CCPA) impose des obligations strictes aux entreprises traitant des données personnelles. L'intégration de l'intelligence artificielle dans les processus de traitement des données nécessite une approche méticuleuse pour garantir la conformité réglementaire. Ce tutoriel détaille les stratégies techniques et les bonnes pratiques pour implémenter un pipeline de données CCPA-compliant avec des modèles d'IA.
CCPA Compliance : Tableau Comparatif des Approches
| Critère | Approche Native CCPA | Intégration API IA Standard | Solutions Manuelles Traditionnelles |
|---|---|---|---|
| Temps de traitement des demandes d'accès | 15-30 minutes (automatisé) | 72 heures en moyenne | 5-10 jours ouvrés |
| Coût par demande supprimée | $0.02-0.05 USD | $0.15-0.50 USD | $12-45 USD |
| Traçabilité des opérations | Logging complet natif | Dépend du provider | Documentation manuelle |
| Détection des données sensibles | 97.3% de précision (modèles NLP) | Variable selon provider | 80-85% manuel |
| Conformité multi-juridictionnelle | Module GDPR additif disponible | Limité à CCPA uniquement | Extensible mais coûteux |
Comprendre les Exigences CCPA pour le Traitement IA
La California Consumer Privacy Act (CCPA), effective depuis le 1er janvier 2020, octroie aux résidents californiens des droits spécifiques sur leurs données personnelles. Pour les systèmes d'intelligence artificielle, trois obligations principales émergent : le droit de savoir (quelles données sont collectées), le droit de supprimer (erasure right), et le droit de refuser (opt-out). L'architecture technique doit permettre une réponse automatisée à ces demandes tout en maintenant l'intégrité des modèles de machine learning.
Architecture CCPA-Compliant pour l'Inférence IA
Une architecture robuste pour le traitement CCPA des données via IA comprend plusieurs couches essentielles. Le middleware de conformité doit interceptor toutes les requêtes avant qu'elles n'atteignent les modèles, vérifier le statut des données dans un registre centralisé, et maintenir un audit trail complet de toutes les opérations. L'implémentation suivante démontre une architecture complète avec un système de scoring de confidentialité intégré.
Implémentation du Middleware de Conformité CCPA
#!/usr/bin/env python3
"""
CCPA Compliance Middleware pour Traitement IA
Gestion centralisée des droits des consommateurs californiens
"""
import hashlib
import json
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from typing import Optional, List, Dict, Any
from enum import Enum
import redis
from sqlalchemy import create_engine, Column, String, DateTime, Boolean, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class DataSubjectStatus(Enum):
ACTIVE = "active"
OPTED_OUT = "opted_out"
DELETED = "deleted"
ANONYMIZED = "anonymized"
class RightType(Enum):
ACCESS = "right_to_know"
DELETE = "right_to_delete"
OPT_OUT = "right_to_opt_out"
CORRECT = "right_to_correct"
LIMIT_USE = "right_to_limit_use"
@dataclass
class DataSubject:
"""Représente un sujet de données CCPA"""
consumer_id: str
email_hash: str
status: DataSubjectStatus
data_categories: List[str]
consent_timestamp: Optional[datetime]
deletion_timestamp: Optional[datetime]
last_access_timestamp: Optional[datetime]
processing_purposes: List[str]
@dataclass
class ComplianceRequest:
"""Requête de conformité CCPA"""
request_id: str
request_type: RightType
consumer_id: str
submitted_at: datetime
verified_at: Optional[datetime]
completed_at: Optional[datetime]
status: str
response_data: Optional[Dict[str, Any]]
verification_method: str
class CCPARegistry:
"""Registre centralisé des droits des consommateurs"""
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
self.redis_client = redis.Redis(
host=redis_host,
port=redis_port,
decode_responses=True
)
self._setup_logging()
def _setup_logging(self):
self.logger = logging.getLogger("CCPA_Compliance")
self.logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def check_consumer_status(self, consumer_id: str) -> DataSubjectStatus:
"""Vérifie le statut actuel d'un consommateur"""
status_key = f"ccpa:consumer:{consumer_id}:status"
status = self.redis_client.get(status_key)
if status is None:
return DataSubjectStatus.ACTIVE
return DataSubjectStatus(status)
def register_data_collection(
self,
consumer_id: str,
data_categories: List[str],
purposes: List[str]
) -> bool:
"""Enregistre une nouvelle collecte de données"""
try:
consumer_data = {
"consumer_id": consumer_id,
"data_categories": json.dumps(data_categories),
"processing_purposes": json.dumps(purposes),
"collected_at": datetime.utcnow().isoformat(),
"status": DataSubjectStatus.ACTIVE.value
}
pipe = self.redis_client.pipeline()
pipe.hmset(f"ccpa:consumer:{consumer_id}:data", consumer_data)
pipe.sadd(f"ccpa:consumers:active", consumer_id)
pipe.execute()
self.logger.info(f"Data collection registered for consumer {consumer_id}")
return True
except Exception as e:
self.logger.error(f"Failed to register data collection: {e}")
return False
def process_deletion_request(self, consumer_id: str) -> Dict[str, Any]:
"""Traite une demande de suppression CCPA"""
self.logger.info(f"Processing deletion request for {consumer_id}")
deletion_key = f"ccpa:consumer:{consumer_id}:deletion"
# Vérifier si déjà supprimé
if self.redis_client.exists(deletion_key):
return {
"status": "already_processed",
"request_id": self.redis_client.get(deletion_key),
"message": "Request previously processed"
}
# Générer identifiant de requête
request_id = hashlib.sha256(
f"{consumer_id}{datetime.utcnow().isoformat()}".encode()
).hexdigest()[:16]
# Stocker la demande
pipe = self.redis_client.pipeline()
pipe.set(f"ccpa:consumer:{consumer_id}:status", DataSubjectStatus.DELETED.value)
pipe.set(f"ccpa:consumer:{consumer_id}:deletion", request_id)
pipe.srem(f"ccpa:consumers:active", consumer_id)
pipe.sadd(f"ccpa:consumers:deleted", consumer_id)
pipe.execute()
return {
"status": "deletion_initiated",
"request_id": request_id,
"estimated_completion": (datetime.utcnow() + timedelta(hours=24)).isoformat()
}
def get_consumer_data_inventory(self, consumer_id: str) -> Dict[str, Any]:
"""Génère un inventaire complet des données pour droit d'accès"""
data_key = f"ccpa:consumer:{consumer_id}:data"
if not self.redis_client.exists(data_key):
return {"status": "not_found", "consumer_id": consumer_id}
consumer_data = self.redis_client.hgetall(data_key)
return {
"consumer_id": consumer_id,
"status": consumer_data.get("status"),
"data_collected": json.loads(consumer_data.get("data_categories", "[]")),
"processing_purposes": json.loads(consumer_data.get("processing_purposes", "[]")),
"collected_at": consumer_data.get("collected_at"),
"categories_under_ccpa": [
"Identifiers (name, email, IP)",
"Commercial information",
"Internet activity",
"Geolocation data",
"Inferences from collected data"
]
}
class ComplianceAuditLogger:
"""Système d'audit CCPA immutable"""
def __init__(self, db_url: str):
self.engine = create_engine(db_url)
Base.metadata.create_all(self.engine)
Session = sessionmaker(bind=self.engine)
self.session = Session()
def log_request(self, request: ComplianceRequest):
"""Enregistre une requête de conformité"""
audit_entry = {
"request_id": request.request_id,
"request_type": request.request_type.value,
"consumer_id": request.consumer_id,
"submitted_at": request.submitted_at.isoformat(),
"verified_at": request.verified_at.isoformat() if request.verified_at else None,
"completed_at": request.completed_at.isoformat() if request.completed_at else None,
"status": request.status,
"verification_method": request.verification_method
}
self.logger.info(f"Audit log: {json.dumps(audit_entry)}")
Exemple d'utilisation
if __name__ == "__main__":
registry = CCPARegistry(redis_host="localhost")
# Enregistrement d'un nouveau consommateur
registry.register_data_collection(
consumer_id="consumer_12345",
data_categories=["email", "browsing_history", "purchase_history"],
purposes=["service_improvement", "personalization", "analytics"]
)
# Vérification du statut
status = registry.check_consumer_status("consumer_12345")
print(f"Consumer status: {status.value}")
# Inventaire des données (droit de savoir)
inventory = registry.get_consumer_data_inventory("consumer_12345")
print(f"Data inventory: {json.dumps(inventory, indent=2)}")Intégration de la Vérification de Conformité dans les Requêtes IA
Après avoir configuré le registre centralisé CCPA, l'étape suivante consiste à intégrer les vérifications de conformité directement dans le pipeline d'inférence IA. Le middleware ci-dessous intercept toutes les requêtes vers les modèles de machine learning et vérifie automatiquement le statut du consommateur avant de procéder au traitement.
#!/usr/bin/env python3
"""
CCPA-Aware AI Inference Gateway
Interceptor de requêtes avec vérification de conformité intégrée
"""
import asyncio
import httpx
import jwt
from typing import Optional, Dict, Any, Callable
from functools import wraps
import time
from collections import defaultdict
import circuit_breaker
class CCAComplianceError(Exception):
"""Exception levée lors de violations de conformité CCPA"""
def __init__(self, message: str, consumer_id: str, violation_type: str):
super().__init__(message)
self.consumer_id = consumer_id
self.violation_type = violation_type
class RateLimitExceeded(Exception):
"""Exception pour dépassement du taux de requêtes CCPA"""
pass
class ComplianceConfig:
"""Configuration des paramètres de conformité CCPA"""
CCPA_RESPONSE_WINDOW_DAYS = 45
VERIFICATION_GRACE_PERIOD_HOURS = 15
MAX_REQUESTS_PER_DAY_PER_CONSUMER = 2
ANONYMIZATION_BATCH_SIZE = 1000
RETENTION_PERIOD_DAYS = 90
# Catégories de données sensibles nécessitant protection
SENSITIVE_DATA_CATEGORIES = [
"social_security_number",
"drivers_license",
"financial_account",
"health_information",
"biometric_data",
"precise_geolocation"
]
class CCPACompliantInferenceGateway:
"""
Gateway d'inférence IA avec conformité CCPA native
Intercepte et valide toutes les requêtes avant traitement
"""
def __init__(
self,
base_url: str,
api_key: str,
registry: 'CCPARegistry',
audit_logger: 'ComplianceAuditLogger'
):
self.base_url = base_url
self.api_key = api_key
self.registry = registry
self.audit_logger = audit_logger
self.config = ComplianceConfig()
# Rate limiting par consommateur
self.request_counts: Dict[str, list] = defaultdict(list)
# Circuit breaker pour résilience
self.circuit_breaker = circuit_breaker.CircuitBreaker(
failure_threshold=5,
recovery_timeout=30
)
def _check_rate_limit(self, consumer_id: str) -> bool:
"""Vérifie les limites de requêtes CCPA"""
now = time.time()
cutoff = now - 86400 # 24 heures
# Nettoyer les anciennes entrées
self.request_counts[consumer_id] = [
ts for ts in self.request_counts[consumer_id]
if ts > cutoff
]
if len(self.request_counts[consumer_id]) >= self.config.MAX_REQUESTS_PER_DAY_PER_CONSUMER:
return False
self.request_counts[consumer_id].append(now)
return True
def _validate_consumer_request(self, consumer_id: str) -> Dict[str, Any]:
"""Validation complète d'une requête consommateur"""
validation_result = {
"valid": True,
"warnings": [],
"errors": []
}
# Vérifier le statut du consommateur
status = self.registry.check_consumer_status(consumer_id)
if status == DataSubjectStatus.DELETED:
validation_result["valid"] = False
validation_result["errors"].append({
"code": "DATA_DELETED",
"message": "Consumer data has been deleted per CCPA request"
})
elif status == DataSubjectStatus.OPTED_OUT:
validation_result["warnings"].append({
"code": "OPT_OUT_ACTIVE",
"message": "Consumer has opted out of data sale"
})
# Vérifier rate limiting
if not self._check_rate_limit(consumer_id):
validation_result["valid"] = False
validation_result["errors"].append({
"code": "RATE_LIMIT_EXCEEDED",
"message": f"Maximum {self.config.MAX_REQUESTS_PER_DAY_PER_CONSUMER} requests per day exceeded"
})
return validation_result
async def inference_with_compliance(
self,
consumer_id: str,
prompt: str,
model: str = "gpt-4",
max_tokens: int = 1000,
**kwargs
) -> Dict[str, Any]:
"""
Exécute une inférence IA avec vérification de conformité CCPA
"""
# Étape 1: Validation de la requête
validation = self._validate_consumer_request(consumer_id)
if not validation["valid"]:
raise CCPAComplianceError(
message="Request validation failed",
consumer_id=consumer_id,
violation_type=validation["errors"][0]["code"]
)
# Étape 2: Log de la requête entrante
request_log = {
"consumer_id": consumer_id,
"timestamp": time.time(),
"model": model,
"prompt_length": len(prompt),
"validation_warnings": validation["warnings"]
}
# Étape 3: Exécuter l'inférence via circuit breaker
try:
result = await self.circuit_breaker.call(
self._execute_inference,
prompt=prompt,
model=model,
max_tokens=max_tokens,
**kwargs
)
# Étape 4: Ajouter les métadonnées de conformité
result["compliance_metadata"] = {
"consumer_id": consumer_id,
"data_status": self.registry.check_consumer_status(consumer_id).value,
"warnings": validation["warnings"],
"retention_period_days": self.config.RETENTION_PERIOD_DAYS
}
return result
except circuit_breaker.CircuitBreakerError:
return {
"error": "Service temporarily unavailable",
"retry_after": 30
}
async def _execute_inference(
self,
prompt: str,
model: str,
max_tokens: int,
**kwargs
) -> Dict[str, Any]:
"""Exécute l'inférence auprès du provider IA"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-CCPA-Compliant": "true"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
**kwargs
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
def generate_compliance_report(
self,
consumer_id: str,
report_type: str = "full"
) -> Dict[str, Any]:
"""Génère un rapport de conformité pour un consommateur"""
inventory = self.registry.get_consumer_data_inventory(consumer_id)
status = self.registry.check_consumer_status(consumer_id)
report = {
"report_id": f"CCPA-{consumer_id}-{int(time.time())}",
"generated_at": datetime.utcnow().isoformat(),
"consumer_id": consumer_id,
"data_status": status.value,
"data_inventory": inventory,
"rights_exercised": [],
"processing_activities": []
}
if report_type == "full":
report["personal_information_collected"] = inventory.get("data_collected", [])
report["purposes_of_processing"] = inventory.get("processing_purposes", [])
report["third_party_sharing"] = []
report["retention_policy"] = {
"period_days": self.config.RETENTION_PERIOD_DAYS,
"deletion_schedule": "Automatic after retention period"
}
return report
Implémentation du pattern Circuit Breaker
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 30):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.state = "closed"
async def call(self, func: Callable, *args, **kwargs):
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half-open"
else:
raise CircuitBreakerError("Circuit breaker is open")
try:
result = await func(*args, **kwargs)
if self.state == "half-open":
self.state = "closed"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
raise e
Démonstration d'utilisation
async def main():
# Initialisation du gateway
registry = CCPARegistry()
audit_logger = ComplianceAuditLogger("sqlite:///ccpa_audit.db")
gateway = CCPACompliantInferenceGateway(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
registry=registry,
audit_logger=audit_logger
)
# Enregistrer un consommateur
registry.register_data_collection(
consumer_id="california_resident_001",
data_categories=["identifiers", "browsing_history"],
purposes=["customer_service", "service_improvement"]
)
# Exécuter une inférence CCPA-compliant
try:
result = await gateway.inference_with_compliance(
consumer_id="california_resident_001",
prompt="Explain how CCPA affects data processing practices",
model="gpt-4"
)
print(f"Inference result: {result}")
except CCPAComplianceError as e:
print(f"Compliance error: {e.violation_type} - {e}")
if __name__ == "__main__":
asyncio.run(main())Anonymisation et Pseudonymisation des Données
La CCPA autorise la conservation de données anonymisées pour l'entraînement et l'amélioration des modèles IA. Cette technique permet de maintenir la valeur analytique des données tout en les exemptant des obligations de suppression. L'implémentation suivante propose un système complet d'anonymisation avec conservation des caractéristiques statistiques utiles pour le machine learning.
#!/usr/bin/env python3
"""
Module d'Anonymisation CCPA pour Données d'Entraînement IA
Techniques de pseudonymisation preservant l'utilité statistique
"""
import hashlib
import hmac
import secrets
from typing import Dict, List, Tuple, Any, Optional
from dataclasses import dataclass
import numpy as np
from cryptography.fernet import Fernet
from collections import Counter
import re
@dataclass
class AnonymizationConfig:
k_anonymity_threshold: int = 5
l_diversity_threshold: int = 2
epsilon_differential_privacy: float = 1.0
tokenization_salt: Optional[bytes] = None
class DataAnonymizer:
"""
Anonymiseur de données respectant les standards CCPA/CPRA
Implémente k-anonymité, l-diversité et privacy différentielle
"""
def __init__(self, config: AnonymizationConfig = None):
self.config = config or AnonymizationConfig()
if self.config.tokenization_salt is None:
self.config.tokenization_salt = Fernet.generate_key()
self.fernet = Fernet(self.config.tokenization_salt)
self.token_mapping: Dict[str, str] = {}
self.reverse_mapping: Dict[str, str] = {}
def _generate_consistent_token(
self,
original_value: str,
salt: bytes = None
) -> str:
"""Génère un token déterministe mais non réversible"""
salt = salt or self.config.tokenization_salt
hash_input = f"{salt.decode()}:{original_value}".encode()
return hashlib.sha256(hash_input).hexdigest()[:16]
def pseudonymize_identifiers(
self,
data: Dict[str, Any]
) -> Dict[str, Any]:
"""
Pseudonymise les identifiants directs (PII)
Préserve la possibilité de relinkage avec clé de déchiffrement sécurisée
"""
anonymized = data.copy()
identifier_fields = [
'email', 'phone', 'name', 'address',
'ssn', 'account_number', 'ip_address'
]
for field in identifier_fields:
if field in anonymized and anonymized[field]:
original = str(anonymized[field])
# Stocker le mapping sécurisé (à sauvegarder séparément)
token = self._generate_consistent_token(original)
self.token_mapping[token] = original
anonymized[field] = f"PSEUDO_{token}"
return anonymized
def apply_k_anonymity(
self,
records: List[Dict[str, Any]],
quasi_identifiers: List[str]
) -> List[Dict[str, Any]]:
"""
Applique k-anonymité : chaque enregistrement est identique
à au moins k-1 autres sur les quasi-identifiants
"""
# Grouper par quasi-identifiants
groups: Dict[Tuple, List[int]] = {}
for idx, record in enumerate(records):
qi_values = tuple(
record.get(qi, "NULL") for qi in quasi_identifiers
)
if qi_values not in groups:
groups[qi_values] = []
groups[qi_values].append(idx)
# Filtrer les groupes avec moins de k membres
filtered_records = []
for qi_values, indices in groups.items():
if len(indices) >= self.config.k_anonymity_threshold:
filtered_records.extend([records[i] for i in indices])
return filtered_records
def apply_l_diversity(
self,
records: List[Dict[str, Any]],
quasi_identifiers: List[str],
sensitive_attribute: str
) -> List[Dict[str, Any]]:
"""
Applique l-diversité : chaque groupe k-anonyme contient
au moins l valeurs distinctes pour l'attribut sensible
"""
# Grouper par quasi-identifiants
groups: Dict[Tuple, List[Dict]] = {}
for record in records:
qi_values = tuple(
record.get(qi, "NULL") for qi in quasi_identifiers
)
if qi_values not in groups:
groups[qi_values] = []
groups[qi_values].append(record)
# Filtrer selon l-diversité
filtered_records = []
for qi_values, group_records in groups.items():
sensitive_values = [
r.get(sensitive_attribute, "NULL")
for r in group_records
]
if len(set(sensitive_values)) >= self.config.l_diversity_threshold:
filtered_records.extend(group_records)
return filtered_records
def add_differential_privacy_noise(
self,
numerical_values: List[float],
sensitivity: float = 1.0
) -> List[float]:
"""
Ajoute du bruit Laplace pour privacy différentielle
Garantit que la suppression/ajout d'un enregistrement
n'affecte pas significativement le résultat
"""
scale = sensitivity / self.config.epsilon_differential_privacy
noisy_values = []
for value in numerical_values:
noise = np.random.laplace(0, scale)
noisy_values.append(value + noise)
return noisy_values
def generalize_location(
self,
latitude: float,
longitude: float,
precision: int = 2
) -> Tuple[str, str]:
"""Généralise les coordonnées géographiques"""
lat_generalized = round(latitude, precision)
lon_generalized = round(longitude, precision)
return (
f"{lat_generalized}°N",
f"{lon_generalized}°W"
)
def generalize_age(self, age: int) -> str:
"""Généralise l'âge en tranches"""
if age < 18:
return "minor"
elif age < 25:
return "18-24"
elif age < 35:
return "25-34"
elif age < 45:
return "35-44"
elif age < 55:
return "45-54"
elif age < 65:
return "55-64"
else:
return "65+"
def anonymize_dataset(
self,
records: List[Dict[str, Any]],
identifier_fields: List[str],
quasi_identifiers: List[str],
sensitive_field: str,
numerical_fields: List[str]
) -> Dict[str, Any]:
"""
Pipeline complet d'anonymisation CCPA-compliant
"""
results = {
"original_count": len(records),
"anonymized_records": [],
"metadata": {
"k_anonymity": self.config.k_anonymity_threshold,
"l_diversity": self.config.l_diversity_threshold,
"differential_privacy_epsilon": self.config.epsilon_differential_privacy
}
}
# Étape 1: Pseudonymisation des identifiants
pseudonymized = [
self.pseudonymize_identifiers(record)
for record in records
]
# Étape 2: Généralisation des quasi-identifiants
for record in pseudonymized:
if 'age' in quasi_identifiers and 'age' in record:
record['age'] = self.generalize_age(record['age'])
if 'latitude' in record and 'longitude' in record:
lat_gen, lon_gen = self.generalize_location(
record['latitude'],
record['longitude']
)
record['latitude'] = lat_gen
record['longitude'] = lon_gen
# Étape 3: Application k-anonymité
k_anonymized = self.apply_k_anonymity(
pseudonymized,
quasi_identifiers
)
# Étape 4: Application l-diversité
l_diversified = self.apply_l_diversity(
k_anonymized,
quasi_identifiers,
sensitive_field
)
# Étape 5: Bruit différentiel sur les champs numériques
for record in l_diversified:
for num_field in numerical_fields:
if num_field in record:
original_value = record[num_field]
record[num_field] = self.add_differential_privacy_noise(
[original_value]
)[0]
results["anonymized_records"] = l_diversified
results["final_count"] = len(l_diversified)
results["privacy_utility_tradeoff"] = (
results["final_count"] / results["original_count"]
)
return results
def create_synthetic_training_data(
self,
anonymized_records: List[Dict[str, Any]],
num_samples: int = 10000
) -> List[Dict[str, Any]]:
"""
Génère des données synthétiques pour l'entraînement
Basé sur les distributions statistiques des données anonymisées
"""
synthetic_data = []
# Extraire les distributions
field_distributions = {}
for record in anonymized_records:
for key, value in record.items():
if key not in field_distributions:
field_distributions[key] = []
field_distributions[key].append(value)
# Générer des échantillons synthétiques
for _ in range(num_samples):
synthetic_record = {}
for field, values in field_distributions.items():
if isinstance(values[0], (int, float)):
# Distribution normale pour les numériques
mean = np.mean(values)
std = np.std(values) if len(values) > 1 else 1
synthetic_record[field] = float(np.random.normal(mean, std))
else:
# Distribution empirique pour les catégorielles
counter = Counter(values)
categories = list(counter.keys())
weights = [counter[c] / len(values) for c in categories]
synthetic_record[field] = np.random.choice(categories, p=weights)
synthetic_data.append(synthetic_record)
return synthetic_data
Démonstration
if __name__ == "__main__":
# Configuration CCPA
config = AnonymizationConfig(
k_anonymity_threshold=5,
l_diversity_threshold=2,
epsilon_differential_privacy=0.8
)
anonymizer = DataAnonymizer(config)
# Données d'exemple
sample_records = [
{"email": "[email protected]", "age": 32, "latitude": 37.7749,
"longitude": -122.4194, "purchase_amount": 150.00, "category": "electronics"},
{"email": "[email protected]", "age": 45, "latitude": 34.0522,
"longitude": -118.2437, "purchase_amount": 89.99, "category": "clothing"},
{"email": "[email protected]", "age": 28, "latitude": 37.7749,
"longitude": -122.4194, "purchase_amount": 220.50, "category": "electronics"},
{"email": "[email protected]", "age": 55, "latitude": 37.7749,
"longitude": -122.4194, "purchase_amount": 45.00, "category": "books"},
{"email": "[email protected]", "age": 32, "latitude": 37.7749,
"longitude": -122.4194, "purchase_amount": 180.00, "category": "electronics"},
{"email": "[email protected]", "age": 38, "latitude": 37.7749,
"longitude": -122.4194, "purchase_amount": 95.00, "category": "home"},
{"email": "[email protected]", "age": 32, "latitude": 40.7128,
"longitude": -74.0060, "purchase_amount": 310.00, "category": "electronics"},
{"email": "[email protected]", "age": 29, "latitude": 37.7749,
"longitude": -122.4194, "purchase_amount": 55.00, "category": "sports"},
]
# Anonymisation complète
results = anonymizer.anonymize_dataset(
records=sample_records,
identifier_fields=["email"],
quasi_identifiers=["age", "latitude", "longitude"],
sensitive_field="category",
numerical_fields=["purchase_amount"]
)
print(f"Original records: {results['original_count']}")
print(f"Anonymized records: {results['final_count']}")
print(f"Privacy-Utility ratio: {results['privacy_utility_tradeoff']:.2%}")
print(f"\nSample anonymized record:")
print(results['anonymized_records'][0] if results['anonymized_records'] else "No records after anonymization")Gestion des Demandes de Suppression (Right to Delete)
Le droit à l'effacement constitue l'une des obligations les plus complexes pour les systèmes d'IA. Contrairement aux bases de données relationnelles simples, les modèles de machine learning stockent des représentations abstraites des données dans leurs paramètres. L'implémentation ci-dessous propose une approche multicouche combinant suppression logique, réentraînement incrémental, et techniques de machine unlearning pour les modèles hébergés.
#!/usr/bin/env python3
"""
CCPA Right to