Pythonで暗号資産のクオンツ取引戦略を構築する方法：マーケットデータ取得からバックテストまで

暗号資産のクオンツ取引戦略を開発する開発者にとって、売買シグナルを書くこと自体は難しくありません。難しいのは、そのシグナルが信頼できる市場データに基づいているか、そしてバックテストの結果が戦略の実際のパフォーマンスを正しく反映しているかを確認することです。

例えば、BTC/USDTの1時間足データを使って移動平均線のクロス戦略をテストする場合、データに重複したタイムスタンプが含まれていたり、戦略が現在のローソク足の終値を事前に利用していたり、取引手数料を無視していたりすると、バックテストの損益曲線が上昇し続けていても、その戦略に実際の価値があるとは限りません。

そのため、Pythonによるクオンツ取引で重要なのは、単に指標を計算することではありません。マーケットデータの取得、データクリーニング、シグナル生成、バックテストを一連のプロセスとしてつなげることが重要です。

信頼できる市場データから始める

戦略を構築する前に、まずどの種類の市場データを使用するかを決める必要があります。トレンド指標を使用する中低頻度の戦略であれば、過去のローソク足データで初期検証を行うことができます。一方、注文板の変化、短期的な価格インパクト、高頻度取引などを研究する場合は、より細かい粒度のティックデータが必要になることがあります。

1時間足を使った移動平均線クロス戦略を例にすると、AllTickの暗号資産K線APIから過去のデータを取得できます。このAPIはHTTP GETリクエストを使用し、基本パスは quote-b-api/kline です。認証情報とリクエスト条件は、token とURLエンコードされたJSON形式のクエリパラメータで渡します。K線データには、タイムスタンプ、始値、高値、安値、終値などのフィールドが含まれます。

データを取得した後、すぐに指標を計算してはいけません。価格フィールドを数値型に変換し、時刻形式を統一し、時間の昇順に並べ替え、重複したタイムスタンプを削除する必要があります。また、ローソク足が欠落している場合も、前の足の終値でそのまま補完するべきではありません。そうすると、実際には発生していない市場の値動きがバックテストに入り込む可能性があります。

Pythonで市場データを取得し取引シグナルを生成する

以下では、20期間と50期間の指数移動平均線（EMA）を使って、シンプルなトレンド戦略を構築します。短期EMAが長期EMAを上回っている場合はロングポジションを保有し、それ以外の場合はポジションを持たないルールです。この例は、クオンツ取引戦略の開発と検証の流れを説明するためのものであり、最適化済みの取引システムを意味するものではありません。

まず、必要なライブラリをインストールします。

pip install requests pandas

環境変数 ALLTICK_API_TOKEN を設定したうえで、次のコードを実行します。

import os
import json
import uuid
import requests
import pandas as pd

BASE_URL = "https://quote.alltick.co/quote-b-api"
TOKEN = os.environ["ALLTICK_API_TOKEN"]


def fetch_klines(code="BTCUSDT", count=500):
    query = {
        "trace": str(uuid.uuid4()),
        "data": {
            "code": code,
            "kline_type": 5,  # 1時間足のK線
            "kline_timestamp_end": 0,
            "query_kline_num": count,
            "adjust_type": 0
        }
    }

    response = requests.get(
        f"{BASE_URL}/kline",
        params={
            "token": TOKEN,
            "query": json.dumps(query)
        },
        timeout=15
    )
    response.raise_for_status()

    result = response.json()
    if result.get("ret") != 200:
        raise RuntimeError(f"API error: {result.get('msg')}")

    rows = result.get("data", {}).get("kline_list", [])
    if not rows:
        raise ValueError("No K-line data returned")

    df = pd.DataFrame(rows).rename(columns={
        "open_price": "open",
        "high_price": "high",
        "low_price": "low",
        "close_price": "close"
    })

    df["timestamp"] = pd.to_numeric(df["timestamp"])
    df["timestamp"] = pd.to_datetime(
        df["timestamp"], unit="s", utc=True
    )

    for col in ["open", "high", "low", "close"]:
        df[col] = pd.to_numeric(df[col], errors="coerce")

    return (
        df.dropna(subset=["close"])
        .sort_values("timestamp")
        .drop_duplicates("timestamp")
        .set_index("timestamp")
    )


df = fetch_klines()

df["ema20"] = df["close"].ewm(
    span=20, adjust=False
).mean()
df["ema50"] = df["close"].ewm(
    span=50, adjust=False
).mean()

df["signal"] = (df["ema20"] > df["ema50"]).astype(int)

# 直前のローソク足で生成したシグナルを使い、現在の足の終値情報の先読みを避ける
df["position"] = df["signal"].shift(1).fillna(0)

df["market_return"] = df["close"].pct_change().fillna(0)
df["turnover"] = (
    df["position"].diff().abs()
    .fillna(df["position"].abs())
)

# 手数料の例：ポジション変更ごとに5ベーシスポイントを見積もる
fee_rate = 0.0005

df["strategy_return"] = (
    df["position"] * df["market_return"]
    - df["turnover"] * fee_rate
)

test = df.iloc[50:].copy()
test["equity"] = (1 + test["strategy_return"]).cumprod()

print("戦略の累積リターン：", round(
    test["equity"].iloc[-1] - 1, 4
))

コード内の BTCUSDT は取引銘柄コードの例です。実際にリクエストする前に、使用中のAllTick製品で利用可能な銘柄コードと一致していることを確認してください。APIでは1回のリクエストで最大500本のK線を取得できます。より長い期間をバックテストする場合は、暗号資産APIがサポートするタイムスタンプパラメータを使用してデータを分割取得し、結合時に重複を削除する必要があります。

バックテストではなぜ収益率だけを見てはいけないのか

上記のコードはEMAでトレンドを判定し、シグナルを1期間ずらして使用します。これにより、同じローソク足の終値でシグナルを生成しながら、あたかもその終値で事前に約定できたかのように扱うことを避けています。

また、手数料の見積もりも加えていますが、これはあくまで簡略化したバックテストです。例で使用している手数料率は仮のパラメータであり、実際の取引所の手数料率ではありません。スリッページ、資金調達率、注文の約定状況、市場インパクトもモデルに含まれていません。そのため、出力された累積リターンを、実際の取引パフォーマンスの予測として直接扱うことはできません。

本格的に戦略を評価する場合は、最大ドローダウン、取引回数、勝率、コスト控除後のリターンなども確認し、サンプル外データを使って戦略が有効かどうかを検証する必要があります。パラメータの多い戦略では、過去の損益曲線が最も良く見えるまで何度も調整することを避け、過剰適合のリスクを抑えることも重要です。

もう一つ見落としやすいのが、データのカバー範囲です。500本の1時間足は約20日分に相当します。このデータだけでは、異なる市場サイクルにおいて戦略が安定して機能するかどうかを判断できません。より長期間の過去データ、異なるボラティリティ環境でのテスト、欠損区間の確認も、バックテストの信頼性を左右する重要な要素です。

過去データのバックテストからリアルタイム市場データへ

過去データは戦略の開発と検証に適しており、リアルタイム市場データは、その後のモニタリングやシグナル更新に利用できます。AllTickは暗号資産の市場データを提供し、RESTとWebSocketによる接続に対応しています。開発者はまず過去のK線データを使って分析プロセスを構築し、その後、戦略に必要な更新頻度やデータ粒度に応じて、リアルタイムのデータストリームが必要かどうかを判断できます。

開発を始める前に、AllTick Crypto API製品ページ と K線API技術ドキュメント を確認し、利用可能な取引銘柄、過去データの範囲、API制限、リクエストパラメータを確認してください。

信頼できるクオンツ取引戦略とは、単に売買シグナルを生成できる戦略ではありません。データの出所を追跡でき、約定に関する前提を説明でき、取引コストを差し引いたうえで検証できる戦略です。 データとバックテストのロジックの両方が検証に耐えてこそ、過去のパフォーマンスが参考になるものになります。