Real-Time Personalization Engine with Hypergraph: Context-Aware Customer Experiences

from dataclasses import dataclass
from typing import Set, List, Dict, Optional
from datetime import datetime
import numpy as np

@dataclass(frozen=True)
class HypergraphVertex:
    """A vertex in the personalization hypergraph."""
    vertex_type: str  # 'customer', 'product', 'context', 'intent', 'time'
    vertex_id: str
    embedding: Optional[np.ndarray] = None

@dataclass
class PersonalizationHyperedge:
    """
    A hyperedge representing a multi-dimensional customer interaction.
    Connects customer, product, context, intent, and temporal vertices.
    """
    hyperedge_id: str
    vertices: Set[HypergraphVertex]
    weight: float  # Interaction strength
    timestamp: datetime
    metadata: Dict[str, any]

    def get_vertices_by_type(self, vertex_type: str) -> List[HypergraphVertex]:
        """Get all vertices of a specific type in this hyperedge."""
        return [v for v in self.vertices if v.vertex_type == vertex_type]

    def to_incidence_vector(self, vertex_index: Dict[str, int]) -> np.ndarray:
        """Convert hyperedge to incidence matrix column."""
        vector = np.zeros(len(vertex_index))
        for vertex in self.vertices:
            key = f"{vertex.vertex_type}:{vertex.vertex_id}"
            if key in vertex_index:
                vector[vertex_index[key]] = 1
        return vector

class PersonalizationHypergraph:
    """
    Hypergraph data structure for personalization.

    Supports efficient operations for:
    - Adding/removing hyperedges
    - Querying vertices and their incident hyperedges
    - Computing hypergraph neural network features
    """

    def __init__(self):
        self.vertices: Dict[str, HypergraphVertex] = {}
        self.hyperedges: Dict[str, PersonalizationHyperedge] = {}
        self.vertex_to_hyperedges: Dict[str, Set[str]] = {}
        self.incidence_matrix: Optional[np.ndarray] = None

    def add_interaction(
        self,
        customer_id: str,
        product_id: str,
        context: Dict[str, str],
        intent: str,
        time_bucket: str,
        weight: float = 1.0,
        metadata: Dict = None
    ) -> str:
        """
        Add a customer interaction as a hyperedge.
        """
        vertices = set()

        # Customer vertex
        customer_vertex = HypergraphVertex('customer', customer_id)
        vertices.add(customer_vertex)
        self._ensure_vertex(customer_vertex)

        # Product vertex
        product_vertex = HypergraphVertex('product', product_id)
        vertices.add(product_vertex)
        self._ensure_vertex(product_vertex)

        # Context vertices
        for context_type, context_value in context.items():
            context_vertex = HypergraphVertex('context', f"{context_type}:{context_value}")
            vertices.add(context_vertex)
            self._ensure_vertex(context_vertex)

        # Intent vertex
        intent_vertex = HypergraphVertex('intent', intent)
        vertices.add(intent_vertex)
        self._ensure_vertex(intent_vertex)

        # Time vertex
        time_vertex = HypergraphVertex('time', time_bucket)
        vertices.add(time_vertex)
        self._ensure_vertex(time_vertex)

        # Create hyperedge
        hyperedge_id = f"he_{len(self.hyperedges)}"
        hyperedge = PersonalizationHyperedge(
            hyperedge_id=hyperedge_id,
            vertices=vertices,
            weight=weight,
            timestamp=datetime.now(),
            metadata=metadata or {}
        )

        self.hyperedges[hyperedge_id] = hyperedge
        return hyperedge_id

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Tuple

class HypergraphConvolution(nn.Module):
    """
    Hypergraph Convolution Layer.

    Performs message passing on hypergraph structure:
    1. Vertex -> Hyperedge aggregation
    2. Hyperedge -> Vertex aggregation
    """

    def __init__(self, in_features: int, out_features: int, bias: bool = True):
        super().__init__()
        self.in_features = in_features
        self.out_features = out_features

        self.weight = nn.Parameter(torch.FloatTensor(in_features, out_features))
        if bias:
            self.bias = nn.Parameter(torch.FloatTensor(out_features))
        else:
            self.register_parameter('bias', None)

        self.reset_parameters()

    def reset_parameters(self):
        nn.init.xavier_uniform_(self.weight)
        if self.bias is not None:
            nn.init.zeros_(self.bias)

    def forward(
        self,
        X: torch.Tensor,
        H: torch.Tensor,
        D_v: torch.Tensor,
        D_e: torch.Tensor,
        W: torch.Tensor = None
    ) -> torch.Tensor:
        """
        Forward pass of hypergraph convolution.

        Args:
            X: Node features [n_vertices, in_features]
            H: Incidence matrix [n_vertices, n_hyperedges]
            D_v: Vertex degree matrix (diagonal)
            D_e: Hyperedge degree matrix (diagonal)
            W: Hyperedge weights (optional)

        Returns:
            Updated node features [n_vertices, out_features]
        """
        if W is not None:
            H = H * W.unsqueeze(0)

        D_v_inv_sqrt = torch.pow(D_v + 1e-8, -0.5)
        D_v_inv_sqrt[torch.isinf(D_v_inv_sqrt)] = 0
        H_normalized = D_v_inv_sqrt.unsqueeze(1) * H

        D_e_inv = torch.pow(D_e + 1e-8, -1)
        D_e_inv[torch.isinf(D_e_inv)] = 0

        # Step 1: Transform input features
        X_transformed = torch.matmul(X, self.weight)

        # Step 2: Vertex -> Hyperedge aggregation
        hyperedge_features = torch.matmul(H.T, X_transformed)
        hyperedge_features = D_e_inv.unsqueeze(1) * hyperedge_features

        # Step 3: Hyperedge -> Vertex propagation
        output = torch.matmul(H_normalized, hyperedge_features)
        output = D_v_inv_sqrt.unsqueeze(1) * output

        if self.bias is not None:
            output = output + self.bias

        return output


class HypergraphRecommender(nn.Module):
    """
    Hypergraph Neural Network for Personalization.
    """

    def __init__(
        self,
        n_customers: int,
        n_products: int,
        n_contexts: int,
        n_intents: int,
        n_time_buckets: int,
        embedding_dim: int = 128,
        hidden_dim: int = 256,
        n_layers: int = 3,
        dropout: float = 0.2
    ):
        super().__init__()

        # Entity embeddings
        self.customer_embedding = nn.Embedding(n_customers, embedding_dim)
        self.product_embedding = nn.Embedding(n_products, embedding_dim)
        self.context_embedding = nn.Embedding(n_contexts, embedding_dim)
        self.intent_embedding = nn.Embedding(n_intents, embedding_dim)
        self.time_embedding = nn.Embedding(n_time_buckets, embedding_dim)

        # Hypergraph convolution layers
        self.conv_layers = nn.ModuleList([
            HypergraphConvolution(
                embedding_dim if i == 0 else hidden_dim,
                hidden_dim
            )
            for i in range(n_layers)
        ])

        # Context attention
        self.context_attention = nn.MultiheadAttention(
            embed_dim=hidden_dim,
            num_heads=8,
            dropout=dropout
        )

        # Prediction head
        self.predictor = nn.Sequential(
            nn.Linear(hidden_dim * 2, hidden_dim),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, 1)
        )

import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Tuple
import lightgbm as lgb

@dataclass
class RankingFeatures:
    """Features for context-aware ranking."""
    customer_product_affinity: float  # From hypergraph similarity
    context_match_score: float  # How well product matches context
    intent_alignment: float  # Product-intent alignment
    temporal_relevance: float  # Time-based relevance
    collaborative_score: float  # Hypergraph collaborative filtering
    popularity_score: float  # Global popularity
    recency_score: float  # Recent interaction recency
    price_sensitivity_match: float  # Price alignment with customer
    category_affinity: float  # Category preference strength

class ContextAwareRanker:
    """
    Two-stage ranking model:
    1. Candidate generation from hypergraph
    2. Context-aware ranking using gradient boosting
    """

    def __init__(self, hypergraph: PersonalizationHypergraph):
        self.hypergraph = hypergraph
        self.model = None

    def extract_features(
        self,
        customer_id: str,
        product_id: str,
        context: Dict[str, str],
        intent: str,
        time_bucket: str
    ) -> RankingFeatures:
        """Extract ranking features for a customer-product-context combination."""
        # Customer-product affinity from hypergraph
        customer_hyperedges = self.hypergraph.vertex_to_hyperedges.get(
            f"customer:{customer_id}", set()
        )
        product_hyperedges = self.hypergraph.vertex_to_hyperedges.get(
            f"product:{product_id}", set()
        )

        # Jaccard similarity of hyperedge sets
        if customer_hyperedges and product_hyperedges:
            intersection = len(customer_hyperedges & product_hyperedges)
            union = len(customer_hyperedges | product_hyperedges)
            affinity = intersection / union if union > 0 else 0
        else:
            affinity = 0

        # Context match score
        context_vertices = [f"context:{k}:{v}" for k, v in context.items()]
        context_hyperedges = set()
        for cv in context_vertices:
            context_hyperedges.update(
                self.hypergraph.vertex_to_hyperedges.get(cv, set())
            )

        context_product_overlap = len(context_hyperedges & product_hyperedges)
        context_match = context_product_overlap / len(context_hyperedges) if context_hyperedges else 0

        return RankingFeatures(
            customer_product_affinity=affinity,
            context_match_score=context_match,
            intent_alignment=0.5,  # Computed similarly
            temporal_relevance=0.5,
            collaborative_score=0.5,
            popularity_score=len(product_hyperedges) / len(self.hypergraph.hyperedges),
            recency_score=0.5,
            price_sensitivity_match=0.5,
            category_affinity=0.5
        )

    def rank(
        self,
        customer_id: str,
        candidate_products: List[str],
        context: Dict[str, str],
        intent: str,
        time_bucket: str
    ) -> List[Tuple[str, float]]:
        """Rank candidate products for a customer in context."""
        features = []
        for product_id in candidate_products:
            feat = self.extract_features(
                customer_id, product_id, context, intent, time_bucket
            )
            features.append([
                feat.customer_product_affinity,
                feat.context_match_score,
                feat.intent_alignment,
                feat.temporal_relevance,
                feat.collaborative_score,
                feat.popularity_score,
                feat.recency_score,
                feat.price_sensitivity_match,
                feat.category_affinity
            ])

        scores = self.model.predict(np.array(features))
        ranked = sorted(zip(candidate_products, scores), key=lambda x: x[1], reverse=True)
        return ranked

import asyncio
from functools import lru_cache
import hashlib
from typing import Optional

class OptimizedPersonalizationEngine:
    """Production-optimized personalization engine. Target: p99 latency < 50ms"""

    def __init__(self, hypergraph, ranker, redis_client, feature_store):
        self.hypergraph = hypergraph
        self.ranker = ranker
        self.redis = redis_client
        self.feature_store = feature_store

    async def get_recommendations(
        self,
        customer_id: str,
        context: Dict[str, str],
        intent: str,
        limit: int = 20
    ) -> Dict:
        """Get personalized recommendations with latency optimization."""
        start_time = asyncio.get_event_loop().time()

        # 1. Check response cache (2ms budget)
        cache_key = self._get_cache_key(customer_id, context, intent)
        cached = await self._check_cache(cache_key)
        if cached:
            return {
                'recommendations': cached,
                'source': 'cache',
                'latency_ms': (asyncio.get_event_loop().time() - start_time) * 1000
            }

        # 2. Get time bucket
        time_bucket = self._get_time_bucket()

        # 3. Parallel execution of independent operations (15ms budget)
        candidate_task = asyncio.create_task(
            self._get_candidates_async(customer_id, context, intent, time_bucket)
        )
        features_task = asyncio.create_task(
            self._get_customer_features_async(customer_id)
        )

        candidates, customer_features = await asyncio.gather(
            candidate_task, features_task
        )

        # 4. Fast ranking (15ms budget)
        ranked = await self._rank_candidates_async(
            customer_id, candidates, context, intent, time_bucket
        )

        # 5. Apply business rules (5ms)
        final = self._apply_business_rules(ranked, customer_features)[:limit]

        # 6. Cache response (async, no wait)
        asyncio.create_task(self._cache_response(cache_key, final))

        total_latency = (asyncio.get_event_loop().time() - start_time) * 1000

        return {
            'recommendations': final,
            'source': 'computed',
            'latency_ms': total_latency
        }

    def _get_cache_key(self, customer_id: str, context: Dict[str, str], intent: str) -> str:
        """Generate cache key from request parameters."""
        context_str = '|'.join(f"{k}:{v}" for k, v in sorted(context.items()))
        raw_key = f"{customer_id}|{context_str}|{intent}"
        return hashlib.md5(raw_key.encode()).hexdigest()

from dataclasses import dataclass
from typing import Dict, List, Optional
from enum import Enum
import hashlib
from scipy import stats
import numpy as np

class ExperimentStatus(Enum):
    DRAFT = "draft"
    RUNNING = "running"
    PAUSED = "paused"
    COMPLETED = "completed"

@dataclass
class ExperimentVariant:
    """A variant in an A/B test."""
    name: str
    description: str
    traffic_percentage: float
    config: Dict[str, any]

@dataclass
class Experiment:
    """A/B test experiment configuration."""
    experiment_id: str
    name: str
    status: ExperimentStatus
    variants: List[ExperimentVariant]
    primary_metric: str
    minimum_sample_size: int

class PersonalizationExperimentFramework:
    """A/B testing framework for personalization experiments."""

    def __init__(self, redis_client, metrics_store):
        self.redis = redis_client
        self.metrics_store = metrics_store
        self.experiments: Dict[str, Experiment] = {}

    def get_variant(self, experiment_id: str, user_id: str) -> Optional[ExperimentVariant]:
        """Get variant assignment for a user using deterministic hashing."""
        experiment = self.experiments.get(experiment_id)
        if not experiment or experiment.status != ExperimentStatus.RUNNING:
            return None

        # Deterministic hash
        hash_input = f"{experiment_id}:{user_id}"
        hash_value = int(hashlib.sha256(hash_input.encode()).hexdigest(), 16)
        bucket = (hash_value % 10000) / 10000.0

        # Assign to variant based on traffic allocation
        cumulative = 0.0
        for variant in experiment.variants:
            cumulative += variant.traffic_percentage
            if bucket < cumulative:
                return variant

        return experiment.variants[-1]

    async def analyze_experiment(self, experiment_id: str) -> Dict:
        """Analyze experiment results with statistical significance."""
        experiment = self.experiments.get(experiment_id)
        if not experiment:
            return {"error": "Experiment not found"}

        results = {}
        variant_data = {}

        # Gather data for each variant
        for variant in experiment.variants:
            exposures = await self.redis.hget(
                f"exp:{experiment_id}:exposures:{variant.name}", "count"
            ) or 0
            conversions = await self.redis.hget(
                f"exp:{experiment_id}:conversions:{variant.name}:{experiment.primary_metric}", "count"
            ) or 0

            variant_data[variant.name] = {
                'exposures': int(exposures),
                'conversions': int(conversions),
                'rate': int(conversions) / int(exposures) if int(exposures) > 0 else 0
            }

        results['variants'] = variant_data
        return results

class ResilientPersonalizationService:
    """Personalization service with graceful degradation."""

    def __init__(self, primary_engine, fallback_strategies):
        self.primary = primary_engine
        self.fallbacks = fallback_strategies
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=30
        )

    async def get_recommendations(
        self,
        customer_id: str,
        context: Dict[str, str],
        intent: str,
        limit: int = 20
    ) -> Dict:
        """Get recommendations with fallback chain."""
        # Try primary personalization
        if self.circuit_breaker.is_closed():
            try:
                result = await asyncio.wait_for(
                    self.primary.get_recommendations(customer_id, context, intent, limit),
                    timeout=0.045  # 45ms timeout
                )
                self.circuit_breaker.record_success()
                return result
            except asyncio.TimeoutError:
                self.circuit_breaker.record_failure()
            except Exception as e:
                self.circuit_breaker.record_failure()

        # Fallback chain
        for fallback in self.fallbacks:
            try:
                result = await fallback.get_recommendations(customer_id, context, limit)
                return {'recommendations': result, 'source': fallback.name, 'degraded': True}
            except Exception:
                continue

        # Final fallback: popular items
        return {
            'recommendations': await self._get_popular_items(limit),
            'source': 'popular_items',
            'degraded': True
        }