イーサリアムで独自の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以下の見積もりを前提として:
2Asset: WETH/USDC
3        2026-01-20T16:34 3016.21
4        .
5        .
6        .
7        2026-02-01T17:49 2299.10
8
9Asset: WBTC/WETH
10        2026-01-20T16:34 29.84
11        .
12        .
13        .
14        2026-02-01T17:50 33.46
15
16
172026-02-02T17:56の時点でWETH/USDCの値はいくらになると予想しますか?
18
19回答は、他のテキストを含まず、小数点以下2桁に丸めた単一の数値として提供してください。
すべて表示

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, "Block is in the future"
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"""
コピー

1Given these quotes:
2{
3    functools.reduce(lambda acc, q: acc + '\n' + q,
4        map(lambda q: format_quotes(q), quotes))
5}
コピー

1What would you expect the value for {asset} to be at time {expected_time}?
2
3Provide your answer as a single number rounded to two decimal places,
4without any other text.
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 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 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 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"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} USD上昇すると予想します")
6    buy(wethusdc_quotes[-1])
7else:
8    print(f"売り、価格が{current_price - expected_price} USD下落すると予想します")
9    sell()
10
11print("取引後の勘定残高:")
12balances()
すべて表示
コピー