Spaces:

yusufcalisir
/

Collaborative-Fraud-Intelligence-Simulator

Running

Collaborative-Fraud-Intelligence-Simulator / backend /app /application /services /graph_engine.py
deploy: Hugging Face space upload
eff6797 about 22 hours ago
63.1 kB
"""Graph engine.
Builds and queries the entity relationship graph. The graph represents
connections between entities (customers, merchants, devices, etc.)
across institutions. Graph operations include neighbor expansion,
connected component detection, and serialization to React Flow format
for interactive visualization.
"""

from __future__ import annotations

import hashlib
import logging
import re
import threading
from collections import defaultdict, deque
from datetime import UTC, datetime, timedelta
from typing import Any

from app.config import get_settings
from app.domain.enums import EntityType, RelationshipType, RiskLevel
from app.domain.investigation_entities import Entity, Relationship
from app.domain.value_objects_investigation import GraphSubgraph
from app.infrastructure.redis_store import RedisStore

logger = logging.getLogger(__name__)


def _entity_to_dict(e: Entity) -> dict[str, Any]:
    return {
        "id": e.id,
        "entity_type": e.entity_type.value,
        "privacy_id": e.privacy_id,
        "bank_id": e.bank_id,
        "display_label": e.display_label,
        "attributes": e.attributes,
        "risk_level": e.risk_level.value,
        "alert_count": e.alert_count,
        "first_seen": e.first_seen.isoformat(),
        "last_seen": e.last_seen.isoformat(),
    }


def _dict_to_entity(d: dict[str, Any]) -> Entity:
    from datetime import datetime

    d_copy = d.copy()
    d_copy["entity_type"] = EntityType(d_copy["entity_type"])
    d_copy["risk_level"] = RiskLevel(d_copy["risk_level"])
    d_copy["first_seen"] = datetime.fromisoformat(d_copy["first_seen"])
    d_copy["last_seen"] = datetime.fromisoformat(d_copy["last_seen"])
    return Entity(**d_copy)


def _relationship_to_dict(r: Relationship) -> dict[str, Any]:
    return {
        "id": r.id,
        "source_entity_id": r.source_entity_id,
        "target_entity_id": r.target_entity_id,
        "relationship_type": r.relationship_type.value,
        "confidence": r.confidence,
        "evidence": r.evidence,
        "created_at": r.created_at.isoformat(),
    }


def _dict_to_relationship(d: dict[str, Any]) -> Relationship:
    from datetime import datetime

    d_copy = d.copy()
    d_copy["relationship_type"] = RelationshipType(d_copy["relationship_type"])
    d_copy["created_at"] = datetime.fromisoformat(d_copy["created_at"])
    return Relationship(**d_copy)


# Color mapping for React Flow node types
_NODE_COLORS: dict[str, str] = {
    EntityType.CUSTOMER: "#6366f1",
    EntityType.MERCHANT: "#f59e0b",
    EntityType.DEVICE: "#14b8a6",
    EntityType.CARD: "#ec4899",
    EntityType.EMAIL: "#8b5cf6",
    EntityType.PHONE: "#06b6d4",
    EntityType.IP_ADDRESS: "#f43f5e",
}

_EDGE_STYLES: dict[str, dict] = {
    RelationshipType.OWNS: {"stroke": "#6366f1", "strokeWidth": 2},
    RelationshipType.USES: {"stroke": "#14b8a6", "strokeWidth": 1.5},
    RelationshipType.TRANSACTS_WITH: {"stroke": "#f59e0b", "strokeWidth": 2},
    RelationshipType.SHARES_DEVICE: {"stroke": "#ec4899", "strokeDasharray": "5,5"},
    RelationshipType.SHARES_IP: {"stroke": "#f43f5e", "strokeDasharray": "5,5"},
    RelationshipType.LINKED_ALERT: {"stroke": "#ef4444", "strokeWidth": 2.5},
    RelationshipType.SAME_ENTITY: {"stroke": "#a855f7", "strokeWidth": 3},
}


RISK_LEVEL_TO_SCORE: dict[str, float] = {
    RiskLevel.CRITICAL.value: 1.0,
    RiskLevel.HIGH.value: 0.8,
    RiskLevel.MEDIUM.value: 0.5,
    RiskLevel.LOW.value: 0.2,
    RiskLevel.MINIMAL.value: 0.1,
}

_MUTATING_CYPHER_RE = re.compile(
    r"\b(CREATE|MERGE|DELETE|SET|REMOVE|DROP|DETACH)\b",
    re.IGNORECASE,
)


def _neo4j_node_to_entity(node) -> Entity:
    import json
    from datetime import datetime

    props = dict(node)
    if isinstance(props.get("first_seen"), str):
        props["first_seen"] = datetime.fromisoformat(props["first_seen"])
    if isinstance(props.get("last_seen"), str):
        props["last_seen"] = datetime.fromisoformat(props["last_seen"])
    props["entity_type"] = EntityType(props["entity_type"])
    props["risk_level"] = RiskLevel(props["risk_level"])
    if isinstance(props.get("attributes"), str):
        props["attributes"] = json.loads(props["attributes"])
    return Entity(**props)


def _neo4j_rel_to_relationship(rel) -> Relationship:
    from datetime import datetime

    props = dict(rel)
    if isinstance(props.get("created_at"), str):
        props["created_at"] = datetime.fromisoformat(props["created_at"])
    props["relationship_type"] = RelationshipType(props["relationship_type"])
    return Relationship(**props)


class GraphEngine:
    """Builds and queries the entity relationship graph.
    The graph is backed by a graph database (Neo4j, Memgraph, etc.)
    if enabled, or falls back to Redis / in-memory adjacency structures.
    """

    def __init__(self) -> None:
        self._entities = RedisStore("entity")
        self._relationships = RedisStore("relationship")
        self._adjacency: defaultdict[str, set[tuple[str, str]]] = defaultdict(set)
        self._lock = threading.RLock()

        settings = get_settings()

        self.db_type = settings.graph_db_type
        self.driver = None
        if self.db_type in ("neo4j", "memgraph"):
            try:
                import importlib

                neo4j_mod = importlib.import_module("neo4j")
                self.driver = neo4j_mod.GraphDatabase.driver(
                    settings.neo4j_uri, auth=(settings.neo4j_user, settings.neo4j_password)
                )
                logger.info("Connected to graph database of type: %s", self.db_type)

            except Exception as e:
                logger.error(
                    "Failed to connect to graph database of type %s: %s. Falling back to redis/in-memory.",
                    self.db_type,
                    e,
                )
                self.db_type = "redis"

    def _build_adjacency_list(self) -> None:
        self._adjacency = defaultdict(set)
        raw_relationships = self._relationships.list_values()
        for r_dict in raw_relationships:
            r = _dict_to_relationship(r_dict)
            self._adjacency[r.source_entity_id].add((r.target_entity_id, r.id))
            self._adjacency[r.target_entity_id].add((r.source_entity_id, r.id))

    def register_entity(self, entity: Entity) -> None:
        """Register an entity in the graph."""
        with self._lock:
            self._entities.set(entity.id, _entity_to_dict(entity))
            if self.db_type in ("neo4j", "memgraph") and self.driver:
                import json

                with self.driver.session() as session:
                    session.run(
                        "MERGE (e:Entity {id: $id}) "
                        "SET e.entity_type = $entity_type, "
                        "    e.privacy_id = $privacy_id, "
                        "    e.bank_id = $bank_id, "
                        "    e.display_label = $display_label, "
                        "    e.risk_level = $risk_level, "
                        "    e.alert_count = $alert_count, "
                        "    e.first_seen = $first_seen, "
                        "    e.last_seen = $last_seen, "
                        "    e.attributes = $attributes",
                        id=entity.id,
                        entity_type=entity.entity_type.value,
                        privacy_id=entity.privacy_id,
                        bank_id=entity.bank_id,
                        display_label=entity.display_label,
                        risk_level=entity.risk_level.value,
                        alert_count=entity.alert_count,
                        first_seen=entity.first_seen.isoformat(),
                        last_seen=entity.last_seen.isoformat(),
                        attributes=json.dumps(entity.attributes),
                    )

    def register_entities(self, entities: list[Entity]) -> None:
        with self._lock:
            for entity in entities:
                self.register_entity(entity)

    def add_relationship(self, relationship: Relationship) -> None:
        """Add a relationship (edge) to the graph."""
        with self._lock:
            self._relationships.set(relationship.id, _relationship_to_dict(relationship))
            # Incrementally update in-memory adjacency
            self._adjacency[relationship.source_entity_id].add(
                (relationship.target_entity_id, relationship.id)
            )
            self._adjacency[relationship.target_entity_id].add(
                (relationship.source_entity_id, relationship.id)
            )
            if self.db_type in ("neo4j", "memgraph") and self.driver:
                rel_type = relationship.relationship_type.value
                query = (
                    f"MATCH (s:Entity {{id: $source_id}}) "
                    f"MATCH (t:Entity {{id: $target_id}}) "
                    f"MERGE (s)-[r:{rel_type} {{id: $id}}]->(t) "
                    "SET r.relationship_type = $relationship_type, "
                    "    r.confidence = $confidence, "
                    "    r.evidence = $evidence, "
                    "    r.created_at = $created_at, "
                    "    r.source_entity_id = $source_id, "
                    "    r.target_entity_id = $target_id"
                )
                with self.driver.session() as session:
                    session.run(
                        query,
                        id=relationship.id,
                        source_id=relationship.source_entity_id,
                        target_id=relationship.target_entity_id,
                        relationship_type=relationship.relationship_type.value,
                        confidence=relationship.confidence,
                        evidence=relationship.evidence,
                        created_at=relationship.created_at.isoformat(),
                    )

    def find_neighbors(
        self,
        entity_id: str,
        depth: int = 1,
        relationship_types: set[RelationshipType] | None = None,
        as_of: datetime | None = None,
    ) -> list[Entity]:
        """BFS traversal to find neighbors up to a given depth.
        Args:
            entity_id: Root entity ID to find neighbors for.
            depth: Maximum traversal depth hops.
            relationship_types: Optional set of allowed RelationshipTypes.
            as_of: Optional temporal cutoff. Relationships created after as_of are excluded.
        """
        with self._lock:
            if self.db_type in ("neo4j", "memgraph") and self.driver:
                rel_filter = ""
                if relationship_types:
                    rel_types_str = "|".join(rt.value for rt in relationship_types)
                    rel_filter = f":{rel_types_str}"

                # Exclude self-node (n.id <> $entity_id) to prevent cyclic return of root entity
                query = (
                    f"MATCH (s:Entity {{id: $entity_id}})-[r{rel_filter}*1..{depth}]-(n:Entity) "
                    "WHERE n.id <> $entity_id "
                )
                params: dict[str, Any] = {"entity_id": entity_id}
                if as_of is not None:
                    query += "  AND all(rel IN relationships(r) WHERE rel.created_at <= $as_of) "
                    params["as_of"] = as_of.isoformat()
                query += "RETURN DISTINCT n"

                neighbors = []
                with self.driver.session() as session:
                    result = session.run(query, **params)
                    for record in result:
                        neighbors.append(_neo4j_node_to_entity(record["n"]))
                return neighbors

            self._build_adjacency_list()
            visited: set[str] = {entity_id}
            queue: deque[tuple[str, int]] = deque([(entity_id, 0)])
            neighbors = []

            while queue:
                current_id, current_depth = queue.popleft()
                if current_depth >= depth:
                    continue

                for neighbor_id, rel_id in self._adjacency.get(current_id, set()):
                    if neighbor_id in visited:
                        continue

                    rel_val = self._relationships.get(rel_id)
                    rel = _dict_to_relationship(rel_val) if rel_val else None
                    if not rel:
                        continue

                    # Filter by temporal boundary if specified
                    if as_of is not None:
                        rel_time = rel.created_at
                        if rel_time.tzinfo is None:
                            rel_time = rel_time.replace(tzinfo=UTC)
                        cutoff = as_of if as_of.tzinfo is not None else as_of.replace(tzinfo=UTC)
                        if rel_time > cutoff:
                            continue

                    # Filter by relationship type if specified
                    if relationship_types and rel.relationship_type not in relationship_types:
                        continue

                    visited.add(neighbor_id)
                    entity_val = self._entities.get(neighbor_id)
                    if entity_val:
                        entity = _dict_to_entity(entity_val)
                        neighbors.append(entity)
                        queue.append((neighbor_id, current_depth + 1))

            return neighbors

    def detect_clusters(self, min_size: int = 3, as_of: datetime | None = None) -> list[list[str]]:
        """Find connected components (clusters) of at least min_size."""
        with self._lock:
            if self.db_type in ("neo4j", "memgraph") and self.driver:
                adjacency = defaultdict(set)
                all_nodes = set()
                query = (
                    "MATCH (n:Entity) "
                    "OPTIONAL MATCH (n)-[r]->(m:Entity) "
                )
                params: dict[str, Any] = {}
                if as_of is not None:
                    query += "WHERE r IS NULL OR r.created_at <= $as_of "
                    params["as_of"] = as_of.isoformat()
                query += "RETURN n.id as source, m.id as target"
                with self.driver.session() as session:
                    result = session.run(query, **params)
                    for record in result:
                        s = record["source"]
                        t = record["target"]
                        all_nodes.add(s)
                        if t:
                            all_nodes.add(t)
                            adjacency[s].add(t)
                            adjacency[t].add(s)

                visited = set()
                clusters = []
                for node_id in all_nodes:
                    if node_id in visited:
                        continue
                    component = []
                    queue = deque([node_id])
                    while queue:
                        curr = queue.popleft()
                        if curr in visited:
                            continue
                        visited.add(curr)
                        component.append(curr)
                        for neighbor in adjacency.get(curr, set()):
                            if neighbor not in visited:
                                queue.append(neighbor)
                    if len(component) >= min_size:
                        clusters.append(component)
                clusters.sort(key=len, reverse=True)
                logger.info("Found %d clusters (min_size=%d) in graph DB", len(clusters), min_size)
                return clusters

            self._build_adjacency_list()
            fallback_visited: set[str] = set()
            fallback_clusters: list[list[str]] = []

            # If as_of is specified, construct a temporally filtered adjacency view
            if as_of is not None:
                adj: defaultdict[str, set[str]] = defaultdict(set)
                raw_rels = [_dict_to_relationship(v) for v in self._relationships.list_values()]
                for rel in raw_rels:
                    rel_time = rel.created_at
                    if rel_time.tzinfo is None:
                        rel_time = rel_time.replace(tzinfo=UTC)
                    cutoff = as_of if as_of.tzinfo is not None else as_of.replace(tzinfo=UTC)
                    if rel_time <= cutoff:
                        adj[rel.source_entity_id].add(rel.target_entity_id)
                        adj[rel.target_entity_id].add(rel.source_entity_id)
            else:
                adj = defaultdict(set)
                for u, nbrs in self._adjacency.items():
                    for v, _ in nbrs:
                        adj[u].add(v)

            raw_entities = [_dict_to_entity(v) for v in self._entities.list_values()]
            for entity in raw_entities:
                entity_id = entity.id
                if entity_id in fallback_visited:
                    continue

                fallback_component: list[str] = []
                fallback_queue: deque[str] = deque([entity_id])

                while fallback_queue:
                    current = fallback_queue.popleft()
                    if current in fallback_visited:
                        continue
                    fallback_visited.add(current)
                    fallback_component.append(current)

                    for neighbor_id in adj.get(current, set()):
                        if neighbor_id not in fallback_visited:
                            fallback_queue.append(neighbor_id)

                if len(fallback_component) >= min_size:
                    fallback_clusters.append(fallback_component)

            fallback_clusters.sort(key=len, reverse=True)
            logger.info("Found %d clusters (min_size=%d)", len(fallback_clusters), min_size)
            return fallback_clusters

    def get_subgraph(
        self,
        center_entity_id: str,
        radius: int = 2,
        max_nodes: int = 100,
        as_of: datetime | None = None,
    ) -> GraphSubgraph:
        """Extract a subgraph centered on an entity with bounded node budget."""
        with self._lock:
            if self.db_type in ("neo4j", "memgraph") and self.driver:
                nodes_dict = {}
                rels_dict = {}
                query = (
                    "MATCH (s:Entity {id: $center_id}) "
                    "OPTIONAL MATCH p = (s)-[r*1..$radius]-(n:Entity) "
                )
                params: dict[str, Any] = {
                    "center_id": center_entity_id,
                    "radius": radius,
                    "max_nodes": max_nodes,
                }
                if as_of is not None:
                    query += "WHERE all(rel IN relationships(p) WHERE rel.created_at <= $as_of) "
                    params["as_of"] = as_of.isoformat()
                query += "RETURN s, collect(p)[..$max_nodes] as paths"

                with self.driver.session() as session:
                    result = session.run(query, **params)
                    record = result.single()
                    if not record:
                        return GraphSubgraph(center_entity_id=center_entity_id, depth=radius)

                    s_node = record["s"]
                    s_entity = _neo4j_node_to_entity(s_node)
                    nodes_dict[s_entity.id] = s_entity

                    paths = record["paths"] or []
                    for path in paths:
                        for node in path.nodes:
                            entity = _neo4j_node_to_entity(node)
                            nodes_dict[entity.id] = entity
                        for rel in path.relationships:
                            relationship = _neo4j_rel_to_relationship(rel)
                            rels_dict[relationship.id] = relationship

                node_ids = list(nodes_dict.keys())[:max_nodes]

                # Build React Flow nodes
                nodes = []
                for i, nid in enumerate(node_ids):
                    entity = nodes_dict[nid]
                    import math

                    if nid == center_entity_id:
                        x: float = 400.0
                        y: float = 300.0
                    else:
                        angle = 2 * math.pi * (i - 1) / max(1, len(node_ids) - 1)
                        layer = 1
                        for depth_check in range(radius):
                            if i > len(node_ids) * (depth_check + 1) / (radius + 1):
                                layer = depth_check + 2
                        r = 150 * layer
                        x = 400 + r * math.cos(angle)
                        y = 300 + r * math.sin(angle)

                    color = _NODE_COLORS.get(entity.entity_type, "#6366f1")
                    is_high_risk = entity.risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL)

                    nodes.append(
                        {
                            "id": nid,
                            "type": "default",
                            "position": {"x": round(x), "y": round(y)},
                            "data": {
                                "label": entity.display_label,
                                "entityType": entity.entity_type.value,
                                "bankId": entity.bank_id,
                                "riskLevel": entity.risk_level.value,
                                "alertCount": entity.alert_count,
                                "isCenter": nid == center_entity_id,
                            },
                            "style": {
                                "background": color if not is_high_risk else "#ef4444",
                                "color": "#ffffff",
                                "border": f"2px solid {'#ef4444' if is_high_risk else color}",
                                "borderRadius": "8px",
                                "padding": "8px 12px",
                                "fontSize": "11px",
                                "fontWeight": "600",
                            },
                        }
                    )

                # Build React Flow edges
                edges = []
                node_id_set = set(node_ids)
                for rel in rels_dict.values():
                    if rel.source_entity_id in node_id_set and rel.target_entity_id in node_id_set:
                        style = _EDGE_STYLES.get(rel.relationship_type, {})
                        edges.append(
                            {
                                "id": rel.id,
                                "source": rel.source_entity_id,
                                "target": rel.target_entity_id,
                                "label": rel.relationship_type.value.replace("_", " "),
                                "type": "smoothstep",
                                "animated": rel.relationship_type == RelationshipType.LINKED_ALERT,
                                "style": style,
                                "data": {
                                    "confidence": rel.confidence,
                                    "relationshipType": rel.relationship_type.value,
                                },
                            }
                        )

                # Detect clusters within subgraph using local adjacency
                sub_adj: defaultdict[str, set[tuple[str, str]]] = defaultdict(set)
                for rel in rels_dict.values():
                    sub_adj[rel.source_entity_id].add((rel.target_entity_id, rel.id))
                    sub_adj[rel.target_entity_id].add((rel.source_entity_id, rel.id))
                clusters = self._detect_subgraph_clusters(node_id_set, adjacency=sub_adj)

                return GraphSubgraph(
                    nodes=nodes,
                    edges=edges,
                    clusters=clusters,
                    center_entity_id=center_entity_id,
                    depth=radius,
                )

            self._build_adjacency_list()
            center_entity_val = self._entities.get(center_entity_id)
            if not center_entity_val:
                return GraphSubgraph(center_entity_id=center_entity_id, depth=radius)

            # Collect nodes via BFS bounded by max_nodes and as_of cutoff
            fb_visited: set[str] = {center_entity_id}
            fb_queue: deque[tuple[str, int]] = deque([(center_entity_id, 0)])
            fb_node_ids: list[str] = [center_entity_id]
            bounded_max_nodes = max(1, min(max_nodes, 200))

            while fb_queue and len(fb_node_ids) < bounded_max_nodes:
                current_id, current_depth = fb_queue.popleft()
                if current_depth >= radius:
                    continue
                for neighbor_id, rel_id in self._adjacency.get(current_id, set()):
                    if as_of is not None:
                        rel_val = self._relationships.get(rel_id)
                        if rel_val:
                            rel_obj = _dict_to_relationship(rel_val)
                            rel_time = rel_obj.created_at
                            if rel_time.tzinfo is None:
                                rel_time = rel_time.replace(tzinfo=UTC)
                            cutoff = as_of if as_of.tzinfo is not None else as_of.replace(tzinfo=UTC)
                            if rel_time > cutoff:
                                continue
                    if neighbor_id not in fb_visited:
                        fb_visited.add(neighbor_id)
                        fb_node_ids.append(neighbor_id)
                        fb_queue.append((neighbor_id, current_depth + 1))
                        if len(fb_node_ids) >= bounded_max_nodes:
                            break

            # Build React Flow nodes
            nodes = []
            for i, nid in enumerate(fb_node_ids):
                entity_val = self._entities.get(nid)
                if not entity_val:
                    continue
                entity = _dict_to_entity(entity_val)

                import math

                if nid == center_entity_id:
                    x = 400.0
                    y = 300.0
                else:
                    angle = 2 * math.pi * (i - 1) / max(1, len(fb_node_ids) - 1)
                    layer = 1
                    for depth_check in range(radius):
                        if i > len(fb_node_ids) * (depth_check + 1) / (radius + 1):
                            layer = depth_check + 2
                    r = 150 * layer
                    x = 400 + r * math.cos(angle)
                    y = 300 + r * math.sin(angle)

                color = _NODE_COLORS.get(entity.entity_type, "#6366f1")
                is_high_risk = entity.risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL)

                nodes.append(
                    {
                        "id": nid,
                        "type": "default",
                        "position": {"x": round(x), "y": round(y)},
                        "data": {
                            "label": entity.display_label,
                            "entityType": entity.entity_type.value,
                            "bankId": entity.bank_id,
                            "riskLevel": entity.risk_level.value,
                            "alertCount": entity.alert_count,
                            "isCenter": nid == center_entity_id,
                        },
                        "style": {
                            "background": color if not is_high_risk else "#ef4444",
                            "color": "#ffffff",
                            "border": f"2px solid {'#ef4444' if is_high_risk else color}",
                            "borderRadius": "8px",
                            "padding": "8px 12px",
                            "fontSize": "11px",
                            "fontWeight": "600",
                        },
                    }
                )

            # Build React Flow edges
            edges = []
            node_id_set = set(fb_node_ids)

            raw_relationships = [_dict_to_relationship(v) for v in self._relationships.list_values()]
            sub_adj_fb: defaultdict[str, set[tuple[str, str]]] = defaultdict(set)
            for rel in raw_relationships:
                if as_of is not None:
                    rel_time = rel.created_at
                    if rel_time.tzinfo is None:
                        rel_time = rel_time.replace(tzinfo=UTC)
                    cutoff = as_of if as_of.tzinfo is not None else as_of.replace(tzinfo=UTC)
                    if rel_time > cutoff:
                        continue

                if rel.source_entity_id in node_id_set and rel.target_entity_id in node_id_set:
                    sub_adj_fb[rel.source_entity_id].add((rel.target_entity_id, rel.id))
                    sub_adj_fb[rel.target_entity_id].add((rel.source_entity_id, rel.id))
                    style = _EDGE_STYLES.get(rel.relationship_type, {})
                    edges.append(
                        {
                            "id": rel.id,
                            "source": rel.source_entity_id,
                            "target": rel.target_entity_id,
                            "label": rel.relationship_type.value.replace("_", " "),
                            "type": "smoothstep",
                            "animated": rel.relationship_type == RelationshipType.LINKED_ALERT,
                            "style": style,
                            "data": {
                                "confidence": rel.confidence,
                                "relationshipType": rel.relationship_type.value,
                            },
                        }
                    )

            # Detect clusters within subgraph using local adjacency
            clusters = self._detect_subgraph_clusters(node_id_set, adjacency=sub_adj_fb)

            return GraphSubgraph(
                nodes=nodes,
                edges=edges,
                clusters=clusters,
                center_entity_id=center_entity_id,
                depth=radius,
            )

    def search_nodes(
        self,
        query: str,
        entity_type: EntityType | None = None,
        limit: int = 20,
    ) -> list[Entity]:
        """Search entities by display label or privacy ID prefix."""
        if self.db_type in ("neo4j", "memgraph") and self.driver:
            et_str = entity_type.value if entity_type else None
            query_lower = query.lower()
            results = []
            with self.driver.session() as session:
                res = session.run(
                    "MATCH (n:Entity) "
                    "WHERE ($entity_type IS NULL OR n.entity_type = $entity_type) "
                    "  AND (toLower(n.display_label) CONTAINS $query OR toLower(n.privacy_id) CONTAINS $query) "
                    "RETURN n LIMIT $limit",
                    entity_type=et_str,
                    query=query_lower,
                    limit=limit,
                )
                for r in res:
                    results.append(_neo4j_node_to_entity(r["n"]))
            return results

        query_lower = query.lower()
        results = []
        raw_entities = [_dict_to_entity(v) for v in self._entities.list_values()]
        for entity in raw_entities:
            if entity_type and entity.entity_type != entity_type:
                continue
            if (
                query_lower in entity.display_label.lower()
                or query_lower in entity.privacy_id.lower()
            ):
                results.append(entity)
            if len(results) >= limit:
                break
        return results

    @property
    def node_count(self) -> int:
        if self.db_type in ("neo4j", "memgraph") and self.driver:
            with self.driver.session() as session:
                res = session.run("MATCH (n:Entity) RETURN count(n) as count")
                return res.single()["count"]
        return len(self._entities.list_values())

    @property
    def edge_count(self) -> int:
        if self.db_type in ("neo4j", "memgraph") and self.driver:
            with self.driver.session() as session:
                res = session.run("MATCH ()-[r]->() RETURN count(r) as count")
                return res.single()["count"]
        return len(self._relationships.list_values())

    def get_stats(self) -> dict:
        """Graph statistics for the dashboard."""
        if self.db_type in ("neo4j", "memgraph") and self.driver:
            with self.driver.session() as session:
                res_type = session.run(
                    "MATCH (n:Entity) RETURN n.entity_type as type, count(n) as count"
                )
                type_counts = {r["type"]: r["count"] for r in res_type}

                res_risk = session.run(
                    "MATCH (n:Entity) RETURN n.risk_level as risk, count(n) as count"
                )
                risk_counts = {r["risk"]: r["count"] for r in res_risk}

                total_nodes = sum(type_counts.values())
                total_edges = self.edge_count
                cluster_count = len(self.detect_clusters(min_size=3))

                return {
                    "total_nodes": total_nodes,
                    "total_edges": total_edges,
                    "nodes_by_type": type_counts,
                    "nodes_by_risk": risk_counts,
                    "cluster_count": cluster_count,
                    "database_backend": self.db_type.capitalize(),
                }

        fb_type_counts: dict[str, int] = defaultdict(int)
        fb_risk_counts: dict[str, int] = defaultdict(int)
        raw_entities = [_dict_to_entity(v) for v in self._entities.list_values()]
        for entity in raw_entities:
            fb_type_counts[entity.entity_type.value] += 1
            fb_risk_counts[entity.risk_level.value] += 1

        return {
            "total_nodes": self.node_count,
            "total_edges": self.edge_count,
            "nodes_by_type": dict(fb_type_counts),
            "nodes_by_risk": dict(fb_risk_counts),
            "cluster_count": len(self.detect_clusters(min_size=3)),
            "database_backend": "Redis (in-memory)",
        }

    def detect_cyclic_mule_rings(
        self,
        min_length: int = 3,
        max_length: int = 7,
        bank_id: str | None = None,
        as_of: datetime | None = None,
    ) -> list[dict[str, Any]]:
        """Detect closed transaction cycles (mule rings) with L in [min_length, max_length].
        Applies canonical rotation deduplication: every cycle C is rotated to start
        with its lexicographically minimum node ID. A deterministic SHA-256 ring ID is assigned.
        """
        with self._lock:
            # 1. If Neo4j driver is active, query Neo4j
            if self.db_type in ("neo4j", "memgraph") and self.driver:
                try:
                    query = (
                        f"MATCH path = (start:Entity)-[r*{min_length}..{max_length}]->(start) "
                    )
                    params: dict[str, Any] = {}
                    if as_of is not None:
                        query += "WHERE all(rel IN relationships(path) WHERE rel.created_at <= $as_of) "
                        params["as_of"] = as_of.isoformat()
                    query += (
                        "RETURN [n IN nodes(path) | n.id] AS cycle_ids, "
                        "       [n IN nodes(path) | n.bank_id] AS cycle_banks, "
                        "       [n IN nodes(path) | n.risk_level] AS cycle_risks, "
                        "       [rel IN relationships(path) | rel.evidence] AS cycle_evidences "
                        "LIMIT 200"
                    )
                    with self.driver.session() as session:
                        result = session.run(query, **params)
                        raw_rings = []
                        seen_ring_ids: set[str] = set()
                        for record in result:
                            node_ids = record["cycle_ids"]
                            if len(node_ids) > 1 and node_ids[0] == node_ids[-1]:
                                node_ids = node_ids[:-1]
                            if len(node_ids) < min_length or len(node_ids) > max_length:
                                continue
                            min_idx = node_ids.index(min(node_ids))
                            canon_nodes = node_ids[min_idx:] + node_ids[:min_idx]
                            ring_id = hashlib.sha256(":".join(canon_nodes).encode("utf-8")).hexdigest()[:16]
                            if ring_id in seen_ring_ids:
                                continue
                            seen_ring_ids.add(ring_id)

                            banks = [b for b in record["cycle_banks"] if b]
                            banks_involved = sorted(list(set(banks)))
                            if bank_id and bank_id not in banks_involved:
                                continue

                            is_cross_bank = len(banks_involved) >= 2
                            total_vol = 0.0
                            for ev in record["cycle_evidences"]:
                                if isinstance(ev, dict):
                                    total_vol += float(ev.get("amount") or ev.get("volume") or 0.0)

                            risk_scores = [RISK_LEVEL_TO_SCORE.get(r, 0.5) for r in record["cycle_risks"]]
                            base_risk = sum(risk_scores) / max(len(risk_scores), 1)
                            cross_boost = 0.2 if is_cross_bank else 0.0
                            len_factor = 0.1 if len(canon_nodes) <= 4 else 0.05
                            risk_score = round(min(1.0, max(0.1, base_risk + cross_boost + len_factor)), 4)

                            raw_rings.append(
                                {
                                    "ring_id": ring_id,
                                    "length": len(canon_nodes),
                                    "entity_ids": canon_nodes,
                                    "banks_involved": banks_involved,
                                    "is_cross_bank": is_cross_bank,
                                    "risk_score": risk_score,
                                    "total_volume": round(total_vol, 2),
                                    "detected_at": datetime.now(UTC),
                                }
                            )
                        return raw_rings
                except Exception as e:
                    logger.warning("Neo4j ring detection query failed (%s), falling back to in-memory DFS", e)

            # In-memory directed DFS cycle finder
            rel_dicts = self._relationships.list_values()
            directed_adj: defaultdict[str, list[tuple[str, float]]] = defaultdict(list)

            cutoff_dt = as_of if as_of is not None else None
            if cutoff_dt is not None and cutoff_dt.tzinfo is None:
                cutoff_dt = cutoff_dt.replace(tzinfo=UTC)

            for r_val in rel_dicts:
                r = _dict_to_relationship(r_val)
                if cutoff_dt is not None:
                    r_time = r.created_at
                    if r_time.tzinfo is None:
                        r_time = r_time.replace(tzinfo=UTC)
                    if r_time > cutoff_dt:
                        continue
                u = r.source_entity_id
                v = r.target_entity_id
                vol = 0.0
                if isinstance(r.evidence, dict):
                    vol = float(r.evidence.get("amount") or r.evidence.get("volume") or 0.0)
                directed_adj[u].append((v, vol))

            all_nodes = sorted(
                list(
                    set(self._entities.list_keys())
                    | set(directed_adj.keys())
                    | {target for targets in directed_adj.values() for target, _ in targets}
                )
            )
            mem_seen_ring_ids: set[str] = set()
            rings: list[dict[str, Any]] = []

            for s in all_nodes:
                stack: list[tuple[str, list[str], float]] = [(s, [s], 0.0)]
                while stack:
                    curr, path, path_vol = stack.pop()
                    if len(path) > max_length:
                        continue

                    for nxt, vol in directed_adj.get(curr, []):
                        if nxt == s:
                            # Closed cycle back to start node
                            if len(path) >= min_length:
                                ring_id = hashlib.sha256(":".join(path).encode("utf-8")).hexdigest()[:16]
                                if ring_id not in mem_seen_ring_ids:
                                    mem_seen_ring_ids.add(ring_id)

                                    node_banks: list[str] = []
                                    node_risk_scores: list[float] = []
                                    for nid in path:
                                        ent_val = self._entities.get(nid)
                                        if ent_val:
                                            ent = _dict_to_entity(ent_val)
                                            if ent.bank_id:
                                                node_banks.append(ent.bank_id)
                                            node_risk_scores.append(RISK_LEVEL_TO_SCORE.get(ent.risk_level.value, 0.5))
                                        else:
                                            node_risk_scores.append(0.5)

                                    banks_involved = sorted(list(set(node_banks)))
                                    if bank_id and bank_id not in banks_involved:
                                        continue

                                    is_cross_bank = len(banks_involved) >= 2
                                    base_risk = sum(node_risk_scores) / max(len(node_risk_scores), 1)
                                    cross_boost = 0.2 if is_cross_bank else 0.0
                                    len_factor = 0.1 if len(path) <= 4 else 0.05
                                    risk_score = round(min(1.0, max(0.1, base_risk + cross_boost + len_factor)), 4)

                                    rings.append(
                                        {
                                            "ring_id": ring_id,
                                            "length": len(path),
                                            "entity_ids": list(path),
                                            "banks_involved": banks_involved,
                                            "is_cross_bank": is_cross_bank,
                                            "risk_score": risk_score,
                                            "total_volume": round(path_vol + vol, 2),
                                            "detected_at": datetime.now(UTC),
                                        }
                                    )
                        elif nxt > s and nxt not in path:
                            stack.append((nxt, path + [nxt], path_vol + vol))

            return sorted(rings, key=lambda x: x["risk_score"], reverse=True)

    def detect_smurfing_patterns(
        self,
        window_hours: int = 24,
        min_fan: int = 3,
        max_depth: int = 3,
        bank_id: str | None = None,
        as_of: datetime | None = None,
    ) -> list[dict[str, Any]]:
        """Detect multi-hop financial smurfing patterns (fan-in, fan-out, layering)."""
        with self._lock:
            now = as_of if as_of is not None else datetime.now(UTC)
            if now.tzinfo is None:
                now = now.replace(tzinfo=UTC)
            cutoff_time = now - timedelta(hours=window_hours)

            rel_dicts = self._relationships.list_values()
            in_edges: defaultdict[str, list[tuple[str, float]]] = defaultdict(list)
            out_edges: defaultdict[str, list[tuple[str, float]]] = defaultdict(list)

            for r_val in rel_dicts:
                r = _dict_to_relationship(r_val)
                r_time = r.created_at
                if r_time.tzinfo is None:
                    r_time = r_time.replace(tzinfo=UTC)
                if r_time < cutoff_time or r_time > now:
                    continue

                u = r.source_entity_id
                v = r.target_entity_id
                vol = 0.0
                if isinstance(r.evidence, dict):
                    vol = float(r.evidence.get("amount") or r.evidence.get("volume") or 0.0)
                out_edges[u].append((v, vol))
                in_edges[v].append((u, vol))

            patterns: list[dict[str, Any]] = []
            seen_pattern_ids: set[str] = set()

            all_node_ids = set(self._entities.list_keys()) | set(in_edges.keys()) | set(out_edges.keys())

            for node_id in sorted(all_node_ids):
                node_val = self._entities.get(node_id)
                node_ent = _dict_to_entity(node_val) if node_val else None
                node_risk_val = node_ent.risk_level.value if node_ent else RiskLevel.MEDIUM.value
                base_risk = RISK_LEVEL_TO_SCORE.get(node_risk_val, 0.5)

                # 1. Fan-In Check (Aggregation Mules)
                in_sources = list({src for src, _ in in_edges.get(node_id, []) if src != node_id})
                if len(in_sources) >= min_fan:
                    spokes = sorted(in_sources)
                    pattern_key = f"fan_in:{node_id}:{':'.join(spokes)}"
                    pattern_id = hashlib.sha256(pattern_key.encode("utf-8")).hexdigest()[:16]
                    if pattern_id not in seen_pattern_ids:
                        seen_pattern_ids.add(pattern_id)

                        banks = set()
                        if node_ent and node_ent.bank_id:
                            banks.add(node_ent.bank_id)
                        for spk in spokes:
                            spk_val = self._entities.get(spk)
                            if spk_val:
                                spk_ent = _dict_to_entity(spk_val)
                                if spk_ent.bank_id:
                                    banks.add(spk_ent.bank_id)
                        banks_involved = sorted(list(banks))

                        if not bank_id or bank_id in banks_involved:
                            is_cross_bank = len(banks_involved) >= 2
                            total_vol = sum(v for src, v in in_edges[node_id] if src in spokes)
                            fan_degree = len(spokes)
                            risk_score = round(
                                min(
                                    1.0,
                                    base_risk
                                    + min(0.3, (fan_degree - min_fan + 1) * 0.05)
                                    + (0.15 if is_cross_bank else 0.0),
                                ),
                                4,
                            )
                            patterns.append(
                                {
                                    "pattern_id": pattern_id,
                                    "pattern_type": "fan_in",
                                    "hub_entity_id": node_id,
                                    "spoke_entity_ids": spokes,
                                    "fan_degree": fan_degree,
                                    "total_volume": round(
                                        total_vol if total_vol > 0 else float(fan_degree * 1000.0), 2
                                    ),
                                    "depth": 1,
                                    "banks_involved": banks_involved,
                                    "is_cross_bank": is_cross_bank,
                                    "risk_score": risk_score,
                                    "detected_at": datetime.now(UTC),
                                }
                            )

                # 2. Fan-Out Check (Dispersion Mules)
                out_targets = list({tgt for tgt, _ in out_edges.get(node_id, []) if tgt != node_id})
                if len(out_targets) >= min_fan:
                    spokes = sorted(out_targets)
                    pattern_key = f"fan_out:{node_id}:{':'.join(spokes)}"
                    pattern_id = hashlib.sha256(pattern_key.encode("utf-8")).hexdigest()[:16]
                    if pattern_id not in seen_pattern_ids:
                        seen_pattern_ids.add(pattern_id)

                        banks = set()
                        if node_ent and node_ent.bank_id:
                            banks.add(node_ent.bank_id)
                        for spk in spokes:
                            spk_val = self._entities.get(spk)
                            if spk_val:
                                spk_ent = _dict_to_entity(spk_val)
                                if spk_ent.bank_id:
                                    banks.add(spk_ent.bank_id)
                        banks_involved = sorted(list(banks))

                        if not bank_id or bank_id in banks_involved:
                            is_cross_bank = len(banks_involved) >= 2
                            total_vol = sum(v for tgt, v in out_edges[node_id] if tgt in spokes)
                            fan_degree = len(spokes)
                            risk_score = round(
                                min(
                                    1.0,
                                    base_risk
                                    + min(0.3, (fan_degree - min_fan + 1) * 0.05)
                                    + (0.15 if is_cross_bank else 0.0),
                                ),
                                4,
                            )
                            patterns.append(
                                {
                                    "pattern_id": pattern_id,
                                    "pattern_type": "fan_out",
                                    "hub_entity_id": node_id,
                                    "spoke_entity_ids": spokes,
                                    "fan_degree": fan_degree,
                                    "total_volume": round(
                                        total_vol if total_vol > 0 else float(fan_degree * 1000.0), 2
                                    ),
                                    "depth": 1,
                                    "banks_involved": banks_involved,
                                    "is_cross_bank": is_cross_bank,
                                    "risk_score": risk_score,
                                    "detected_at": datetime.now(UTC),
                                }
                            )

                # 3. Multi-hop Layering Check (Transit Mule Hub with In-Flow and Out-Flow)
                if len(in_sources) >= 2 and len(out_targets) >= 2:
                    all_spokes = sorted(list(set(in_sources) | set(out_targets)))
                    pattern_key = f"layering:{node_id}:{':'.join(all_spokes)}"
                    pattern_id = hashlib.sha256(pattern_key.encode("utf-8")).hexdigest()[:16]
                    if pattern_id not in seen_pattern_ids:
                        seen_pattern_ids.add(pattern_id)

                        banks = set()
                        if node_ent and node_ent.bank_id:
                            banks.add(node_ent.bank_id)
                        for spk in all_spokes:
                            spk_val = self._entities.get(spk)
                            if spk_val:
                                spk_ent = _dict_to_entity(spk_val)
                                if spk_ent.bank_id:
                                    banks.add(spk_ent.bank_id)
                        banks_involved = sorted(list(banks))

                        if not bank_id or bank_id in banks_involved:
                            is_cross_bank = len(banks_involved) >= 2
                            fan_degree = len(in_sources) + len(out_targets)
                            in_vol = sum(v for src, v in in_edges[node_id] if src in in_sources)
                            out_vol = sum(v for tgt, v in out_edges[node_id] if tgt in out_targets)
                            total_vol = in_vol + out_vol
                            risk_score = round(min(1.0, base_risk + 0.2 + (0.15 if is_cross_bank else 0.0)), 4)
                            patterns.append(
                                {
                                    "pattern_id": pattern_id,
                                    "pattern_type": "multi_hop_layering",
                                    "hub_entity_id": node_id,
                                    "spoke_entity_ids": all_spokes,
                                    "fan_degree": fan_degree,
                                    "total_volume": round(
                                        total_vol if total_vol > 0 else float(fan_degree * 1000.0), 2
                                    ),
                                    "depth": 2,
                                    "banks_involved": banks_involved,
                                    "is_cross_bank": is_cross_bank,
                                    "risk_score": risk_score,
                                    "detected_at": datetime.now(UTC),
                                }
                            )

            return sorted(patterns, key=lambda x: x["risk_score"], reverse=True)

    def execute_cypher(
        self,
        query: str,
        params: dict[str, Any] | None = None,
        read_only: bool = True,
    ) -> list[dict[str, Any]]:
        """Safely execute a parameterized Cypher query.
        If read_only=True, rejects any query containing mutating keywords
        (CREATE, MERGE, DELETE, SET, REMOVE, DROP, DETACH).
        """
        if read_only:
            match = _MUTATING_CYPHER_RE.search(query)
            if match:
                raise ValueError(
                    f"Cypher execution rejected: query contains mutating keyword '{match.group(0)}' while read_only=True"
                )

        with self._lock:
            if self.db_type in ("neo4j", "memgraph") and self.driver:
                with self.driver.session() as session:
                    result = session.run(query, **(params or {}))
                    return [record.data() for record in result]

            # In-memory Redis fallback emulation
            logger.info("Executing Cypher query in in-memory fallback mode: %s", query[:80])
            upper_q = query.upper()
            if "RETURN N.ENTITY_TYPE" in upper_q or "COUNT(N)" in upper_q:
                counts: dict[str, int] = defaultdict(int)
                for val in self._entities.list_values():
                    e = _dict_to_entity(val)
                    counts[e.entity_type.value] += 1
                return [{"type": k, "count": v} for k, v in counts.items()]

            if "MATCH (N:ENTITY)" in upper_q and "RETURN N" in upper_q:
                return [{"n": _entity_to_dict(_dict_to_entity(v))} for v in self._entities.list_values()]

            return []

    # ── Private helpers ────────────────────────

    def _detect_subgraph_clusters(
        self,
        node_ids: set[str],
        adjacency: dict[str, set[tuple[str, str]]] | None = None,
    ) -> list[list[str]]:
        """Find clusters within a subset of nodes."""
        visited: set[str] = set()
        clusters: list[list[str]] = []
        adj = adjacency if adjacency is not None else self._adjacency

        for nid in node_ids:
            if nid in visited:
                continue
            component: list[str] = []
            queue: deque[str] = deque([nid])
            while queue:
                current = queue.popleft()
                if current in visited or current not in node_ids:
                    continue
                visited.add(current)
                component.append(current)
                for neighbor_id, _ in adj.get(current, set()):
                    if neighbor_id not in visited and neighbor_id in node_ids:
                        queue.append(neighbor_id)
            if len(component) >= 2:
                clusters.append(component)

        return clusters

    # ── MinHash LSH Fuzzy Entity Resolution ──────────────────────

    def find_fuzzy_matches(
        self,
        entity_id: str,
        similarity_threshold: float = 0.40,
        num_hashes: int = 64,
        num_bands: int = 16,
        rows_per_band: int = 4,
        cross_bank_only: bool = True,
    ) -> list[dict[str, Any]]:
        """Finds candidate matching entities in the graph using MinHash LSH.
        Extracts display labels and text attributes from the target entity,
        computes MinHash signatures, and queries across all registered entities.
        """
        from app.domain.minhash_lsh import (
            compute_minhash_signature,
            estimate_jaccard_similarity,
            partition_into_lsh_bands,
        )

        with self._lock:
            target_data = self._entities.get(entity_id)
            if not target_data:
                return []
            target_ent = _dict_to_entity(target_data)
            target_text = target_ent.display_label or target_ent.id

            target_sig = compute_minhash_signature(target_text, num_hashes=num_hashes)
            target_bands = set(
                partition_into_lsh_bands(
                    target_sig, num_bands=num_bands, rows_per_band=rows_per_band
                )
            )

            candidates: list[dict[str, Any]] = []
            for val in self._entities.list_values():
                cand_ent = _dict_to_entity(val)
                if cand_ent.id == entity_id:
                    continue
                if cross_bank_only and cand_ent.bank_id == target_ent.bank_id:
                    continue

                cand_text = cand_ent.display_label or cand_ent.id
                cand_sig = compute_minhash_signature(cand_text, num_hashes=num_hashes)
                cand_bands = partition_into_lsh_bands(
                    cand_sig, num_bands=num_bands, rows_per_band=rows_per_band
                )

                collided_bands = sum(1 for b in cand_bands if b in target_bands)
                if collided_bands > 0:
                    sim = estimate_jaccard_similarity(target_sig, cand_sig)
                    if sim >= similarity_threshold:
                        candidates.append(
                            {
                                "entity_id": cand_ent.id,
                                "bank_id": cand_ent.bank_id,
                                "display_label": cand_ent.display_label,
                                "entity_type": cand_ent.entity_type.value,
                                "estimated_jaccard": sim,
                                "collided_bands": collided_bands,
                                "total_bands": num_bands,
                            }
                        )

            candidates.sort(key=lambda x: x["estimated_jaccard"], reverse=True)
            return candidates

    def link_fuzzy_entities(
        self,
        similarity_threshold: float = 0.50,
        num_hashes: int = 64,
        num_bands: int = 16,
        rows_per_band: int = 4,
        auto_add_relationship: bool = False,
    ) -> list[dict[str, Any]]:
        """Scans the graph for cross-institution entities with high fuzzy similarity.
        Returns detected duplicate / mule candidate pairs. If auto_add_relationship=True,
        creates RelationshipType.SAME_ENTITY edges between matched entities.
        """
        from app.domain.minhash_lsh import (
            compute_minhash_signature,
            estimate_jaccard_similarity,
            partition_into_lsh_bands,
        )

        with self._lock:
            entities = [_dict_to_entity(v) for v in self._entities.list_values()]
            if len(entities) < 2:
                return []

            # Index all entities by LSH bands
            band_index: dict[str, list[tuple[Entity, list[int]]]] = defaultdict(list)
            for ent in entities:
                text = ent.display_label or ent.id
                sig = compute_minhash_signature(text, num_hashes=num_hashes)
                bands = partition_into_lsh_bands(
                    sig, num_bands=num_bands, rows_per_band=rows_per_band
                )
                for b in bands:
                    band_index[b].append((ent, sig))

            # Candidate pair tracking
            candidate_pairs: dict[
                tuple[str, str], tuple[Entity, Entity, list[int], list[int], int]
            ] = {}
            for entries in band_index.values():
                for i in range(len(entries)):
                    for j in range(i + 1, len(entries)):
                        e1, s1 = entries[i]
                        e2, s2 = entries[j]
                        if e1.bank_id != e2.bank_id:
                            pair_key = (min(e1.id, e2.id), max(e1.id, e2.id))
                            if pair_key not in candidate_pairs:
                                candidate_pairs[pair_key] = (e1, e2, s1, s2, 1)
                            else:
                                prior = candidate_pairs[pair_key]
                                candidate_pairs[pair_key] = (
                                    prior[0],
                                    prior[1],
                                    prior[2],
                                    prior[3],
                                    prior[4] + 1,
                                )

            matches: list[dict[str, Any]] = []
            for (id1, id2), (e1, e2, s1, s2, band_cnt) in candidate_pairs.items():
                sim = estimate_jaccard_similarity(s1, s2)
                if sim >= similarity_threshold:
                    match_info = {
                        "source_entity_id": id1,
                        "source_bank_id": e1.bank_id,
                        "target_entity_id": id2,
                        "target_bank_id": e2.bank_id,
                        "estimated_jaccard": sim,
                        "collided_bands": band_cnt,
                        "total_bands": num_bands,
                    }
                    matches.append(match_info)

                    if auto_add_relationship:
                        rel = Relationship(
                            id=f"fuzzy_same_{id1[:8]}_{id2[:8]}",
                            source_entity_id=id1,
                            target_entity_id=id2,
                            relationship_type=RelationshipType.SAME_ENTITY,
                            confidence=sim,
                            evidence=[
                                f"MinHash LSH fuzzy match J={sim:.4f} across {band_cnt}/{num_bands} bands"
                            ],
                        )
                        self.add_relationship(rel)

            matches.sort(key=lambda x: x["estimated_jaccard"], reverse=True)
            return matches