为什么要去挂单

在加密货币市场中，交易机会往往转瞬即逝，尤其是套利机会，可能只持续几分钟。如果依赖手动操作，用户可能无法及时抓住最佳时机，或者错过最优价格。例如，前段时间由于Bybit遭受黑客攻击，USDe出现脱锚现象，导致其价格大幅波动。当时，通过买入sUSDe并赎回的操作，年化收益率一度超过300%。这种高收益窗口通常持续时间极短，若没有提前设置挂单，手动交易很难跟上市场节奏。因此，挂单功能成为去中心化交易所（DEX）用户的重要工具，显著提高了交易效率。

DEX挂单的原理和缺点

多个DEX提供限价订单功能，每种实现方式和费用结构各有不同。以Cow Swap和Odos为例。核心原理是利用其聚合器功能，实时监控多个DEX的流动性和价格。当市场价格满足限价条件时，订单会被接单方（taker）触发，智能合约自动执行交易并由支付Gas费。Cow还会将用户的订单存储在链下，由一组被称为“求解者”（Solvers）的去中心化节点进行竞争撮合，但最终在链上执行。总之虽然都免除GAS，只不过是这些DEX替你完成的交易产生的额外利润都能覆盖GAS费。

但这就带来了一个问题：如果你的一笔订单以90U的价格挂单等待成交，而聚合器监控到某个DEX的价格下跌到80U，此时会替你执行这笔交易，你的订单最终以90U的价格成交，中间产生了10U的利润，这部分利润如何分配？在实际操作中，不同平台的处理方式存在差异。以Cow Swap为例，其机制明确规定，当执行价格优于限价时，产生的盈余（即这10U）会被平台与用户分成，Cow Swap抽取50%，用户获得剩余50%。而Odos则会把所有的盈余存入金库。本质上你的订单被DEX交易所无风险套利了。

另外DEX挂单为了节约手续费，都是聚合交易，即每隔一段时间把众多的交易都打包在一起，而ETH 12s一个区块，会导致错过一些可能的机会。当然，DEX还是有很多优势的，比如搜索更多路径、线下撮合、节约GAS等，大部分用户也够用了。

自己程序挂单的优势

相比依赖DEX的聚合挂单，自己通过智能合约直接交易具有独特优势。首先，用户可以完全掌控订单执行逻辑和所有盈余。其次，自己挂单避免了聚合交易的打包延迟，能更快响应市场变化，尤其在高波动时抓住12秒内的机会。此外，自定义挂单可灵活设置复杂条件（如多资产组合交易或止盈止损），不受平台预设功能的限制。然而，这需要一定的编程能力，并需自行支付Gas费，且链上可能带来安全风险。因此，自己挂单适合技术能力强、追求最大收益的高级用户。

私钥的保护

想要自己用程序操作智能合约，私钥的安全无疑是最关心的。我目前想出的方案是，自己用Python离线加密自己的私钥，并且将加密后的密文存在运行托管者的服务器上，解密的密码用FMZ平台机器人的参数传入，程序运行在托管者上读取后解密(可以在程序运行后删除)。这样即使你的服务器被黑，也没有问题, 但仍要注意FMZ账户不要泄露。由于只涉到一次离网后明文，安全性可以接受。 具体的代码如下，可以在本地离网的情况下运行

from web3 import Web3  # Web3.py 是与以太坊区块链交互的Python库
import json  # 用于处理JSON数据
import time  # 用于设置时间间隔
import requests  # 用于发送HTTP请求
from cryptography.fernet import Fernet
import base64
import hashlib
from datetime import datetime
from hexbytes import HexBytes

def generate_key(password: str) -> bytes:
    """通过用户密码生成 AES 密钥"""
    return base64.urlsafe_b64encode(hashlib.sha256(password.encode()).digest())

def encrypt_private_key(private_key: str, password: str) -> str:
    """使用密码加密私钥"""
    key = generate_key(password)
    cipher = Fernet(key)
    return cipher.encrypt(private_key.encode()).decode()

def decrypt_private_key(encrypted_key: str, password: str) -> str:
    """使用密码解密私钥"""
    key = generate_key(password)
    cipher = Fernet(key)
    return cipher.decrypt(encrypted_key.encode()).decode()

def save_key_to_file(key, file_path):
    # 将加密密钥保存到txt文件
    with open(file_path, 'w') as file:
        file.write(key) 

def load_key_from_file(file_path):
    # 从txt文件读取加密密钥
    with open(file_path, 'r') as file:
        return str(file.read())

def main():
    encrypt_key = encrypt_private_key('my_private_key', 'password') # my_private_key是自己的私钥，password是自己设置的密码
    save_key_to_file(encrypt_key,'encrypt_key.txt')
    print("加密后的私钥", encrypt_key)
    decrypt_key = decrypt_private_key(load_key_from_file('encrypt_key.txt'), 'password')
        print("解密的私钥", decrypt_key)

链接ETH前的准备

Web3.py是一个强大的Python库，用于与以太坊网络交互。通过以下命令安装：pip install web3。它的作用是简化开发者与以太坊节点的通信，支持查询余额、调用智能合约、发送交易等操作。例如，检查账户余额只需几行代码,这使得Web3.py成为构建DApp或自动化交易的理想工具。

python
from web3 import Web3
w3 = Web3(Web3.HTTPProvider('你的RPC地址'))
balance = w3.eth.get_balance('0x某个地址')
print(w3.from_wei(balance, 'ether'))

RPC（远程过程调用）是以太坊节点提供的通信接口，通过HTTP或WebSocket协议发送JSON请求与区块链交互。例如，eth_blockNumber可查询最新区块高度。由于运行本地节点成本高，开发者通常依赖第三方RPC提供商。常见选择包括：

• Infura：MetaMask默认服务，易用但免费额度低（每天10万次请求）。
• Alchemy：性能优越，免费额度高，支持更多功能。
• QuickNode：适合高性能需求，但偏向付费用户。

from web3 import Web3
w3 = Web3(Web3.HTTPProvider('https://eth-mainnet.g.alchemy.com/v2/你的API密钥'))
print(w3.is_connected())

推荐使用Alchemy,相比MetaMask的Infura，Alchemy提供更高免费额度, 注册后可获取RPC地址, 如 https://eth-mainnet.g.alchemy.com/v2/你的API密钥 ,配置到Web3.py就可以使用。

Curve合约地址和ABI

以sDAI/sUSDe的Crve池子为例 https://curve.fi/dex/ethereum/pools/factory-stable-ng-102/swap/ ，可以很轻松的找到，两个代币的地址，以及池子的地址。

ABI定义了如何和合约进行交互，因此也是必须获取的，在ethscan上查看合约https://etherscan.io/address/0x167478921b907422f8e88b43c4af2b8bea278d3a#code ，在contract页面下可以看到abi，直接复制可用。

链接合约获取价格

首先链接ETH钱包，如果打印出自己的钱包地址，就说明成功了。

def main():
    # 文件路径
    file_path = 'encrypted_key.txt'

    # 从文件中读取加密的私钥
    encrypted_private_key = load_key_from_file(file_path)
    private_key = decrypt_private_key(encrypted_private_key, Password) #Password为密码，定义为策略的参数
    web3 = Web3(Web3.HTTPProvider(HTTPProvider)) # HTTPProvider为RPC的链接，定义为参数
    account = web3.eth.account.from_key(private_key)
    address = account.address  # 获取账户的公开地址
    Log('链接账户', address)

下面是获取价格的过程，最后计算在不考虑GAS费的情况下，目前投入100000 SDAI一周后会收获335U的收益。会看起来有点复杂，但其实也不难理解。

    # --------------- 合约设置 ---------------
    # Curve.fi sDAI/sUSDe 池合约地址和ABI
    pool_address = '0x167478921b907422F8E88B43C4Af2B8BEa278d3A'  # Curve池子的合约地址
    # 以下是简化的ABI，仅包含我们需要的函数
    pool_abi = json.loads('''[{"stateMutability":"view","type":"function","name":"get_dy","inputs":[{"name":"i","type":"int128"},{"name":"j","type":"int128"},{"name":"dx","type":"uint256"}],"outputs":[{"name":"","type":"uint256"}]}]''')
    # 创建池子合约对象
    pool_contract = web3.eth.contract(address=pool_address, abi=pool_abi)

    # ERC20 代币标准合约ABI
    erc20_abi = json.loads('''[
        {"constant":true,"inputs":[{"name":"_owner","type":"address"}],"name":"balanceOf","outputs":[{"name":"balance","type":"uint256"}],"type":"function"},
        {"constant":false,"inputs":[{"name":"_spender","type":"address"},{"name":"_value","type":"uint256"}],"name":"approve","outputs":[{"name":"success","type":"bool"}],"type":"function"},
        {"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"stateMutability":"view","type":"function"}
    ]''')

    sdai_address = '0x83F20F44975D03b1b09e64809B757c47f942BEeA'  # sDAI代币的合约地址
    susde_address = '0x9D39A5DE30e57443BfF2A8307A4256c8797A3497'  # sUSDE代币的合约地址
    # 池子中代币的索引
    SDAI_INDEX = 0  # sDAI代币在池子中的索引
    SUSDE_INDEX = 1  # sUSDE代币在池子中的索引
    # 创建代币合约对象
    sdai_contract = web3.eth.contract(address=sdai_address, abi=erc20_abi)
    susde_contract = web3.eth.contract(address=susde_address, abi=erc20_abi)
    SUSDE_PRICE = 1.1623 #这个价格是ethena官网价格，1周后可赎回
    SDAI_PRICE = 1.15 #sDAI是收益代币，价值会累计，目前价格1.15

    try:
        SDAI_DECIMALS = sdai_contract.functions.decimals().call()
        SUSDE_DECIMALS = susde_contract.functions.decimals().call()
    except:
        # 如果无法获取，假设为标准的18位精度
        SDAI_DECIMALS = 18
        SUSDE_DECIMALS = 18
    amount_in = 100000
    amount_out = pool_contract.functions.get_dy(
                    SDAI_INDEX,  # 输入代币索引
                    SUSDE_INDEX,  # 输出代币索引
                    int(amount_in *  10**SDAI_DECIMALS)   # 输入数量(wei)
                ).call()
    profit =  SUSDE_PRICE * amount_out / 10**SUSDE_DECIMALS -  amount_in * SDAI_PRICE
    Log(round(profit, 2), round(amount_out / 10**SUSDE_DECIMALS, 2))

完整程序

最后在用轮询的方式不断获取价格，当达到预期的利润后下单，注意这个代码仅为示例代码，不要直接使用，读者在实践中间可能会遇到各种问题，但目前AI非常强大，基本上可以解答出各种疑问，也可以直接帮帮写代码，FMZ的代码编辑器也集成了ChatGPT, 可以多使用。

def execute_trade(amount_in, min_amount_out, direction):
    gas_price = web3.eth.gas_price
    index_in = SUSDE_INDEX
    index_out = SDAI_INDEX
    if direction == 'buy':
        index_in = SDAI_INDEX
        index_out = SUSDE_INDEX
    # 第二步：执行代币交换交易
    swap_tx = pool_contract.functions.exchange(
        index_in,  # 输入代币索引
        index_out,  # 输出代币索引
        int(amount_in*10**SDAI_DECIMALS),  # 输入数量
        int(min_amount_out*10**SUSDE_DECIMALS)  # 最小输出数量(考虑滑点)
    ).build_transaction({
        'from': address,  # 交易发送方
        'gas': 600000,  # gas限制
        'gasPrice': int(2*gas_price) ,
        'nonce': web3.eth.get_transaction_count(address),  # 获取新的nonce值
    })
    
    # 签名并发送交换交易
    signed_swap_tx = web3.eth.account.sign_transaction(swap_tx, private_key)
    swap_tx_hash = web3.eth.send_raw_transaction(signed_swap_tx.rawTransaction)
    
    Log(f"交换交易已提交，交易哈希: {swap_tx_hash.hex()}")

def get_buy_profit(amount_in):
    amount_out = pool_contract.functions.get_dy(
                    SDAI_INDEX,  # 输入代币索引
                    SUSDE_INDEX,  # 输出代币索引
                    int(amount_in *  10**SDAI_DECIMALS)   # 输入数量(wei)
                ).call()
    return  SUSDE_PRICE * amount_out / 10**SUSDE_DECIMALS -  amount_in * SDAI_PRICE, amount_out / 10**SUSDE_DECIMALS

def main():
    while True:
        try:
            sdai_balance = sdai_contract.functions.balanceOf(address).call() / 10**SDAI_DECIMALS
            susde_balance = susde_contract.functions.balanceOf(address).call() / 10**SUSDE_DECIMALS
            amount_in = 100000 #交易的DAI数量
            profit, amount_out = get_buy_profit(amount_in)
            LogStatus(f"SDAI数量：{sdai_balance}, SUSDE数量：{susde_balance}, 收益：{profit}")
            if profit > 1000 and sdai_balance > amount_in: #利润空间
                Log("\n开始执行SDAI->SUSDE交易...")
                execute_trade(amount_in, 0.999*amount_out, 'buy') #一定要设置滑点
            wait_time = 3  # 等待时间(秒)
            time.sleep(wait_time)
            
        except Exception as e:
            # 全局错误处理
            print(f"程序发生错误: {e}")
            print("60秒后重试...")
            time.sleep(60)  # 出错后等待更长时间

风险提醒

链上操作对新手风险比较高，除了刚才提到的私钥泄露风险外，还有各种风险：、

• MEV机器人，执行交易时一定要设置好最小输出min_amount_out，确保再最差的情况下也能有利润，否则会被MEV剥削。今天就有一个人用22万usdc再uniswap只换出了5272usdt，原因就是amountOutMinimum设置为0.
• 策略出错，和交易所的API一样，如果链上程序交易中出现Bug，也会频繁消耗GAS

对于链上交易新手，需要学习基础知识：理解Gas、滑点、MEV等概念。 始终从低金额开始，逐步增加。使用Etherscan等监控交易。宁可错过机会，也不要冒险亏损本金。