仮想通貨について、以前からやりたかったこと。

ブロックチェーンは透明性が高いので、デプロイされたばかりの新規トークンを発見することはできるのか？

これができると、色々と夢がありそうです。
実際の売買やエアドロップはTelegramやDiscordで情報を得る必要があると思いますが、できれば自分で発見したいです。

この記事では、Etherscan APIを利用してEthereumネットワーク上に新しくデプロイされたトークンを探索してみます。

要件

Etherscan APIを使用して、以下の手順で新規トークンを探索します。

1. 最新のブロック番号を取得
Etherscan APIを使用して最新のブロック番号を取得。

2. 最後に処理したブロック番号を読み込む
last_block.txtファイルから最後に処理したブロック番号を読み込む。
ファイルが存在しない場合は、Noneを返す。

3. 最新のブロック番号との差分を計算し、未処理のブロックを処理対象にする
初回実行時（last_blockがNoneの場合）は、最新ブロック番号の10ブロック前から処理を開始する。

4. 各未処理ブロックのトランザクションを取得
各未処理のブロック番号に対し、Etherscan APIを使用してトランザクションデータを取得する。

5. コントラクト作成トランザクションを検出する
各トランザクションのtoアドレスがNoneの場合、そのトランザクションがコントラクト作成であると判断する。

6. コントラクトアドレスに対してtotalSupplyを取得する
検出したコントラクトアドレスについて、Etherscan APIを使用してtotalSupplyメソッドを呼び出す。

7. トークンコントラクトである可能性を判定する
totalSupplyが0より大きい場合、新しいトークンコントラクトである可能性が高いと判断し、コントラクトアドレスやtotalSupplyを出力する。

8. 処理したブロック番号を保存する
処理が完了したブロック番号をlast_block.txtファイルに保存する。

9. 一定時間待機して再実行する
指定された間隔（INTERVAL秒）で待機し、手順1から繰り返す。

スクリプト

必要なライブラリ:

pip install requests

コード:

token_scan.py

import requests
import time
import os

# Etherscan APIキー（取得用のリンクは後述）
ETHERSCAN_API_KEY = "xxxxxxxxxx"

# 監視間隔 (秒) - 10秒ごとにブロックをチェック
INTERVAL = 10

# 最後に確認したブロック番号を保存するファイル
LAST_BLOCK_FILE = "last_block.txt"


def get_latest_block():
    """
    最新のブロック番号を取得します。

    Returns:
        int: 最新のブロック番号 (10進数)。エラーが発生した場合は None。
    """
    url = f"https://api.etherscan.io/api?module=proxy&action=eth_blockNumber&apikey={ETHERSCAN_API_KEY}"
    try:
        # Etherscan APIにリクエストを送信
        response = requests.get(url)  
        response.raise_for_status()
        data = response.json()
        # レート制限対策 - Etherscan APIへのリクエスト間隔を1秒空ける
        time.sleep(1)

        # レスポンスが有効な形式であることを確認
        if data["jsonrpc"] == "2.0" and data["id"] is not None and data["result"]:
            # 16進数のブロック番号を10進数に変換して返す
            return int(data["result"], 16)
        else:
            print(f"Error getting latest block: Unexpected response format: {data}")
            return None
    except requests.exceptions.RequestException as e:
        # リクエストエラーが発生した場合のエラーメッセージ
        print(f"Error getting latest block: {e}")
        return None


def get_block_transactions(block_number):
    """
    指定されたブロックに含まれるトランザクションを取得します。

    Args:
        block_number (int): ブロック番号 (10進数)。

    Returns:
        list: トランザクションのリスト。エラーが発生した場合は None。
    """
    url = f"https://api.etherscan.io/api?module=proxy&action=eth_getBlockByNumber&tag=0x{hex(block_number)[2:]}&boolean=true&apikey={ETHERSCAN_API_KEY}"
    try:
        # Etherscan APIにリクエストを送信
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        # レート制限対策 - Etherscan APIへのリクエスト間隔を1秒空ける
        time.sleep(1)

        # レスポンスが有効な形式であることを確認
        if data["jsonrpc"] == "2.0" and data["id"] is not None and data["result"]:
            return data["result"]["transactions"]
        else:
            print(f"Error getting block transactions: Unexpected response format: {data}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"Error getting block transactions: {e}")
        return None


def check_for_new_tokens(transactions, block_number):
    """
    トランザクションリストから新しいトークンコントラクトを探します。

    Args:
        transactions (list): トランザクションのリスト。
        block_number (int):  チェック対象のブロック番号。
    """
    # 処理中のブロック番号を表示
    print(f"Checking block {block_number}")
    from_to_count = 0

    for tx in transactions:
        # 数が多いので、5件以上出力したらループを抜ける
        if from_to_count >= 5:
            break

        from_addr = None
        to_addr = None

        # fromアドレスとtoアドレスを最大5つまで表示
        if "from" in tx:
            from_addr = tx['from']

        if "to" in tx and tx["to"] is not None:
            to_addr = tx['to']

        # 新しいコントラクトが作成されたトランザクション(toアドレスがNone)を検出
        if tx["to"] is None and "contractAddress" in tx:
            # 新しいコントラクトのアドレスを取得
            contract_address = tx["contractAddress"]

            try:
                totalSupply_url = f"https://api.etherscan.io/api?module=proxy&action=eth_call&to={contract_address}&data=0x18160ddd&apikey={ETHERSCAN_API_KEY}"
                #totalSupplyの取得をEtherscan APIで行う
                totalSupply_response = requests.get(totalSupply_url)
                totalSupply_data = totalSupply_response.json()

                # API呼び出しが成功したことを確認
                if totalSupply_data.get("status") == "1":
                    total_supply_hex = totalSupply_data["result"]
                    # totalSupplyが0x(空)の場合のエラー処理
                    if total_supply_hex == "0x":
                        total_supply = 0
                    else:
                        total_supply = int(total_supply_hex, 16)

                    # totalSupplyが0より大きい場合、新しいトークンとして表示
                    if total_supply > 0:
                        print(
                            f"Potential new token contract (totalSupply={total_supply}): {contract_address} from: {tx.get('from')}")
                else:
                     print(f"Error getting totalSupply for {contract_address}: {totalSupply_data.get('message', 'Unknown error')}")
            # 予期せぬエラーが発生した場合の処理
            except Exception as e:
                print(f"Error checking totalSupply for {contract_address}: {e}")

        # コントラクト作成トランザクションだがcontractAddressがない場合
        elif tx["to"] is None:
            # print(f"Contract creation transaction without contractAddress: {tx['hash']} from: {tx.get('from')}")
            pass

        # 通常のトランザクションの場合、fromとtoアドレスを表示
        elif from_addr is not None and to_addr is not None:
            print(f"Regular transaction: from: {from_addr} to: {to_addr}")
            from_to_count += 1


def load_last_block():
    """最後に処理したブロック番号を読み込みます。"""
    try:
        with open(LAST_BLOCK_FILE, "r") as f:
            return int(f.read())
    except FileNotFoundError:
        return None


def save_last_block(block_number):
    """最後に処理したブロック番号を保存します。"""
    with open(LAST_BLOCK_FILE, "w") as f:
        f.write(str(block_number))


if __name__ == "__main__":
    while True:
        # 最新のブロック番号を取得
        latest_block = get_latest_block()
        # 取得に失敗した場合はcontinue
        if latest_block is None:
            continue

        # 最後に処理したブロック番号を読み込む
        last_block = load_last_block()

        # 初めての実行の場合は、最新のブロック番号から10ブロック前から開始
        if last_block is None:
            last_block = latest_block - 10

        # 最後に処理したブロックの次から最新のブロックまでループ
        for block_number in range(last_block + 1, latest_block + 1):
            # ブロック内のトランザクションを取得
            transactions = get_block_transactions(block_number)
            # 取得に失敗した場合はcontinue
            if transactions is None:
                continue

            # 新しいトークンコントラクトを探す
            check_for_new_tokens(transactions, block_number)
            # 処理したブロック番号を保存
            save_last_block(block_number)

        # 指定された間隔で待機
        time.sleep(INTERVAL)

EtherScan API はこちら

実行結果:

% python token_scan.py
Checking block 21408558
Regular transaction: from: 0x958536d6e8eed9bc64a97d67a2416a8f7d111991 to: 0x95ad61b0a150d79219dcf64e1e6cc01f0b64c4ce
Regular transaction: from: 0xd34d928ee62a44dd21885a75b1a6596fd0f088a0 to: 0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48
Regular transaction: from: 0x21debfa81fc74415383cfbc597f77ccf6b61334b to: 0xcf0c122c6b73ff809c693db761e7baebe62b6a2e
Regular transaction: from: 0xb5a46bc8b76fd2825aeb43db9c9e89e89158ecde to: 0x9f8f72aa9304c8b593d555f12ef6589cc3a579a2
Regular transaction: from: 0x48f18653beae232ac20574247357362dd88f6e00 to: 0x090185f2135308bad17527004364ebcc2d37e5f6
Checking block 21408559
Regular transaction: from: 0x080086911d8c78008800fae75871a657b77d0082 to: 0x0000e0ca771e21bd00057f54a68c30d400000000
Regular transaction: from: 0xfba0014d3a9dbe8a0cda6affd3da7b541a1ec32f to: 0xea8cf32e4ac03acab2babb9028bac5c853e0ce80
Regular transaction: from: 0x4066e9bd5618373d2da7a1cb7bba03ef800875ee to: 0x6719c6ebf80d6499ca9ce170cda72beb3f1d1a54
Regular transaction: from: 0xa40c08732ea74411e404bb108043ccf2992fbf87 to: 0x0000000000007f150bd6f54c40a34d7c3d5e9f56
Regular transaction: from: 0x6000cf9e08c900e2da21f04f00a0430000fad259 to: 0x509fde8900840c1f8900f3325c0000bdaa00b500
Checking block 21408560
Regular transaction: from: 0xffa45a7c2d6c1e73a4306d6448a24768de467232 to: 0x055c48651015cf5b21599a4ded8c402fdc718058
Regular transaction: from: 0x315d2ee4fccda0def532ef4108ff57204f8d9eba to: 0x738e79fbc9010521763944ddf13aad7f61502221
Regular transaction: from: 0xa7b326ad42dfef8507f0e6543afe65e99c3115ef to: 0x4fa98b4640d73991007e109c41814310431ada58
Regular transaction: from: 0x2a0eff97bb68298e36bea8c86852436e96af92b9 to: 0x73a8a6f5d9762ea5f1de193ec19cdf476c7e86b1
Regular transaction: from: 0x5b43453fce04b92e190f391a83136bfbecedefd1 to: 0x68d3a973e7272eb388022a5c6518d9b2a2e66fbf
Checking block 21408561
Regular transaction: from: 0x4066e9bd5618373d2da7a1cb7bba03ef800875ee to: 0x6719c6ebf80d6499ca9ce170cda72beb3f1d1a54
Regular transaction: from: 0xb0fc682a69b37abf85cd519f9e4fd94a8c7c8669 to: 0x1bf621aa9cee3f6154881c25041bb39aed4ca7cc
Regular transaction: from: 0x11235534a66a33c366b84933d5202c841539d1c9 to: 0x22ce84a7f86662b78e49c6ec9e51d60fdde7b70a
Regular transaction: from: 0x1057c855a823203b32f49e1c1741a77fbe13ae31 to: 0x51c72848c68a965f66fa7a88855f9f7784502a7f
Regular transaction: from: 0xc16157e00b1bff1522c6f01246b4fb621da048d0 to: 0x9a15bb3a8fec8d0d810691bafe36f6e5d42360f7
Checking block 21408562
Regular transaction: from: 0x111206594f2fb5927f719b4417ab1da3d41b14b1 to: 0x000000d40b595b94918a28b27d1e2c66f43a51d3
Regular transaction: from: 0x0dfd32320af10f58cd4a5c2b567b7739ea2d691c to: 0x3fc91a3afd70395cd496c647d5a6cc9d4b2b7fad
Regular transaction: from: 0x1114c78d5de672996d812dc2e1a05b5f33eacdfb to: 0x000000d40b595b94918a28b27d1e2c66f43a51d3
Regular transaction: from: 0x93793bd1f3e35a0efd098c30e486a860a0ef7551 to: 0x68d3a973e7272eb388022a5c6518d9b2a2e66fbf
Regular transaction: from: 0x448166a91e7bc50d0ac720c2fbed29e0963f5af8 to: 0x68d3a973e7272eb388022a5c6518d9b2a2e66fbf

新規トークンを見つけることはできませんでしたが、トランザクションログを表示することはできましたね。

このスクリプトでは一部のログしか取得できないので、新規トークンを見つけるのは難しいかもしれません。しかし、可能性はゼロではないので、見つけたときは、夢が広がりますね。

今後は、スクリプトの改良も視野に、宝探しを続けたいと思います🙂