Stablecoin Depeg Alert in Python With Chainlink and Telegram

The message you want on your phone looks like this: USDT DEPEG: Chainlink reports $0.9987 (updated 0.8h ago). One message when the price leaves its band, one when it comes back, nothing in between. This post builds that as a single Python file that reads USDC, USDT and DAI from Chainlink in one request and runs from cron.

Prerequisites

• Python 3.10 or newer and pip install requests.
• An evmquery API key from the free tier, saved as EVMQUERY_API_KEY in a .env file next to the script.
• A Telegram bot token from @BotFather and the chat id of the chat it should write to, saved as TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID in the same .env.
• A machine with cron. A small VPS, a Raspberry Pi or your laptop all work, as long as it is on when the job fires.

If you only want the feed addresses for other pairs, the Chainlink price feed tool lists them per chain. This post is about what to do with the numbers once you have them.

What a Chainlink stablecoin feed can and cannot tell you

A Chainlink price feed is not a stream. A new round is written to the chain when the aggregated price moves more than the deviation threshold away from the last on-chain answer, or when the heartbeat interval has passed, whichever comes first. Chainlink publishes both numbers per feed in its feed directory.

For the Ethereum stablecoin feeds the directory says:

Read that table as a statement about blind spots. If USDT trades at $0.9980 and the last on-chain answer was $0.9995, the move is 0.15%, below the threshold, so the feed stays silent for up to 24 hours. An alert that fires at “price below $0.999” would be waiting on a feed that is not allowed to report that yet.

So the band has to be wider than the deviation threshold. I use twice the threshold, 0.5% for these three, which means a feed has to publish at least one fresh round before the alert can fire. This is a rule of thumb, not a Chainlink recommendation. If you want to catch smaller moves, a Chainlink feed is the wrong instrument and you need a DEX price instead.

With that band the script alerts when the price drops below $0.995 or rises above $1.005. The readings from the run below sit well inside it, and well inside the zone where the feeds stay silent.

One detail about the feed directory: it contains several entries named USDC / USD and USDT / USD. The extra ones have paths ending in -svr or -shared-svr and no ens name. They are separate proxy contracts. I used the entry whose ens is usdc-usd (and likewise for the other two), which matches the address on the tool page.

Reading all three feeds in one request

Each feed is an AggregatorV3Interface proxy, and the call you want is latestRoundData(), which returns the price as answer and the time of the last update as updatedAt. Here is the whole expression for the three feeds:

curl -s -X POST https://api.evmquery.com/api/v1/query \
  -H "x-api-key: $EVMQUERY_API_KEY" \
  -H "content-type: application/json" \
  -d '{
    "chain": "evm_ethereum",
    "schema": { "contracts": {
      "usdc": { "address": "0x8fFfFfd4AfB6115b954Bd326cbe7B4BA576818f6" },
      "usdt": { "address": "0x3E7d1eAB13ad0104d2750B8863b489D65364e32D" },
      "dai":  { "address": "0xAed0c38402a5d19df6E4c03F4E2DceD6e29c1ee9" }
    } },
    "expression": "[usdc.latestRoundData().answer, usdc.latestRoundData().updatedAt, usdt.latestRoundData().answer, usdt.latestRoundData().updatedAt, dai.latestRoundData().answer, dai.latestRoundData().updatedAt]"
  }'

When I ran it, result.value came back as this list:

["99975591", "1794207155", "99872736", "1794235799", "99949875", "1794236855"]

The expression returns a flat list of integers: pull answer and updatedAt out of each round and put those in the list. Although it calls latestRoundData() twice per feed, the response metadata said totalCalls: 3, because identical calls are deduplicated and you pay for one read per feed.

The values are strings, because int256 can exceed what a JSON number holds safely. The price is answer / 10**8 since these feeds use 8 decimals, so 99872736 is $0.99872736. Your Python code should convert with int() before dividing.

If you would rather not use the API, the same read through web3.py or viem means encoding three calls and wrapping them in Multicall3 yourself. For a cron script that is more code than the alert logic.

Deciding when a feed is lying by omission

Two conditions make a reading untrustworthy for an alert, and they need different messages.

The first is the depeg itself: the price is further than the band from $1. The second is staleness: updatedAt is older than the heartbeat should allow. A stale USDT feed that still says $1.0000 is worse than a depeg alert, because it fails silently. I treat a feed as stale once its age exceeds the heartbeat plus one hour of grace. The Chainlink staleness and decimals post covers why that comparison matters and the extra sequencer check you need on Base. On Ethereum mainnet there is no sequencer, so the age check is all there is.

Stale is checked first. A depeg reading from a feed that has not updated in two days is not evidence of anything.

The script

This is the complete file. Save it as depeg_alert.py:

import json
import os
import time
from pathlib import Path

import requests

API_URL = "https://api.evmquery.com/api/v1/query"
STATE_FILE = Path(__file__).with_name("state.json")

# symbol: (proxy address, heartbeat in seconds, decimals, deviation threshold in %)
# Copied from Chainlink's feed directory, Ethereum mainnet entries.
FEEDS = {
    "USDC": ("0x8fFfFfd4AfB6115b954Bd326cbe7B4BA576818f6", 82800, 8, 0.25),
    "USDT": ("0x3E7d1eAB13ad0104d2750B8863b489D65364e32D", 86400, 8, 0.25),
    "DAI": ("0xAed0c38402a5d19df6E4c03F4E2DceD6e29c1ee9", 3600, 8, 0.25),
}
STALE_GRACE = 3600  # seconds allowed past the heartbeat before a feed counts as stale


def send_telegram(text: str) -> None:
    token = os.environ["TELEGRAM_BOT_TOKEN"]
    resp = requests.post(
        f"https://api.telegram.org/bot{token}/sendMessage",
        json={"chat_id": os.environ["TELEGRAM_CHAT_ID"], "text": text},
        timeout=10,
    )
    resp.raise_for_status()


def read_feeds() -> dict[str, tuple[float, int]]:
    """Return {symbol: (price, seconds since last update)} from one request."""
    contracts = {s.lower(): {"address": cfg[0]} for s, cfg in FEEDS.items()}
    expression = "[" + ", ".join(
        f"{s.lower()}.latestRoundData().answer, {s.lower()}.latestRoundData().updatedAt"
        for s in FEEDS
    ) + "]"
    resp = requests.post(
        API_URL,
        headers={"x-api-key": os.environ["EVMQUERY_API_KEY"]},
        json={
            "chain": "evm_ethereum",
            "schema": {"contracts": contracts},
            "expression": expression,
        },
        timeout=30,
    )
    resp.raise_for_status()
    values = resp.json()["result"]["value"]  # int256 values arrive as strings

    now = time.time()
    feeds = {}
    for i, symbol in enumerate(FEEDS):
        answer, updated_at = int(values[2 * i]), int(values[2 * i + 1])
        decimals = FEEDS[symbol][2]
        feeds[symbol] = (answer / 10**decimals, int(now - updated_at))
    return feeds


def classify(symbol: str, price: float, age: int) -> str:
    _, heartbeat, _, deviation = FEEDS[symbol]
    if age > heartbeat + STALE_GRACE:
        return "stale"
    if abs(price - 1) > 2 * deviation / 100:  # band is twice the feed's deviation threshold
        return "depeg"
    return "ok"


def main() -> None:
    state = json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else {}
    for symbol, (price, age) in read_feeds().items():
        status = classify(symbol, price, age)
        previous = state.get(symbol, "ok")
        if status != previous:
            hours = age / 3600
            if status == "depeg":
                message = f"{symbol} DEPEG: Chainlink reports ${price:.4f} (updated {hours:.1f}h ago)"
            elif status == "stale":
                message = f"{symbol} feed is stale: last update {hours:.1f}h ago, price ${price:.4f}"
            elif previous == "stale":
                message = f"{symbol} feed is updating again: ${price:.4f}, last update {hours:.1f}h ago"
            else:
                message = f"{symbol} is back inside its band: ${price:.4f}"
            send_telegram(message)
            state[symbol] = status
            STATE_FILE.write_text(json.dumps(state))


if __name__ == "__main__":
    main()

The lines that deserve an explanation:

• contracts and expression are generated from FEEDS, so adding a coin is one new dict entry. The keys are lowercased because they become variable names inside the expression.
• previous = state.get(symbol, "ok") means the very first run is silent when everything is healthy. If you start the script during a depeg, previous defaults to ok and you get the alert immediately.
• The state file holds one word per coin: ok, depeg or stale. A message goes out only when the word changes, so a four-day depeg produces one alert and one recovery, not 1,152 messages at a five-minute interval. A feed that goes from stale back to ok gets its own “updating again” message, since the price did not move, the feed did.
• The message is sent before the state is written, and the write happens per coin. If Telegram fails, the script crashes with the status and the same message is retried on the next run.
• age is measured against your machine’s clock, not the block timestamp. With a heartbeat measured in hours, a few seconds of drift does not matter.
• resp.raise_for_status() lets a failed API call crash the script, and cron will report that. A monitor that swallows its own errors looks healthy while it is blind.

Running it every five minutes

Put the three secrets in a .env file next to the script as plain KEY=value lines:

EVMQUERY_API_KEY=...
TELEGRAM_BOT_TOKEN=...
TELEGRAM_CHAT_ID=...

Then add this to crontab -e, using the interpreter path from which python3:

*/5 * * * * cd /home/you/depeg-alert && set -a; . ./.env; set +a; /usr/bin/python3 depeg_alert.py >> run.log 2>&1

Cron has no PATH or shell profile, so the line uses an absolute interpreter path and loads the variables itself. The cd makes .env and run.log resolve; state.json is found from the script’s own location. Each run costs one request, so five minutes is cheap, and the feeds update far less often than that.

Or skip the cron: run it as a monitor

The price band also works as an evmquery monitor. This is the downside check for USDT with the 0.5% band from above, on evm_ethereum:

formatUnits(usdt.latestRoundData().answer, 8) < 0.995

Use a condition trigger. It fires once when the price drops below $0.995 and stays quiet while it remains below. The upside alert is a second monitor. Keep the staleness check in the script. Your send_telegram function can sit behind a small webhook receiver. evmquery retries failed deliveries and pauses the monitor with an alert after repeated errors. The Free plan covers 3 monitors with 15-minute checks.

Testing it without waiting for a depeg

I did not want to wait for a real depeg, so I tested the state machine against live feeds in a separate harness that replaces send_telegram with print and points STATE_FILE at a scratch file. With the real configuration the three feeds read USDC $0.9998, USDT $0.9987 and DAI $0.9995, all inside the band, and no message was sent.

Then I tightened the USDT deviation from 0.25 to 0.05 in memory, which makes its band 0.1%. USDT at $0.9987 is outside that, and the next run printed the depeg message from the top of this post. Running it again printed nothing. Restoring the setting printed USDT is back inside its band: $0.9987. Setting STALE_GRACE to a large negative number made all three feeds report stale, and restoring it sent three “feed is updating again” messages.

The ages were the interesting part. At the time of the test the USDC feed had last updated 8.7 hours earlier, while DAI’s was 29 minutes old. USDC was nowhere near its 23 hour heartbeat and its price had not moved 0.25%, so it simply had nothing to say. A feed that has been quiet for hours is the normal state for a healthy stablecoin, and that is the practical meaning of the blind spot above.

Feed directory and first-run pitfalls

• Duplicate feed names. Pick the directory entry with an ens name, not the -svr variants, and confirm the address against the tool page.
• The upside alert. The band is symmetric, so a price above $1.005 alerts too. The directory flags the USDC and USDT feeds as stablecoinCapped, so the downside is the direction to watch.
• First run during an incident. A missing state file is treated as ok, so you will be told about a depeg that started before the script did.

Extending it to stETH

stETH is not a dollar stablecoin, but it is a pegged asset, and the same script covers it. The directory lists an STETH / ETH feed at 0x86392dC19c0b719886221c78AB11eb8Cf5c52812 with 18 decimals, a 0.5% deviation threshold and a 24 hour heartbeat. Add one line to FEEDS:

    "stETH": ("0x86392dC19c0b719886221c78AB11eb8Cf5c52812", 86400, 18, 0.5),

I ran it that way and the feed read 0.9997 ETH per stETH. The band scales on its own: twice a 0.5% threshold is 1%. The price here is in ETH, not dollars, so price - 1 measures the stETH discount against ETH. For the staking exchange rate itself, which is a different number, see the Lido stETH exchange rate post.

Next steps

• Chainlink staleness, sequencer uptime and decimals for the checks this script only touches on.
• Lido stETH exchange rate via API for the rate behind the stETH extension.
• evmquery for automation if you want this running in n8n or another scheduler instead of cron.
• Chainlink price feed addresses by chain for feeds on Base, BNB Chain and Polygon.
• Monitors to run the price band as a hosted condition that sends an email or a webhook to your Telegram bot code.