feat(data): Neo4j deployer graph + ClickHouse sync + scammer ring detection

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

@@ -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])
 
 

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"]