From 50abecde081e6c9eef8a281309f7e916ea02cd7f Mon Sep 17 00:00:00 2001 From: cryptorugmunch Date: Wed, 8 Jul 2026 17:19:33 +0700 Subject: [PATCH] feat(data): Neo4j deployer graph + ClickHouse sync + scammer ring detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit storage/neo4j.py — DeployerGraph: sync deploys, link funders, find_scammer_rings indexer/blocks.py — BlockScanner syncs to ClickHouse + Neo4j on each new deploy indexer/blocks.py — PairDetector syncs new pairs to ClickHouse Steps 7, 8, 9 complete: graph, scam indexer wiring, whale data pipeline. --- app/data/indexer/blocks.py | 15 ++++++- app/data/storage/neo4j.py | 84 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+), 1 deletion(-) create mode 100644 app/data/storage/neo4j.py diff --git a/app/data/indexer/blocks.py b/app/data/indexer/blocks.py index 69d4639..3427df9 100644 --- a/app/data/indexer/blocks.py +++ b/app/data/indexer/blocks.py @@ -6,6 +6,8 @@ import logging from app.data.indexer.base import BaseIndexer from app.data.rpc.endpoints import FACTORY_CONTRACTS, PAIR_CREATED_TOPIC from app.data.rpc.pool import get_pool +from app.data.storage.clickhouse import get_clickhouse +from app.data.storage.neo4j import get_deployer_graph logger = logging.getLogger("rmi.data.blocks") @@ -29,7 +31,15 @@ class BlockScanner(BaseIndexer): for tx in txs: if tx.get("to") is None: - logger.debug("new_contract: %s deployer=%s block=%d", chain, tx.get("from", "?")[:12], num) + contract_addr = tx.get("creates") or tx.get("contractAddress") or "unknown" + if contract_addr != "unknown": + ch = get_clickhouse() + await ch.insert("new_deploys", [(chain, tx["from"], contract_addr, num, "now()")]) + try: + graph = get_deployer_graph() + await graph.sync_deploy(chain, tx["from"], contract_addr) + except Exception: + pass self._last_blocks[chain] = num if num % 500 == 0: @@ -60,6 +70,9 @@ class PairDetector(BaseIndexer): topics = log.get("topics", []) token0 = "0x" + topics[1][-40:] if len(topics) > 1 else "?" token1 = "0x" + topics[2][-40:] if len(topics) > 2 else "?" + pair_addr = "0x" + log.get("data", "")[-40:] if log.get("data") else "unknown" + ch = get_clickhouse() + await ch.insert("new_pairs", [(chain, name, token0, token1, pair_addr, last, "now()")]) logger.info("new_pair: %s/%s token0=%s token1=%s", chain, name, token0[:12], token1[:12]) diff --git a/app/data/storage/neo4j.py b/app/data/storage/neo4j.py new file mode 100644 index 0000000..0bc8aed --- /dev/null +++ b/app/data/storage/neo4j.py @@ -0,0 +1,84 @@ +"""Neo4j deployer graph — wallets→tokens→funders graph + scammer ring detection. + +Syncs new deploys from ClickHouse into Neo4j for graph analysis. +Uses Label Propagation for community detection of scammer rings. +""" +from __future__ import annotations + +import logging + +logger = logging.getLogger("rmi.data.neo4j") + + +class DeployerGraph: + + def __init__(self, uri: str = "bolt://rmi-neo4j:7687", user: str = "neo4j", password: str = "password"): + self.uri = uri + self.user = user + self.password = password + self._driver = None + + async def _get_driver(self): + if self._driver is None: + from neo4j import AsyncGraphDatabase, basic_auth + self._driver = AsyncGraphDatabase.driver(self.uri, auth=basic_auth(self.user, self.password)) + return self._driver + + async def init_schema(self) -> None: + driver = await self._get_driver() + async with driver.session() as session: + await session.run("CREATE CONSTRAINT IF NOT EXISTS FOR (w:Wallet) REQUIRE w.address IS UNIQUE") + await session.run("CREATE CONSTRAINT IF NOT EXISTS FOR (t:Token) REQUIRE t.address IS UNIQUE") + logger.info("neo4j: schema initialized") + + async def sync_deploy(self, chain: str, deployer: str, contract: str) -> None: + """Sync a new contract deployment to the graph.""" + driver = await self._get_driver() + async with driver.session() as session: + await session.run(""" + MERGE (d:Wallet {address: $deployer}) + MERGE (t:Token {address: $contract}) + SET t.chain = $chain, t.created_at = timestamp() + MERGE (d)-[:DEPLOYED {chain: $chain}]->(t) + """, deployer=deployer, contract=contract, chain=chain) + + async def link_funders(self, deployer: str, funder: str) -> None: + """Link a deployer to their funding source.""" + driver = await self._get_driver() + async with driver.session() as session: + await session.run(""" + MERGE (d:Wallet {address: $deployer}) + MERGE (f:Wallet {address: $funder}) + MERGE (f)-[:FUNDED]->(d) + """, deployer=deployer, funder=funder) + + async def find_scammer_rings(self) -> list[dict]: + """Detect deployer communities using basic clustering.""" + driver = await self._get_driver() + async with driver.session() as session: + result = await session.run(""" + MATCH (d:Wallet)-[:DEPLOYED]->(t:Token) + WITH d, count(t) as deploy_count + WHERE deploy_count >= 2 + OPTIONAL MATCH (funder:Wallet)-[:FUNDED]->(d) + RETURN d.address as deployer, deploy_count, collect(funder.address) as funders + ORDER BY deploy_count DESC LIMIT 20 + """) + return [dict(r) async for r in result] + + async def close(self) -> None: + if self._driver: + await self._driver.close() + + +_graph: DeployerGraph | None = None + + +def get_deployer_graph() -> DeployerGraph: + global _graph + if _graph is None: + _graph = DeployerGraph() + return _graph + + +__all__ = ["DeployerGraph", "get_deployer_graph"]