Data Collection Architecture & Implementation

Building Robust Data Pipelines for Marketing Analytics

# Example: Real-time customer event processing
import kafka
from kafka import KafkaConsumer
import json
import pandas as pd

def process_customer_events():
    consumer = KafkaConsumer(
        'customer_events',
        bootstrap_servers=['localhost:9092'],
        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
    )

    for message in consumer:
        event_data = message.value
        # Process event for real-time personalization
        customer_profile = update_customer_profile(event_data)
        trigger_marketing_automation(customer_profile)
                    

The Three Whys: Technical Implementation

Advanced Machine Learning Models for Customer Behavior Analysis

# Engagement prediction model
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

# Feature engineering for engagement prediction
def build_engagement_model(data):
    features = ['page_views', 'session_duration', 'email_opens',
                'social_interactions', 'device_type_encoded']
    X = data[features]
    y = data['engaged']

    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    model = LogisticRegression(random_state=42)
    model.fit(X_scaled, y)
    return model, scaler
                                

# Conversion decision tree implementation
from sklearn.ensemble import RandomForestClassifier
import xgboost as xgb
from sklearn.tree import export_text

def build_conversion_model(data):
    features = ['engagement_score', 'price_sensitivity', 'product_affinity',
                'channel_preference', 'purchase_history']
    X = data[features]
    y = data['converted']

    # Random Forest for interpretability
    rf_model = RandomForestClassifier(n_estimators=100, max_depth=10)
    rf_model.fit(X, y)

    # XGBoost for performance
    xgb_model = xgb.XGBClassifier()
    xgb_model.fit(X, y)

    return rf_model, xgb_model
                                

# Churn prediction with survival analysis
from lifelines import CoxPHFitter
from lifelines import KaplanMeierFitter
import dowhy

def build_churn_model(data):
    # Survival analysis for time-to-churn
    cph = CoxPHFitter()
    cph.fit(data, duration_col='days_to_churn', event_col='churned')

    # Causal analysis for intervention strategies
    causal_graph = """
    digraph {
    satisfaction -> churn;
    support_interactions -> churn;
    product_usage -> churn;
    satisfaction -> product_usage;
    }
    """

    model = dowhy.CausalModel(data, causal_graph=causal_graph,
                             treatment='support_interactions',
                             outcome='churn')
    return cph, model
                                

Time Series Analysis for Marketing Optimization

Advanced Forecasting and Seasonality Detection

# Comprehensive time series analysis pipeline
import pandas as pd
import numpy as np
from prophet import Prophet
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.seasonal import seasonal_decompose
import matplotlib.pyplot as plt

class MarketingTimeSeriesAnalyzer:
    def __init__(self, data):
        self.data = data
        self.models = {}

    def detect_seasonality(self, column):
        """Detect seasonal patterns in marketing data"""
        decomposition = seasonal_decompose(
            self.data[column],
            model='multiplicative',
            period=12
        )
        return decomposition

    def build_prophet_model(self, column, holidays_df=None):
        """Build Prophet model with marketing-specific considerations"""
        df = self.data.reset_index()
        df.columns = ['ds', 'y']

        model = Prophet(
            changepoint_prior_scale=0.05,
            seasonality_prior_scale=10,
            holidays_prior_scale=10,
            daily_seasonality=False,
            weekly_seasonality=True,
            yearly_seasonality=True
        )

        if holidays_df is not None:
            model.add_country_holidays(country_name='US')

        model.fit(df)
        self.models['prophet'] = model
        return model

    def forecast_with_confidence(self, periods=30):
        """Generate forecasts with confidence intervals"""
        future = self.models['prophet'].make_future_dataframe(periods=periods)
        forecast = self.models['prophet'].predict(future)
        return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']]
                    

Production-Ready Sentiment Analysis System

NLP Pipeline with Real-Time Processing and Geospatial Analysis

# Production sentiment analysis pipeline
import pandas as pd
import numpy as np
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import spacy
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
import redis
import geopandas as gpd
from sklearn.cluster import DBSCAN

class ProductionSentimentAnalyzer:
    def __init__(self):
        # Load pre-trained models
        self.bert_analyzer = pipeline(
            "sentiment-analysis",
            model="cardiffnlp/twitter-roberta-base-sentiment-latest"
        )
        self.vader_analyzer = SentimentIntensityAnalyzer()
        self.nlp = spacy.load("en_core_web_sm")
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)

    def preprocess_text(self, text):
        """Advanced text preprocessing for marketing content"""
        doc = self.nlp(text)

        # Remove noise, normalize, handle negations
        tokens = []
        for token in doc:
            if not token.is_stop and not token.is_punct and token.is_alpha:
                tokens.append(token.lemma_.lower())

        return " ".join(tokens)

    def analyze_sentiment_ensemble(self, text):
        """Multi-model sentiment analysis with confidence scoring"""
        preprocessed = self.preprocess_text(text)

        # BERT-based analysis
        bert_result = self.bert_analyzer(preprocessed)[0]

        # VADER analysis
        vader_scores = self.vader_analyzer.polarity_scores(text)

        # Ensemble scoring
        final_score = (bert_result['score'] * 0.7 +
                      abs(vader_scores['compound']) * 0.3)

        return {
            'sentiment': bert_result['label'],
            'confidence': final_score,
            'compound_score': vader_scores['compound']
        }

    def geospatial_sentiment_clustering(self, sentiment_data_with_coords):
        """Cluster sentiment data geographically"""
        coords = np.array([(row['lat'], row['lon'])
                          for _, row in sentiment_data_with_coords.iterrows()])

        # DBSCAN clustering for geographical sentiment hotspots
        clustering = DBSCAN(eps=0.01, min_samples=5).fit(coords)
        sentiment_data_with_coords['cluster'] = clustering.labels_

        return sentiment_data_with_coords
                    

Advanced A/B Testing Framework

Statistical Rigor and Machine Learning Model Comparison

# Advanced A/B testing framework with ML model comparison
import numpy as np
import pandas as pd
from scipy import stats
from sklearn.ensemble import RandomForestClassifier
import xgboost as xgb
from sklearn.neural_network import MLPClassifier
from sklearn.model_selection import cross_val_score, StratifiedKFold
from sklearn.metrics import roc_auc_score, precision_recall_curve
import mlflow
import mlflow.sklearn

class MLModelABTester:
    def __init__(self, experiment_name):
        self.experiment_name = experiment_name
        mlflow.set_experiment(experiment_name)
        self.models = {
            'random_forest': RandomForestClassifier(n_estimators=100, random_state=42),
            'xgboost': xgb.XGBClassifier(random_state=42),
            'neural_network': MLPClassifier(hidden_layer_sizes=(100, 50), random_state=42)
        }

    def calculate_sample_size(self, baseline_rate, minimum_effect, alpha=0.05, power=0.8):
        """Calculate required sample size for A/B test"""
        from statsmodels.stats.power import ttest_power
        effect_size = minimum_effect / np.sqrt(baseline_rate * (1 - baseline_rate))
        n = ttest_power(effect_size, power, alpha, alternative='two-sided')
        return int(n * 2)  # For two groups

    def compare_models(self, X, y, cv_folds=5):
        """Compare multiple ML models with statistical significance testing"""
        results = {}

        with mlflow.start_run(run_name="model_comparison"):
            cv = StratifiedKFold(n_splits=cv_folds, shuffle=True, random_state=42)

            for model_name, model in self.models.items():
                with mlflow.start_run(run_name=model_name, nested=True):
                    # Cross-validation scores
                    cv_scores = cross_val_score(model, X, y, cv=cv, scoring='roc_auc')

                    # Train final model for feature importance
                    model.fit(X, y)
                    y_pred_proba = model.predict_proba(X)[:, 1]
                    auc_score = roc_auc_score(y, y_pred_proba)

                    results[model_name] = {
                        'cv_scores': cv_scores,
                        'mean_auc': cv_scores.mean(),
                        'std_auc': cv_scores.std(),
                        'final_auc': auc_score
                    }

                    # Log metrics to MLflow
                    mlflow.log_metrics({
                        'mean_cv_auc': cv_scores.mean(),
                        'std_cv_auc': cv_scores.std(),
                        'final_auc': auc_score
                    })

                    # Log model
                    mlflow.sklearn.log_model(model, f"{model_name}_model")

        return self.statistical_significance_test(results)

    def statistical_significance_test(self, results):
        """Perform pairwise statistical significance testing"""
        model_names = list(results.keys())
        significance_matrix = pd.DataFrame(index=model_names, columns=model_names)

        for i, model1 in enumerate(model_names):
            for j, model2 in enumerate(model_names):
                if i != j:
                    scores1 = results[model1]['cv_scores']
                    scores2 = results[model2]['cv_scores']
                    t_stat, p_value = stats.ttest_rel(scores1, scores2)
                    significance_matrix.loc[model1, model2] = p_value
                else:
                    significance_matrix.loc[model1, model2] = 1.0

        return results, significance_matrix
                    

Market Basket Analysis & Collaborative Filtering

Advanced Recommendation Systems for Personalized Marketing

# Advanced recommendation system implementation
import pandas as pd
import numpy as np
from mlxtend.frequent_patterns import apriori, association_rules
from sklearn.metrics.pairwise import cosine_similarity
from surprise import Dataset, Reader, SVD, KNNBasic
from surprise.model_selection import cross_validate
import redis
import pickle

class AdvancedRecommendationSystem:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.models = {}

    def market_basket_analysis(self, transactions_df):
        """Implement market basket analysis with association rules"""
        # Convert transactions to basket format
        basket = transactions_df.groupby(['transaction_id', 'product_id'])['quantity'].sum().unstack().fillna(0)

        # Convert to binary matrix
        basket_binary = basket.applymap(lambda x: 1 if x > 0 else 0)

        # Find frequent itemsets
        frequent_itemsets = apriori(basket_binary, min_support=0.01, use_colnames=True)

        # Generate association rules
        rules = association_rules(frequent_itemsets, metric="lift", min_threshold=1.0)

        return frequent_itemsets, rules

    def collaborative_filtering_svd(self, ratings_df):
        """Implement SVD-based collaborative filtering"""
        reader = Reader(rating_scale=(1, 5))
        data = Dataset.load_from_df(ratings_df[['user_id', 'item_id', 'rating']], reader)

        # Train SVD model
        svd_model = SVD(n_factors=100, random_state=42)
        cross_validate(svd_model, data, measures=['RMSE', 'MAE'], cv=5, verbose=True)

        # Train on full dataset
        trainset = data.build_full_trainset()
        svd_model.fit(trainset)

        self.models['svd'] = svd_model
        return svd_model

    def user_based_collaborative_filtering(self, user_item_matrix):
        """Implement user-based collaborative filtering"""
        # Calculate user similarity matrix
        user_similarity = cosine_similarity(user_item_matrix)
        user_similarity_df = pd.DataFrame(
            user_similarity,
            index=user_item_matrix.index,
            columns=user_item_matrix.index
        )

        def get_user_recommendations(user_id, n_recommendations=10):
            # Find similar users
            similar_users = user_similarity_df[user_id].sort_values(ascending=False)[1:11]

            # Get items rated by similar users but not by target user
            user_items = set(user_item_matrix.loc[user_id][user_item_matrix.loc[user_id] > 0].index)

            recommendations = {}
            for similar_user, similarity_score in similar_users.items():
                similar_user_items = user_item_matrix.loc[similar_user]
                for item, rating in similar_user_items[similar_user_items > 0].items():
                    if item not in user_items:
                        if item not in recommendations:
                            recommendations[item] = 0
                        recommendations[item] += similarity_score * rating

            return sorted(recommendations.items(), key=lambda x: x[1], reverse=True)[:n_recommendations]

        return get_user_recommendations

    def real_time_recommendation_cache(self, user_id, recommendations):
        """Cache recommendations in Redis for real-time serving"""
        cache_key = f"recommendations:{user_id}"
        self.redis_client.setex(
            cache_key,
            3600,  # 1 hour expiry
            pickle.dumps(recommendations)
        )

    def get_cached_recommendations(self, user_id):
        """Retrieve cached recommendations"""
        cache_key = f"recommendations:{user_id}"
        cached_data = self.redis_client.get(cache_key)

        if cached_data:
            return pickle.loads(cached_data)
        return None
                    

Generative AI Implementation for Marketing

Production-Ready LLM Integration and RAG Systems

# Production-ready RAG system for marketing content generation
import openai
import pinecone
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import pandas as pd
from typing import List, Dict
import re

class MarketingRAGSystem:
    def __init__(self, pinecone_api_key, openai_api_key, index_name):
        # Initialize connections
        pinecone.init(api_key=pinecone_api_key, environment="us-west1-gcp")
        openai.api_key = openai_api_key

        self.embeddings = OpenAIEmbeddings()
        self.index_name = index_name
        self.vectorstore = None

        # Brand voice configuration
        self.brand_guidelines = {
            'tone': 'professional yet approachable',
            'voice': 'confident and helpful',
            'keywords': ['innovative', 'reliable', 'customer-focused'],
            'avoid_words': ['cheap', 'discount', 'basic']
        }

    def setup_knowledge_base(self, documents: List[str]):
        """Setup vector database with marketing knowledge"""
        # Split documents into chunks
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len
        )

        chunks = []
        for doc in documents:
            doc_chunks = text_splitter.split_text(doc)
            chunks.extend(doc_chunks)

        # Create vector store
        self.vectorstore = Pinecone.from_texts(
            chunks,
            self.embeddings,
            index_name=self.index_name
        )

    def generate_marketing_content(self, prompt: str, content_type: str) -> Dict:
        """Generate marketing content with RAG enhancement"""
        # Retrieve relevant context
        retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})

        # Content-specific prompt templates
        templates = {
            'email': """
            Create a marketing email with the following requirements:
            - Subject line that drives opens
            - Personalized greeting
            - Clear value proposition
            - Strong call-to-action

            Context: {context}
            Requirements: {prompt}

            Brand Guidelines: {brand_guidelines}
            """,
            'social_media': """
            Create engaging social media content:
            - Platform-appropriate length
            - Engaging hook
            - Relevant hashtags
            - Visual content suggestions

            Context: {context}
            Requirements: {prompt}

            Brand Guidelines: {brand_guidelines}
            """,
            'product_description': """
            Create compelling product descriptions:
            - Key features and benefits
            - Target audience appeal
            - SEO-optimized content
            - Conversion-focused language

            Context: {context}
            Product Details: {prompt}

            Brand Guidelines: {brand_guidelines}
            """
        }

        # Build the QA chain
        qa_chain = RetrievalQA.from_chain_type(
            llm=OpenAI(temperature=0.7),
            chain_type="stuff",
            retriever=retriever,
            return_source_documents=True
        )

        # Generate content
        formatted_prompt = templates[content_type].format(
            context="{context}",
            prompt=prompt,
            brand_guidelines=str(self.brand_guidelines)
        )

        result = qa_chain({"query": formatted_prompt})

        # Content quality validation
        quality_score = self.validate_content_quality(result['result'])

        return {
            'content': result['result'],
            'quality_score': quality_score,
            'source_documents': result['source_documents'],
            'brand_compliance': self.check_brand_compliance(result['result'])
        }

    def validate_content_quality(self, content: str) -> float:
        """Validate generated content quality"""
        quality_metrics = {
            'length_appropriate': 0.2,
            'readability': 0.3,
            'brand_alignment': 0.3,
            'call_to_action_present': 0.2
        }

        score = 0.0

        # Length check
        if 50 <= len(content.split()) <= 500:
            score += quality_metrics['length_appropriate']

        # CTA check
        cta_patterns = [r'click here', r'shop now', r'learn more', r'get started', r'contact us']
        if any(re.search(pattern, content.lower()) for pattern in cta_patterns):
            score += quality_metrics['call_to_action_present']

        # Brand keyword check
        brand_keywords_present = sum(1 for keyword in self.brand_guidelines['keywords']
                                   if keyword.lower() in content.lower())
        score += (brand_keywords_present / len(self.brand_guidelines['keywords'])) * quality_metrics['brand_alignment']

        return min(score, 1.0)

    def check_brand_compliance(self, content: str) -> Dict:
        """Check content compliance with brand guidelines"""
        compliance = {
            'avoid_words_used': [],
            'brand_keywords_included': [],
            'tone_appropriate': True
        }

        # Check for avoided words
        for word in self.brand_guidelines['avoid_words']:
            if word.lower() in content.lower():
                compliance['avoid_words_used'].append(word)

        # Check for brand keywords
        for keyword in self.brand_guidelines['keywords']:
            if keyword.lower() in content.lower():
                compliance['brand_keywords_included'].append(keyword)

        return compliance
                    

MLOps Pipeline for Marketing AI Systems

Production Deployment, Monitoring, and Continuous Learning

# MLOps pipeline for marketing AI deployment
import docker
import kubernetes
from kubernetes import client, config
import mlflow
import prometheus_client
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import joblib
import pandas as pd

class MarketingMLOps:
    def __init__(self):
        # Initialize MLflow tracking
        mlflow.set_tracking_uri("http://mlflow-server:5000")

        # Kubernetes configuration
        config.load_incluster_config()  # For in-cluster deployment
        self.k8s_client = client.AppsV1Api()

        # Prometheus metrics
        self.model_predictions = prometheus_client.Counter(
            'marketing_model_predictions_total',
            'Total predictions made by marketing models',
            ['model_name', 'model_version']
        )
        self.model_latency = prometheus_client.Histogram(
            'marketing_model_latency_seconds',
            'Model prediction latency'
        )

    def deploy_model_to_kubernetes(self, model_name, model_version, image_tag):
        """Deploy ML model as Kubernetes deployment"""
        deployment = {
            "apiVersion": "apps/v1",
            "kind": "Deployment",
            "metadata": {"name": f"{model_name}-{model_version}"},
            "spec": {
                "replicas": 3,
                "selector": {"matchLabels": {"app": model_name}},
                "template": {
                    "metadata": {"labels": {"app": model_name}},
                    "spec": {
                        "containers": [{
                            "name": model_name,
                            "image": f"marketing-ml-models:{image_tag}",
                            "ports": [{"containerPort": 8080}],
                            "env": [
                                {"name": "MODEL_NAME", "value": model_name},
                                {"name": "MODEL_VERSION", "value": model_version}
                            ],
                            "resources": {
                                "requests": {"cpu": "100m", "memory": "256Mi"},
                                "limits": {"cpu": "500m", "memory": "512Mi"}
                            },
                            "livenessProbe": {
                                "httpGet": {"path": "/health", "port": 8080},
                                "initialDelaySeconds": 30,
                                "periodSeconds": 10
                            }
                        }]
                    }
                }
            }
        }

        self.k8s_client.create_namespaced_deployment(
            namespace="marketing-ml",
            body=deployment
        )

    def create_model_monitoring_pipeline(self):
        """Create Airflow DAG for model monitoring and retraining"""
        default_args = {
            'owner': 'marketing-team',
            'depends_on_past': False,
            'start_date': datetime(2025, 1, 1),
            'email_on_failure': True,
            'email_on_retry': False,
            'retries': 1,
            'retry_delay': timedelta(minutes=5)
        }

        dag = DAG(
            'marketing_model_monitoring',
            default_args=default_args,
            description='Monitor and retrain marketing ML models',
            schedule_interval=timedelta(hours=6),
            catchup=False
        )

        def check_data_drift():
            """Check for data drift in model inputs"""
            # Load reference dataset
            reference_data = pd.read_csv("reference_data.csv")
            current_data = pd.read_csv("current_data.csv")

            # Calculate drift metrics (using KS test as example)
            from scipy.stats import ks_2samp

            drift_detected = False
            for column in reference_data.columns:
                if reference_data[column].dtype in ['int64', 'float64']:
                    statistic, p_value = ks_2samp(
                        reference_data[column],
                        current_data[column]
                    )
                    if p_value < 0.05:  # Significant drift detected
                        drift_detected = True
                        break

            return drift_detected

        def retrain_model_if_needed():
            """Retrain model if performance degradation detected"""
            # Check model performance metrics
            latest_metrics = mlflow.get_run(mlflow.search_runs().iloc[0].run_id).data.metrics

            if latest_metrics.get('accuracy', 0) < 0.85:  # Performance threshold
                # Trigger retraining
                from marketing_model_trainer import train_new_model
                train_new_model()

        # Define tasks
        drift_check = PythonOperator(
            task_id='check_data_drift',
            python_callable=check_data_drift,
            dag=dag
        )

        model_retrain = PythonOperator(
            task_id='retrain_model_if_needed',
            python_callable=retrain_model_if_needed,
            dag=dag
        )

        # Set task dependencies
        drift_check >> model_retrain

        return dag

    def implement_canary_deployment(self, model_name, new_version):
        """Implement canary deployment for model updates"""
        # Deploy new version with limited traffic
        canary_service = {
            "apiVersion": "v1",
            "kind": "Service",
            "metadata": {"name": f"{model_name}-canary"},
            "spec": {
                "selector": {"app": model_name, "version": new_version},
                "ports": [{"port": 80, "targetPort": 8080}],
                "type": "ClusterIP"
            }
        }

        # Istio VirtualService for traffic splitting
        virtual_service = {
            "apiVersion": "networking.istio.io/v1alpha3",
            "kind": "VirtualService",
            "metadata": {"name": f"{model_name}-vs"},
            "spec": {
                "hosts": [f"{model_name}.marketing-ml.svc.cluster.local"],
                "http": [{
                    "match": [{"headers": {"canary": {"exact": "true"}}}],
                    "route": [{"destination": {"host": f"{model_name}-canary"}}]
                }, {
                    "route": [{
                        "destination": {"host": f"{model_name}-stable"},
                        "weight": 90
                    }, {
                        "destination": {"host": f"{model_name}-canary"},
                        "weight": 10
                    }]
                }]
            }
        }

        return canary_service, virtual_service
                    

AI Ethics & Security Implementation

Building Responsible and Secure Marketing AI Systems

# Comprehensive AI ethics and security implementation
import pandas as pd
import numpy as np
from fairlearn.metrics import demographic_parity_difference, equalized_odds_difference
from fairlearn.reductions import ExponentiatedGradient, DemographicParity
import shap
from cryptography.fernet import Fernet
import hashlib
from datetime import datetime, timedelta
import logging

class AIEthicsAndSecurity:
    def __init__(self):
        self.encryption_key = Fernet.generate_key()
        self.cipher_suite = Fernet(self.encryption_key)
        self.audit_log = []

        # Setup logging for compliance
        logging.basicConfig(
            filename='ai_ethics_audit.log',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )

    def implement_differential_privacy(self, data, epsilon=1.0):
        """Add differential privacy noise to protect individual privacy"""
        sensitivity = 1.0  # Assuming normalized data
        noise_scale = sensitivity / epsilon

        # Add Laplace noise
        noise = np.random.laplace(0, noise_scale, data.shape)
        private_data = data + noise

        self.log_privacy_action(f"Applied differential privacy with epsilon={epsilon}")
        return private_data

    def detect_and_mitigate_bias(self, X, y, sensitive_features, model):
        """Comprehensive bias detection and mitigation"""
        # Train baseline model
        model.fit(X, y)
        y_pred = model.predict(X)

        # Calculate bias metrics
        dp_diff = demographic_parity_difference(
            y_true=y,
            y_pred=y_pred,
            sensitive_features=sensitive_features
        )

        eo_diff = equalized_odds_difference(
            y_true=y,
            y_pred=y_pred,
            sensitive_features=sensitive_features
        )

        bias_metrics = {
            'demographic_parity_difference': dp_diff,
            'equalized_odds_difference': eo_diff
        }

        # Mitigate bias if detected
        if abs(dp_diff) > 0.1 or abs(eo_diff) > 0.1:
            # Use ExponentiatedGradient for bias mitigation
            constraint = DemographicParity()
            mitigator = ExponentiatedGradient(model, constraint)

            mitigator.fit(X, y, sensitive_features=sensitive_features)
            y_pred_mitigated = mitigator.predict(X)

            # Recalculate metrics
            dp_diff_mitigated = demographic_parity_difference(
                y_true=y,
                y_pred=y_pred_mitigated,
                sensitive_features=sensitive_features
            )

            bias_metrics['mitigated_demographic_parity'] = dp_diff_mitigated

            self.log_bias_mitigation(bias_metrics)
            return mitigator, bias_metrics

        return model, bias_metrics

    def generate_model_explanations(self, model, X_test, feature_names):
        """Generate SHAP explanations for model decisions"""
        explainer = shap.Explainer(model)
        shap_values = explainer(X_test)

        # Generate summary
        explanation_summary = {
            'feature_importance': dict(zip(feature_names, np.abs(shap_values.values).mean(axis=0))),
            'top_features': [feature_names[i] for i in np.argsort(np.abs(shap_values.values).mean(axis=0))[-5:]],
            'explanation_quality_score': self.calculate_explanation_quality(shap_values)
        }

        self.log_explanation_generation(explanation_summary)
        return shap_values, explanation_summary

    def implement_gdpr_compliance(self, user_data, user_id, action):
        """Implement GDPR compliance mechanisms"""
        compliance_actions = {
            'consent_given': self.record_consent,
            'consent_withdrawn': self.remove_consent,
            'data_access_request': self.provide_data_access,
            'data_deletion_request': self.delete_user_data,
            'data_portability_request': self.export_user_data
        }

        if action in compliance_actions:
            result = compliance_actions[action](user_data, user_id)
            self.log_gdpr_action(user_id, action, result)
            return result
        else:
            raise ValueError(f"Unknown GDPR action: {action}")

    def anonymize_data(self, data, quasi_identifiers):
        """Implement k-anonymity for data anonymization"""
        # Simple k-anonymity implementation
        k = 5  # Minimum group size

        # Group by quasi-identifiers
        grouped = data.groupby(quasi_identifiers)

        anonymized_data = []
        for name, group in grouped:
            if len(group) >= k:
                anonymized_data.append(group)
            else:
                # Generalize or suppress small groups
                generalized_group = self.generalize_quasi_identifiers(group, quasi_identifiers)
                anonymized_data.append(generalized_group)

        result = pd.concat(anonymized_data, ignore_index=True)
        self.log_anonymization(len(data), len(result))
        return result

    def secure_model_storage(self, model, model_id):
        """Securely store ML models with encryption"""
        import pickle

        # Serialize model
        model_bytes = pickle.dumps(model)

        # Encrypt model
        encrypted_model = self.cipher_suite.encrypt(model_bytes)

        # Create integrity hash
        model_hash = hashlib.sha256(model_bytes).hexdigest()

        # Store with metadata
        storage_metadata = {
            'model_id': model_id,
            'encrypted_model': encrypted_model,
            'hash': model_hash,
            'created_at': datetime.now().isoformat(),
            'encryption_key_id': self.encryption_key[:16].hex()  # Key identifier
        }

        self.log_secure_storage(model_id, model_hash)
        return storage_metadata

    def audit_ai_system(self, model, test_data, sensitive_features):
        """Comprehensive AI system audit"""
        audit_results = {
            'timestamp': datetime.now().isoformat(),
            'model_performance': {},
            'bias_metrics': {},
            'privacy_compliance': {},
            'security_assessment': {}
        }

        # Performance audit
        predictions = model.predict(test_data)
        audit_results['model_performance'] = {
            'accuracy': (predictions == test_data['target']).mean(),
            'prediction_distribution': np.bincount(predictions).tolist()
        }

        # Bias audit
        _, bias_metrics = self.detect_and_mitigate_bias(
            test_data.drop('target', axis=1),
            test_data['target'],
            sensitive_features,
            model
        )
        audit_results['bias_metrics'] = bias_metrics

        # Privacy audit
        audit_results['privacy_compliance'] = {
            'differential_privacy_enabled': True,
            'data_minimization_score': self.calculate_data_minimization_score(test_data),
            'consent_compliance': self.check_consent_compliance()
        }

        # Security assessment
        audit_results['security_assessment'] = {
            'model_encrypted': True,
            'access_controls_enabled': True,
            'audit_logging_active': True
        }

        self.log_system_audit(audit_results)
        return audit_results

    def log_privacy_action(self, action):
        """Log privacy-related actions for audit trail"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'action_type': 'privacy',
            'details': action
        }
        self.audit_log.append(log_entry)
        logging.info(f"Privacy action: {action}")

    def log_bias_mitigation(self, metrics):
        """Log bias mitigation actions"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'action_type': 'bias_mitigation',
            'metrics': metrics
        }
        self.audit_log.append(log_entry)
        logging.info(f"Bias mitigation performed: {metrics}")

    def log_gdpr_action(self, user_id, action, result):
        """Log GDPR compliance actions"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'user_id': hashlib.sha256(str(user_id).encode()).hexdigest()[:16],  # Anonymized user ID
            'action': action,
            'result': result
        }
        self.audit_log.append(log_entry)
        logging.info(f"GDPR action {action} for user {user_id[:8]}...")