DeFi监控仪表盘 用Python实时追踪借贷协议——关键指标与告警逻辑

DeFi监控仪表盘

用Python实时追踪借贷协议——关键指标与告警逻辑

系列十二「DeFi机制原理」第5篇（完结篇）

💡链上数据怎么查？健康因子怎么算？如何搭建实时监控脚本？

⏱阅读时间：约12分钟

1. DeFi数据分析的入口

链上数据是公开的——任何人都可以用RPC节点或第三方API查询。这一篇教你用Python搭建一个DeFi监控脚本，追踪Compound/Aave的关键指标。

两个主流数据获取方式：

●公共RPC + eth_call：免费但速度慢，适合小量数据

●The Graph： subgraph索引服务，查询快，支持复杂筛选

2. 用Web3.py查询链上数据

contract.sol

# DeFi关键指标监控脚本

from web3 import Web3

import requests, time

# 连接到Ethereum主网（Alchemy/Infura RPC）

RPC_URL = "https://eth-mainnet.g.alchemy.com/v2/YOUR_API_KEY"

w3 = Web3(Web3.HTTPProvider(RPC_URL))

assert w3.is_connected(), "RPC连接失败"

print(f"当前区块高度: {w3.eth.block_number}")

# Compound V2 cETH合约：查询当前汇率

CETH_ADDRESS = "0x4Ddc2D193948926D02f9B1fE9e1daa0718270ED5"

COMPTROLLER = "0x3d9819210A31b4961b30EF54bE2aeD79B9c9Cd3B"

def get_compound_supply_apr(underlying_address, ctoken_address):

"""

获取Compound供应APY

公式:supply_rate_per_block × blocks_per_year × underlying_price

"""

ctoken_abi= [

{"name": "supplyRatePerBlock", "outputs": [{"type": "uint256"}], "type": "function", "stateMutability": "view"},

{"name": "exchangeRateCurrent", "outputs": [{"type": "uint256"}], "type": "function", "stateMutability": "nonpayable"},

]

ctoken= w3.eth.contract(cToken_address, abi=ctoken_abi)

#每区块供应利率

supply_rate= ctoken.functions.supplyRatePerBlock().call()

#当前汇率（ctoken → underlying）

exchange_rate= ctoken.functions.exchangeRateCurrent().call()

#ETH区块时间约12秒，一年区块数

blocks_per_year= 365 * 24 * 60 * 5# ~2628000

supply_apy= (supply_rate * blocks_per_year) / 10**18

return supply_apy

def get_account_health_factor(account, comptroller):

"""查询账户健康因子"""

comp_abi= [

{"name": "getAccountLiquidity", "outputs": [{"type": "uint256"}], "type": "function", "stateMutability": "view"},

{"name": "markets", "outputs": [{"type": "uint256"}, {"type": "bool"}], "type": "function", "stateMutability": "view"},

]

comp= w3.eth.contract(comptroller, abi=comp_abi)

liquidity,shortfall = comp.functions.getAccountLiquidity(account).call()

if shortfall > 0:

return0.0# 健康因子=0，即资不抵债

return None if liquidity == 0else liquidity / 10**18

# 示例：查询ETH供应APY

eth_apr = get_compound_supply_apr("ETH", CETH_ADDRESS)

print(f"cETH 供应年化收益率: {eth_apr*100:.2f}%")

defi_monitor.py

3. 用The Graph查询历史数据

The Graph通过subgraph提供索引好的链上数据，比直接用RPC快很多。下面是查询Compound清算历史的GraphQL：

contract.sol

# 用requests调用The Graph API

import requests

COMPOUND_SUBGRAPH = (

"https://api.thegraph.com/subgraphs/name/compound-finance/compound-v2"

)

query = """

query LiquidationEvents($last: Int!) {

liquidationEvents(

first:$last,

orderBy:blockNumber,

orderDirection:desc

){

id

borrower{ id }

liquidator{ id }

repayAmount

cTokenCollateral

collateralAmount

blockNumber

blockTime

}

}

"""

def get_latest_liquidations(limit=20):

"""获取最近N条清算事件"""

r= requests.post(

COMPOUND_SUBGRAPH,

json={"query": query, "variables": {"last": limit}},

timeout=10

)

data= r.json()

if"errors" in data:

print("GraphQL错误:", data["errors"])

return []

return data["data"]["liquidationEvents"]

events = get_latest_liquidations(10)

print(f" 最近10条Compound清算事件：")

for e in events:

print(f"借款人:{e['borrower']['id'][:10]}... "

f"还款:${int(e['repayAmount'])/1e18:.2f} "

f"抵押:${int(e['collateralAmount'])/1e18:.4f} "

f"区块:{e['blockNumber']}")

graph_query.py

4. 健康因子实时监控脚本

搭建一个持续监控脚本，当健康因子接近清算线时发送告警：

contract.sol

# 健康因子告警脚本

import time, requests

from web3 import Web3

class DeFiMonitor:

def__init__(self, rpc_url):

self.w3= Web3(Web3.HTTPProvider(rpc_url))

self.alert_threshold= 1.2# HF < 1.2 时告警

defcheck_positions(self, addresses):

"""批量检查多个地址的健康因子"""

results= []

for addr in addresses:

hf= self._get_hf(addr)

status= self._classify(hf)

results.append({"address": addr, "hf": hf, "status": status})

if hf < self.alert_threshold and hf > 0:

self._send_alert(addr,hf)

return results

def_get_hf(self, addr):

#简化版：实际应从合约读取完整数据

return1.5# placeholder

def_classify(self, hf):

if hf == 0: return"无借款"

if hf < 1.0: return"⚠️ 即将清算"

if hf < 1.2: return"⚠️ 告警：接近清算"

if hf < 2.0: return"🟡 注意：HF偏低"

return"🟢 安全"

def_send_alert(self, addr, hf):

msg = f"🚨 健康因子告警！ 地址: {addr} HF: {hf:.3f}"

print(f"[ALERT] {msg}")

#实际可接入Telegram Bot / Slack Webhook发送通知

# 使用示例

monitor = DeFiMonitor("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY")

my_positions = [

"0x1234...",# 要监控的地址列表

]

results = monitor.check_positions(my_positions)

for r in results:

print(f"{r['status']} {r['address'][:10]}... HF={r['hf']:.3f}")

defi_alert.py

5. 系列十二完结总结

✅ 12-1 清算机制：HF<1触发自动清算，清算奖金激励快速处置坏账

✅ 12-2 预言机：Chainlink多节点聚合+去除极端值抗操纵，TWAP提高操纵成本

✅ 12-3 AMM：常数乘积定价，V3集中流动性，V4 Hooks开放定制逻辑

✅ 12-4 借贷清算线：Compound vs Aave阈值设计差异，利率非线性曲线调节资金供需

✅ 12-5 监控仪表盘：Web3.py链上数据查询，The Graph历史事件，HF告警脚本

🎉 系列十二「DeFi机制原理」完结

系列十二涵盖：清算机制、预言机、AMM机制、借贷清算线、监控仪表盘

━━━━━━━━━━━━━━━━━━━━━

📢 本文由「区块链编程」原创出品

未经授权，禁止转载

如有转载需求，请联系作者

👉 关注「区块链编程」

关注我，解锁更多可能