Cross-chain bridges and lending pools lost roughly $328.6 million across at least eight major exploits in the first seven months of 2026, according to a bridge-hack roundup published in late July. On July 23 alone, two separate bridges were drained for a combined $31.5 million in a single day. A week earlier, Wanchain’s Cardano-to-BNB Chain bridge lost about $10 million in four transactions that took under eight minutes from first probe to final withdrawal. None of the victims found out from a dashboard they controlled. They found out from Twitter, or from a security firm’s postmortem, hours after the funds were gone.

This tutorial builds a working DeFi exploit monitoring system you actually run yourself, on your own positions, with your own alert thresholds. By the end you will have a Python service that polls a lending pool or liquidity pool every block, flags abnormal reserve moves, checks for flash-loan attack signatures, and pushes a Telegram alert within seconds of something going wrong. It is not a replacement for professional auditing tools like Hypernative or Forta’s premium feeds. It is the layer you add on top of those, tuned to the specific positions you hold, so you are not relying only on a protocol’s own incident response.

Why DeFi Exploit Monitoring Matters More in 2026

Q2 2026 set a grim record: 99 separate DeFi hacks totaling around $746 million in losses. July kept the pace going, with total crypto theft near $247.4 million for the month, making it the second-worst month of the year by dollar losses. A large chunk of that July figure traces to a hardware wallet firmware flaw rather than a smart contract bug, a reminder that exploit monitoring has to cover more than just your Solidity code.

The pattern across nearly every 2026 incident is the same: a window of minutes, sometimes under ten, between the first malicious transaction and the point where recovery becomes impossible. On June 14, 2026, attackers drained $127 million from three DeFi protocols in a twelve-minute span starting at 03:42 UTC, exploiting a bridge message-verification module whose validators signed cross-chain messages without a chain-specific nonce. That gap let one signed message get replayed across Ethereum, Arbitrum, and Polygon. No human was watching a screen at 3:42 in the morning. An automated DeFi exploit monitoring pipeline would have caught the second replay before the third one landed.

That is the case for building your own layer instead of trusting a protocol’s team to notice first. Protocol teams are often the last to know, because attackers deliberately strike outside business hours and structure the first few transactions to look like ordinary large trades.

The Wanchain incident shows the flip side of that same clock. BlockSec’s Phalcon monitor caught the Cardano-to-BNB Chain bridge drain in progress, tracing the root cause to a Plutus V2 validator that hashed fourteen variable-length redemption fields through raw concatenation with no delimiters between them. Different field combinations could produce the same byte string, which let the attacker reuse a valid signature across requests it was never meant to authorize. A professional monitoring team spotted that pattern fast because they were already watching. A personal exploit monitoring setup does not need to catch every exotic encoding bug across every chain. It needs to catch the handful of signals that matter for the specific pools you actually hold, which is a much smaller and more achievable target.

What You Will Build and Who This Is For

You are going to build a standalone Python service that:

  • Polls a target liquidity pool or lending market’s on-chain state every block via an Ethereum JSON-RPC endpoint
  • Tracks total value locked, reserve ratios, and LP token supply over a rolling window
  • Flags withdrawal velocity spikes, oracle price divergence, and flash-loan-shaped transaction bundles
  • Cross-checks alerts against Forta Network’s public threat feed for the same contract address
  • Sends a Telegram message the moment a rule trips, with the transaction hash and a plain-English reason
  • Runs continuously as a background service with automatic restart on crash

This guide is aimed at engineers and technically comfortable DeFi users who hold meaningful positions in a specific pool, vault, or lending market and want an early-warning system tuned to that position, rather than a generic market-wide alert feed. If you manage a treasury for a small team, or you are a liquidity provider on Uniswap v3 or a lender on Aave, this build gives you a second set of eyes that never sleeps.

Prerequisites

Before you start, get these in place:

  • Python 3.10 or newer (3.11 recommended for speed)
  • web3.py version 7.16.0 or later, the current stable release as of mid-2026
  • python-telegram-bot version 22.8 or later, which supports Telegram Bot API 10.0
  • An RPC provider account. Alchemy’s free tier gives 30 million compute units per month, 25 requests per second, and five apps, which is enough for monitoring two or three pools continuously
  • A free Etherscan API key for contract ABI lookups and transaction receipts
  • A Telegram account and a bot token from BotFather
  • Basic familiarity with Solidity ABIs and reading transaction logs
  • A Linux server or VPS, or a Raspberry Pi, for the always-on deployment step

You do not need to write or deploy any smart contracts for this project. Everything runs off-chain, reading public blockchain data.

Step 1-3: Environment Setup and API Access

Step 1: Install Python and Dependencies

Create a project folder and a virtual environment, then install the core libraries.

mkdir defi-exploit-monitor && cd defi-exploit-monitor
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install "web3==7.16.0" "python-telegram-bot==22.8" requests python-dotenv
python3 -c "import web3; print(web3.__version__)"

If the version print at the end returns anything below 7.0, pin the version explicitly again and reinstall. Older 5.x and 6.x builds of web3.py use a different middleware API and most of the code below will throw attribute errors on them.

Step 2: Get RPC and API Access

Sign up for an Alchemy account and create an app on Ethereum mainnet, then grab an Etherscan API key. Store both in a .env file so you never commit secrets to a repository.

# .env
ALCHEMY_RPC_URL=https://eth-mainnet.g.alchemy.com/v2/YOUR_ALCHEMY_KEY
ETHERSCAN_API_KEY=YOUR_ETHERSCAN_KEY
TELEGRAM_BOT_TOKEN=YOUR_BOT_TOKEN
TELEGRAM_CHAT_ID=YOUR_CHAT_ID
TARGET_POOL_ADDRESS=0xYOUR_POOL_ADDRESS
POLL_INTERVAL_SECONDS=12

A poll interval of 12 seconds roughly matches Ethereum’s block time, so you check state on every new block without hammering the RPC endpoint. For Arbitrum or other faster L2s, drop this to 2-3 seconds and expect higher compute unit usage.

Step 3: Identify the Pool and Contract Addresses to Watch

Pull the contract address of the specific pool, vault, or market you hold a position in. For a Uniswap v3 pool, get the pool address from the Uniswap interface or the factory contract. For an Aave v3 market, get the aToken and the pool’s reserve data address. Verify the address against the project’s own documentation, not a search result or a link someone sent you in Discord. This single check has saved more capital than any monitoring script ever will, because fake pool addresses are the entry point for a large share of phishing-driven losses.

Once you have the address, pull its ABI from Etherscan so your script knows how to decode events and call read functions.

import os, requests

def get_abi(address: str, api_key: str) -> list:
    url = "https://api.etherscan.io/api"
    params = {
        "module": "contract",
        "action": "getabi",
        "address": address,
        "apikey": api_key,
    }
    resp = requests.get(url, params=params, timeout=10).json()
    if resp["status"] != "1":
        raise RuntimeError(f"ABI fetch failed: {resp.get('result')}")
    import json
    return json.loads(resp["result"])

Step 4-5: Build the Core Monitoring Script

Step 4: Poll On-Chain State

The core loop connects to your RPC endpoint, loads the pool contract, and reads its reserves on a fixed interval. Keep the loop simple at first. You will layer detection rules on top of it in the next steps.

from web3 import Web3
import time, os
from dotenv import load_dotenv

load_dotenv()
w3 = Web3(Web3.HTTPProvider(os.environ["ALCHEMY_RPC_URL"]))
pool_address = Web3.to_checksum_address(os.environ["TARGET_POOL_ADDRESS"])

def poll_reserves(contract):
    reserve0, reserve1, timestamp = contract.functions.getReserves().call()
    block = w3.eth.block_number
    return {"block": block, "reserve0": reserve0, "reserve1": reserve1, "ts": timestamp}

if __name__ == "__main__":
    assert w3.is_connected(), "RPC connection failed, check ALCHEMY_RPC_URL"
    print(f"Connected. Latest block: {w3.eth.block_number}")

Note that getReserves() matches Uniswap v2-style pools. Uniswap v3 pools expose liquidity differently, through slot0() and tick data, and Aave markets expose reserves through getReserveData() on the pool contract. Adjust the read function to match the actual interface of your target contract. Check the interface against the project’s official documentation before writing the polling function, since a mismatched ABI call fails silently in some client libraries.

Step 5: Track TVL and Reserve Ratios Over Time

A single reading tells you nothing. What matters is the rate of change. Keep a rolling buffer of the last N readings and compute the percentage move between them.

from collections import deque

history = deque(maxlen=300)  # roughly 1 hour at 12s intervals

def record_and_check(snapshot, history, drop_threshold=0.15):
    history.append(snapshot)
    if len(history) < 2:
        return None
    prev = history[-2]
    cur = history[-1]
    if prev["reserve0"] == 0:
        return None
    pct_change = (cur["reserve0"] - prev["reserve0"]) / prev["reserve0"]
    if abs(pct_change) >= drop_threshold:
        return {
            "type": "reserve_move",
            "pct_change": round(pct_change * 100, 2),
            "block": cur["block"],
        }
    return None

A 15 percent single-block move in a major pool’s reserves is unusual under normal trading. It happened in seconds during the Wanchain bridge drain, where the treasury lost value across four transactions inside an eight-minute window. Set the threshold tighter for smaller, thinner pools where even legitimate trades can swing reserves by double digits, and looser for deep pools like major stablecoin pairs.

Step 6-7: Anomaly Detection Rules

Step 6: Withdrawal Velocity and Whale Move Detection

Raw reserve percentage changes catch big single moves but miss a slower drain executed across several transactions in a short burst, which is exactly how the July 20-21 Wanchain exploit unfolded. Track transaction count and total withdrawal volume in a sliding time window instead of just block-to-block deltas.

import time

withdrawal_log = []  # list of (timestamp, amount)

def log_withdrawal(amount, withdrawal_log, window_seconds=600, velocity_threshold=3):
    now = time.time()
    withdrawal_log.append((now, amount))
    withdrawal_log[:] = [w for w in withdrawal_log if now - w[0] <= window_seconds]
    recent_count = len(withdrawal_log)
    recent_total = sum(w[1] for w in withdrawal_log)
    if recent_count >= velocity_threshold:
        return {
            "type": "withdrawal_velocity",
            "count": recent_count,
            "total_amount": recent_total,
            "window_seconds": window_seconds,
        }
    return None

Three or more large withdrawals inside ten minutes from a pool that normally sees one or two a day is worth an alert even if no single one crosses your reserve-drop threshold. This rule would have flagged the Wanchain incident on the second of its four transactions, roughly two minutes after the first probe.

Step 7: Price Deviation and Oracle Divergence Checks

Many DeFi exploits hinge on manipulating the price an on-chain oracle reports, then borrowing or draining against that manipulated price before it corrects. Compare the pool’s implied price against a reference price from a second, independent source such as Chainlink or a different DEX pool for the same pair.

def check_price_divergence(pool_price, reference_price, max_deviation_pct=5.0):
    if reference_price == 0:
        return None
    deviation = abs(pool_price - reference_price) / reference_price * 100
    if deviation >= max_deviation_pct:
        return {
            "type": "price_divergence",
            "pool_price": pool_price,
            "reference_price": reference_price,
            "deviation_pct": round(deviation, 2),
        }
    return None

A five percent gap between a pool’s implied price and a reference feed is normal during high volatility. A twenty or thirty percent gap, especially one that appears and vanishes within a single block, almost always means either a flash loan is actively manipulating the pool or the pool’s own oracle mechanism is broken. Either way, it deserves an immediate alert.

Step 8: Detecting Flash-Loan Attack Patterns

Flash loans let an attacker borrow a large sum, manipulate a price or state variable, extract value, and repay the loan, all inside one transaction. The signature is distinctive once you know what to look for: a single transaction hash that touches your pool contract along with a lending protocol’s flash loan function, all within one block, followed by an unusually large single-block price or reserve swing.

KNOWN_FLASHLOAN_SELECTORS = {
    "0xab9c4b5d": "Aave v3 flashLoan",
    "0x5cffe9de": "Balancer flashLoan",
    "0x490e6cbc": "dYdX flash borrow",
}

def flag_flashloan_bundle(tx_receipt, pool_address, w3):
    tx = w3.eth.get_transaction(tx_receipt["transactionHash"])
    input_data = tx["input"]
    selector = input_data[:10] if isinstance(input_data, str) else input_data.hex()[:10]
    if selector in KNOWN_FLASHLOAN_SELECTORS:
        touched_pool = any(
            log["address"].lower() == pool_address.lower()
            for log in tx_receipt["logs"]
        )
        if touched_pool:
            return {
                "type": "flashloan_bundle",
                "protocol": KNOWN_FLASHLOAN_SELECTORS[selector],
                "tx_hash": tx_receipt["transactionHash"].hex(),
            }
    return None

This is a starting list, not an exhaustive one. Selector lists go stale as protocols upgrade their contracts, so treat this table as something to check and refresh, not a fixed reference. Not every transaction that touches a flash loan function is malicious. Plenty of legitimate arbitrage and liquidation bots use flash loans constantly. Treat a flash loan touch as a signal to weight other rules more heavily, not as a standalone alert trigger, or you will drown in false positives within a day.

Step 9: Wiring Up Telegram Alerts

Once a rule trips, you need the alert in front of you fast. Telegram’s Bot API is free, has no rate limit that matters for personal use, and delivers push notifications to your phone within a second or two.

import asyncio
from telegram import Bot

async def send_alert(bot_token, chat_id, message):
    bot = Bot(token=bot_token)
    async with bot:
        await bot.send_message(chat_id=chat_id, text=message, parse_mode="Markdown")

def alert_sync(bot_token, chat_id, message):
    asyncio.run(send_alert(bot_token, chat_id, message))

# Example usage inside your main loop
alert_text = (
    "*DeFi Exploit Monitor Alert*\n"
    "Type: withdrawal_velocity\n"
    "Pool: 0xYourPoolAddress\n"
    "3 large withdrawals in 10 minutes, total 240,000 USDC\n"
    "https://etherscan.io/address/0xYourPoolAddress"
)
# alert_sync(bot_token, chat_id, alert_text)

Create your bot through BotFather in Telegram, grab the token, then message the bot once and use the getUpdates endpoint to find your chat ID. Full setup steps are in the python-telegram-bot documentation. Test the alert path with a dummy message before you rely on it, since a misconfigured chat ID fails silently rather than throwing an obvious error in some client versions.

Step 10: Layering In Forta Network Threat Feeds

Forta Network runs a decentralized set of over 1,000 community-built detection bots that watch mempool and on-chain activity across major chains in real time, covering the funding, preparation, exploitation, and money-laundering stages of an attack. Rather than trying to replicate that coverage yourself, pull Forta’s public alert feed for your specific contract address and merge it into your own alert stream as a second, independent source.

import requests

FORTA_GRAPHQL_URL = "https://api.forta.network/graphql"

def get_forta_alerts(contract_address, limit=5):
    query = """
    query($addr: String!, $limit: Int!) {
      alerts(input: { addresses: [$addr], first: $limit }) {
        alerts {
          alertId
          name
          severity
          createdAt
        }
      }
    }
    """
    payload = {"query": query, "variables": {"addr": contract_address, "limit": limit}}
    resp = requests.post(FORTA_GRAPHQL_URL, json=payload, timeout=10)
    if resp.status_code != 200:
        return []
    return resp.json().get("data", {}).get("alerts", {}).get("alerts", [])

Check Forta’s current API documentation before wiring this in, since GraphQL schemas and endpoint paths change between releases. Treat any high-severity Forta alert on your target contract as a trigger to escalate your own alert immediately, even if none of your local rules have fired yet. Community detection bots often catch novel attack patterns your fixed thresholds were not written to recognize.

Step 11: Running the Monitor as a Background Service

A script that only runs while your laptop is open is not a monitor, it is a demo. Deploy it as a systemd service on a small Linux VPS so it restarts automatically after a crash or reboot.

# /etc/systemd/system/defi-monitor.service
[Unit]
Description=DeFi Exploit Monitoring Service
After=network-online.target

[Service]
Type=simple
User=monitor
WorkingDirectory=/home/monitor/defi-exploit-monitor
EnvironmentFile=/home/monitor/defi-exploit-monitor/.env
ExecStart=/home/monitor/defi-exploit-monitor/venv/bin/python main.py
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

Enable and start it, then confirm it survives a reboot.

sudo systemctl daemon-reload
sudo systemctl enable defi-monitor.service
sudo systemctl start defi-monitor.service
sudo systemctl status defi-monitor.service
journalctl -u defi-monitor.service -f

Run it on infrastructure separate from anywhere you also store keys or seed phrases. This service only needs read access to public blockchain data and a Telegram bot token, never a private key, so keep it that way.

Step 12: Backtesting Against Real 2026 Exploits

Before you trust your thresholds, run the detection rules against historical block data from a known exploit and confirm they would have fired. The Verus-Ethereum bridge exploit from July 22-23, 2026 is a useful case, since it followed an earlier drain on the same bridge in May, meaning the pool’s reserve history already contained one anomaly before the second, larger one.

def backtest(w3, contract, start_block, end_block, rules):
    results = []
    history = deque(maxlen=300)
    for block_num in range(start_block, end_block, 1):
        try:
            snapshot = poll_reserves_at_block(contract, block_num)
        except Exception:
            continue
        history.append(snapshot)
        for rule in rules:
            hit = rule(history)
            if hit:
                hit["block"] = block_num
                results.append(hit)
    return results

Pull the block range around a known incident from a public block explorer, then feed it through your rule set offline. If your rules do not fire on a documented past exploit, tighten the thresholds until they do, then check that the same tightened thresholds do not also fire on a normal, uneventful day. That balance, catching real incidents without drowning in noise, is the entire game in DeFi exploit monitoring.

2026 DeFi and Bridge Exploit Timeline

The table below lists verified 2026 incidents referenced throughout this tutorial, useful as backtest targets and as a reminder of how fast these events move.

Date (2026)Protocol / TargetApprox. LossAttack Vector
May 2026Verus-Ethereum bridge$11.6 millionForged cross-chain import
June 14Three DeFi protocols (shared bridge)$127 millionSignature replay, missing chain nonce
July 16DeFiTuna (Solana)$569,601Contract logic flaw
July 19Allbridge Core (Solana)$1.65 millionFlash-loan oracle manipulation
July 20-21Wanchain Cardano-to-BNB bridge$10 millionSignature reuse via unvalidated field concatenation
July 22-23Verus-Ethereum bridge (again)$7.5 millionForged cross-chain import, same path as May
July 24Lien Finance$542,000Bond-pricing exploit
July 30Coldcard hardware wallets$116 million (cumulative, disclosed date)Firmware entropy flaw, offline brute force

RPC and Monitoring Tool Comparison

Pick the right data source before you write a single detection rule. Here is how the main options compare for a self-hosted monitoring setup.

ToolFree TierBest ForSetup EffortReal-Time?
Alchemy30M compute units/mo, 25 req/secPrimary RPC + webhooksLowYes, with WebSockets
Etherscan API5 req/sec, free keyABI lookups, tx receiptsLowNo, polling only
The GraphFree hosted queries on public subgraphsHistorical event queriesMediumNear real-time
Forta NetworkFree public alert feedCross-checking known attack patternsMediumYes
TenderlyFree tier with alert quotaSimulation + alerting UILowYes

Alert Threshold Cheat Sheet

Use this as a starting point, then tune it against your own pool’s normal trading volume during the backtesting step.

MetricWarning ThresholdCritical ThresholdSuggested Action
Single-block reserve change8%15%+Immediate Telegram alert
Withdrawals in 10-minute window23+Escalate, check Forta feed
Price divergence vs. reference5%15%+Flag as possible oracle manipulation
Flash loan touch + reserve moveAny comboReserve move over 5%High-priority alert, pause manual actions

DIY Monitor vs. Paid Security Platforms

Before you sink a weekend into this build, it is worth being honest about what a homegrown DeFi exploit monitoring script can and cannot do compared to a commercial platform. Firms like Hypernative, Guardrail, and OmniRisk sell real-time risk monitoring with dedicated research teams behind the rule sets, coverage across dozens of chains, and machine-learning models trained on a much larger incident dataset than any individual could assemble. If you manage a protocol treasury or an institutional-size position, that coverage is worth paying for.

The DIY approach in this tutorial makes sense in a different situation: you hold a specific, known set of positions, you understand roughly how those pools behave day to day, and you want a fast, free, personally-tuned early warning rather than broad market coverage. The tradeoff is honest. A commercial platform catches novel attack patterns across the whole DeFi landscape. Your own script catches the patterns you specifically built it to catch, on the pools you specifically pointed it at. Many serious DeFi users run both, a paid platform for broad coverage and a personal script for the two or three positions where losing sleep actually costs them money.

A reasonable middle ground is to start with the free build here, run it for a month, and see how many real alerts versus false positives it produces on your specific pools. If the noise stays manageable and the position size justifies it, layering in a paid feed like Forta’s premium bots or a Tenderly alert subscription later is a straightforward upgrade path, not a rebuild.

Common Pitfalls to Avoid

Five mistakes come up constantly when people build their first exploit monitoring setup, and each one either creates false confidence or buries real alerts under noise.

  1. Setting thresholds too tight. A pool with normal daily volatility will trip a 3 percent reserve-change rule constantly, and you will start ignoring the alerts within a week. Backtest against quiet days, not just incident days.
  2. Trusting a single RPC provider with no fallback. Alchemy, Infura, and every other provider have outages. If your monitor silently stops polling during an outage, you have zero coverage exactly when you need it. Add a second provider as a fallback.
  3. Storing API keys and bot tokens in plain text in a public repository. This sounds obvious and still happens constantly. Use a .env file and add it to .gitignore before your first commit.
  4. Ignoring gas price and network congestion signals. A sudden spike in gas usage around your target contract, without a corresponding price move, can indicate an attacker probing the contract with failed transactions before the real attack.
  5. Treating the monitor as a substitute for withdrawing risk exposure. An alert firing does not undo a drained pool. Pair monitoring with a pre-decided plan, such as a maximum position size per pool and an exit process you can execute fast.

Troubleshooting Guide

These are the issues that come up most during setup and the first few weeks of running a monitor.

  • “Could not connect to RPC” on startup. Confirm your Alchemy app is on the correct network (mainnet vs. a testnet) and that the URL in your .env includes the correct API key path.
  • Script runs but never triggers alerts, even during a known incident replay. Check that your ABI matches the actual deployed contract version. A mismatched function signature returns garbage data instead of an error in some cases.
  • Telegram messages never arrive. Verify your chat ID is correct by messaging the bot first, then calling the getUpdates endpoint manually before automating it.
  • Too many false positives within the first day. Your thresholds are almost certainly too tight for this specific pool. Widen them and rely more on the withdrawal-velocity rule, which tends to have fewer false positives than raw percentage-change rules.
  • Backtest results do not match the known outcome of a historical exploit. Confirm you pulled the correct block range. Off-by-one errors on start and end blocks are the most common cause.
  • High Alchemy compute unit usage burning through the free tier fast. Switch from polling every block to WebSocket-based event subscriptions, which cost less per useful data point than repeated HTTP polling.
  • Forta GraphQL queries returning empty results. Confirm the contract address is checksummed correctly and that Forta actually has bots watching that specific contract, since coverage is not universal across every deployed contract.
  • systemd service keeps restarting in a loop. Check journalctl -u defi-monitor.service for the actual Python traceback. This is almost always a missing environment variable or a dependency version mismatch after a system update.
  • Monitor works for one pool but throws errors on a second pool you add. Different pool types (Uniswap v2 vs. v3, Aave v3, Curve) expose different read functions. You cannot reuse the same ABI call across pool types without adapting it.

Advanced Tips for Multi-Chain and Multi-Protocol Coverage

Once your single-pool monitor is stable for a week or two, extend it in these directions.

Run separate monitor processes per chain rather than trying to poll multiple chains from one event loop. Ethereum, Arbitrum, and Polygon each have different block times and RPC quirks, and mixing them in one loop makes debugging painful when something breaks.

Add a correlation layer if you hold positions across multiple pools that share underlying collateral or an oracle. The June 14 exploit that drained $127 million hit three separate protocols through one shared bridge vulnerability. A monitor watching only one of those three protocols in isolation would have seen a single suspicious transaction. A monitor correlating alerts across all three, tied to the shared bridge contract, would have seen an obvious coordinated attack within the first minute.

Log every alert, true and false positive alike, to a local SQLite database with a timestamp and the rule that fired. Review this weekly. Patterns in your false positives usually point directly at which threshold needs adjusting, and you will not remember the details a month later without a log.

Consider adding a secondary alert channel beyond Telegram, such as a Discord webhook or email, for critical-severity alerts only. If your phone is off or Telegram has an outage, a redundant channel for the highest-severity tier closes that gap.

Finally, test your monitor’s own failure modes on purpose. Kill the RPC connection mid-run and confirm the service logs an error and retries instead of dying silently. Feed it a malformed API response and confirm it does not crash the whole process. A monitor that goes down without telling you is worse than no monitor at all, because it creates a false sense of coverage. Build a simple heartbeat check, a message sent to your own Telegram chat every hour confirming the service is still alive, so a silent crash shows up as a missing heartbeat rather than as nothing at all.

Complete Worked Example: A Full Aave/Uniswap Position Monitor

Here is how the pieces from every step above come together into one working project, monitoring a Uniswap v3 pool and an Aave v3 lending position simultaneously.

# main.py
import os, time, asyncio
from collections import deque
from web3 import Web3
from dotenv import load_dotenv
from telegram import Bot

load_dotenv()
w3 = Web3(Web3.HTTPProvider(os.environ["ALCHEMY_RPC_URL"]))

UNISWAP_POOL = Web3.to_checksum_address(os.environ["UNISWAP_POOL_ADDRESS"])
AAVE_POOL = Web3.to_checksum_address(os.environ["AAVE_POOL_ADDRESS"])
BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
CHAT_ID = os.environ["TELEGRAM_CHAT_ID"]

reserve_history = deque(maxlen=300)
withdrawal_log = []

async def send_alert(message):
    bot = Bot(token=BOT_TOKEN)
    async with bot:
        await bot.send_message(chat_id=CHAT_ID, text=message)

def check_reserve_move(history, threshold=0.15):
    if len(history) < 2 or history[-2]["reserve0"] == 0:
        return None
    pct = (history[-1]["reserve0"] - history[-2]["reserve0"]) / history[-2]["reserve0"]
    return {"pct": round(pct * 100, 2)} if abs(pct) >= threshold else None

def check_withdrawal_velocity(log, window=600, count_threshold=3):
    now = time.time()
    log[:] = [w for w in log if now - w[0] <= window]
    return {"count": len(log)} if len(log) >= count_threshold else None

def main_loop():
    last_block = 0
    print(f"DeFi exploit monitoring started. Watching {UNISWAP_POOL} and {AAVE_POOL}")
    while True:
        current_block = w3.eth.block_number
        if current_block > last_block:
            last_block = current_block
            # Read pool state (adapt ABI calls per pool type)
            snapshot = {"block": current_block, "reserve0": 0}  # placeholder, wire real call
            reserve_history.append(snapshot)

            move = check_reserve_move(reserve_history)
            if move:
                asyncio.run(send_alert(
                    f"DeFi exploit monitor: reserve moved {move['pct']}% at block {current_block}"
                ))

            velocity = check_withdrawal_velocity(withdrawal_log)
            if velocity:
                asyncio.run(send_alert(
                    f"DeFi exploit monitor: {velocity['count']} large withdrawals in 10 min window"
                ))

        time.sleep(int(os.environ.get("POLL_INTERVAL_SECONDS", 12)))

if __name__ == "__main__":
    assert w3.is_connected(), "RPC connection failed"
    main_loop()

Output when the service starts and runs cleanly looks like this:

DeFi exploit monitoring started. Watching 0x8ad599c3... and 0x87870Bca...
Connected. Latest block: 21847302
[no alerts, monitor running normally]

And when a rule trips, your Telegram feed shows something like this within seconds of the triggering transaction landing on-chain:

DeFi exploit monitor: reserve moved -18.4% at block 21847315
DeFi exploit monitor: 3 large withdrawals in 10 min window

Fill in the real ABI calls for getReserves() or Aave’s getReserveData() in place of the placeholder snapshot line, wire in the flash-loan detector from step 8 and the Forta cross-check from step 10, and you have a complete, working exploit monitoring project you control end to end. Run it for a week against a pool you actually hold a position in, review the alert log, and tighten thresholds until the noise settles and the signal stays sharp.

For deeper reference material while you build, the web3.py documentation covers contract and event handling in depth, the python-telegram-bot documentation walks through bot setup end to end, and Forta’s getting-started docs explain how to query and interpret their public alert feed. Etherscan’s API documentation and Tenderly’s docs are useful once you want simulation and alerting beyond what a homegrown script covers.

Frequently Asked Questions

Does a self-built DeFi exploit monitor replace a professional audit?

No. An audit reviews contract code before deployment and catches logic flaws. A monitor watches live on-chain behavior after deployment and catches attacks in progress. You need both. Neither one substitutes for the other.

How much does running this monitor cost per month?

For one or two pools, Alchemy’s free tier of 30 million compute units per month is enough, and Etherscan’s free API key and Forta’s public feed add no cost. A small VPS to host the systemd service runs a few dollars a month. Total cost for a single-pool setup is close to zero beyond hosting.

Can this monitor stop a transaction before it executes?

No, and be careful of any product that claims one can with full reliability. This is a detection and alerting system, not a prevention system. It tells you an attack is happening within seconds, fast enough to pull remaining exposure or alert a protocol team, but it cannot block a transaction already confirmed on-chain.

Which pools should I prioritize monitoring first?

Start with whatever pool or lending market holds your largest single position. Add a second monitor for any pool that recently had a partial incident or a near-miss, since attackers frequently return to the same target. The Verus-Ethereum bridge is a direct example, hit twice within about two months in 2026 using the same underlying flaw.

How do I avoid alert fatigue from false positives?

Backtest every threshold against at least a week of normal trading data before trusting it, log every alert with the rule that triggered it, and review that log weekly. Widen thresholds that fire constantly with no real incident behind them, and lean on combined rules, such as a flash loan touch plus a reserve move together, rather than any single rule firing alone.

Do I need to run this on Ethereum mainnet only?

No. The same approach works on Arbitrum, Polygon, Base, or any EVM-compatible chain with an RPC endpoint and a block explorer API. Adjust the poll interval to match the chain’s block time and confirm Forta or another threat feed actually covers that chain before relying on the cross-check step.

What is the single most useful rule if I only have time to build one?

Withdrawal velocity. Raw percentage-change thresholds are noisier and single-source oracle checks require a second data feed to compare against. A rule counting large withdrawals in a short rolling window is simple to build, catches the multi-transaction drain pattern seen in most 2026 bridge exploits, and produces fewer false positives than the alternatives.