Identity Resolution with Hypergraph: Building True Customer 360

from dataclasses import dataclass, field
from typing import Dict, List, Set, Optional
from datetime import datetime
from collections import defaultdict
import hashlib

@dataclass
class IdentityVertex:
    """An identity signal (email, phone, device, etc.)"""
    id: str
    type: str  # email, phone, device, cookie, etc.
    value_hash: str  # Hashed PII value
    first_seen: datetime
    last_seen: datetime
    source: str
    confidence: float = 1.0

@dataclass
class IdentityHyperedge:
    """A hyperedge connecting multiple identity signals"""
    id: str
    vertices: Set[str]  # Set of vertex IDs
    event_type: str  # login, purchase, registration, etc.
    timestamp: datetime
    source: str
    confidence: float = 1.0

@dataclass
class IdentityCluster:
    """A resolved identity (person/household)"""
    id: str
    vertices: Set[str]
    hyperedges: Set[str]
    golden_record: Dict[str, str]
    confidence: float
    created_at: datetime
    updated_at: datetime

class HypergraphIdentityResolver:
    """Identity resolution using hypergraph model"""

    def __init__(self):
        self.vertices: Dict[str, IdentityVertex] = {}
        self.hyperedges: Dict[str, IdentityHyperedge] = {}
        self.clusters: Dict[str, IdentityCluster] = {}

        # Adjacency: vertex -> set of hyperedges containing it
        self.vertex_to_edges: Dict[str, Set[str]] = defaultdict(set)

        # Adjacency: vertex -> set of co-occurring vertices
        self.vertex_adjacency: Dict[str, Set[str]] = defaultdict(set)

    def add_vertex(self, vertex: IdentityVertex) -> None:
        """Add an identity signal vertex"""
        self.vertices[vertex.id] = vertex

    def add_hyperedge(self, hyperedge: IdentityHyperedge) -> None:
        """Add an identity event hyperedge"""
        self.hyperedges[hyperedge.id] = hyperedge

        # Update adjacency structures
        for vid in hyperedge.vertices:
            self.vertex_to_edges[vid].add(hyperedge.id)

            # Update vertex-to-vertex adjacency
            for other_vid in hyperedge.vertices:
                if other_vid != vid:
                    self.vertex_adjacency[vid].add(other_vid)

    def find_connected_component(self, start_vertex: str) -> Set[str]:
        """Find all vertices connected to start_vertex via hyperedges"""
        visited = set()
        queue = [start_vertex]

        while queue:
            current = queue.pop(0)
            if current in visited:
                continue
            visited.add(current)

            # Add all adjacent vertices
            for adjacent in self.vertex_adjacency.get(current, set()):
                if adjacent not in visited:
                    queue.append(adjacent)

        return visited

    def compute_cluster_confidence(self, vertex_ids: Set[str]) -> float:
        """Compute confidence score for a cluster"""
        if len(vertex_ids) <= 1:
            return 1.0

        # Count direct co-occurrences
        direct_connections = 0
        total_pairs = 0

        vertex_list = list(vertex_ids)
        for i, v1 in enumerate(vertex_list):
            for v2 in vertex_list[i+1:]:
                total_pairs += 1
                if v2 in self.vertex_adjacency.get(v1, set()):
                    direct_connections += 1

        if total_pairs == 0:
            return 0.0

        # Base confidence from connectivity
        connectivity_score = direct_connections / total_pairs

        # Boost for diverse identifier types
        types = set(self.vertices[vid].type for vid in vertex_ids
                   if vid in self.vertices)
        diversity_boost = min(len(types) / 5, 1.0) * 0.2

        return min(connectivity_score + diversity_boost, 1.0)

    def resolve_identities(self) -> List[IdentityCluster]:
        """Resolve all identities into clusters"""
        resolved = []
        visited_vertices = set()

        for vertex_id in self.vertices:
            if vertex_id in visited_vertices:
                continue

            # Find connected component
            component = self.find_connected_component(vertex_id)
            visited_vertices.update(component)

            # Get hyperedges for this component
            component_edges = set()
            for vid in component:
                component_edges.update(self.vertex_to_edges.get(vid, set()))

            # Compute confidence
            confidence = self.compute_cluster_confidence(component)

            # Generate golden record
            golden = self._generate_golden_record(component)

            # Create cluster
            cluster = IdentityCluster(
                id=f"cluster_{hashlib.md5(str(sorted(component)).encode()).hexdigest()[:8]}",
                vertices=component,
                hyperedges=component_edges,
                golden_record=golden,
                confidence=confidence,
                created_at=datetime.utcnow(),
                updated_at=datetime.utcnow(),
            )

            resolved.append(cluster)
            self.clusters[cluster.id] = cluster

        return resolved

    def _generate_golden_record(self, vertex_ids: Set[str]) -> Dict[str, str]:
        """Generate golden record from cluster vertices"""
        golden = {}

        # Group vertices by type
        by_type = defaultdict(list)
        for vid in vertex_ids:
            if vid in self.vertices:
                vertex = self.vertices[vid]
                by_type[vertex.type].append(vertex)

        # Select best value for each type (most recent, highest confidence)
        for vtype, vertices in by_type.items():
            best = max(vertices, key=lambda v: (v.confidence, v.last_seen))
            golden[vtype] = best.value_hash

        return golden

    def lookup(self, identifier_type: str, identifier_hash: str) -> Optional[IdentityCluster]:
        """Look up cluster by identifier"""
        for vid, vertex in self.vertices.items():
            if vertex.type == identifier_type and vertex.value_hash == identifier_hash:
                for cid, cluster in self.clusters.items():
                    if vid in cluster.vertices:
                        return cluster
        return None


# Example usage
resolver = HypergraphIdentityResolver()

# Add identity signals
now = datetime.utcnow()
vertices = [
    IdentityVertex("v1", "email", "hash_email1", now, now, "web", 1.0),
    IdentityVertex("v2", "phone", "hash_phone1", now, now, "crm", 0.9),
    IdentityVertex("v3", "device", "hash_device1", now, now, "mobile", 0.8),
    IdentityVertex("v4", "cookie", "hash_cookie1", now, now, "web", 0.7),
    IdentityVertex("v5", "email", "hash_email2", now, now, "support", 1.0),
]

for v in vertices:
    resolver.add_vertex(v)

# Add identity events (hyperedges)
hyperedges = [
    IdentityHyperedge("e1", {"v1", "v3", "v4"}, "login", now, "web", 1.0),
    IdentityHyperedge("e2", {"v1", "v2"}, "registration", now, "crm", 0.95),
    IdentityHyperedge("e3", {"v2", "v5"}, "support_call", now, "support", 0.9),
]

for e in hyperedges:
    resolver.add_hyperedge(e)

# Resolve identities
clusters = resolver.resolve_identities()

print(f"Resolved {len(clusters)} identity clusters:")
for cluster in clusters:
    print(f"\n  Cluster {cluster.id}:")
    print(f"    Vertices: {len(cluster.vertices)}")
    print(f"    Confidence: {cluster.confidence:.2f}")
    print(f"    Golden Record: {cluster.golden_record}")