在以太坊上创建你自己的 AI 交易代理

1git checkout 02-read-quote
2uv run agent.py

1from web3 import Web3
2from web3.contract import Contract
3from decimal import Decimal, ROUND_HALF_UP
4from dataclasses import dataclass
5from datetime import datetime, timezone
6from pprint import pprint
7import time
8import functools
9import sys
显示全部
复制

1print = functools.partial(print, flush=True)
复制

1MAINNET_URL = "https://eth.drpc.org"
复制

1BLOCK_TIME_SECONDS = 12
2MINUTE_BLOCKS = int(60 / BLOCK_TIME_SECONDS)
3HOUR_BLOCKS = MINUTE_BLOCKS * 60
4DAY_BLOCKS = HOUR_BLOCKS * 24
复制

1CYCLE_BLOCKS = DAY_BLOCKS
复制

1# 我们正在读取的池子的地址
2WETHUSDC_ADDRESS = Web3.to_checksum_address("0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640")
复制

1POOL_ABI = [
2    { "name": "slot0", ... },
3    { "name": "token0", ... },
4    { "name": "token1", ... },
5]
6
7ERC20_ABI = [
8    { "name": "symbol", ... },
9    { "name": "decimals", ... }
10]
显示全部
复制

1w3 = Web3(Web3.HTTPProvider(MAINNET_URL))
复制

1@dataclass(frozen=True)
2class ERC20Token:
3    address: str
4    symbol: str
5    decimals: int
6    contract: Contract
复制

1@dataclass(frozen=True)
2class PoolInfo:
3    address: str
4    token0: ERC20Token
5    token1: ERC20Token
6    contract: Contract
7    asset: str
8    decimal_factor: Decimal = 1
复制

1    def get_price(self, block: int) -> Decimal:
复制

1        assert block <= w3.eth.block_number, "区块在未来"
复制

1        sqrt_price_x96 = Decimal(self.contract.functions.slot0().call(block_identifier=block)[0])
复制

1        raw_price = (sqrt_price_x96 / Decimal(2**96)) ** 2
复制

1         # (每个 token0 对应的 token1 数量)
2        return 1/(raw_price * self.decimal_factor)
复制

1@dataclass(frozen=True)
2class Quote:
3    timestamp: str
4    price: Decimal
5    asset: str
复制

1def read_token(address: str) -> ERC20Token:
2    token = w3.eth.contract(address=address, abi=ERC20_ABI)
3    symbol = token.functions.symbol().call()
4    decimals = token.functions.decimals().call()
5
6    return ERC20Token(
7        address=address,
8        symbol=symbol,
9        decimals=decimals,
10        contract=token
11    )
显示全部
复制

1def read_pool(address: str) -> PoolInfo:
2    pool_contract = w3.eth.contract(address=address, abi=POOL_ABI)
3    token0Address = pool_contract.functions.token0().call()
4    token1Address = pool_contract.functions.token1().call()
5    token0 = read_token(token0Address)
6    token1 = read_token(token1Address)
7
8    return PoolInfo(
9        address=address,
10        asset=f"{token1.symbol}/{token0.symbol}",
11        token0=token0,
12        token1=token1,
13        contract=pool_contract,
14        decimal_factor=Decimal(10) ** Decimal(token0.decimals - token1.decimals)
15    )
显示全部
复制

1def get_quote(pool: PoolInfo, block_number: int = None) -> Quote:
复制

1    if block_number is None:
2        block_number = w3.eth.block_number
复制

1    block = w3.eth.get_block(block_number)
2    price = pool.get_price(block_number)
3    return Quote(
4        timestamp=datetime.fromtimestamp(block.timestamp, timezone.utc).isoformat(),
5        price=price.quantize(Decimal("0.01")),
6        asset=pool.asset
7    )
复制

1def get_quotes(pool: PoolInfo, start_block: int, end_block: int, step: int) -> list[Quote]:
复制

1    quotes = []
2    for block in range(start_block, end_block + 1, step):
复制

1        quote = get_quote(pool, block)
2        quotes.append(quote)
3    return quotes
复制

1pool = read_pool(WETHUSDC_ADDRESS)
2quotes = get_quotes(
3    pool,
4    w3.eth.block_number - 12*CYCLE_BLOCKS,
5    w3.eth.block_number,
6    CYCLE_BLOCKS
7)
8
9pprint(quotes)
显示全部
复制

1git checkout 03-create-prompt
2uv run agent.py

1给定以下报价：
2资产：WETH/USDC
3        2026-01-20T16:34 3016.21
4        .
5        .
6        .
7        2026-02-01T17:49 2299.10
8
9资产：WBTC/WETH
10        2026-01-20T16:34 29.84
11        .
12        .
13        .
14        2026-02-01T17:50 33.46
15
16
17您预计在 2026-02-02T17:56 时 WETH/USDC 的价值会是多少？
18
19请以一个四舍五入到两位小数的数字作为您的答案，
20不要包含任何其他文本。
显示全部

1from datetime import datetime, timezone, timedelta
复制

1# 我们正在读取的池子的地址
2WETHUSDC_ADDRESS = Web3.to_checksum_address("0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640")
3WETHWBTC_ADDRESS = Web3.to_checksum_address("0xCBCdF9626bC03E24f779434178A73a0B4bad62eD")
复制

1@dataclass(frozen=True)
2class PoolInfo:
3    .
4    .
5    .
6    reverse: bool = False
7
8    def get_price(self, block: int) -> Decimal:
9        assert block <= w3.eth.block_number, "区块在未来"
10        sqrt_price_x96 = Decimal(self.contract.functions.slot0().call(block_identifier=block)[0])
11        raw_price = (sqrt_price_x96 / Decimal(2**96)) ** 2  # (token1 per token0)
12        if self.reverse:
13            return 1/(raw_price * self.decimal_factor)
14        else:
15            return raw_price * self.decimal_factor
显示全部
复制

1def read_pool(address: str, reverse: bool = False) -> PoolInfo:
2    .
3    .
4    .
5
6    return PoolInfo(
7        .
8        .
9        .
10
11        asset= f"{token1.symbol}/{token0.symbol}" if reverse else f"{token0.symbol}/{token1.symbol}",
12        reverse=reverse
13    )
显示全部
复制

1def format_quotes(quotes: list[Quote]) -> str:
2    result = f"Asset: {quotes[0].asset}\n"
3    for quote in quotes:
4        result += f"\t{quote.timestamp[0:16]} {quote.price.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)}\n"
5    return result
复制

1def make_prompt(quotes: list[list[Quote]], expected_time: str, asset: str) -> str:
2    return f"""
复制

1给定以下报价：
2{
3    functools.reduce(lambda acc, q: acc + '\n' + q,
4        map(lambda q: format_quotes(q), quotes))
5}
复制

1您预计 {asset} 在 {expected_time} 的价值会是多少？
2
3请以一个四舍五入到两位小数的数字作为您的答案，
4不要包含任何其他文本。
5    """
复制

1wethusdc_pool = read_pool(WETHUSDC_ADDRESS, True)
2wethusdc_quotes = get_quotes(
3    wethusdc_pool,
4    w3.eth.block_number - 12*CYCLE_BLOCKS,
5    w3.eth.block_number,
6    CYCLE_BLOCKS,
7)
8
9wethwbtc_pool = read_pool(WETHWBTC_ADDRESS)
10wethwbtc_quotes = get_quotes(
11    wethwbtc_pool,
12    w3.eth.block_number - 12*CYCLE_BLOCKS,
13    w3.eth.block_number,
14    CYCLE_BLOCKS
15)
显示全部
复制

1future_time = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()[0:16]
2
3print(make_prompt(wethusdc_quotes + wethwbtc_quotes, future_time, wethusdc_pool.asset))
复制

1from openai import OpenAI
2
3open_ai = OpenAI()  # 客户端读取 OPENAI_API_KEY 环境变量
复制

1response = open_ai.chat.completions.create(
2    model="gpt-4-turbo",
3    messages=[
4        {"role": "user", "content": prompt}
5    ],
6    temperature=0.0,
7    max_tokens=16,
8)
复制

1expected_price = Decimal(response.choices[0].message.content.strip())
2current_price = wethusdc_quotes[-1].price
3
4print ("Current price:", wethusdc_quotes[-1].price)
5print(f"In {future_time}, expected price: {expected_price} USD")
6
7if (expected_price > current_price):
8    print(f"Buy, I expect the price to go up by {expected_price - current_price} USD")
9else:
10    print(f"Sell, I expect the price to go down by {current_price - expected_price} USD")
显示全部
复制

1uv run test-predictor.py

12026-01-05T19:50 的预测：预测 3138.93 美元，实际 3218.92 美元，误差 79.99 美元
22026-01-06T19:56 的预测：预测 3243.39 美元，实际 3221.08 美元，误差 22.31 美元
32026-01-07T20:02 的预测：预测 3223.24 美元，实际 3146.89 美元，误差 76.35 美元
42026-01-08T20:11 的预测：预测 3150.47 美元，实际 3092.04 美元，误差 58.43 美元
5.
6.
7.
82026-01-31T22:33 的预测：预测 2637.73 美元，实际 2417.77 美元，误差 219.96 美元
92026-02-01T22:41 的预测：预测 2381.70 美元，实际 2318.84 美元，误差 62.86 美元
102026-02-02T22:49 的预测：预测 2234.91 美元，实际 2349.28 美元，误差 114.37 美元
1129 次预测的平均预测误差：83.87103448275862068965517241 美元
12每个建议的平均变化：4.787931034482758620689655172 美元
13变化的 标准方差：104.42 美元
14盈利天数：51.72%
15亏损天数：48.28%
显示全部

1CYCLES_FOR_TEST = 40 # 对于回测，我们测试多少个周期
2
3# 获取大量报价
4wethusdc_pool = read_pool(WETHUSDC_ADDRESS, True)
5wethusdc_quotes = get_quotes(
6    wethusdc_pool,
7    w3.eth.block_number - CYCLE_BLOCKS*CYCLES_FOR_TEST,
8    w3.eth.block_number,
9    CYCLE_BLOCKS,
10)
11
12wethwbtc_pool = read_pool(WETHWBTC_ADDRESS)
13wethwbtc_quotes = get_quotes(
14    wethwbtc_pool,
15    w3.eth.block_number - CYCLE_BLOCKS*CYCLES_FOR_TEST,
16    w3.eth.block_number,
17    CYCLE_BLOCKS
18)
显示全部
复制

1# 创建预测并与真实历史进行比较
2
3total_error = Decimal(0)
4changes = []
复制

1for index in range(0,len(wethusdc_quotes)-CYCLES_BACK):
复制

1    wethusdc_slice = wethusdc_quotes[index:index+CYCLES_BACK]
2    wethwbtc_slice = wethwbtc_quotes[index:index+CYCLES_BACK]
复制

1    predicted_price = Decimal(response.choices[0].message.content.strip())
2    real_price = wethusdc_quotes[index+CYCLES_BACK].price
3    prediction_time_price = wethusdc_quotes[index+CYCLES_BACK-1].price
复制

1    error = abs(predicted_price - real_price)
2    total_error += error
3    print (f"Prediction for {prediction_time}: predicted {predicted_price} USD, real {real_price} USD, error {error} USD")
复制

1    recomended_action = 'buy' if predicted_price > prediction_time_price else 'sell'
2    price_increase = real_price - prediction_time_price
3    changes.append(price_increase if recomended_action == 'buy' else -price_increase)
复制

1print (f"Mean prediction error over {len(wethusdc_quotes)-CYCLES_BACK} predictions: {total_error / Decimal(len(wethusdc_quotes)-CYCLES_BACK)} USD")
2
3length_changes = Decimal(len(changes))
4mean_change = sum(changes, Decimal(0)) / length_changes
5print (f"Mean change per recommendation: {mean_change} USD")
6var = sum((x - mean_change) ** 2 for x in changes) / length_changes
7print (f"Standard variance of changes: {var.sqrt().quantize(Decimal("0.01"))} USD")
复制

1print (f"Profitable days: {len(list(filter(lambda x: x > 0, changes)))/length_changes:.2%}")
2print (f"Losing days: {len(list(filter(lambda x: x < 0, changes)))/length_changes:.2%}")
复制

1git checkout 05-trade
2uv run agent.py

1(ai-trading-agent) qbzzt@Ori-Cloudnomics:~/260215-ai-agent$ uv run agent.py
2当前价格：1843.16
32026-02-06T23:07，预期价格：1724.41 美元
4交易前帐户余额：
5USDC 余额：927301.578272
6WETH 余额：500
7卖出，我预计价格将下跌 118.75 美元
8批准交易已发送：74e367ddbb407c1aaf567d87aa5863049991b1d2aa092b6b85195d925e2bd41f
9批准交易已挖出。
10卖出交易已发送：fad1bcf938585c9e90364b26ac7a80eea9efd34c37e5db81e58d7655bcae28bf
11卖出交易已挖出。
12交易后帐户余额：
13USDC 余额：929143.797116
14WETH 余额：499
显示全部

1SWAP_ROUTER_ADDRESS=Web3.to_checksum_address("0xE592427A0AEce92De3Edee1F18E0157C05861564")
2WETH_TO_USDC=bytes.fromhex("C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc20001F4A0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")
3USDC_TO_WETH=bytes.fromhex("A0b86991c6218b36c1d19D4a2e9Eb0cE3606eB480001F4C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
4PRIVATE_KEY="0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"
复制

1WETH_TRADE_AMOUNT=1
复制

1ERC20_ABI = [
2    { "name": "symbol", ... },
3    { "name": "decimals", ... },
4    { "name": "balanceOf", ...},
5    { "name": "approve", ...}
6]
复制

1SWAP_ROUTER_ABI = [
2  { "name": "exactInput", ...},
3]
复制

1account = w3.eth.account.from_key(PRIVATE_KEY)
2swap_router = w3.eth.contract(
3    address=SWAP_ROUTER_ADDRESS,
4    abi=SWAP_ROUTER_ABI
5)
复制

1def txn_params() -> dict:
2    return {
3        "from": account.address,
4        "value": 0,
5        "gas": 300000,
6        "nonce": w3.eth.get_transaction_count(account.address),
7    }
复制

1def approve_token(contract: Contract, amount: int):
复制

1    txn = contract.functions.approve(SWAP_ROUTER_ADDRESS, amount).build_transaction(txn_params())
2    signed_txn = w3.eth.account.sign_transaction(txn, private_key=PRIVATE_KEY)
3    tx_hash = w3.eth.send_raw_transaction(signed_txn.raw_transaction)
复制

1    print(f"Approve transaction sent: {tx_hash.hex()}")
2    w3.eth.wait_for_transaction_receipt(tx_hash)
3    print("Approve transaction mined.")
复制

1SELL_PARAMS = {
2    "path": WETH_TO_USDC,
3    "recipient": account.address,
4    "deadline": 2**256 - 1,
5    "amountIn": WETH_TRADE_AMOUNT * 10 ** wethusdc_pool.token1.decimals,
6    "amountOutMinimum": 0,
7}
复制

1def make_buy_params(quote: Quote) -> dict:
2    return {
3        "path": USDC_TO_WETH,
4        "recipient": account.address,
5        "deadline": 2**256 - 1,
6        "amountIn": int(quote.price*WETH_TRADE_AMOUNT) * 10**wethusdc_pool.token0.decimals,
7        "amountOutMinimum": 0,
8    }
复制

1def buy(quote: Quote):
2    buy_params = make_buy_params(quote)
3    approve_token(wethusdc_pool.token0.contract, buy_params["amountIn"])
4    txn = swap_router.functions.exactInput(buy_params).build_transaction(txn_params())
5    signed_txn = w3.eth.account.sign_transaction(txn, private_key=PRIVATE_KEY)
6    tx_hash = w3.eth.send_raw_transaction(signed_txn.raw_transaction)
7    print(f"Buy transaction sent: {tx_hash.hex()}")
8    w3.eth.wait_for_transaction_receipt(tx_hash)
9    print("Buy transaction mined.")
10
11
12def sell():
13    approve_token(wethusdc_pool.token1.contract,
14                  WETH_TRADE_AMOUNT * 10**wethusdc_pool.token1.decimals)
15    txn = swap_router.functions.exactInput(SELL_PARAMS).build_transaction(txn_params())
16    signed_txn = w3.eth.account.sign_transaction(txn, private_key=PRIVATE_KEY)
17    tx_hash = w3.eth.send_raw_transaction(signed_txn.raw_transaction)
18    print(f"Sell transaction sent: {tx_hash.hex()}")
19    w3.eth.wait_for_transaction_receipt(tx_hash)
20    print("Sell transaction mined.")
显示全部
复制

1def balances():
2    token0_balance = wethusdc_pool.token0.contract.functions.balanceOf(account.address).call()
3    token1_balance = wethusdc_pool.token1.contract.functions.balanceOf(account.address).call()
4
5    print(f"{wethusdc_pool.token0.symbol} Balance: {Decimal(token0_balance) / Decimal(10 ** wethusdc_pool.token0.decimals)}")
6    print(f"{wethusdc_pool.token1.symbol} Balance: {Decimal(token1_balance) / Decimal(10 ** wethusdc_pool.token1.decimals)}")
复制

1print("交易前帐户余额：")
2balances()
3
4if (expected_price > current_price):
5    print(f"买入，我预计价格将上涨 {expected_price - current_price} 美元")
6    buy(wethusdc_quotes[-1])
7else:
8    print(f"卖出，我预计价格将下跌 {current_price - expected_price} 美元")
9    sell()
10
11print("交易后帐户余额：")
12balances()
显示全部
复制