app/wallet_memory/clustering.py

"""
Wallet Clustering Engine — 6 heuristics with Union-Find merge.
================================================================
Upgraded from the original 4-heuristic engine in wallet_clustering.py.
Added: gas payer linkage (EVM deployer networks) and deposit-withdraw patterns (CEX tracing).

Heuristics (confidence 0-1):
  1. Temporal proximity        — 0.60  (coordinated tx timing)
  2. Common counterparties     — 0.70  (shared senders/recipients)
  3. Behavioral patterns      — 0.65  (similar fingerprint)
  4. Common funding source     — 0.80  (same funder = same operator)
  5. Gas payer linkage         — 0.85  (one wallet pays for multiple deploys)
  6. Deposit-withdraw pattern  — 0.80  (exchange flow tracing)

Union-Find for O(α(n)) merges. Results persisted to storage.py.
"""

import logging
from collections import defaultdict
from datetime import datetime

logger = logging.getLogger("wallet_memory.clustering")


# ── Union-Find (Disjoint Set) ────────────────────────────────────────


class UnionFind:
    """Weighted Union-Find with path compression. O(α(n)) per op."""

    def __init__(self):
        self._parent: dict[str, str] = {}
        self._rank: dict[str, int] = {}
        self._size: dict[str, int] = {}

    def find(self, x: str) -> str:
        if x not in self._parent:
            self._parent[x] = x
            self._rank[x] = 0
            self._size[x] = 1
            return x
        # Path compression
        if self._parent[x] != x:
            self._parent[x] = self.find(self._parent[x])
        return self._parent[x]

    def union(self, x: str, y: str) -> str:
        """Merge sets containing x and y. Returns root of merged set."""
        rx, ry = self.find(x), self.find(y)
        if rx == ry:
            return rx

        # Weighted union by rank
        if self._rank[rx] < self._rank[ry]:
            rx, ry = ry, rx
        self._parent[ry] = rx
        self._size[rx] += self._size[ry]
        if self._rank[rx] == self._rank[ry]:
            self._rank[rx] += 1
        return rx

    def connected(self, x: str, y: str) -> bool:
        return self.find(x) == self.find(y)

    def get_members(self, x: str) -> set[str]:
        """Get all members of x's set. O(n) — use sparingly."""
        root = self.find(x)
        return {k for k, v in self._parent.items() if v and self.find(k) == root}

    def get_all_clusters(self) -> dict[str, set[str]]:
        """Return {root: set_of_members} for all clusters."""
        clusters: dict[str, set[str]] = defaultdict(set)
        for key in self._parent:
            root = self.find(key)
            clusters[root].add(key)
        return dict(clusters)

    @property
    def total_nodes(self) -> int:
        return len(self._parent)


# ── Data classes ─────────────────────────────────────────────────────


class WalletProfile:
    """Behavioral profile of a wallet."""

    __slots__ = (
        "address",
        "avg_tx_size",
        "chain_id",
        "first_seen",
        "last_seen",
        "preferred_hours",
        "programs_used",
        "total_transactions",
        "total_volume",
        "transaction_times",
        "tx_frequency",
        "unique_counterparties",
    )

    def __init__(self, address: str, chain_id: str = ""):
        self.address = address
        self.chain_id = chain_id
        self.first_seen: datetime | None = None
        self.last_seen: datetime | None = None
        self.total_transactions = 0
        self.total_volume = 0.0
        self.unique_counterparties: set[str] = set()
        self.transaction_times: list[datetime] = []
        self.programs_used: set[str] = set()
        self.preferred_hours: list[int] = []
        self.avg_tx_size = 0.0
        self.tx_frequency = 0.0

    def add_tx(self, timestamp: datetime, counterparty: str, amount: float = 0.0, program: str = ""):
        self.total_transactions += 1
        self.total_volume += amount
        self.unique_counterparties.add(counterparty)
        self.transaction_times.append(timestamp)
        if program:
            self.programs_used.add(program)

    def calculate_metrics(self):
        if not self.transaction_times:
            return
        self.transaction_times.sort()
        self.first_seen = self.transaction_times[0]
        self.last_seen = self.transaction_times[-1]
        days = max((self.last_seen - self.first_seen).days, 1)
        self.tx_frequency = len(self.transaction_times) / days
        if self.total_transactions > 0:
            self.avg_tx_size = self.total_volume / self.total_transactions
        hours = [t.hour for t in self.transaction_times]
        hour_counts = defaultdict(int)
        for h in hours:
            hour_counts[h] += 1
        self.preferred_hours = sorted(hour_counts.keys(), key=lambda h: hour_counts[h], reverse=True)[:3]


class Cluster:
    """A detected wallet cluster (entity)."""

    def __init__(self, cluster_id: str):
        self.cluster_id = cluster_id
        self.wallets: set[str] = set()
        self.confidence: float = 0.0
        self.detection_methods: list[str] = []
        self.center_wallet: str | None = None
        self.label: str = ""
        self.category: str = "unknown"

    def to_dict(self) -> dict:
        return {
            "cluster_id": self.cluster_id,
            "wallet_count": len(self.wallets),
            "wallets": list(self.wallets),
            "confidence": round(self.confidence, 3),
            "detection_methods": self.detection_methods,
            "center_wallet": self.center_wallet,
            "label": self.label,
            "category": self.category,
        }


# ── Clustering Engine ───────────────────────────────────────────────

HEURISTIC_WEIGHTS = {
    "temporal_proximity": 0.60,
    "common_counterparties": 0.70,
    "behavioral_pattern": 0.65,
    "common_funding": 0.80,
    "gas_payer": 0.85,
    "deposit_withdraw": 0.80,
}


class WalletClusteringEngine:
    """6-heuristic clustering engine with Union-Find merge."""

    TEMPORAL_WINDOW_MINUTES = 5
    MIN_COMMON_COUNTERPARTIES = 3
    MIN_FINGERPRINT_SIMILARITY = 0.7
    MIN_GAS_PAYER_RECIPIENTS = 3

    def __init__(self):
        self.wallets: dict[str, WalletProfile] = {}
        self.uf = UnionFind()
        self._heuristic_links: dict[str, set[str]] = defaultdict(set)
        # Track which heuristic linked each pair
        self._pair_heuristics: dict[tuple[str, str], set[str]] = defaultdict(set)

    def add_transaction(
        self,
        address: str,
        counterparty: str,
        timestamp: datetime,
        amount: float = 0.0,
        chain_id: str = "",
        program: str = "",
    ):
        """Feed a transaction into the engine."""
        addr = address.lower()
        cp = counterparty.lower()

        if addr not in self.wallets:
            self.wallets[addr] = WalletProfile(addr, chain_id)
        self.wallets[addr].add_tx(timestamp, cp, amount, program)

        # Ensure both exist in Union-Find
        self.uf.find(addr)
        self.uf.find(cp)

    # ── Heuristic 1: Temporal Proximity ──────────────────────────

    def detect_temporal_clusters(self, time_window_min: int | None = None) -> list[Cluster]:
        """Wallets transacting in the same time window may be coordinated."""
        window = time_window_min or self.TEMPORAL_WINDOW_MINUTES
        time_groups: dict[datetime, set[str]] = defaultdict(set)

        for addr, profile in self.wallets.items():
            for ts in profile.transaction_times:
                key = ts.replace(
                    minute=(ts.minute // window) * window,
                    second=0,
                    microsecond=0,
                )
                time_groups[key].add(addr)

        clusters = []
        for window_time, addrs in time_groups.items():
            if len(addrs) < 2:
                continue
            # Link all wallets in this window
            addr_list = list(addrs)
            for i in range(len(addr_list)):
                for j in range(i + 1, len(addr_list)):
                    pair = tuple(sorted([addr_list[i], addr_list[j]]))
                    self._pair_heuristics[pair].add("temporal_proximity")
                    self._heuristic_links["temporal_proximity"].add(f"{pair[0]}:{pair[1]}")
                    self.uf.union(addr_list[i], addr_list[j])

            c = Cluster(f"temporal_{window_time.isoformat()}")
            c.wallets = addrs
            c.detection_methods = ["temporal_proximity"]
            c.confidence = min(0.60, 0.3 + len(addrs) * 0.05)
            clusters.append(c)

        return clusters

    # ── Heuristic 2: Common Counterparties ────────────────────────

    def detect_counterparty_clusters(self) -> list[Cluster]:
        """Wallets sharing multiple counterparties may be related."""
        # Map: counterparty -> set of wallets that interacted with it
        cp_wallets: dict[str, set[str]] = defaultdict(set)
        for addr, profile in self.wallets.items():
            for cp in profile.unique_counterparties:
                cp_wallets[cp].add(addr)

        # Find pairs sharing >= MIN_COMMON_COUNTERPARTIES
        pair_shared: dict[tuple[str, str], int] = defaultdict(int)
        for cp, wallets in cp_wallets.items():
            wl = list(wallets)
            for i in range(len(wl)):
                for j in range(i + 1, len(wl)):
                    pair = tuple(sorted([wl[i], wl[j]]))
                    pair_shared[pair] += 1

        # Link pairs exceeding threshold
        for pair, count in pair_shared.items():
            if count >= self.MIN_COMMON_COUNTERPARTIES:
                self._pair_heuristics[pair].add("common_counterparties")
                self._heuristic_links["common_counterparties"].add(f"{pair[0]}:{pair[1]}")
                self.uf.union(pair[0], pair[1])

        # Return cluster view
        raw = self.uf.get_all_clusters()
        clusters = []
        for root, members in raw.items():
            # Only include if counterparty heuristic contributed
            counterparty_pairs = [
                (a, b)
                for a in members
                for b in members
                if a < b and "common_counterparties" in self._pair_heuristics.get(tuple(sorted([a, b])), set())
            ]
            if counterparty_pairs and len(members) >= 2:
                c = Cluster(f"counterparty_{root[:12]}")
                c.wallets = members
                c.detection_methods = ["common_counterparties"]
                c.confidence = min(0.70, 0.4 + len(members) * 0.05)
                c.center_wallet = self._find_center(members)
                clusters.append(c)
        return clusters

    # ── Heuristic 3: Behavioral Fingerprint ───────────────────────

    def detect_behavioral_clusters(self) -> list[Cluster]:
        """Wallets with similar behavioral fingerprints may share an operator."""
        fingerprints = {}
        for addr, profile in self.wallets.items():
            if profile.total_transactions < 5:
                continue
            profile.calculate_metrics()
            fingerprints[addr] = {
                "avg_size": profile.avg_tx_size,
                "frequency": profile.tx_frequency,
                "preferred_hours": set(profile.preferred_hours),
                "program_diversity": len(profile.programs_used),
                "counterparty_count": len(profile.unique_counterparties),
            }

        addrs = list(fingerprints.keys())
        for i in range(len(addrs)):
            for j in range(i + 1, len(addrs)):
                sim = self._fingerprint_similarity(fingerprints[addrs[i]], fingerprints[addrs[j]])
                if sim >= self.MIN_FINGERPRINT_SIMILARITY:
                    pair = tuple(sorted([addrs[i], addrs[j]]))
                    self._pair_heuristics[pair].add("behavioral_pattern")
                    self._heuristic_links["behavioral_pattern"].add(f"{pair[0]}:{pair[1]}")
                    self.uf.union(addrs[i], addrs[j])

        # Build cluster view
        clusters = []
        raw = self.uf.get_all_clusters()
        for root, members in raw.items():
            beh_pairs = [
                1
                for a in members
                for b in members
                if a < b and "behavioral_pattern" in self._pair_heuristics.get(tuple(sorted([a, b])), set())
            ]
            if beh_pairs and len(members) >= 2:
                c = Cluster(f"behavioral_{root[:12]}")
                c.wallets = members
                c.detection_methods = ["behavioral_pattern"]
                c.confidence = min(0.65, 0.35 + len(members) * 0.04)
                c.center_wallet = self._find_center(members)
                clusters.append(c)
        return clusters

    # ── Heuristic 4: Common Funding Source ───────────────────────

    def detect_funding_clusters(self) -> list[Cluster]:
        """Wallets funded from the same source are likely same operator."""
        # Find first incoming tx for each wallet (the funder)
        funders: dict[str, str] = {}
        for addr, profile in self.wallets.items():
            if profile.transaction_times:
                min(profile.transaction_times)
                # The counterparty at the earliest tx is the funder
                for cp in profile.unique_counterparties:
                    funders[addr] = cp
                    break

        # Group by funder
        funder_groups: dict[str, set[str]] = defaultdict(set)
        for wallet, funder in funders.items():
            funder_groups[funder].add(wallet)

        clusters = []
        for funder, wallets in funder_groups.items():
            if len(wallets) < 2:
                continue
            # Link all wallets funded by same source
            wl = list(wallets)
            for i in range(len(wl)):
                for j in range(i + 1, len(wl)):
                    pair = tuple(sorted([wl[i], wl[j]]))
                    self._pair_heuristics[pair].add("common_funding")
                    self._heuristic_links["common_funding"].add(f"{pair[0]}:{pair[1]}")
                    self.uf.union(wl[i], wl[j])

            c = Cluster(f"funding_{funder[:12]}")
            c.wallets = wallets
            c.detection_methods = ["common_funding"]
            c.confidence = 0.80 if len(wallets) >= 5 else 0.60
            c.center_wallet = funder
            clusters.append(c)

        return clusters

    # ── Heuristic 5: Gas Payer Linkage (NEW — for deployer networks) ─

    def detect_gas_payer_clusters(self, gas_payer_map: dict[str, list[str]] | None = None) -> list[Cluster]:
        """
        Link wallets where one address pays gas for multiple contract
        deployments or token transfers. Primary heuristic for identifying
        serial deployer networks.

        Args:
            gas_payer_map: Optional external data {gas_payer: [recipients]}.
                          If not provided, inferred from transaction data.
        """
        if gas_payer_map is None:
            # Infer from existing data: from_address paid for to_address
            gas_payer_map = defaultdict(set)
            for addr, profile in self.wallets.items():
                for cp in profile.unique_counterparties:
                    gas_payer_map[addr].add(cp)

        clusters = []
        for payer, recipients in gas_payer_map.items():
            recipients_set = {r.lower() for r in recipients}
            payer_lower = payer.lower()
            recipients_set.discard(payer_lower)  # Don't link to self

            if len(recipients_set) < self.MIN_GAS_PAYER_RECIPIENTS:
                continue

            # Link payer to all recipients
            self.uf.find(payer_lower)
            for r in recipients_set:
                self.uf.find(r)
                pair = tuple(sorted([payer_lower, r]))
                self._pair_heuristics[pair].add("gas_payer")
                self._heuristic_links["gas_payer"].add(f"{pair[0]}:{pair[1]}")
                self.uf.union(payer_lower, r)

            c = Cluster(f"gas_payer_{payer_lower[:12]}")
            c.wallets = recipients_set | {payer_lower}
            c.detection_methods = ["gas_payer"]
            c.confidence = 0.85
            c.center_wallet = payer_lower
            clusters.append(c)

        return clusters

    # ── Heuristic 6: Deposit-Withdraw Pattern (NEW — CEX tracing) ─

    def detect_deposit_withdraw_clusters(
        self,
        exchange_deposit_map: dict[str, list[str]] | None = None,
        exchange_withdrawal_map: dict[str, list[str]] | None = None,
    ) -> list[Cluster]:
        """
        When funds flow: wallet_A → exchange deposit → exchange hot wallet →
        exchange withdrawal → wallet_B, link wallet_A and wallet_B to same entity.

        Args:
            exchange_deposit_map: {exchange_hot_wallet: [depositing_addresses]}
            exchange_withdrawal_map: {exchange_hot_wallet: [withdrawal_addresses]}
        """
        if not exchange_deposit_map or not exchange_withdrawal_map:
            # Without external CEX data, we can't do this heuristic
            return []

        clusters = []
        for hot_wallet in set(exchange_deposit_map.keys()) & set(exchange_withdrawal_map.keys()):
            depositors = {a.lower() for a in exchange_deposit_map[hot_wallet]}
            withdrawers = {a.lower() for a in exchange_withdrawal_map[hot_wallet]}

            if len(depositors) < 2 or len(withdrawers) < 2:
                continue

            # Link all depositors and withdrawers that share the same hot wallet
            all_addrs = depositors | withdrawers
            al = list(all_addrs)
            for i in range(len(al)):
                for j in range(i + 1, len(al)):
                    pair = tuple(sorted([al[i], al[j]]))
                    self._pair_heuristics[pair].add("deposit_withdraw")
                    self._heuristic_links["deposit_withdraw"].add(f"{pair[0]}:{pair[1]}")
                    self.uf.union(al[i], al[j])

            c = Cluster(f"cex_flow_{hot_wallet[:12]}")
            c.wallets = all_addrs
            c.detection_methods = ["deposit_withdraw"]
            c.confidence = 0.80
            c.center_wallet = hot_wallet.lower()
            clusters.append(c)

        return clusters

    # ── Run all heuristics ────────────────────────────────────────

    def run_all_heuristics(
        self,
        gas_payer_map: dict[str, list[str]] | None = None,
        exchange_deposit_map: dict[str, list[str]] | None = None,
        exchange_withdrawal_map: dict[str, list[str]] | None = None,
    ) -> list[Cluster]:
        """Run all 6 heuristics and return merged clusters."""
        all_clusters = []

        all_clusters.extend(self.detect_temporal_clusters())
        all_clusters.extend(self.detect_counterparty_clusters())
        all_clusters.extend(self.detect_behavioral_clusters())
        all_clusters.extend(self.detect_funding_clusters())
        all_clusters.extend(self.detect_gas_payer_clusters(gas_payer_map))
        all_clusters.extend(self.detect_deposit_withdraw_clusters(exchange_deposit_map, exchange_withdrawal_map))

        # The Union-Find already merged everything — now extract the final clusters
        final = self._build_final_clusters()
        return final

    def _build_final_clusters(self) -> list[Cluster]:
        """Extract final merged clusters from Union-Find with accumulated heuristics."""
        raw = self.uf.get_all_clusters()
        clusters = []

        for root, members in raw.items():
            if len(members) < 2:
                continue

            # Collect all heuristics that contributed to this cluster
            all_heuristics: set[str] = set()
            members_list = list(members)
            for i in range(len(members_list)):
                for j in range(i + 1, len(members_list)):
                    pair = tuple(sorted([members_list[i], members_list[j]]))
                    all_heuristics |= self._pair_heuristics.get(pair, set())

            # Confidence = weighted sum of contributing heuristics
            total_weight = sum(HEURISTIC_WEIGHTS.get(h, 0.5) for h in all_heuristics)
            # Normalize: max confidence when all 6 heuristics contribute
            max_possible = sum(HEURISTIC_WEIGHTS.values())
            confidence = min(0.99, total_weight / max_possible) if max_possible else 0.5

            c = Cluster(f"entity_{root[:16]}")
            c.wallets = members
            c.confidence = round(confidence, 3)
            c.detection_methods = sorted(all_heuristics)
            c.center_wallet = self._find_center(members)
            clusters.append(c)

        return clusters

    # ── Helpers ───────────────────────────────────────────────────

    def _fingerprint_similarity(self, fp1: dict, fp2: dict) -> float:
        scores = []
        if fp1["avg_size"] > 0 and fp2["avg_size"] > 0:
            scores.append(min(fp1["avg_size"], fp2["avg_size"]) / max(fp1["avg_size"], fp2["avg_size"]))
        if fp1["frequency"] > 0 and fp2["frequency"] > 0:
            scores.append(min(fp1["frequency"], fp2["frequency"]) / max(fp1["frequency"], fp2["frequency"]))
        h1, h2 = fp1["preferred_hours"], fp2["preferred_hours"]
        if h1 and h2:
            scores.append(len(h1 & h2) / len(h1 | h2))
        if fp1["program_diversity"] > 0 and fp2["program_diversity"] > 0:
            scores.append(
                min(fp1["program_diversity"], fp2["program_diversity"])
                / max(fp1["program_diversity"], fp2["program_diversity"])
            )
        return sum(scores) / len(scores) if scores else 0

    def _find_center(self, wallets: set[str]) -> str | None:
        if not wallets:
            return None
        best, best_count = None, 0
        for w in wallets:
            if w in self.wallets:
                count = len(self.wallets[w].unique_counterparties & wallets)
                if count > best_count:
                    best, best_count = w, count
        return best or next(iter(wallets))

    def get_cluster_for_wallet(self, address: str) -> Cluster | None:
        """Get the cluster containing a specific wallet."""
        root = self.uf.find(address.lower())
        members = self.uf.get_members(address.lower())
        if len(members) < 2:
            return None
        c = Cluster(f"entity_{root[:16]}")
        c.wallets = members
        c.center_wallet = root
        return c

    def get_stats(self) -> dict:
        """Return engine statistics."""
        clusters = self.uf.get_all_clusters()
        multi = {k: v for k, v in clusters.items() if len(v) >= 2}
        return {
            "total_wallets": self.uf.total_nodes,
            "total_clusters": len(multi),
            "total_singletons": len(clusters) - len(multi),
            "largest_cluster": max((len(v) for v in multi.values()), default=0),
            "heuristic_links": {k: len(v) for k, v in self._heuristic_links.items()},
        }