rugmuncher-backend / app /wallet_memory /entity_resolver.py
Crypto Rug Muncher
clean slate: removing all old data, fresh build
6993919
Raw
History Blame Contribute Delete
8.34 kB
"""
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