DeFi Bot開発入門 — web3.py/ethers.jsでオンチェーン自動取引

はじめに

DeFi（分散型金融）のBot開発は、CEX（中央集権型取引所）のBot開発とは根本的に異なります。取引所APIの代わりにブロックチェーンのスマートコントラクトと直接対話し、注文の概念がなく、トランザクションの送信とガス代の支払いが取引の基本単位となります。

本記事では、Ethereum（web3.py/ethers.js）とSolana（Anchor/Raydium SDK）の両方をカバーし、オンチェーン自動取引システムの実践的な構築手法を解説します。

CEX Bot vs DeFi Botの比較

---

web3.py / ethers.js の基礎

web3.py（Python）

from web3 import Web3
from eth_account import Account
import os

# RPC接続（Alchemy/Infura推奨、公開RPCは不安定）
w3 = Web3(Web3.HTTPProvider(os.getenv("ETH_RPC_URL")))
print(f"接続状態: {w3.is_connected()}")
print(f"最新ブロック: {w3.eth.block_number}")

# ウォレット設定
account = Account.from_key(os.getenv("PRIVATE_KEY"))
print(f"アドレス: {account.address}")

# ETH残高確認
balance = w3.eth.get_balance(account.address)
print(f"ETH残高: {w3.from_wei(balance, 'ether')} ETH")

# ERC-20トークン残高確認
ERC20_ABI = [
    {
        "inputs": [{"name": "account", "type": "address"}],
        "name": "balanceOf",
        "outputs": [{"name": "", "type": "uint256"}],
        "stateMutability": "view",
        "type": "function",
    }
]
USDC_ADDRESS = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
usdc = w3.eth.contract(address=USDC_ADDRESS, abi=ERC20_ABI)
usdc_balance = usdc.functions.balanceOf(account.address).call()
print(f"USDC残高: {usdc_balance / 1e6}")

ethers.js（TypeScript/JavaScript）

import { ethers } from "ethers";

// RPC接続
const provider = new ethers.JsonRpcProvider(process.env.ETH_RPC_URL);
const wallet = new ethers.Wallet(process.env.PRIVATE_KEY!, provider);

console.log(`アドレス: ${wallet.address}`);

// 残高確認
const balance = await provider.getBalance(wallet.address);
console.log(`ETH残高: ${ethers.formatEther(balance)} ETH`);

// ERC-20トークン残高
const usdcAbi = ["function balanceOf(address) view returns (uint256)"];
const usdc = new ethers.Contract(
  "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
  usdcAbi,
  provider
);
const usdcBalance = await usdc.balanceOf(wallet.address);
console.log(`USDC残高: ${ethers.formatUnits(usdcBalance, 6)}`);

RPCプロバイダーの比較

---

DEXインタラクション

Uniswap V3でのスワップ実装

from web3 import Web3
import json

# Uniswap V3 SwapRouter02
SWAP_ROUTER = "0x68b3465833fb72A70ecDF485E0e4C7bD8665Fc45"

SWAP_ROUTER_ABI = json.loads("""[{
    "inputs": [{
        "components": [
            {"name": "tokenIn", "type": "address"},
            {"name": "tokenOut", "type": "address"},
            {"name": "fee", "type": "uint24"},
            {"name": "recipient", "type": "address"},
            {"name": "amountIn", "type": "uint256"},
            {"name": "amountOutMinimum", "type": "uint256"},
            {"name": "sqrtPriceLimitX96", "type": "uint160"}
        ],
        "name": "params",
        "type": "tuple"
    }],
    "name": "exactInputSingle",
    "outputs": [{"name": "amountOut", "type": "uint256"}],
    "stateMutability": "payable",
    "type": "function"
}]""")

def swap_exact_input(
    w3: Web3,
    account,
    token_in: str,
    token_out: str,
    amount_in: int,
    fee: int = 3000,
    slippage_pct: float = 0.5,
) -> str:
    """Uniswap V3でトークンをスワップ"""
    router = w3.eth.contract(
        address=SWAP_ROUTER, abi=SWAP_ROUTER_ABI
    )

    # 最小受取量を計算（スリッページ考慮）
    # 実運用ではオラクル価格から計算する
    amount_out_min = 0  # 簡略化のため0（本番では必ず設定）

    # Approve（初回のみ）
    token_contract = w3.eth.contract(
        address=token_in,
        abi=[{
            "inputs": [
                {"name": "spender", "type": "address"},
                {"name": "amount", "type": "uint256"},
            ],
            "name": "approve",
            "outputs": [{"name": "", "type": "bool"}],
            "stateMutability": "nonpayable",
            "type": "function",
        }],
    )
    approve_tx = token_contract.functions.approve(
        SWAP_ROUTER, amount_in
    ).build_transaction({
        "from": account.address,
        "nonce": w3.eth.get_transaction_count(account.address),
        "gas": 60000,
        "maxFeePerGas": w3.eth.gas_price * 2,
        "maxPriorityFeePerGas": w3.to_wei(2, "gwei"),
    })
    signed = account.sign_transaction(approve_tx)
    w3.eth.send_raw_transaction(signed.raw_transaction)

    # スワップ実行
    swap_params = {
        "tokenIn": token_in,
        "tokenOut": token_out,
        "fee": fee,
        "recipient": account.address,
        "amountIn": amount_in,
        "amountOutMinimum": amount_out_min,
        "sqrtPriceLimitX96": 0,
    }

    tx = router.functions.exactInputSingle(swap_params).build_transaction({
        "from": account.address,
        "nonce": w3.eth.get_transaction_count(account.address),
        "gas": 250000,
        "maxFeePerGas": w3.eth.gas_price * 2,
        "maxPriorityFeePerGas": w3.to_wei(2, "gwei"),
    })
    signed = account.sign_transaction(tx)
    tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
    receipt = w3.eth.wait_for_transaction_receipt(tx_hash)

    return receipt["transactionHash"].hex()

Uniswap手数料ティアの選択

---

ガス最適化

EIP-1559ガス戦略

def estimate_gas_params(w3: Web3) -> dict:
    """EIP-1559ガスパラメータを最適化"""
    latest_block = w3.eth.get_block("latest")
    base_fee = latest_block["baseFeePerGas"]

    # 過去5ブロックの優先手数料を分析
    priority_fees = []
    for i in range(5):
        block = w3.eth.get_block(latest_block["number"] - i)
        if block["transactions"]:
            # 中央値の優先手数料を推定
            priority_fees.append(w3.to_wei(2, "gwei"))

    avg_priority = sum(priority_fees) // len(priority_fees)

    return {
        "maxFeePerGas": base_fee * 2 + avg_priority,
        "maxPriorityFeePerGas": avg_priority,
    }

def estimate_transaction_cost(
    w3: Web3, gas_used: int
) -> dict:
    """トランザクションコストを見積もり"""
    gas_params = estimate_gas_params(w3)
    eth_price_usd = 3500  # 外部APIから取得推奨

    cost_wei = gas_used * gas_params["maxFeePerGas"]
    cost_eth = w3.from_wei(cost_wei, "ether")
    cost_usd = float(cost_eth) * eth_price_usd

    return {
        "gas_used": gas_used,
        "cost_eth": float(cost_eth),
        "cost_usd": cost_usd,
    }

ガスコストの目安

---

フラッシュローンの実装

Aaveフラッシュローン

フラッシュローンは、同一トランザクション内で借入と返済を完了する無担保ローンです。アービトラージや清算に活用されます。

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;

import "@aave/v3-core/contracts/flashloan/base/FlashLoanSimpleReceiverBase.sol";
import "@aave/v3-core/contracts/interfaces/IPoolAddressesProvider.sol";
import "@openzeppelin/contracts/token/ERC20/IERC20.sol";

contract FlashLoanArbitrage is FlashLoanSimpleReceiverBase {
    address public owner;

    constructor(address _addressProvider)
        FlashLoanSimpleReceiverBase(
            IPoolAddressesProvider(_addressProvider)
        )
    {
        owner = msg.sender;
    }

    function executeOperation(
        address asset,
        uint256 amount,
        uint256 premium,
        address initiator,
        bytes calldata params
    ) external override returns (bool) {
        // ここにアービトラージロジックを実装
        // 例: DEX AでトークンAを買い、DEX Bで売る

        // 返済額 = 借入額 + 手数料(0.05%)
        uint256 amountOwed = amount + premium;
        IERC20(asset).approve(
            address(POOL), amountOwed
        );

        return true;
    }

    function requestFlashLoan(
        address asset,
        uint256 amount
    ) external {
        require(msg.sender == owner, "Not owner");
        POOL.flashLoanSimple(
            address(this),
            asset,
            amount,
            "",
            0
        );
    }
}

Balancerフラッシュローン

// Balancerのフラッシュローンは手数料0
import "@balancer-labs/v2-interfaces/contracts/vault/IVault.sol";
import "@balancer-labs/v2-interfaces/contracts/vault/IFlashLoanRecipient.sol";

contract BalancerFlashLoan is IFlashLoanRecipient {
    IVault private constant vault =
        IVault(0xBA12222222228d8Ba445958a75a0704d566BF2C8);

    function receiveFlashLoan(
        IERC20[] memory tokens,
        uint256[] memory amounts,
        uint256[] memory feeAmounts,
        bytes memory userData
    ) external override {
        require(msg.sender == address(vault), "Not vault");

        // アービトラージロジック

        // 返済（Balancerは手数料0）
        for (uint256 i = 0; i < tokens.length; i++) {
            tokens[i].transfer(
                address(vault),
                amounts[i] + feeAmounts[i]
            );
        }
    }
}

フラッシュローンプロバイダーの比較

---

MEV保護

MEVとは

MEV（Maximal Extractable Value）は、トランザクションの順序操作によってバリデーターや検索者が得られる利益です。一般ユーザーにとってはサンドイッチ攻撃などの形で損失の原因となります。

サンドイッチ攻撃の仕組み

1. あなた: USDC → ETH のスワップTXをメンプールに送信
2. 攻撃者: あなたのTXの前に ETH を大量購入（フロントラン）
3. あなた: ETH価格が上昇した状態で購入（不利な価格で約定）
4. 攻撃者: あなたのTXの後に ETH を売却（バックラン）→ 利益確保

Flashbots Protectの活用

from web3 import Web3

# Flashbots RPCを使用（トランザクションがメンプールに公開されない）
FLASHBOTS_RPC = "https://rpc.flashbots.net"
w3_flashbots = Web3(Web3.HTTPProvider(FLASHBOTS_RPC))

def send_private_transaction(w3, account, tx_data):
    """Flashbots経由でプライベートトランザクションを送信"""
    signed_tx = account.sign_transaction(tx_data)
    tx_hash = w3.eth.send_raw_transaction(
        signed_tx.raw_transaction
    )
    return tx_hash

MEV保護の選択肢

---

Multicallバッチ処理

複数コントラクト呼び出しの一括実行

from multicall import Call, Multicall

# 複数のトークン残高を1回のRPC呼び出しで取得
def batch_balance_check(
    w3: Web3,
    wallet: str,
    tokens: list[dict],
) -> dict:
    """複数トークン残高を一括取得（ガス節約）"""
    calls = [
        Call(
            token["address"],
            ["balanceOf(address)(uint256)", wallet],
            [[token["symbol"], lambda x: x]],
        )
        for token in tokens
    ]
    result = Multicall(calls, _w3=w3)()
    return result

# 使用例
tokens = [
    {"symbol": "USDC", "address": "0xA0b8..."},
    {"symbol": "WETH", "address": "0xC02a..."},
    {"symbol": "DAI", "address": "0x6B17..."},
]
balances = batch_balance_check(w3, account.address, tokens)

---

イベント監視：清算候補の検出

import asyncio
from web3 import Web3

async def monitor_liquidation_candidates(w3: Web3):
    """Aaveの清算候補をリアルタイム監視"""
    AAVE_POOL = "0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2"

    # Borrowイベントを監視
    borrow_filter = w3.eth.filter({
        "address": AAVE_POOL,
        "topics": [
            w3.keccak(text="Borrow(address,address,address,uint256,uint8,uint256,uint16)")
        ],
    })

    while True:
        events = borrow_filter.get_new_entries()
        for event in events:
            # ヘルスファクターをチェック
            user = event["args"]["onBehalfOf"]
            user_data = aave_pool.functions.getUserAccountData(
                user
            ).call()
            health_factor = user_data[5] / 1e18

            if health_factor < 1.1:
                print(
                    f"清算候補検出: {user} "
                    f"HF={health_factor:.4f}"
                )
                # 清算トランザクションの送信ロジック

        await asyncio.sleep(1)

---

Solana版：Anchor + Raydium SDK

Solana開発環境

# Solana CLI + Anchor インストール
sh -c "$(curl -sSfL https://release.anza.xyz/stable/install)"
cargo install --git https://github.com/coral-xyz/anchor anchor-cli

# プロジェクト作成
anchor init my-defi-bot
cd my-defi-bot

Raydium SDKでのスワップ

import { Connection, Keypair, PublicKey } from "@solana/web3.js";
import { Raydium } from "@raydium-io/raydium-sdk-v2";

const connection = new Connection(process.env.SOLANA_RPC_URL!);
const wallet = Keypair.fromSecretKey(
  Buffer.from(JSON.parse(process.env.SOLANA_PRIVATE_KEY!))
);

async function swapOnRaydium(
  inputMint: string,
  outputMint: string,
  amountIn: number,
  slippageBps: number = 50, // 0.5%
) {
  const raydium = await Raydium.load({
    connection,
    owner: wallet,
    cluster: "mainnet",
  });

  // ルート計算
  const routes = await raydium.api.fetchSwapRoutes({
    inputMint,
    outputMint,
    amount: amountIn,
    slippageBps,
  });

  // トランザクション構築・送信
  const { execute } = await raydium.swap({
    swapResponse: routes,
  });

  const txId = await execute();
  console.log(`スワップ完了: https://solscan.io/tx/${txId}`);
  return txId;
}

// SOL → USDC スワップ
await swapOnRaydium(
  "So11111111111111111111111111111111111111112", // SOL
  "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", // USDC
  1_000_000_000, // 1 SOL (lamports)
);

Ethereum vs Solanaの比較

---

テストネット検証フロー

本番デプロイ前のチェックリスト

Foundryによるフォークテスト

# メインネットをフォークしてテスト実行
forge test --fork-url $ETH_RPC_URL \
  --fork-block-number 19500000 \
  -vvv

# ガスレポートの出力
forge test --gas-report

---

ガスコスト vs 利益のシミュレーション

def simulate_profitability(
    gross_profit_usd: float,
    gas_cost_usd: float,
    success_rate: float = 0.6,
    attempts_per_day: int = 10,
) -> dict:
    """DeFi Bot運用の損益シミュレーション"""
    expected_profit = gross_profit_usd * success_rate
    expected_cost = gas_cost_usd  # 失敗してもガス代は発生
    net_per_attempt = expected_profit - expected_cost

    daily_pnl = net_per_attempt * attempts_per_day
    monthly_pnl = daily_pnl * 30
    annual_pnl = daily_pnl * 365

    return {
        "net_per_attempt": net_per_attempt,
        "daily_pnl": daily_pnl,
        "monthly_pnl": monthly_pnl,
        "annual_pnl": annual_pnl,
        "breakeven_success_rate": gas_cost_usd / gross_profit_usd,
    }

# アービトラージBotのシミュレーション
result = simulate_profitability(
    gross_profit_usd=50,    # 成功時の利益
    gas_cost_usd=20,        # ガス代
    success_rate=0.5,       # 成功率50%
    attempts_per_day=20,    # 1日20回試行
)
print(f"日次PnL: ${result['daily_pnl']:.2f}")
print(f"月次PnL: ${result['monthly_pnl']:.2f}")
print(f"損益分岐成功率: {result['breakeven_success_rate']:.1%}")

---

まとめ

DeFi Bot開発は、CEX Botと比較して技術的な複雑性が高く、リスクも大きい反面、透明性が高く、フラッシュローンなどの独自機能を活用できるメリットがあります。

開発の推奨ステップ

1. テストネットでweb3.py/ethers.jsの基本操作を習得
2. フォークテストでスマートコントラクトの動作を検証
3. ガスコスト vs 期待利益のシミュレーションで収益性を確認
4. MEV保護（Flashbots）を必ず導入
5. 少額（$100以下）で本番検証
6. 段階的にスケールアップ

重要な注意点

• 秘密鍵の管理: 環境変数やシークレットマネージャーで管理。コードにハードコードしない
• Approve量の制限: 無制限approveを避け、必要最小限に設定
• コントラクト検証: 相互作用するコントラクトのソースコードを必ず確認
• ガス代の変動: Ethereumのガスはネットワーク混雑で10倍以上変動する可能性がある
• スマートコントラクトリスク: ハッキングやバグによる資金喪失リスクが常に存在する