Customer Data Platform Architecture with Hypergraph: Unified Customer Intelligence

from dataclasses import dataclass, field
from typing import List, Dict, Set, Optional, Any
from datetime import datetime
from enum import Enum
import uuid
import numpy as np
from scipy import sparse

class VertexType(Enum):
    """Types of vertices in the CDP hypergraph."""
    CUSTOMER = "customer"
    PRODUCT = "product"
    CHANNEL = "channel"
    DEVICE = "device"
    LOCATION = "location"
    CAMPAIGN = "campaign"
    SESSION = "session"
    CONTENT = "content"
    IDENTIFIER = "identifier"
    HOUSEHOLD = "household"
    SEGMENT = "segment"
    TIME_BUCKET = "time_bucket"

class HyperedgeType(Enum):
    """Types of hyperedges representing different interaction contexts."""
    PRODUCT_VIEW = "product_view"
    PURCHASE = "purchase"
    ADD_TO_CART = "add_to_cart"
    SEARCH = "search"
    CAMPAIGN_EXPOSURE = "campaign_exposure"
    CAMPAIGN_RESPONSE = "campaign_response"
    IDENTITY_LINK = "identity_link"
    HOUSEHOLD_MEMBERSHIP = "household_membership"
    SUPPORT_INTERACTION = "support_interaction"
    CONTENT_CONSUMPTION = "content_consumption"
    SOCIAL_INTERACTION = "social_interaction"

@dataclass
class Vertex:
    """
    Represents a vertex in the CDP hypergraph.
    Vertices are the fundamental entities: customers, products, channels, etc.
    """
    id: str
    vertex_type: VertexType
    properties: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)

    def __hash__(self):
        return hash(self.id)

    def __eq__(self, other):
        return isinstance(other, Vertex) and self.id == other.id

@dataclass
class Hyperedge:
    """
    Represents a hyperedge connecting multiple vertices.
    A hyperedge captures a complete interaction context.
    """
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    edge_type: HyperedgeType = HyperedgeType.PRODUCT_VIEW
    vertices: Set[str] = field(default_factory=set)  # Set of vertex IDs
    properties: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.utcnow)
    confidence: float = 1.0
    source: str = "unknown"

    # Temporal properties for sequence analysis
    session_id: Optional[str] = None
    sequence_number: Optional[int] = None
    duration_ms: Optional[int] = None

    def cardinality(self) -> int:
        """Return the number of vertices in this hyperedge."""
        return len(self.vertices)

    def contains_vertex_type(self, vertex_type: VertexType,
                              vertex_registry: Dict[str, Vertex]) -> bool:
        """Check if hyperedge contains a vertex of given type."""
        for vid in self.vertices:
            if vid in vertex_registry:
                if vertex_registry[vid].vertex_type == vertex_type:
                    return True
        return False

class CDPHypergraph:
    """
    Core hypergraph data structure for Customer Data Platform.
    Implements efficient storage and querying using incidence matrix representation.
    """

    def __init__(self):
        self.vertices: Dict[str, Vertex] = {}
        self.hyperedges: Dict[str, Hyperedge] = {}

        # Incidence matrix: rows = vertices, columns = hyperedges
        # H[i,j] = 1 if vertex i is in hyperedge j
        self._vertex_index: Dict[str, int] = {}
        self._edge_index: Dict[str, int] = {}
        self._incidence_matrix: Optional[sparse.csr_matrix] = None
        self._matrix_dirty: bool = True

    def add_vertex(self, vertex: Vertex) -> None:
        """Add a vertex to the hypergraph."""
        self.vertices[vertex.id] = vertex
        if vertex.id not in self._vertex_index:
            self._vertex_index[vertex.id] = len(self._vertex_index)
        self._matrix_dirty = True

    def add_hyperedge(self, hyperedge: Hyperedge) -> None:
        """Add a hyperedge to the hypergraph."""
        # Validate all vertices exist
        for vid in hyperedge.vertices:
            if vid not in self.vertices:
                raise ValueError(f"Vertex {vid} not found in hypergraph")

        self.hyperedges[hyperedge.id] = hyperedge
        if hyperedge.id not in self._edge_index:
            self._edge_index[hyperedge.id] = len(self._edge_index)
        self._matrix_dirty = True

    def get_vertex_adjacency(self) -> sparse.csr_matrix:
        """
        Compute vertex adjacency matrix: H * H^T
        Result[i,j] = number of hyperedges containing both vertex i and j
        """
        H = self.incidence_matrix
        return H @ H.T

    def get_hyperedge_adjacency(self) -> sparse.csr_matrix:
        """
        Compute hyperedge adjacency matrix: H^T * H
        Result[i,j] = number of vertices shared between hyperedge i and j
        """
        H = self.incidence_matrix
        return H.T @ H

    def get_customer_journey(self, customer_id: str) -> List[Hyperedge]:
        """
        Get all hyperedges for a customer, ordered by timestamp.
        This represents the customer's complete interaction journey.
        """
        journey = []
        for edge in self.hyperedges.values():
            if customer_id in edge.vertices:
                journey.append(edge)
        return sorted(journey, key=lambda e: e.timestamp)

    def compute_hyperedge_similarity(self, edge1_id: str,
                                       edge2_id: str) -> float:
        """
        Compute Jaccard similarity between two hyperedges.
        Useful for finding similar interaction patterns.
        """
        if edge1_id not in self.hyperedges or edge2_id not in self.hyperedges:
            return 0.0

        e1_vertices = self.hyperedges[edge1_id].vertices
        e2_vertices = self.hyperedges[edge2_id].vertices

        intersection = len(e1_vertices & e2_vertices)
        union = len(e1_vertices | e2_vertices)

        return intersection / union if union > 0 else 0.0

-- HQL: Hypergraph Query Language for CDP

-- Find all customers who viewed Product A via Mobile
-- AFTER receiving Campaign X via Email
SELECT DISTINCT c.id as customer_id
FROM HYPEREDGE h1
JOIN HYPEREDGE h2 ON h1.customer = h2.customer
WHERE h1.type = 'CAMPAIGN_EXPOSURE'
  AND h1.channel = 'email'
  AND h1.campaign = 'campaign_X'
  AND h2.type = 'PRODUCT_VIEW'
  AND h2.product = 'product_A'
  AND h2.channel = 'mobile'
  AND h2.timestamp > h1.timestamp;

-- Find household members who influence purchase decisions
SELECT DISTINCT h_member.id as influencer_id,
       h_member.properties->>'relationship' as relationship
FROM HYPEREDGE consideration
JOIN HYPEREDGE purchase ON consideration.customer = purchase.customer
                       AND consideration.product = purchase.product
JOIN VERTEX h_member ON h_member.id IN consideration.vertices
                    AND h_member.type = 'HOUSEHOLD_MEMBER'
                    AND h_member.id NOT IN purchase.vertices
WHERE consideration.type = 'PRODUCT_VIEW'
  AND purchase.type = 'PURCHASE';

-- Multi-touch attribution query using hyperedge paths
WITH journey AS (
    SELECT customer,
           product,
           ARRAY_AGG(hyperedge_id ORDER BY timestamp) as touchpoint_path,
           ARRAY_AGG(channel ORDER BY timestamp) as channel_path
    FROM HYPEREDGE
    WHERE type IN ('CAMPAIGN_EXPOSURE', 'PRODUCT_VIEW', 'SEARCH', 'PURCHASE')
    GROUP BY customer, product
    HAVING ARRAY_CONTAINS(ARRAY_AGG(type), 'PURCHASE')
)
SELECT channel,
       COUNT(*) as touchpoints,
       SUM(CASE WHEN position = 1 THEN 1 ELSE 0 END) as first_touch,
       SUM(CASE WHEN position = array_length THEN 1 ELSE 0 END) as last_touch
FROM journey, UNNEST(channel_path) WITH ORDINALITY as t(channel, position)
GROUP BY channel;