On-Chain Analytics for AI Trading Agents: Reading the Blockchain

Why On-Chain Data Matters for AI Agents

Traditional financial markets obscure order flow behind closed exchange APIs and institutional dark pools. Blockchains expose everything: every trade, every wallet balance change, every pending transaction, every liquidation. For AI trading agents, this transparency is a structural edge that centralized markets simply cannot offer.

On-chain analytics transforms raw blockchain data — millions of transactions per day — into actionable trading signals. Agents that master this data source can anticipate liquidation cascades before they happen, follow smart money wallets in near real-time, detect DEX volume anomalies that precede price moves, and front-run obvious on-chain events with precision timing.

This guide covers the full analytics stack: which data sources to use, how to structure async pipelines, and complete Python implementations for each major signal type. All examples integrate with the Purple Flea infrastructure for agent registration, wallet management, and position execution.

On-Chain Data Sources

Six core platforms provide the majority of actionable on-chain analytics. Each has different strengths, rate limits, and cost structures. Effective agents aggregate across multiple sources rather than relying on any single API.

Setting Up Multi-Source Client

Agents need a unified client abstraction that handles authentication, rate limiting, and fallback across providers. Here is the base client structure:

import asyncio
import aiohttp
import time
from dataclasses import dataclass, field
from typing import Optional, Dict, Any

@dataclass
class RateLimiter:
    calls_per_second: float
    _last_call: float = field(default=0.0, init=False)

    async def acquire(self):
        now = time.monotonic()
        gap = 1.0 / self.calls_per_second
        wait = gap - (now - self._last_call)
        if wait > 0:
            await asyncio.sleep(wait)
        self._last_call = time.monotonic()


class OnChainClient:
    def __init__(self):
        self.session: Optional[aiohttp.ClientSession] = None
        self.rate_limiters = {
            "etherscan": RateLimiter(4.5),     # 5/s limit with buffer
            "dune": RateLimiter(0.5),           # conservative for free tier
            "thegraph": RateLimiter(10.0),
            "alchemy": RateLimiter(25.0),
        }
        self.api_keys = {
            "etherscan": "YOUR_ETHERSCAN_KEY",
            "dune": "YOUR_DUNE_KEY",
            "alchemy": "YOUR_ALCHEMY_KEY",
        }

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=10),
            connector=aiohttp.TCPConnector(limit=50)
        )
        return self

    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()

    async def get(self, provider: str, url: str,
                  params: Dict = None) -> Dict[str, Any]:
        await self.rate_limiters[provider].acquire()
        async with self.session.get(url, params=params) as resp:
            resp.raise_for_status()
            return await resp.json()

Whale Wallet Identification

Whale wallets — addresses controlling more than $1M in a given asset — disproportionately move markets. Tracking their activity gives agents early warning of major sell-offs, accumulation phases, and rotation between assets.

Identification requires two steps: balance threshold scanning (which addresses hold large positions?) and behavioral classification (are they a long-term holder, active trader, exchange hot wallet, or protocol treasury?). Nansen provides pre-classified labels; building your own classification system from raw data is slower but cheaper.

Whale Detection Pipeline

from decimal import Decimal
from enum import Enum

class WalletType(Enum):
    WHALE_HOLDER = "whale_holder"
    SMART_MONEY = "smart_money"
    EXCHANGE_HOT = "exchange_hot"
    PROTOCOL_TREASURY = "protocol_treasury"
    UNKNOWN = "unknown"

@dataclass
class WalletProfile:
    address: str
    balance_usd: float
    wallet_type: WalletType
    avg_hold_days: float
    win_rate_30d: float        # % of trades profitable
    last_active_block: int
    known_label: Optional[str] = None

class WhaleTracker:
    def __init__(self, client: OnChainClient):
        self.client = client
        self.whale_cache: Dict[str, WalletProfile] = {}
        self.WHALE_THRESHOLD_USD = 1_000_000

    async def get_token_holders(self, token_address: str,
                                 top_n: int = 100) -> list[WalletProfile]:
        """Fetch top holders from Etherscan token API."""
        url = "https://api.etherscan.io/api"
        params = {
            "module": "token",
            "action": "tokenholderlist",
            "contractaddress": token_address,
            "page": 1,
            "offset": top_n,
            "apikey": self.client.api_keys["etherscan"]
        }
        data = await self.client.get("etherscan", url, params)
        holders = []
        for h in data.get("result", []):
            profile = WalletProfile(
                address=h["TokenHolderAddress"],
                balance_usd=float(h["TokenHolderQuantity"]) * 0,  # multiply by price
                wallet_type=WalletType.UNKNOWN,
                avg_hold_days=0.0,
                win_rate_30d=0.0,
                last_active_block=0
            )
            holders.append(profile)
        return holders

    async def classify_wallet(self, address: str) -> WalletType:
        """Classify wallet based on transaction patterns."""
        # Known exchange hot wallets (simplified list)
        EXCHANGE_WALLETS = {
            "0x28c6c06298d514db089934071355e5743bf21d60",  # Binance
            "0x21a31ee1afc51d94c2efccaa2092ad1028285549",  # Binance 2
            "0xdfd5293d8e347dfe59e90efd55b2956a1343963d",  # Binance 3
        }
        if address.lower() in EXCHANGE_WALLETS:
            return WalletType.EXCHANGE_HOT

        # Fetch recent tx history to classify behavior
        url = "https://api.etherscan.io/api"
        params = {
            "module": "account",
            "action": "txlist",
            "address": address,
            "page": 1,
            "offset": 50,
            "sort": "desc",
            "apikey": self.client.api_keys["etherscan"]
        }
        data = await self.client.get("etherscan", url, params)
        txs = data.get("result", [])
        if not txs:
            return WalletType.UNKNOWN

        # High frequency = active trader; low frequency = holder
        tx_count_30d = sum(
            1 for tx in txs
            if int(tx["timeStamp"]) > time.time() - 30 * 86400
        )
        return (WalletType.SMART_MONEY if tx_count_30d > 20
                else WalletType.WHALE_HOLDER)

    async def watch_whale_moves(self, addresses: list[str],
                                 from_block: int) -> list[dict]:
        """Poll for new transactions from tracked whale addresses."""
        alerts = []
        for addr in addresses:
            url = "https://api.etherscan.io/api"
            params = {
                "module": "account",
                "action": "tokentx",
                "address": addr,
                "startblock": from_block,
                "sort": "desc",
                "apikey": self.client.api_keys["etherscan"]
            }
            data = await self.client.get("etherscan", url, params)
            for tx in data.get("result", [])[:5]:
                value_eth = float(tx["value"]) / 1e18
                if value_eth > 100:  # only large moves
                    alerts.append({
                        "address": addr,
                        "direction": "in" if tx["to"].lower() == addr.lower() else "out",
                        "value_eth": value_eth,
                        "token": tx["tokenSymbol"],
                        "tx_hash": tx["hash"],
                        "block": int(tx["blockNumber"])
                    })
        return alerts

DEX Volume and Flow Analysis

Decentralized exchanges publish every trade on-chain, including exact amounts, execution price, and the initiating wallet. This makes DEX flow one of the cleanest leading indicators available: abnormal volume in a pair consistently precedes price moves by 2-15 minutes on the underlying asset.

Key signals: volume surge ratio (current vs. rolling 24h average), buy/sell pressure imbalance, and large single-trade outliers. The Graph protocol provides indexed Uniswap V3 swap data via GraphQL — far more structured than raw event logs.

import json

UNISWAP_V3_SUBGRAPH = (
    "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"
)

class DEXFlowAnalyzer:
    def __init__(self, client: OnChainClient):
        self.client = client
        self.volume_baseline: Dict[str, float] = {}

    async def graphql_query(self, query: str,
                             variables: dict = None) -> dict:
        """Execute GraphQL query against The Graph."""
        await self.client.rate_limiters["thegraph"].acquire()
        payload = {"query": query}
        if variables:
            payload["variables"] = variables
        async with self.client.session.post(
            UNISWAP_V3_SUBGRAPH,
            json=payload,
            headers={"Content-Type": "application/json"}
        ) as resp:
            return await resp.json()

    async def get_recent_swaps(self, pool_address: str,
                                limit: int = 100) -> list[dict]:
        """Fetch recent swaps for a specific Uniswap V3 pool."""
        query = """
        query RecentSwaps($pool: String!, $first: Int!) {
          swaps(
            first: $first,
            orderBy: timestamp,
            orderDirection: desc,
            where: { pool: $pool }
          ) {
            timestamp
            amount0
            amount1
            amountUSD
            sender
            origin
            transaction { id }
          }
        }
        """
        data = await self.graphql_query(
            query, {"pool": pool_address.lower(), "first": limit}
        )
        return data.get("data", {}).get("swaps", [])

    async def compute_flow_metrics(self, pool_address: str) -> dict:
        """Compute buy/sell pressure and volume anomaly score."""
        swaps = await self.get_recent_swaps(pool_address, 200)
        if not swaps:
            return {}

        total_volume_usd = sum(float(s["amountUSD"]) for s in swaps)
        buy_volume = sum(
            float(s["amountUSD"]) for s in swaps
            if float(s["amount0"]) < 0   # token0 sold = token1 bought
        )
        sell_volume = total_volume_usd - buy_volume

        buy_pressure = buy_volume / total_volume_usd if total_volume_usd > 0 else 0.5
        baseline = self.volume_baseline.get(pool_address, total_volume_usd)
        volume_ratio = total_volume_usd / baseline if baseline > 0 else 1.0
        self.volume_baseline[pool_address] = (
            baseline * 0.95 + total_volume_usd * 0.05  # EMA update
        )

        # Detect large single trades (whale activity)
        large_trades = [
            s for s in swaps if float(s["amountUSD"]) > 100_000
        ]

        return {
            "pool": pool_address,
            "total_volume_usd": total_volume_usd,
            "buy_pressure": buy_pressure,
            "volume_anomaly_ratio": volume_ratio,
            "large_trade_count": len(large_trades),
            "signal": self._classify_signal(buy_pressure, volume_ratio)
        }

    def _classify_signal(self, buy_pressure: float,
                          volume_ratio: float) -> str:
        if volume_ratio > 2.5 and buy_pressure > 0.65:
            return "STRONG_BUY"
        elif volume_ratio > 2.5 and buy_pressure < 0.35:
            return "STRONG_SELL"
        elif volume_ratio > 1.5 and buy_pressure > 0.6:
            return "MODERATE_BUY"
        elif volume_ratio > 1.5 and buy_pressure < 0.4:
            return "MODERATE_SELL"
        return "NEUTRAL"

Liquidation Cascade Detection

DeFi lending protocols (Aave, Compound, MakerDAO) publish liquidation events on-chain. When collateral values drop, positions become undercollateralized and liquidations trigger. A cascade occurs when liquidations themselves cause price drops, triggering more liquidations in a self-reinforcing loop.

Agents can monitor at-risk positions before they are liquidated, estimate cascade impact, and position accordingly — either shorting ahead of anticipated cascades or buying the fear-driven overshoots.

@dataclass
class LiquidationRisk:
    protocol: str
    user_address: str
    collateral_asset: str
    debt_asset: str
    collateral_usd: float
    debt_usd: float
    health_factor: float          # < 1.0 = liquidatable
    liquidation_threshold_usd: float

class LiquidationMonitor:
    AAVE_V3_SUBGRAPH = (
        "https://api.thegraph.com/subgraphs/name/aave/protocol-v3"
    )

    def __init__(self, client: OnChainClient):
        self.client = client

    async def get_at_risk_positions(self,
                                     max_health_factor: float = 1.15,
                                     min_debt_usd: float = 50_000
                                     ) -> list[LiquidationRisk]:
        """Fetch positions close to liquidation on Aave V3."""
        query = """
        query AtRisk($hf: BigDecimal!, $minDebt: BigDecimal!) {
          users(
            first: 100,
            where: {
              healthFactor_lt: $hf,
              totalBorrowsUSD_gt: $minDebt
            },
            orderBy: healthFactor,
            orderDirection: asc
          ) {
            id
            healthFactor
            totalCollateralUSD
            totalBorrowsUSD
            reserves {
              reserve {
                symbol
                liquidationThreshold
              }
              currentATokenBalance
              currentVariableDebt
            }
          }
        }
        """
        await self.client.rate_limiters["thegraph"].acquire()
        async with self.client.session.post(
            self.AAVE_V3_SUBGRAPH,
            json={"query": query, "variables": {
                "hf": str(max_health_factor),
                "minDebt": str(min_debt_usd)
            }}
        ) as resp:
            data = await resp.json()

        positions = []
        for user in data.get("data", {}).get("users", []):
            positions.append(LiquidationRisk(
                protocol="aave_v3",
                user_address=user["id"],
                collateral_asset="mixed",
                debt_asset="mixed",
                collateral_usd=float(user["totalCollateralUSD"]),
                debt_usd=float(user["totalBorrowsUSD"]),
                health_factor=float(user["healthFactor"]),
                liquidation_threshold_usd=float(user["totalCollateralUSD"]) * 0.8
            ))
        return positions

    def estimate_cascade_impact(self,
                                 at_risk: list[LiquidationRisk],
                                 price_drop_pct: float) -> dict:
        """Estimate additional liquidations from a price drop."""
        newly_liquidatable = []
        for pos in at_risk:
            # Approximate: if collateral drops by price_drop_pct
            new_collateral = pos.collateral_usd * (1 - price_drop_pct)
            new_hf = new_collateral / (pos.debt_usd * 1.05)  # 5% penalty
            if new_hf < 1.0:
                newly_liquidatable.append(pos)

        total_debt_at_risk = sum(p.debt_usd for p in newly_liquidatable)
        return {
            "positions_at_risk": len(newly_liquidatable),
            "total_debt_at_risk_usd": total_debt_at_risk,
            "cascade_probability": min(total_debt_at_risk / 5_000_000, 1.0),
            "recommended_action": (
                "SHORT" if total_debt_at_risk > 2_000_000 else "WATCH"
            )
        }

Smart Money Signal Extraction

Smart money wallets — addresses with documented histories of profitable trades — provide one of the most reliable on-chain signals. The key insight is that these wallets consistently accumulate before price appreciation and distribute before declines, presumably due to informational advantages or superior analysis.

Tracking smart money involves three steps: identify wallets with statistically significant win rates over 90+ days, monitor their new positions in real-time, and weight signals by the wallet's historical accuracy on similar assets.

import statistics

@dataclass
class SmartMoneySignal:
    wallet: str
    asset: str
    direction: str        # "accumulating" or "distributing"
    confidence: float     # 0.0-1.0 based on wallet history
    position_size_usd: float
    detected_block: int

class SmartMoneyTracker:
    # Pre-curated list of historically accurate wallets
    # In production, build this dynamically from backtested performance
    SMART_MONEY_WATCHLIST = [
        "0x9d6d5e9cfef8bb2bd42b8e4b04c22b0b1d8a2f97",
        "0x3e3a1e2e3d4b9e0f1a2b3c4d5e6f7a8b9c0d1e2f",
        # ... extend with your own tracked wallets
    ]

    def __init__(self, client: OnChainClient):
        self.client = client
        self.wallet_histories: Dict[str, list[dict]] = {}

    async def score_wallet_accuracy(self, address: str,
                                     lookback_days: int = 90) -> float:
        """Calculate win rate for a wallet over lookback period."""
        # Simplified: count profitable vs losing positions
        # Production: integrate with DEX trade history and token price data
        trades = self.wallet_histories.get(address, [])
        if len(trades) < 10:
            return 0.0  # insufficient data

        wins = sum(1 for t in trades if t.get("pnl_pct", 0) > 0)
        return wins / len(trades)

    async def detect_new_positions(self,
                                    from_block: int) -> list[SmartMoneySignal]:
        """Scan watchlist for new ERC-20 token accumulation."""
        signals = []
        tasks = [
            self._check_wallet_activity(addr, from_block)
            for addr in self.SMART_MONEY_WATCHLIST
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for result in results:
            if isinstance(result, list):
                signals.extend(result)
        return signals

    async def _check_wallet_activity(self, address: str,
                                      from_block: int) -> list[SmartMoneySignal]:
        """Check a single wallet for new token activity."""
        url = "https://api.etherscan.io/api"
        params = {
            "module": "account",
            "action": "tokentx",
            "address": address,
            "startblock": from_block,
            "sort": "desc",
            "apikey": self.client.api_keys["etherscan"]
        }
        data = await self.client.get("etherscan", url, params)
        signals = []
        seen_tokens = {}

        for tx in data.get("result", []):
            token = tx["tokenSymbol"]
            value = float(tx["value"]) / (10 ** int(tx["tokenDecimal"]))
            is_buy = tx["to"].lower() == address.lower()

            if token not in seen_tokens:
                seen_tokens[token] = {"buys": 0, "sells": 0}
            seen_tokens[token]["buys" if is_buy else "sells"] += value

        for token, flows in seen_tokens.items():
            net = flows["buys"] - flows["sells"]
            if abs(net) > 1000:  # minimum position size filter
                accuracy = await self.score_wallet_accuracy(address)
                signals.append(SmartMoneySignal(
                    wallet=address,
                    asset=token,
                    direction="accumulating" if net > 0 else "distributing",
                    confidence=accuracy,
                    position_size_usd=abs(net),  # needs price lookup
                    detected_block=from_block
                ))
        return signals

Mempool Pending Transaction Analysis

The mempool — the pool of pending transactions waiting to be included in a block — provides a 5-15 second preview of on-chain activity. Large pending trades, liquidations, and governance votes are visible before they execute, giving agents a narrow but real predictive window.

Mempool access requires a direct node connection (Alchemy or Infura provide this via WebSocket). The key events to watch: large pending token approvals (often precede swaps), pending governance votes, large ETH transfers to exchange addresses, and pending liquidation calls.

import websockets
import json
from collections import deque

class MempoolMonitor:
    ALCHEMY_WS = "wss://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"

    def __init__(self, min_value_eth: float = 50.0):
        self.min_value_eth = min_value_eth
        self.pending_queue: deque = deque(maxlen=1000)
        self.signal_callbacks = []

    def on_signal(self, callback):
        """Register callback for high-value pending tx detection."""
        self.signal_callbacks.append(callback)
        return self

    async def subscribe(self):
        """Connect to Alchemy WebSocket and subscribe to pending tx."""
        async with websockets.connect(self.ALCHEMY_WS) as ws:
            # Subscribe to pending transactions
            await ws.send(json.dumps({
                "id": 1,
                "method": "eth_subscribe",
                "params": ["newPendingTransactions"]
            }))
            response = await ws.recv()
            sub_id = json.loads(response)["result"]
            print(f"Subscribed to mempool: {sub_id}")

            async for message in ws:
                data = json.loads(message)
                if "params" in data:
                    tx_hash = data["params"]["result"]
                    await self._process_pending(ws, tx_hash)

    async def _process_pending(self, ws, tx_hash: str):
        """Fetch and analyze a pending transaction."""
        await ws.send(json.dumps({
            "id": 2,
            "method": "eth_getTransactionByHash",
            "params": [tx_hash]
        }))
        # In practice, batch these requests
        # and parse value/to/input data fields

    def classify_pending_tx(self, tx: dict) -> Optional[dict]:
        """Classify a pending tx into signal categories."""
        value_eth = int(tx.get("value", "0x0"), 16) / 1e18
        if value_eth < self.min_value_eth:
            return None

        # Large ETH transfer to known exchange = selling pressure
        to_addr = tx.get("to", "").lower()
        EXCHANGE_HOT_WALLETS = {
            "0x28c6c06298d514db089934071355e5743bf21d60",
        }
        if to_addr in EXCHANGE_HOT_WALLETS:
            return {
                "type": "EXCHANGE_INFLOW",
                "value_eth": value_eth,
                "signal": "SELL_PRESSURE",
                "urgency": "HIGH",
                "tx_hash": tx["hash"]
            }

        # Input data starts with swap selector?
        input_data = tx.get("input", "0x")
        UNISWAP_SELECTORS = {"0x38ed1739", "0x7ff36ab5", "0x18cbafe5"}
        if input_data[:10] in UNISWAP_SELECTORS and value_eth > 100:
            return {
                "type": "LARGE_DEX_SWAP",
                "value_eth": value_eth,
                "signal": "PRICE_IMPACT_IMMINENT",
                "urgency": "VERY_HIGH",
                "tx_hash": tx["hash"]
            }
        return None

Cross-Chain Flow Tracking

Capital flows between blockchains via bridges — and bridge inflows/outflows reveal where liquidity is moving. When large amounts of USDC bridge from Ethereum to a Layer 2 network, DEX liquidity and trading activity on that L2 tends to increase within hours. When capital bridges back to mainnet, it often signals distribution of profits.

Key bridges to monitor: Arbitrum bridge, Optimism bridge, Polygon PoS bridge, and Stargate Finance (cross-chain USDC routing). Each publishes deposit/withdrawal events that can be indexed via their respective subgraphs or event logs.

STARGATE_SUBGRAPH = (
    "https://api.thegraph.com/subgraphs/name/stargate-finance/stargate"
)

class CrossChainFlowTracker:
    CHAIN_IDS = {
        1: "ethereum",
        42161: "arbitrum",
        10: "optimism",
        137: "polygon",
        8453: "base",
    }

    def __init__(self, client: OnChainClient):
        self.client = client

    async def get_bridge_flows(self, hours: int = 24) -> dict:
        """Get net capital flows between chains via Stargate."""
        query = """
        query BridgeFlows($since: Int!) {
          swaps(
            where: { timestamp_gt: $since },
            orderBy: timestamp,
            orderDirection: desc,
            first: 500
          ) {
            srcChainId
            dstChainId
            amountUSD
            timestamp
          }
        }
        """
        since = int(time.time()) - hours * 3600
        await self.client.rate_limiters["thegraph"].acquire()
        async with self.client.session.post(
            STARGATE_SUBGRAPH,
            json={"query": query, "variables": {"since": since}}
        ) as resp:
            data = await resp.json()

        flows = {}
        for swap in data.get("data", {}).get("swaps", []):
            src = self.CHAIN_IDS.get(swap["srcChainId"], str(swap["srcChainId"]))
            dst = self.CHAIN_IDS.get(swap["dstChainId"], str(swap["dstChainId"]))
            key = f"{src}->{dst}"
            flows[key] = flows.get(key, 0) + float(swap["amountUSD"])

        # Net flow per chain
        chain_net = {}
        for route, volume in flows.items():
            src, dst = route.split("->")
            chain_net[src] = chain_net.get(src, 0) - volume
            chain_net[dst] = chain_net.get(dst, 0) + volume

        return {
            "route_volumes": flows,
            "chain_net_flows": chain_net,
            "largest_inflow_chain": max(chain_net, key=chain_net.get),
            "largest_outflow_chain": min(chain_net, key=chain_net.get)
        }

    async def interpret_flows(self, flows: dict) -> list[str]:
        """Generate trading signals from cross-chain flow data."""
        signals = []
        net = flows.get("chain_net_flows", {})
        for chain, net_usd in net.items():
            if net_usd > 10_000_000:
                signals.append(
                    f"BULLISH {chain.upper()}: +${net_usd/1e6:.1f}M net inflow"
                )
            elif net_usd < -10_000_000:
                signals.append(
                    f"BEARISH {chain.upper()}: ${net_usd/1e6:.1f}M net outflow"
                )
        return signals

Python OnChainAnalyticsAgent — Full Implementation

The complete agent ties all signals together into an async pipeline with configurable signal weighting, Purple Flea integration for position execution, and a cooldown system to prevent over-trading on correlated signals.

import asyncio
import aiohttp
import time
import json
import logging
from dataclasses import dataclass, field, asdict
from typing import Optional, Dict, Any, List
from enum import Enum

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger("OnChainAgent")

PURPLE_FLEA_API = "https://purpleflea.com/api/v1"
AGENT_API_KEY = "YOUR_PURPLE_FLEA_KEY"

@dataclass
class TradeSignal:
    source: str
    asset: str
    direction: str       # "long" or "short"
    confidence: float    # 0.0-1.0
    size_pct: float      # % of balance to allocate
    reasoning: str
    expires_at: float    # Unix timestamp

class OnChainAnalyticsAgent:
    """
    Autonomous trading agent using on-chain signals.
    Aggregates whale moves, DEX flow, liquidation risk,
    smart money, mempool, and cross-chain flows into
    actionable trade signals.
    """
    SIGNAL_WEIGHTS = {
        "whale": 0.25,
        "dex_flow": 0.20,
        "liquidation": 0.20,
        "smart_money": 0.25,
        "mempool": 0.10,
    }
    CONFIDENCE_THRESHOLD = 0.55   # minimum to trade
    MAX_CONCURRENT_SIGNALS = 3
    COOLDOWN_SECONDS = 300        # 5 min between same-asset trades

    def __init__(self, agent_id: str, api_key: str):
        self.agent_id = agent_id
        self.api_key = api_key
        self.active_signals: List[TradeSignal] = []
        self.last_trade: Dict[str, float] = {}
        self.session: Optional[aiohttp.ClientSession] = None

        # Initialize sub-analyzers
        self.dex_analyzer = None
        self.whale_tracker = None
        self.liq_monitor = None
        self.sm_tracker = None
        self.flow_tracker = None

    async def initialize(self):
        self.session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=aiohttp.ClientTimeout(total=15)
        )
        client = OnChainClient()
        await client.__aenter__()
        self.dex_analyzer = DEXFlowAnalyzer(client)
        self.whale_tracker = WhaleTracker(client)
        self.liq_monitor = LiquidationMonitor(client)
        self.sm_tracker = SmartMoneyTracker(client)
        self.flow_tracker = CrossChainFlowTracker(client)
        logger.info(f"Agent {self.agent_id} initialized on-chain analytics pipeline")

    async def collect_signals(self, target_assets: list[str]) -> List[TradeSignal]:
        """Run all signal collectors in parallel."""
        current_block = await self._get_current_block()

        tasks = {
            "dex_flow": self._collect_dex_signals(target_assets),
            "whale": self._collect_whale_signals(current_block),
            "liquidation": self._collect_liquidation_signals(target_assets),
            "smart_money": self._collect_smart_money_signals(current_block),
            "cross_chain": self._collect_flow_signals(),
        }
        results = await asyncio.gather(*tasks.values(), return_exceptions=True)
        all_signals = []
        for key, result in zip(tasks.keys(), results):
            if isinstance(result, list):
                all_signals.extend(result)
                logger.info(f"[{key}] collected {len(result)} signals")
            elif isinstance(result, Exception):
                logger.warning(f"[{key}] failed: {result}")
        return all_signals

    async def _collect_dex_signals(self,
                                    assets: list[str]) -> List[TradeSignal]:
        signals = []
        # ETH/USDC Uniswap V3 pool (0.3% fee tier)
        POOLS = {
            "ETH": "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8",
            "BTC": "0x4585fe77225b41b697c938b018e2ac67ac5a20c0",
        }
        for asset in assets:
            pool = POOLS.get(asset)
            if not pool:
                continue
            metrics = await self.dex_analyzer.compute_flow_metrics(pool)
            if metrics.get("signal") in ("STRONG_BUY", "STRONG_SELL"):
                signals.append(TradeSignal(
                    source="dex_flow",
                    asset=asset,
                    direction="long" if "BUY" in metrics["signal"] else "short",
                    confidence=self.SIGNAL_WEIGHTS["dex_flow"] * (
                        0.9 if "STRONG" in metrics["signal"] else 0.7
                    ),
                    size_pct=2.5,
                    reasoning=(
                        f"DEX volume {metrics['volume_anomaly_ratio']:.1f}x baseline, "
                        f"buy pressure {metrics['buy_pressure']:.0%}"
                    ),
                    expires_at=time.time() + 900   # 15 min expiry
                ))
        return signals

    async def _collect_liquidation_signals(self,
                                            assets: list[str]) -> List[TradeSignal]:
        at_risk = await self.liq_monitor.get_at_risk_positions(1.08, 100_000)
        if not at_risk:
            return []
        cascade = self.liq_monitor.estimate_cascade_impact(at_risk, 0.05)
        if cascade["cascade_probability"] > 0.5:
            return [TradeSignal(
                source="liquidation",
                asset="ETH",
                direction="short",
                confidence=cascade["cascade_probability"] * 0.9,
                size_pct=3.0,
                reasoning=(
                    f"{cascade['positions_at_risk']} positions at risk, "
                    f"${cascade['total_debt_at_risk_usd']/1e6:.1f}M debt exposed"
                ),
                expires_at=time.time() + 600
            )]
        return []

    async def _collect_whale_signals(self,
                                      from_block: int) -> List[TradeSignal]:
        moves = await self.whale_tracker.watch_whale_moves(
            WhaleTracker.SMART_MONEY_WATCHLIST[:5] if hasattr(WhaleTracker, 'SMART_MONEY_WATCHLIST')
            else [], from_block
        )
        signals = []
        for move in moves:
            if move["value_eth"] > 500:
                signals.append(TradeSignal(
                    source="whale",
                    asset=move["token"],
                    direction="short" if move["direction"] == "out" else "long",
                    confidence=0.6,
                    size_pct=2.0,
                    reasoning=(
                        f"Whale {move['address'][:10]}... moved "
                        f"{move['value_eth']:.0f} ETH worth of {move['token']}"
                    ),
                    expires_at=time.time() + 1800
                ))
        return signals

    async def _collect_smart_money_signals(self,
                                            from_block: int) -> List[TradeSignal]:
        sm_signals = await self.sm_tracker.detect_new_positions(from_block)
        return [
            TradeSignal(
                source="smart_money",
                asset=s.asset,
                direction="long" if s.direction == "accumulating" else "short",
                confidence=s.confidence * self.SIGNAL_WEIGHTS["smart_money"],
                size_pct=3.5,
                reasoning=(
                    f"Smart money {s.wallet[:10]}... "
                    f"{s.direction} ${s.position_size_usd:,.0f} of {s.asset}"
                ),
                expires_at=time.time() + 3600
            )
            for s in sm_signals if s.confidence > 0.6
        ]

    async def _collect_flow_signals(self) -> List[TradeSignal]:
        flows = await self.flow_tracker.get_bridge_flows(6)
        interpretations = await self.flow_tracker.interpret_flows(flows)
        signals = []
        for interp in interpretations:
            if "BULLISH" in interp:
                chain = interp.split()[1].lower()
                signals.append(TradeSignal(
                    source="cross_chain",
                    asset="ETH",
                    direction="long",
                    confidence=0.45,
                    size_pct=1.5,
                    reasoning=interp,
                    expires_at=time.time() + 7200
                ))
        return signals

    def aggregate_signals(self, signals: List[TradeSignal],
                          asset: str) -> Optional[TradeSignal]:
        """Combine signals for the same asset into a consensus view."""
        asset_signals = [s for s in signals if s.asset == asset]
        if not asset_signals:
            return None

        long_conf = sum(
            s.confidence for s in asset_signals if s.direction == "long"
        )
        short_conf = sum(
            s.confidence for s in asset_signals if s.direction == "short"
        )
        net_confidence = long_conf - short_conf

        if abs(net_confidence) < self.CONFIDENCE_THRESHOLD:
            return None

        dominant_direction = "long" if net_confidence > 0 else "short"
        dominant_signals = [
            s for s in asset_signals if s.direction == dominant_direction
        ]
        avg_size = sum(s.size_pct for s in dominant_signals) / len(dominant_signals)

        return TradeSignal(
            source="aggregated",
            asset=asset,
            direction=dominant_direction,
            confidence=abs(net_confidence),
            size_pct=min(avg_size, 5.0),  # cap at 5% of balance
            reasoning=f"{len(dominant_signals)} signals aligned: " + "; ".join(
                s.reasoning[:50] for s in dominant_signals[:3]
            ),
            expires_at=min(s.expires_at for s in dominant_signals)
        )

    async def execute_signal(self, signal: TradeSignal) -> dict:
        """Submit trade via Purple Flea perpetuals API."""
        last = self.last_trade.get(signal.asset, 0)
        if time.time() - last < self.COOLDOWN_SECONDS:
            return {"status": "skipped", "reason": "cooldown"}

        payload = {
            "agent_id": self.agent_id,
            "asset": signal.asset,
            "direction": signal.direction,
            "size_pct": signal.size_pct,
            "source": signal.source,
            "reasoning": signal.reasoning
        }
        async with self.session.post(
            f"{PURPLE_FLEA_API}/trades",
            json=payload
        ) as resp:
            result = await resp.json()
            if resp.status == 200:
                self.last_trade[signal.asset] = time.time()
                logger.info(
                    f"Trade executed: {signal.direction} {signal.asset} "
                    f"({signal.size_pct:.1f}%) conf={signal.confidence:.2f}"
                )
            return result

    async def _get_current_block(self) -> int:
        """Get latest Ethereum block number via Etherscan."""
        url = "https://api.etherscan.io/api"
        async with self.session.get(url, params={
            "module": "proxy",
            "action": "eth_blockNumber",
            "apikey": AGENT_API_KEY
        }) as resp:
            data = await resp.json()
            return int(data.get("result", "0x0"), 16)

    async def run(self, target_assets: list[str],
                  scan_interval: int = 60):
        """Main agent loop."""
        await self.initialize()
        logger.info(
            f"OnChainAnalyticsAgent starting — "
            f"tracking: {', '.join(target_assets)}"
        )
        while True:
            try:
                signals = await self.collect_signals(target_assets)
                logger.info(f"Total signals collected: {len(signals)}")

                for asset in target_assets:
                    consensus = self.aggregate_signals(signals, asset)
                    if consensus:
                        result = await self.execute_signal(consensus)
                        logger.info(f"Execution result: {result}")
                    else:
                        logger.debug(f"No consensus signal for {asset}")

            except Exception as e:
                logger.error(f"Agent loop error: {e}", exc_info=True)

            await asyncio.sleep(scan_interval)


async def main():
    agent = OnChainAnalyticsAgent(
        agent_id="onchain-agent-001",
        api_key=AGENT_API_KEY
    )
    await agent.run(
        target_assets=["ETH", "BTC", "ARB", "OP"],
        scan_interval=60
    )

if __name__ == "__main__":
    asyncio.run(main())

Integration with Purple Flea

Purple Flea provides the execution layer for on-chain analytics agents. The wallet API handles balance management, the perpetuals API handles position entry and exit, and the escrow service enables agent-to-agent coordination for coordinated strategies.

New agents can claim free starting capital from the Purple Flea Faucet to bootstrap their on-chain analytics strategy without initial capital risk. The faucet provides enough to test signal validation across 10-20 trades before committing real funds.

Summary: Building Your On-Chain Edge

On-chain analytics gives AI trading agents structural advantages unavailable in traditional markets. The key components covered in this guide:

• Data sources: Dune, Nansen, Etherscan, The Graph, Alchemy, CryptoQuant — each with specific strengths
• Whale tracking: Identify, classify, and monitor large-balance addresses for early move detection
• DEX flow: Volume anomaly ratios and buy/sell pressure from Uniswap V3 swap data
• Liquidation cascades: Aave health factor monitoring for cascade prediction and execution
• Smart money: Backtested wallet performance scoring with real-time accumulation detection
• Mempool: 5-15 second preview window via WebSocket pending transaction subscription
• Cross-chain flows: Bridge volume analysis for inter-chain capital rotation signals
• Aggregation: Confidence-weighted signal fusion with cooldown protection

The complete OnChainAnalyticsAgent class provides a production-ready foundation. Extend it with your own signal sources, refine the confidence thresholds based on backtesting, and connect to Purple Flea for seamless execution.