在以太坊上打造你自己的 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  # (每 token0 的 token1 數量)
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您預期在 {expected_time} 時，{asset} 的價值會是多少？
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 ("目前價格：", wethusdc_quotes[-1].price)
5print(f"在 {future_time}，預期價格：{expected_price} USD")
6
7if (expected_price > current_price):
8    print(f"買入，我預期價格會上漲 {expected_price - current_price} USD")
9else:
10    print(f"賣出，我預期價格會下跌 {current_price - expected_price} USD")
顯示全部
複製

1uv run test-predictor.py

12026-01-05T19:50 的預測：預測 3138.93 USD，實際 3218.92 USD，誤差 79.99 USD
22026-01-06T19:56 的預測：預測 3243.39 USD，實際 3221.08 USD，誤差 22.31 USD
32026-01-07T20:02 的預測：預測 3223.24 USD，實際 3146.89 USD，誤差 76.35 USD
42026-01-08T20:11 的預測：預測 3150.47 USD，實際 3092.04 USD，誤差 58.43 USD
5.
6.
7.
82026-01-31T22:33 的預測：預測 2637.73 USD，實際 2417.77 USD，誤差 219.96 USD
92026-02-01T22:41 的預測：預測 2381.70 USD，實際 2318.84 USD，誤差 62.86 USD
102026-02-02T22:49 的預測：預測 2234.91 USD，實際 2349.28 USD，誤差 114.37 USD
1129 次預測的平均預測誤差：83.87103448275862068965517241 USD
12每次建議的平均變動：4.787931034482758620689655172 USD
13變動的標準差：104.42 USD
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_time} 的預測：預測 {predicted_price} USD，實際 {real_price} USD，誤差 {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"在 {len(wethusdc_quotes)-CYCLES_BACK} 次預測中的平均預測誤差：{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} USD")
6var = sum((x - mean_change) ** 2 for x in changes) / length_changes
7print (f"變動的標準差：{var.sqrt().quantize(Decimal("0.01"))} USD")
複製

1print (f"獲利天數：{len(list(filter(lambda x: x > 0, changes)))/length_changes:.2%}")
2print (f"虧損天數：{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
3在 2026-02-06T23:07，預期價格：1724.41 USD
4交易前帳戶餘額：
5USDC 餘額：927301.578272
6WETH 餘額：500
7賣出，我預期價格會下跌 118.75 USD
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"授權交易已發送：{tx_hash.hex()}")
2    w3.eth.wait_for_transaction_receipt(tx_hash)
3    print("授權交易已上鏈。")
複製

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"買入交易已發送：{tx_hash.hex()}")
8    w3.eth.wait_for_transaction_receipt(tx_hash)
9    print("買入交易已上鏈。")
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"賣出交易已發送：{tx_hash.hex()}")
19    w3.eth.wait_for_transaction_receipt(tx_hash)
20    print("賣出交易已上鏈。")
顯示全部
複製

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} 餘額：{Decimal(token0_balance) / Decimal(10 ** wethusdc_pool.token0.decimals)}")
6    print(f"{wethusdc_pool.token1.symbol} 餘額：{Decimal(token1_balance) / Decimal(10 ** wethusdc_pool.token1.decimals)}")
複製

1print("交易前帳戶餘額：")
2balances()
3
4if (expected_price > current_price):
5    print(f"買入，我預期價格會上漲 {expected_price - current_price} USD")
6    buy(wethusdc_quotes[-1])
7else:
8    print(f"賣出，我預期價格會下跌 {current_price - expected_price} USD")
9    sell()
10
11print("交易後帳戶餘額：")
12balances()
顯示全部
複製