DeFiスマートコントラクトをPythonで読む：ABI解析入門【詐欺検出付き】

最近、web3.pyを使ったDeFi開発が注目されていますね！どうやら、web3.pyを使うと、スマートコントラクトのABIを解析してオンチェーンデータを自動取得・詐欺検出まで自動化できるようです！

そこで今回は、Python環境で「web3.py」のABI解析・イベント監視・詐欺コントラクト検出の実装を行ってみました！コード1行ずつ丁寧に解説していますので、ぜひ皆さんも記事を読んで試してみてください！

ABIとは何か：コントラクトとの「会話方法」

「スマートコントラクトって難しそう…」と感じていませんか？実は、ABIさえ理解すれば、あとはPythonの関数を呼ぶのと変わりません。この記事を読めば、初心者の方でもオンチェーンデータを自在に取得できるようになります。

ABI（Application Binary Interface）とは、スマートコントラクトと外部アプリケーションが「どうやり取りするか」を定義した仕様書です。つまり、コントラクトが持つ関数の名前・引数・返り値の型を一覧化したJSON形式のドキュメントのこと。これがないと、EVMバイトコードを直接読む必要があり、現実的ではありません。

バイトコードとABIの関係をわかりやすく説明

Solidityで書かれたスマートコントラクトはコンパイルされると、2つのものが生成されます。①EVMが実行する「バイトコード」と、②人間やアプリが読むための「ABI」です。たとえば、レストランに例えると、バイトコードは厨房の調理手順（機械語）、ABIはお客様向けのメニュー表（関数一覧）に相当します。Pythonからは必ずABIを通じてコントラクトと会話します。

ABIのJSON構造を読み解く

ABIはJSON配列で記述されており、各要素がひとつの関数またはイベントを表します。主なフィールドは name（関数名）、inputs（引数）、outputs（返り値）、type（function / event / constructor）です。EtherscanでコントラクトページのContract→ABIタブを開くと、すぐにコピーできます。たとえば、USDCコントラクト（0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48）のABIはEtherscanから無料で取得可能です。

// abi_sample.json - ERC-20 balanceOf 関数のABI例
[
  {
    "name": "balanceOf",        // 関数名
    "type": "function",         // 種別
    "stateMutability": "view",  // 読み取り専用
    "inputs": [
      { "name": "account", "type": "address" }
    ],
    "outputs": [
      { "name": "", "type": "uint256" }
    ]
  }
]

web3.pyのセットアップ

ここではweb3.pyのインストールからチェーンへの接続確認まで、順を追って解説します。RPCエンドポイント選びで詰まる方が多いので、おすすめの選び方も紹介します。

インストールとRPCエンドポイントの選び方

まずはライブラリのインストールから始めましょう。

# requirements.txt - 必要パッケージ
pip install web3==6.20.3
pip install python-dotenv  # APIキーを環境変数で管理する

次に、RPCエンドポイントの選択です。RPCとは、つまりEthereumノードへのアクセス窓口のこと。主な選択肢を比較すると以下のとおりです。

サービス	無料枠	特徴
Infura	100,000 req/日	最も有名・安定性高い
Alchemy	300M compute units/月	デバッグ機能が充実
Cloudflare（無料RPC）	無制限（低速）	APIキー不要・学習用途向き

本番運用にはInfuraまたはAlchemyがおすすめです。学習・検証目的であれば、APIキー不要の公開RPCでも十分動作します。

チェーンへの接続確認コード

環境変数にAPIキーを設定したら、接続確認を行いましょう。

# connect_check.py - Ethereumメインネットへの接続確認
from web3 import Web3
from dotenv import load_dotenv
import os

load_dotenv()

# Infuraを使う場合
RPC_URL = f"https://mainnet.infura.io/v3/{os.getenv('INFURA_API_KEY')}"
# 無料RPCを使う場合は以下に置き換え可能
# RPC_URL = "https://cloudflare-eth.com"

w3 = Web3(Web3.HTTPProvider(RPC_URL))

# 接続確認
if w3.is_connected():
    print(f"✅ 接続成功！最新ブロック: {w3.eth.block_number}")
else:
    print("❌ 接続失敗。RPC URLを確認してください")

✅ 接続成功！最新ブロック: XXXXXXXX と表示されれば準備完了です。

ABIで関数を呼び出す

接続できたら、いよいよコントラクトの関数を呼び出してみましょう。実在するUSDCコントラクトを使って、残高確認から価格取得まで実装します。

ERC-20トークンの残高確認（実装コード）

USDCのコントラクトアドレス（0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48）を使って残高を取得します。ERC-20の標準ABIは最小限の定義で十分です。

# erc20_balance.py - ERC-20トークン残高確認
from web3 import Web3

# 最小限のERC-20 ABI（balanceOf / decimals / symbol）
ERC20_ABI = [
    {"name": "balanceOf", "type": "function", "stateMutability": "view",
     "inputs": [{"name": "account", "type": "address"}],
     "outputs": [{"name": "", "type": "uint256"}]},
    {"name": "decimals", "type": "function", "stateMutability": "view",
     "inputs": [], "outputs": [{"name": "", "type": "uint8"}]},
    {"name": "symbol", "type": "function", "stateMutability": "view",
     "inputs": [], "outputs": [{"name": "", "type": "string"}]},
]

# USDCコントラクト（Ethereum Mainnet）
USDC_ADDRESS = Web3.to_checksum_address("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")

contract = w3.eth.contract(address=USDC_ADDRESS, abi=ERC20_ABI)

# 調べたいウォレットアドレス
wallet = Web3.to_checksum_address("0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045")

symbol   = contract.functions.symbol().call()
decimals = contract.functions.decimals().call()
raw_bal  = contract.functions.balanceOf(wallet).call()
balance  = raw_bal / (10 ** decimals)  # 人間が読める単位に変換

print(f"{symbol} 残高: {balance:,.6f}")

Uniswap V2で価格を取得する

Uniswap V2のDAI/USDCペアコントラクト（0xAE461cA67B15dc8dc81CE7615e0320dA1A9aB8D5）を使って、リアルタイムの価格比率を取得します。getReserves()関数でプール内の残高比から価格を計算するのがポイントです。

# uniswap_price.py - Uniswap V2でDAI/USDC価格を取得
PAIR_ABI = [
    {"name": "getReserves", "type": "function", "stateMutability": "view",
     "inputs": [],
     "outputs": [
         {"name": "_reserve0", "type": "uint112"},
         {"name": "_reserve1", "type": "uint112"},
         {"name": "_blockTimestampLast", "type": "uint32"},
     ]},
]

# DAI/USDC ペアアドレス（Uniswap V2, Ethereum Mainnet）
PAIR_ADDRESS = Web3.to_checksum_address("0xAE461cA67B15dc8dc81CE7615e0320dA1A9aB8D5")
pair = w3.eth.contract(address=PAIR_ADDRESS, abi=PAIR_ABI)

r0, r1, _ = pair.functions.getReserves().call()

# token0=DAI(18 decimals), token1=USDC(6 decimals) の比率
dai_reserve  = r0 / 1e18
usdc_reserve = r1 / 1e6

price_dai_in_usdc = usdc_reserve / dai_reserve
print(f"1 DAI = {price_dai_in_usdc:.6f} USDC")

callとtransactの違い

web3.pyでコントラクト関数を呼ぶ方法は2種類あります。つまり、読む（call）と書く（transact）の違いです。

メソッド	用途	ガス費用
.call()	状態を読み取るのみ（view/pure関数）	不要
.transact()	状態を変更する（送金・承認など）	必要
.build_transaction()	署名前のトランザクション構築	必要（後払い）

残高確認・価格取得などは必ず .call() を使い、秘密鍵やガスを消費しないようにしましょう。

イベントログを監視する

オンチェーンデータの醍醐味は、リアルタイムのイベント監視にあります。ERC-20のTransferイベントを例に、ライブ検知と過去ブロックの一括取得を実装します。

Transferイベントのリアルタイム検知

イベント監視には create_filter を使います。たとえば、大口送金を検知してアラートを出すBotを作るときに活躍します。WebSocketエンドポイントを使うと、より低遅延な監視が可能です。

# event_monitor.py - ERC-20 Transferイベントのリアルタイム監視
import time

ERC20_ABI_EVENT = [
    {"name": "Transfer", "type": "event",
     "inputs": [
         {"name": "from",  "type": "address", "indexed": True},
         {"name": "to",    "type": "address", "indexed": True},
         {"name": "value", "type": "uint256", "indexed": False},
     ]},
]

usdc = w3.eth.contract(address=USDC_ADDRESS, abi=ERC20_ABI_EVENT)

# 最新ブロックからフィルタ作成
transfer_filter = usdc.events.Transfer.create_filter(from_block="latest")

print("監視開始... Ctrl+Cで停止")
while True:
    for event in transfer_filter.get_new_entries():
        amount = event["args"]["value"] / 1e6  # USDC decimals=6
        print(f"Transfer: {amount:,.2f} USDC  from={event['args']['from'][:10]}...")
    time.sleep(2)  # 2秒ごとにポーリング

過去ブロックのイベントを一括取得する

直近N日分のイベントをまとめて取得したい場合は get_logs を使います。無料RPCでは1リクエストあたり取得できるブロック範囲に上限があるため、範囲を分割してループするのがポイントです。

# event_historical.py - 過去ブロックのTransferイベントを一括取得
from_block = w3.eth.block_number - 1000  # 過去1000ブロック分
to_block   = w3.eth.block_number

# get_logsで過去イベントを取得（フリーRPCは範囲を2000ブロック以内に分割推奨）
logs = usdc.events.Transfer.get_logs(
    from_block=from_block,
    to_block=to_block,
)

print(f"取得イベント数: {len(logs)}")
for log in logs[:5]:  # 最初の5件を表示
    amount = log["args"]["value"] / 1e6
    print(f"  Block {log['blockNumber']}: {amount:,.2f} USDC")

怪しいコントラクトを見抜く3つのチェック

ここからは、投資前の「予防知識」として身につけておきたい詐欺検出の手法を紹介します。これらのチェックを習慣にするだけで、rug pullやhoneypotによる被害を大幅に減らせます。

① mint権限が残っていないか確認する方法

Ownerがいつでも新しいトークンを大量発行できる状態は危険信号です。つまり、供給量を意のままに増やしてトークン価値を希薄化できるということ。owner()関数でオーナーアドレスを確認し、ゼロアドレスかどうかをチェックしましょう。

# check_mint.py - mint権限（owner）の確認
OWNER_ABI = [
    {"name": "owner", "type": "function", "stateMutability": "view",
     "inputs": [], "outputs": [{"name": "", "type": "address"}]},
]

ZERO_ADDRESS = "0x0000000000000000000000000000000000000000"

def check_ownership(token_address: str) -> dict:
    contract = w3.eth.contract(
        address=Web3.to_checksum_address(token_address),
        abi=OWNER_ABI
    )
    try:
        owner = contract.functions.owner().call()
        renounced = owner.lower() == ZERO_ADDRESS
        return {
            "owner": owner,
            "renounced": renounced,
            # ゼロアドレスなら所有権放棄 = mint不可（安全）
            "safe": renounced,
        }
    except Exception:
        # owner()関数が存在しない場合
        return {"owner": "N/A", "renounced": False, "safe": None}

result = check_ownership("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")
print(result)

② 流動性ロックの確認手順

流動性（Liquidity）がロックされていないプロジェクトは、開発者がいつでもプールからETHを引き出せます。これがrug pullの典型的な手口です。Pythonからは、LPトークンのホルダーリストにUnicryptやTeam Financeなどの既知ロッカーアドレスが含まれているか確認できます。

# check_liquidity_lock.py - LPトークンの主要ホルダー確認
LP_ABI = [
    {"name": "balanceOf", "type": "function", "stateMutability": "view",
     "inputs": [{"name": "account", "type": "address"}],
     "outputs": [{"name": "", "type": "uint256"}]},
    {"name": "totalSupply", "type": "function", "stateMutability": "view",
     "inputs": [], "outputs": [{"name": "", "type": "uint256"}]},
]

# 主要な流動性ロッカーの既知アドレス（例）
KNOWN_LOCKERS = {
    "0x663A5C229c09b049E36dCc11a9B0d4a8Eb9db214": "Unicrypt",
    "0xe2fe530c047f2d85298b07d9333c05737f1435fb": "Team Finance",
    "0x0000000000000000000000000000000000000000": "Dead(Burned)",
}

def check_lp_lock(lp_address: str, locker_address: str) -> dict:
    lp = w3.eth.contract(
        address=Web3.to_checksum_address(lp_address), abi=LP_ABI
    )
    total   = lp.functions.totalSupply().call()
    locked  = lp.functions.balanceOf(
        Web3.to_checksum_address(locker_address)
    ).call()
    ratio   = locked / total * 100 if total > 0 else 0
    locker_name = KNOWN_LOCKERS.get(locker_address.lower(), "Unknown")
    return {
        "locker": locker_name,
        "locked_pct": f"{ratio:.1f}%",
        # 80%以上ロックされていれば比較的安全
        "safe": ratio >= 80,
    }

③ honeypot（売れないトークン）の検出ロジック

honeypotとは、買うことはできても売ることができないように細工されたトークンのことです。コントラクトのABIに transfer / approve が存在するか確認し、さらにブラックリスト関数や手数料を異常に高くする関数が定義されていないかをチェックします。

# check_honeypot.py - honeypotの疑いをABIから静的チェック
import requests
import json

# EtherscanのAPIでABIを取得（無料APIキー要）
def fetch_abi_from_etherscan(address: str, api_key: str) -> list:
    url = (
        f"https://api.etherscan.io/api"
        f"?module=contract&action=getabi"
        f"&address={address}&apikey={api_key}"
    )
    resp = requests.get(url, timeout=10)
    data = resp.json()
    if data["status"] == "1":
        return json.loads(data["result"])
    return []

# 危険フラグとなりうる関数名リスト
DANGER_FUNCTIONS = {
    "blacklist", "addToBlacklist", "setMaxTxAmount",
    "setFee", "pause", "setTaxFee", "setBuyFee", "setSellFee",
}

def check_honeypot_static(abi: list) -> dict:
    func_names = {
        item["name"] for item in abi
        if item.get("type") == "function"
    }
    # 基本関数の存在確認
    has_transfer = "transfer" in func_names
    has_approve  = "approve"  in func_names
    # 危険関数の検出
    danger_found = DANGER_FUNCTIONS & func_names

    return {
        "has_transfer": has_transfer,
        "has_approve":  has_approve,
        "danger_functions": list(danger_found),
        # 危険関数が0件 かつ transfer/approveが存在すれば静的チェックはOK
        "static_ok": has_transfer and has_approve and len(danger_found) == 0,
    }

# 使用例
# abi = fetch_abi_from_etherscan("0x...", "YOUR_API_KEY")
# print(check_honeypot_static(abi))

なお、静的チェックだけでは限界もあります。より精度の高い検証には honeypot.is のようなシミュレーションAPIと組み合わせることをおすすめします。

まとめ：DeFi開発の次のステップへ

今回は、web3.pyによるABI解析・関数呼び出し・イベント監視・詐欺検出の実装に挑戦してみました。

ethers.jsも便利ですが、「web3.py」はPythonエコシステム（pandas・numpy・機械学習ライブラリ）とシームレスに連携できるため、オンチェーンデータ分析やBot開発においては更に便利です。これは使わないのはもったいない！と感じました。

導入も pip install web3 一発ですので、みなさんも今回の記事を参考に、ぜひ「web3.py」を活用してみてください！

関連記事