一、从普通人的视角搞懂AMM和无常损失

1.1 先把AMM翻译成大白话

很多人第一次接触AMM,会被“自动做市商”这个名字吓退,其实它本质就是一个自动卖货的小店。你可以想象成小区门口有个卖苹果和梨的小店,老板定了个死规矩:店里苹果的总价值必须永远等于梨的总价值。比如一开始店里有100个苹果(每个1块钱)和100个梨(每个1块钱),那不管谁来买苹果或梨,都得先保证苹果总价值=梨总价值。

比如有人来买苹果,买走50个,那店里剩下50个苹果,总价值还是100块(因为原来100个值100块),所以苹果单价会涨到2块钱一个。这时候梨的总价值也得是100块,原来100个梨的价值没动,所以梨单价还是1块钱。要是有人来换,用50个梨换苹果,那换完梨变成50个(总价值100块,单价2块),苹果变成100个(总价值100块,单价1块)。这就是AMM最核心的逻辑,不用老板喊价,全靠公式自动算价格。

1.2 无常损失到底亏在哪?

无常损失,很多人以为是真的亏了钱,其实它更像“本该赚的没赚到”。还是用刚才的苹果梨小店举例子: 假设一开始你投了100个苹果和100个梨(总价值200块),过了一个月,苹果涨价到每个2块,梨还是1块。这时候有两种选择: 第一种:你没投AMM,自己拿着100个苹果和100个梨,总价值是100×2 + 100×1 = 300块,赚了100块。 第二种:你投了AMM,这时候AMM里苹果和梨的总价值要相等。我们来算AMM里的数量: AMM里的数量乘积永远不变(这个是AMM的核心公式,比如原来100个苹果×100个梨=10000),设现在苹果数量是A,梨是B,那A×B=10000,同时A×2(苹果单价)=B×1(梨单价),解出来A≈70.7,B≈141.4。总价值是70.7×2 + 141.4×1≈282.8块。

你会发现,投AMM的总价值282.8块,比自己拿着的300块少了17.2块,这少的部分就是无常损失。本质就是:当两种资产价格偏离原来的比例时,AMM里的资产组合价值,会比你自己持有初始比例资产的价值低,低的部分就是无常损失。而且这个损失是“无常”的,要是后来苹果和梨的价格又回到1:1,损失就会消失。

二、无常损失的本质拆解

2.1 从公式看本质

刚才的例子里,AMM的核心公式是“数量乘积不变”,用数学公式表示就是:x×y=k(x是A资产数量,y是B资产数量,k是固定值)。而资产的总价值V = x×P_x + y×P_y(P_x是A资产价格,P_y是B资产价格)。

当两种资产的价格比发生变化时,x和y的比例也会跟着变,导致V的变化幅度和“持有初始比例资产”的变化幅度不一样。我们可以推导一个通用的无常损失公式:假设初始价格比是P0,后来的价格比是P,无常损失率(就是损失的比例)是: 无常损失率 = 2×√(P) / (1 + P) - 1 比如刚才的例子,P是苹果价格从1涨到2,所以P=2,代入公式:2×√2/(1+2) -1 ≈ 2×1.414/3 -1≈0.9427-1=-0.0573,也就是损失了约5.73%,和刚才算的17.2/300≈5.73%一致。

这个公式能看出来两个关键点:第一,只有当P=1(价格比和初始一样)时,损失率为0;第二,P离1越远,损失率越高,也就是两种资产价格差越大,无常损失越严重。

2.2 本质总结

无常损失的本质不是“资产贬值”,而是“机会成本损失”。它是AMM的“自动做市”逻辑带来的固有缺陷:为了保证两种资产的价值相等,当一种资产涨价时,AMM会自动卖出更多的该资产,导致涨价后该资产的持有量比你自己持有的少,赚的差价自然就少了;当一种资产降价时,AMM会自动买入更多的该资产,导致降价后该资产的持有量比你自己持有的多,亏的差价自然就多了。

三、链上流动性池实时监控方案的设计思路

既然无常损失是AMM的固有问题,那我们可以通过实时监控流动性池的状态,提前预警甚至自动调整,来降低损失。实时监控的核心是:实时获取流动性池的资产数量、价格,计算无常损失,当损失达到预警线时,触发告警或自动撤资。

3.1 监控方案的核心模块

一个完整的实时监控方案,需要三个核心模块:

  1. 数据采集模块:实时从区块链上获取流动性池的资产数量、两种资产的价格。
  2. 损失计算模块:实时计算当前的无常损失率和损失金额。
  3. 告警/执行模块:当损失达到预设的阈值时,发送告警(比如短信、邮件),或者自动执行撤资、调整比例等操作。

四、监控方案的落地实践

4.1 技术栈选择

我们选择的技术栈是:Python(用于数据处理、计算)+ Web3.py(用于连接区块链,获取链上数据)+ Aiohttp(用于异步获取价格数据,提高效率)。这个技术栈的优点是:Python语法简单,容易上手;Web3.py是连接以太坊等区块链的主流库,功能完善;Aiohttp可以同时获取多个价格数据,速度快。

4.2 完整的代码实现

我们以以太坊上的USDT-ETH流动性池为例,实现一个实时监控程序。代码的功能是:每10秒获取一次流动性池的状态,计算无常损失,当损失率超过5%时,发送告警。

# 导入需要的库
import asyncio
import web3
from web3 import Web3
import aiohttp
import time

# -------------------------- 配置参数 --------------------------
# 以太坊主网节点地址(可以用Infura的免费节点)
INFURA_URL = "https://mainnet.infura.io/v3/你的Infura密钥"
# USDT-ETH流动性池的合约地址(以Uniswap V2为例)
POOL_ADDRESS = "0x0d4a11d5eeaac28ec3f61d100daf4d40471f1852"
# 流动性池的ABI(简化版,只保留获取资产数量的函数)
POOL_ABI = [
    {
        "constant": True,
        "inputs": [],
        "name": "getReserves",
        "outputs": [
            {"name": "_reserve0", "type": "uint112"},
            {"name": "_reserve1", "type": "uint112"},
            {"name": "_blockTimestampLast", "type": "uint32"}
        ],
        "payable": False,
        "stateMutability": "view",
        "type": "function"
    }
]
# 初始价格比(假设初始时USDT和ETH的价格比是1:2000,也就是1ETH=2000USDT)
INITIAL_PRICE_RATIO = 2000
# 预警损失率(5%)
ALERT_LOSS_RATE = 0.05
# 监控间隔(10秒)
MONITOR_INTERVAL = 10

# -------------------------- 核心函数 --------------------------
async def get_pool_reserves():
    """异步获取流动性池的资产数量(USDT和ETH)"""
    w3 = Web3(Web3.HTTPProvider(INFURA_URL))
    # 检查连接是否正常
    if not w3.is_connected():
        raise Exception("无法连接到以太坊节点")
    # 加载流动性池合约
    pool_contract = w3.eth.contract(address=POOL_ADDRESS, abi=POOL_ABI)
    # 调用合约的getReserves函数获取资产数量
    reserves = pool_contract.functions.getReserves().call()
    # 注意:Uniswap V2的资产顺序是按地址排序的,USDT的地址比ETH小,所以reserve0是USDT,reserve1是ETH
    # USDT的小数位数是6,ETH的小数位数是18,所以要转换成实际数量
    usdt_amount = reserves[0] / 10**6  # USDT数量(单位:个)
    eth_amount = reserves[1] / 10**18  # ETH数量(单位:个)
    return usdt_amount, eth_amount

async def get_current_price_ratio():
    """异步获取当前的USDT和ETH的价格比(1ETH=多少USDT)"""
    # 用CoinGecko的API获取ETH和USDT的价格
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.coingecko.com/api/v3/simple/price?ids=ethereum,tether&vs_currencies=usd") as resp:
            data = await resp.json()
            eth_price = data["ethereum"]["usd"]  # 1ETH的价格(USDT)
            usdt_price = data["tether"]["usd"]    # 1USDT的价格(USDT)
            # 价格比=ETH价格/USDT价格
            return eth_price / usdt_price

def calculate_loss_rate(current_price_ratio):
    """计算无常损失率"""
    # 代入无常损失公式:2*√P/(1+P) -1,P是当前价格比/初始价格比
    P = current_price_ratio / INITIAL_PRICE_RATIO
    loss_rate = 2 * (P**0.5) / (1 + P) - 1
    return loss_rate

# -------------------------- 主监控循环 --------------------------
async def monitor():
    print("开始监控USDT-ETH流动性池的无常损失...")
    while True:
        try:
            # 1. 获取流动性池的资产数量
            usdt_amount, eth_amount = await get_pool_reserves()
            print(f"当前流动性池:USDT数量={usdt_amount:.2f},ETH数量={eth_amount:.4f}")
            
            # 2. 获取当前的价格比
            current_price_ratio = await get_current_price_ratio()
            print(f"当前价格比:1ETH={current_price_ratio:.2f}USDT")
            
            # 3. 计算无常损失率
            loss_rate = calculate_loss_rate(current_price_ratio)
            print(f"当前无常损失率:{loss_rate:.2%}")
            
            # 4. 检查是否达到预警线
            if abs(loss_rate) > ALERT_LOSS_RATE:
                print(f"【告警】无常损失率达到{loss_rate:.2%},超过预警线{ALERT_LOSS_RATE:.2%}!")
                # 这里可以添加自动撤资的代码,比如调用合约的removeLiquidity函数
                # 比如:await remove_liquidity()
            
            # 等待监控间隔
            await asyncio.sleep(MONITOR_INTERVAL)
        except Exception as e:
            print(f"监控出错:{str(e)}")
            await asyncio.sleep(MONITOR_INTERVAL)

# 运行监控程序
if __name__ == "__main__":
    asyncio.run(monitor())

4.3 代码的使用说明

  1. 替换参数:把代码中的你的Infura密钥替换成自己的Infura密钥(可以在Infura官网免费申请)。
  2. 安装依赖:需要安装web3、aiohttp等库,执行命令:
pip install web3 aiohttp
  1. 运行程序:执行python monitor.py,程序就会开始实时监控,每10秒输出一次流动性池的状态和无常损失率,当损失率超过5%时,会打印告警信息。

4.4 代码的扩展方向

这个代码只是一个基础版本,实际使用时可以扩展很多功能:

  1. 告警方式:可以把打印告警改成发送短信、邮件,或者接入企业微信、钉钉的机器人,实时推送给用户。
  2. 自动撤资:可以调用流动性池合约的removeLiquidity函数,自动把自己的流动性撤出来,避免损失扩大。
  3. 多池监控:可以同时监控多个流动性池(比如USDT-ETH、USDC-ETH等),只需要修改POOL_ADDRESS参数,添加多个监控任务即可。
  4. 历史数据:可以把每次计算的无常损失率保存到数据库(比如MySQL、Redis),方便后续分析损失的变化趋势。

五、方案的应用场景、优缺点和注意事项

5.1 应用场景

这个实时监控方案主要适用于两类人群:

  1. 流动性提供者(LP):就是把资产投到流动性池里赚手续费的人,他们可以通过监控实时掌握自己的无常损失,避免损失过大。
  2. 量化交易员:可以把这个监控方案和自动交易策略结合,当无常损失达到一定程度时,自动调整流动性的比例,或者切换到其他流动性池,提高收益。

5.2 方案的优缺点

优点:

  1. 实时性强:可以每几秒就获取一次数据,实时掌握流动性池的状态,不会错过预警时机。
  2. 成本低:不需要额外的硬件设备,只需要一台电脑或者云服务器就可以运行,而且Infura的免费节点足够满足基础的监控需求。
  3. 可扩展性强:可以根据自己的需求,添加告警、自动撤资、多池监控等功能。

缺点:

  1. 依赖节点:如果以太坊节点出现故障,就无法获取数据,导致监控中断。
  2. 数据延迟:链上数据的获取会有一定的延迟(比如1-2秒),如果市场价格波动特别快,可能会导致预警不及时。
  3. 无法完全避免损失:监控方案只能提前预警,不能完全消除无常损失,只是降低损失的风险。

5.3 注意事项

  1. 初始价格比的设置:初始价格比要和你投入流动性池时的价格比一致,否则计算出来的无常损失率会不准确。
  2. 合约地址和ABI的正确性:不同的流动性池(比如Uniswap V2、Uniswap V3、SushiSwap)的合约地址和ABI不一样,要根据自己监控的流动性池进行修改。
  3. 节点的稳定性:如果要长时间运行监控程序,建议使用付费的以太坊节点,保证连接的稳定性。
  4. 自动撤资的安全性:如果添加自动撤资的功能,要注意私钥的安全,不要把私钥硬编码在代码里,建议使用硬件钱包或者密钥管理服务(KMS)来管理私钥。

六、文章总结

无常损失是AMM的固有缺陷,它的本质是机会成本损失,当两种资产的价格比偏离初始比例时,流动性池的资产组合价值会比自己持有初始比例资产的价值低。实时监控流动性池的无常损失,可以帮助流动性提供者提前预警,降低损失风险。

我们通过一个具体的Python程序,实现了实时监控USDT-ETH流动性池的无常损失,这个方案简单易上手,成本低,可扩展性强。在实际使用时,可以根据自己的需求,扩展告警、自动撤资等功能,更好地管理自己的流动性资产。