File size: 8,337 Bytes
6993919 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | """
Entity Resolver — Cross-chain entity linking.
================================================================
Links wallets across chains using:
1. Bridge mappings (wallet A on chain X → wallet B on chain Y via bridge tx)
2. Same address heuristic (EVM addresses are portable across EVM chains)
3. Gas payer networks (one wallet paying for deploys across chains)
4. Deposit-withdraw patterns (CEX flow tracing)
Produces entity_id (deterministic hash) for each resolved entity group.
"""
import hashlib
import logging
from typing import Any
from .storage import WalletStorage, get_storage
logger = logging.getLogger("wallet_memory.entity_resolver")
class EntityResolver:
"""
Cross-chain entity linking. Resolves wallets to entities.
"""
def __init__(self, storage: WalletStorage | None = None):
self.storage = storage or get_storage()
self._entity_cache: dict[str, str] = {} # addr:chain -> entity_id
async def resolve(self, address: str, chain: str) -> dict[str, Any]:
"""
Resolve a wallet address to its cross-chain entity.
Returns:
{
"entity_id": str,
"addresses": [{"address": str, "chain": str, "confidence": float, "method": str}],
"primary_chain": str,
"total_chains": int,
}
"""
addr = address.lower().strip()
entity_id = self._make_entity_id(addr, chain)
# Already resolved in storage?
stored_entity = await self.storage.get_entity_for_wallet(addr, chain)
if stored_entity and stored_entity.get("entity_id"):
entity_id = stored_entity["entity_id"]
addresses = [{"address": addr, "chain": chain, "confidence": 1.0, "method": "direct"}]
# ── Method 1: Same EVM address heuristic ──────────────────
# EVM addresses are portable: same private key works on all EVM chains
evm_chains = [
"ethereum",
"bsc",
"polygon",
"arbitrum",
"optimism",
"base",
"avalanche",
"fantom",
]
from app.chain_registry import is_evm
if is_evm(chain):
for evm_chain in evm_chains:
if evm_chain != chain:
# Same address = same entity on all EVM chains
addresses.append(
{
"address": addr,
"chain": evm_chain,
"confidence": 0.90,
"method": "evm_address_portability",
}
)
# ── Method 2: Bridge mappings ────────────────────────────
try:
bridges = await self.storage.get_bridge_links(addr, chain)
for bridge in bridges:
dest_addr = bridge.get("dest_address", "").lower()
dest_chain = bridge.get("dest_chain", "")
if dest_addr and dest_chain:
addresses.append(
{
"address": dest_addr,
"chain": dest_chain,
"confidence": bridge.get("confidence", 0.8),
"method": f"bridge_{bridge.get('bridge_name', 'unknown')}",
}
)
except Exception as e:
logger.debug(f"Bridge resolution failed for {addr}: {e}")
# ── Method 3: Check existing entity membership ───────────
try:
if stored_entity and stored_entity.get("entity_id"):
members = await self.storage.get_entity_wallets(stored_entity["entity_id"])
for m in members:
m_addr = m.get("wallet_address", "").lower()
m_chain = m.get("chain_id", "")
existing = {(a["address"], a["chain"]) for a in addresses}
if (m_addr, m_chain) not in existing:
addresses.append(
{
"address": m_addr,
"chain": m_chain,
"confidence": m.get("confidence_score", 0.5),
"method": "entity_membership",
}
)
except Exception as e:
logger.debug(f"Entity membership lookup failed for {addr}: {e}")
# Persist new entity links
for addr_entry in addresses:
if addr_entry["method"] != "direct": # Don't re-save the original
try:
await self.storage.save_entity_link(
entity_id=entity_id,
wallet_address=addr_entry["address"],
chain_id=addr_entry["chain"],
heuristic_type=addr_entry["method"],
confidence=addr_entry["confidence"],
)
except Exception as e:
logger.debug(f"Entity link persist failed: {e}")
# Determine primary chain (most active chain for this entity)
chain_counts: dict[str, int] = {}
for a in addresses:
chain_counts[a["chain"]] = chain_counts.get(a["chain"], 0) + 1
primary_chain = max(chain_counts, key=chain_counts.get) if chain_counts else chain
# Deduplicate
seen = set()
unique_addresses = []
for a in addresses:
key = (a["address"], a["chain"])
if key not in seen:
seen.add(key)
unique_addresses.append(a)
return {
"entity_id": entity_id,
"addresses": unique_addresses,
"primary_chain": primary_chain,
"total_chains": len({a["chain"] for a in unique_addresses}),
}
async def link_via_bridge(
self,
source_addr: str,
source_chain: str,
dest_addr: str,
dest_chain: str,
bridge_name: str,
tx_hash: str,
confidence: float = 1.0,
) -> str:
"""Explicitly link two addresses via a bridge transaction."""
# Save the bridge mapping
await self.storage.save_bridge_link(
source_addr, source_chain, dest_addr, dest_chain, bridge_name, tx_hash, confidence
)
# Create/merge entity
entity_id = self._make_entity_id(source_addr, source_chain)
# Link both addresses
await self.storage.save_entity_link(entity_id, source_addr, source_chain, "bridge_mapping", confidence)
await self.storage.save_entity_link(entity_id, dest_addr, dest_chain, "bridge_mapping", confidence)
# Check if dest_addr already has an entity and merge
dest_entity = await self.storage.get_entity_for_wallet(dest_addr, dest_chain)
if dest_entity and dest_entity.get("entity_id") and dest_entity["entity_id"] != entity_id:
# Merge entities: both are the same operator
dest_members = await self.storage.get_entity_wallets(dest_entity["entity_id"])
for m in dest_members:
await self.storage.save_entity_link(
entity_id,
m["wallet_address"],
m["chain_id"],
"entity_merge",
m.get("confidence_score", 0.5),
)
logger.info(f"Merged entity {dest_entity['entity_id']} into {entity_id}")
return entity_id
@staticmethod
def _make_entity_id(address: str, chain: str) -> str:
"""Deterministic entity ID from address + chain."""
raw = f"{address.lower()}:{chain}"
return "ent_" + hashlib.sha256(raw.encode()).hexdigest()[:16]
# ── Global singleton ────────────────────────────────────────────
_resolver: EntityResolver | None = None
def get_entity_resolver() -> EntityResolver:
global _resolver
if _resolver is None:
_resolver = EntityResolver()
return _resolver
|