暗号資産市場における統計的裁定取引とペアトレード：共和分からカルマンフィルターまで

1987年、モルガン・スタンレーの物理学者グループは、銀行経営陣に完全に説明できないアルゴリズムを使って株式のペアトレードを行い、1年で5000万ドルを稼いだ。経営陣は異議を唱えなかった。2026年、あなたは同じアイデアを暗号資産取引所で実行できる——永久先物、24時間365日のマーケット、そしてNunzio Tartagliaも羨むほどの流動性がある。しかし注意点がある：インターネット以前の時代にフォードとGMの株式で機能したものは、BTCが一晩で20%下落し、ファンディングレートが1ブロックで反転する世界では、本格的な適応が必要となる。

本記事は、暗号資産市場における統計的裁定取引とペアトレードの完全な解説である。数学理論（共和分、オルンシュタイン＝ウーレンベック過程、カルマンフィルター）から、実データで実行可能なPythonコードまで。スタイルはエンジニアリング志向：数式を説明し、コードを示し、落とし穴を隠さない。

1. 略史：イエズス会士からクオンツへ

現代的な統計的裁定取引は、1980年代半ばのモルガン・スタンレーのトレーディングデスクで生まれた。Nunzio Tartaglia——物理学のPhDを持つ元イエズス会司祭——が数学者、物理学者、コンピュータ科学者からなるチームを編成した。目的：従来のトレーダーには見えない株価のパターンを見つけること。

アイデアは驚くほど単純だった。コカ・コーラとペプシの株価が歴史的に連動して動くなら（当然だ——異なる色の同じ甘い水を売っている）、それらの価格の乖離は一時的な異常である。遅れている方を買い、先行している方を売り、収束を待って、利益を確定する。マーケットニュートラル戦略：市場の方向は関係ない。

Tartagliaのチームには、後にウォール街全体を変えることになる人物が含まれていた：

• David Shaw — 後にD.E. Shaw & Co.を設立、最大級のクオンツヘッジファンドの一つ
• Peter Muller — モルガン・スタンレー内の伝説的な統計的裁定グループPDT Partnersを設立
• Robert Frey — 後にJim SimonsのRenaissance Technologiesに移籍

グループは投資銀行内の研究所のように運営された。自動化は最先端で、VAXクラスターがシグナルを生成し、端末を通じて執行された。最良の年（1987-1988）には、戦略は数千万ドルを稼いだ。その後2年連続で損失が出て、1989年にモルガン・スタンレーはデスクを閉鎖した。

しかしアイデアは解き放たれた。グループの卒業生がペアトレードの概念をウォール街全体に広めた。Gatev、Goetzmann、Rouwenhorstは2006年に古典的な学術論文「Pairs Trading: Performance of a Relative-Value Arbitrage Rule」を発表し、単純なペアトレード戦略が1962年から2002年まで米国株式市場で安定的に年間約11%のリターンを生み出したことを示した。これは効率的市場仮説への説得力のある反証だった：市場全体は効率的かもしれないが、特定の資産のペアは体系的に均衡から乖離する。

今日、統計的裁定取引は運用資産数千億ドルの産業であり、暗号資産市場は特に肥沃な土壌を提供している：分散化された流動性、未成熟なマイクロストラクチャー、24時間取引、そしてファンディングレート付きの永久先物——従来の市場には存在しない商品である。

2. 数学的基礎：相関は罠である

なぜ相関が機能しないのか

初心者クオンツの半数が犯す間違いから始めよう：「BTCとETHの相関係数は0.85だから、ペアを取引できる」。いいえ、できない。正確には、できるが——お金を失う。

相関は2つの資産のリターン間の線形関係を測定する。2つの資産が完全に相関していても、それらの価格は永遠に発散し得る。古典的な例：相関した増分を持つ2つのランダムウォーク——高い相関にもかかわらず、無限に発散する。「収束」を期待してポジションを開くが、それは来ない。

共和分：正しいアプローチ

共和分はリターンではなく、価格系列の性質である。2つの非定常系列X(t)とY(t)は、線形結合：

S(t) = Y(t) - β · X(t)

が定常——つまり平均値に回帰する場合、共和分と呼ばれる。係数βはヘッジ比率、S(t)はスプレッドと呼ばれる。

直感：BTCとETHは月まで上昇するか奈落の底に落ちるかもしれないが、それらの差分（適切にスケーリングされた）が固定されたレベルの周りで振動するなら——それが共和分である。そしてそれこそがトレードに必要なものだ。

エングル＝グレンジャー検定（1987）

Robert EngleとClive Grangerが2003年にノーベル経済学賞を受賞した2段階手続き：

ステップ1. OLS回帰：Y(t) = α + β · X(t) + ε(t)。ヘッジ比率βと残差ε(t)を取得する。

ステップ2. 残差ε(t)に対するADF（拡張ディッキー＝フラー）検定。帰無仮説：ε(t)は単位根を持つ（非定常）。p値 < 0.05ならH₀を棄却——系列は共和分である。

重要：共和分検定には標準のADF臨界値を使用できない。エングル＝グレンジャーの臨界値はモンテカルロシミュレーションで導出され、OLS回帰における変数間の依存性を考慮している。statsmodelsではcoint()関数で正しく実装されている。

ヨハンセン検定

2つ以上の変数のシステム（例えば、BTC、ETH、SOLを同時に）には、ヨハンセン検定が使用される。システム内のすべての共和分関係を見つけ、複数資産のポートフォリオを構築できる。検定はVAR（ベクトル自己回帰）モデルに基づき、トレース統計量と最大固有値統計量の2つの基準を使用する。

オルンシュタイン＝ウーレンベック過程

スプレッドが共和分であれば、その動態はオルンシュタイン＝ウーレンベック（OU）過程としてモデル化できる：

dS(t) = θ(μ - S(t))dt + σ dW(t)

ここで：

• θ — 平均回帰速度
• μ — 長期平均水準
• σ — ボラティリティ
• W(t) — ウィーナー過程（ブラウン運動）

OU過程のパラメータから平均回帰の半減期が計算される：

t½ = ln(2) / θ

半減期は極めて重要な指標である。t½ = 5日なら、スプレッドは約5日で平均に回帰する。t½ = 200日なら、収束を待って半年ポジションを保持することになる。暗号資産戦略の最適な半減期は1〜30日。それより短い——速すぎて手数料が利益を食い潰す。それより長い——遅すぎて構造的レジーム変化のリスクがある。

実務ではθは回帰で推定される：

ΔS(t) = a + b · S(t-1) + ε(t)

ここでθ = -b、t½ = -ln(2) / b。

Zスコア正規化

取引シグナル生成のため、スプレッドを正規化する：

z(t) = (S(t) - μ̂) / σ̂

ここでμ̂とσ̂はスプレッドの移動平均と標準偏差。Zスコアはスプレッドが平均から何標準偏差乖離しているかを示す。典型的なエントリー閾値：|z| > 2.0、エグジット閾値：|z| < 0.5。

3. 暗号資産市場でのペア選択

BTC-ETH：（時々）機能する古典的ペア

BTCとETHは最も明白で最も流動性の高いペアである。リターンの相関は安定的に0.7を超える。しかし共和分は別の話だ。それは現れたり消えたりする：

• 2023年のレンジ相場ではBTC/ETHは確実に共和分だった（p値 < 0.01）
• 2024-2025年の乖離期間（BTCがETFで上昇、ETHが遅れた）では共和分が崩壊した
• 2026年初頭、ETH ETFの開始とETH/BTC比率の回復後、共和分は再び安定化した

結論：共和分は継続的にモニタリングする必要がある。回帰パラメータはローリングウィンドウで再計算され、ADF検定のp値が閾値を超えた場合、戦略は自動的に停止する。

セクターペア

暗号資産市場はセクター別に便利に分類されており、セクター内のペアはしばしば安定した共和分を示す：

統計的裁定に最適なペアは3つの特性を持つ：(1) 6ヶ月以上の履歴ウィンドウでの安定した共和分、(2) 十分な流動性——各資産の日次取引高 > 1000万ドル、(3) 適切な半減期——1〜30日。

現物 vs 永久先物（ベーシス）

「ペア」の別のカテゴリーは、現物市場と先物市場における同一資産である。永久先物価格と現物価格の差（ベーシス）は定義上定常的である：ファンディングレートメカニズムがそれをゼロに向けて圧縮する。これにより、ベーシストレードは暗号資産における最も信頼性の高い統計的裁定の形態の一つとなる。

4. 3つの取引アプローチ

A. ベーシストレード：現物-先物とファンディングレートキャリー

暗号資産における最も「純粋な」統計的裁定の形態。メカニクス：

1. 現物で資産を買い（例：1 BTC）
2. 永久先物でショートを開く（1 BTC）
3. ファンディングレートがプラスなら（ロングがショートに支払う）——8時間ごとにファンディングを受け取る

平均ファンディングレートが8時間ごとに0.01%の場合、これは1日あたり約0.03%、年率約11%——方向性リスクなし。強気相場では、ファンディングレートは8時間ごとに0.05-0.1%まで上昇することがある——それはすでに年率55-110%だ。

リスク：マイナスのファンディング（市場の反転）、急激な価格上昇時のショートポジションの清算（証拠金バッファーが必要）、取引所手数料。

2026年3月時点で、BTCの平均ファンディングレートは8時間あたり約0.015%で安定している——2024年の水準より約50%高い。

B. クロス取引所裁定

同じコイン、2つの取引所、異なる価格。原因——流動性の違い、トレーダー構成、オーダーブック更新速度の違い。

例： BinanceでBTC：$87,150。BybitでBTC：$87,175。スプレッド：$25（0.029%）。

戦略：Binanceで買い、Bybitで売る。問題：両方の注文が約定するまでにスプレッドが消失する可能性がある。解決策：両方の取引所に残高を持ち、同時に約定させる。

典型的な手数料：

• Binance：テイカー約0.075%（割引後約0.05%）
• Bybit：テイカー約0.03%（VIP）
• 合計：約0.08%

つまり、戦略が利益を上げるにはスプレッドが0.08%を超える必要がある。2026年には、このようなスプレッドが発生する：

• 流動性の低いペア（アルトコイン）——定期的に
• 主要ペア（BTC、ETH）——高ボラティリティの瞬間のみ
• CEXとDEXの間——より頻繁だが、MEVリスクとスリッページあり

コロケーションなしでは、APIレイテンシーは10-100ミリ秒。最適化されたネットワークでは約1ミリ秒。ほとんどの個人トレーダーは100-500ミリ秒の範囲で操作しており、多くの裁定戦略には十分だが、機関投資家と競争するには不十分である。

C. レバレッジ付きペアトレード

2つの異なる資産にレバレッジを使用した古典的なペアトレード。3つの戦略の中で最も複雑——そして最も潜在的に利益が高い。

SOL/AVAXペアを例にしたメカニクス：

1. ヘッジ比率βを計算（例：β = 1.3）
2. Zスコア > +2の場合：SOLをショート、AVAX × βをロング
3. Zスコア < -2の場合：SOLをロング、AVAX × βをショート
4. エグジット：|Zスコア| < 0.5 またはタイムアウト（例：30日）

各レッグに3倍レバレッジ、平均スプレッド回帰2σ → 3σの場合：

• 1取引あたりの目標リターン：約3-6%
• 平均頻度：ペアあたり月2-4取引
• 期待年間リターン：30-60%（手数料とスリッページ前）

主なリスク：相関が最悪のタイミングで崩壊する可能性がある（通常、市場暴落時）。詳細はセクション8を参照。

5. 適応型ヘッジ比率のためのカルマンフィルター

静的ヘッジ比率が問題である理由

古典的アプローチ：履歴ウィンドウでOLSによりβを推定して固定する。問題：βは時間とともに変化する。暗号資産市場は特に非定常的で——ナラティブの変化（DeFi Summer → NFTハイプ → AIトークン）が資産間の根本的な関係を変える。

ローリングOLS（ローリング回帰）の使用は中途半端な対策だ。ウィンドウ長を選択する必要がある：短すぎる——ノイズ、長すぎる——遅延。カルマンフィルターはこの問題をエレガントに解決する。

状態空間モデル

Y(t)とX(t)の関係を時変係数を持つ線形モデルとして表現する：

観測方程式：

Y(t) = α(t) + β(t) · X(t) + ε(t),   ε(t) ~ N(0, R)

状態方程式：

[α(t+1), β(t+1)]ᵀ = [α(t), β(t)]ᵀ + w(t),   w(t) ~ N(0, Q)

パラメータα(t)とβ(t)は、ゆっくりとドリフトする（ランダムウォーク）隠れ状態として扱われる。カルマンフィルターはノイズのある観測からこの隠れ状態を最適に推定する。

• R（観測ノイズ）——観測ノイズの分散。Rが大きいほど、フィルターの新データへの反応は遅くなる。
• Q（状態ノイズ）——状態ノイズの共分散行列。Qが大きいほど、フィルターの適応は速くなる。

Q/R比はフィルターの「滑らかさ」を決定する——ローリングOLSでのウィンドウ長の選択に類似するが、データの硬い切り捨てなし。

ローリングOLSに対する優位性

カルマンフィルターで計算されたスプレッドは、ローリング回帰からのスプレッドよりも著しくより定常的で平均回帰的である。カルマンフィルターは固定ウィンドウ長でデータを切り捨てるのではなく、指数的に減衰する重みですべての過去の観測を使用する。さらに、カルマンフィルターは「ウィンドウ長」パラメータの調整を必要としない——代わりにQとR行列を通じて慣性と適応性のバランスを自動的にキャリブレーションする。

filterpyによる実装

import numpy as np
from filterpy.kalman import KalmanFilter

def create_kalman_filter(
    delta: float = 1e-4,
    obs_noise: float = 1.0
) -> KalmanFilter:
    """
    適応型ヘッジ比率推定用のカルマンフィルターを作成。

    delta: 状態ノイズ分散 (Q = delta * I)。
           deltaが大きい → 適応が速い、ノイズが多い。
    obs_noise: 観測ノイズ分散 (R)。
    """
    kf = KalmanFilter(dim_x=2, dim_z=1)

    kf.x = np.zeros((2, 1))

    kf.F = np.eye(2)

    kf.P = np.eye(2) * 1000

    kf.Q = np.eye(2) * delta

    kf.R = np.array([[obs_noise]])

    return kf

def estimate_hedge_ratio(
    prices_y: np.ndarray,
    prices_x: np.ndarray,
    delta: float = 1e-4,
    obs_noise: float = 1.0
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """
    カルマンフィルターで適応型ヘッジ比率を推定。

    戻り値:
        alphas: 切片の配列 (α)
        betas: ヘッジ比率の配列 (β)
        spreads: スプレッドの配列 Y - α - β*X
    """
    n = len(prices_y)
    kf = create_kalman_filter(delta, obs_noise)

    alphas = np.zeros(n)
    betas = np.zeros(n)
    spreads = np.zeros(n)

    for t in range(n):
        kf.H = np.array([[1.0, prices_x[t]]])

        kf.predict()

        kf.update(np.array([[prices_y[t]]]))

        alphas[t] = kf.x[0, 0]
        betas[t] = kf.x[1, 0]
        spreads[t] = prices_y[t] - kf.x[0, 0] - kf.x[1, 0] * prices_x[t]

    return alphas, betas, spreads

deltaパラメータが鍵。ボラティリティの高い暗号資産ペア（ミームコイン、小型アルト）にはdelta = 1e-3を使用。安定したペア（BTC/ETH、SOL/AVAX）にはdelta = 1e-5。

6. エントリーとエグジットシグナル

Zスコア閾値

基本的なシグナルロジック：

def generate_signals(
    spreads: np.ndarray,
    lookback: int = 60,
    entry_z: float = 2.0,
    exit_z: float = 0.5,
    stop_z: float = 4.0
) -> np.ndarray:
    """
    スプレッドのZスコアに基づいて取引シグナルを生成。

    戻り値の配列: +1（スプレッドロング）、-1（スプレッドショート）、0（ポジションなし）
    """
    signals = np.zeros(len(spreads))
    position = 0

    for t in range(lookback, len(spreads)):
        window = spreads[t - lookback:t]
        mu = np.mean(window)
        sigma = np.std(window)

        if sigma < 1e-10:
            continue

        z = (spreads[t] - mu) / sigma

        if position == 0:
            if z > entry_z:
                position = -1  # スプレッドショート（Yショート、Xロング）
            elif z < -entry_z:
                position = 1   # スプレッドロング（Yロング、Xショート）
        else:
            if position == 1 and z > -exit_z:
                position = 0
            elif position == -1 and z < exit_z:
                position = 0
            elif abs(z) > stop_z:
                position = 0

        signals[t] = position

    return signals

モメンタムフィルター

純粋な平均回帰シグナルはフィルターで改善できる：

1. モメンタムフィルター： スプレッドが発散し続けている場合はポジションを開かない。エントリー前にスプレッドの反転を待つ。技術的には：Zスコアは閾値を超えたが、現在のスプレッド変化はすでに平均の方向に向かっている。

2. ボラティリティフィルター： 高ボラティリティ期間にはエントリー閾値を引き上げる。市場がパニックになると、Zスコアは数週間3σ以上に留まることがある。

3. 共和分フィルター： 各取引の前に、共和分がまだ有効かどうかを確認する（ローリングADF検定）。p値 > 0.1なら取引を一時停止。

時間ベースのエグジット

ポジションが半減期の2倍以上開いており、スプレッドが回帰していない場合——強制的にクローズする。スプレッドが期待時間の2倍以内に回帰しない場合、共和分はおそらく崩壊しており、待つ意味はない。

7. バックテスト：正しいやり方

ウォークフォワード分析

標準的なバックテスト（全データで訓練 → 全データでテスト）は統計的裁定には無意味だ。回帰パラメータがデータに過学習し、結果は楽観的になる。

ウォークフォワードアプローチ：

1. データを期間に分割：[訓練₁ → テスト₁] → [訓練₂ → テスト₂] → ...
2. 各訓練期間で：共和分を推定、ヘッジ比率を計算、Zスコア閾値を選択
3. テスト期間で：固定パラメータで取引
4. すべてのテスト期間を結合して最終評価

暗号資産の典型的な設定：訓練 = 180日、テスト = 30日、ステップ = 30日。

取引コストモデル

暗号資産では以下を考慮する必要がある：

ペアポジションのエントリーとエグジットには4回の取引が発生（2レッグ × エントリー + エグジット）。合計コスト：ラウンドトリップあたり約0.3-0.5%。つまり、正の期待値を持つには、1取引あたりの平均利益が0.5%を超える必要がある。

スリッページモデル

線形モデル：slippage = k × (order_size / ADV)、ここでADVは平均日次取引高。暗号資産では、トップ10コインでk ≈ 0.1、アルトコインでk ≈ 0.3-0.5。

より現実的なモデルはスクエアルートインパクト：slippage = k × sqrt(order_size / ADV)。実際の市場マイクロストラクチャーをより良く反映する。

メトリクス

def calculate_metrics(returns: np.ndarray, rf: float = 0.04) -> dict:
    """
    戦略の主要メトリクスを計算。
    rf: 無リスク金利（年率）
    """
    daily_rf = rf / 365
    excess = returns - daily_rf

    ann_return = np.mean(returns) * 365
    ann_vol = np.std(returns) * np.sqrt(365)

    sharpe = (ann_return - rf) / ann_vol if ann_vol > 0 else 0

    cumulative = np.cumprod(1 + returns)
    running_max = np.maximum.accumulate(cumulative)
    drawdowns = (cumulative - running_max) / running_max
    max_dd = np.min(drawdowns)

    calmar = ann_return / abs(max_dd) if max_dd != 0 else 0

    win_rate = np.mean(returns > 0) if len(returns) > 0 else 0

    gains = returns[returns > 0].sum()
    losses = abs(returns[returns < 0].sum())
    profit_factor = gains / losses if losses > 0 else float('inf')

    return {
        'annual_return': f'{ann_return:.1%}',
        'annual_volatility': f'{ann_vol:.1%}',
        'sharpe_ratio': f'{sharpe:.2f}',
        'max_drawdown': f'{max_dd:.1%}',
        'calmar_ratio': f'{calmar:.2f}',
        'win_rate': f'{win_rate:.1%}',
        'profit_factor': f'{profit_factor:.2f}',
    }

暗号資産統計的裁定のベンチマーク：

• シャープレシオ > 1.5 — 良い戦略
• 最大ドローダウン < 15% — 許容可能なリスク
• カルマーレシオ > 2.0 — 優れたリターン/ドローダウン比
• プロフィットファクター > 1.5 — 持続的なエッジ

8. 現実世界の問題

スリッページと流動性

バックテストでは中間価格で即座にエントリーする。現実では——そうはいかない。日次取引高500万ドルのアルトコインでは、5万ドルの注文が価格を0.2-0.5%動かす可能性がある。ペア戦略では2倍のスリッページ（2レッグ）となり、利益をすべて消費する可能性がある。

解決策：指値注文を使用（テイカーではなくメイカー）、注文を分割（TWAP/VWAP）、ADVに対するポジションサイズを厳密に制限（日次取引高の最大1-2%）。

ファンディングレートリスク

ベーシストレードではファンディングレートを受け取るが、マイナスになることがある。2022年12月の弱気市場では、BTCのファンディングレートは8時間ごとに-0.02%だった——「ロング現物 + ショート永久先物」のポジションを持っていたなら、10万ドルのポジションにつき1日60ドルを支払っていた。

防御策：ファンディングレートをリアルタイムでモニタリングし、レートが反転したらポジションをクローズする。より高度なアプローチは取引所間のファンディングレート裁定（低ファンディングの取引所でロング、高ファンディングの取引所でショート）。

危機時の相関崩壊

2020年3月、2021年5月、2022年11月、2024年8月——すべての暗号資産暴落で相関が崩壊する。より正確には、相関は強まる（すべてが一緒に下落する）が、共和分は破壊される——スプレッドは10σまで飛んで戻ってこない可能性がある。

これがペアトレードのアキレス腱だ。戦略は小さな金額を安定的に稼ぎ、その後1日で大きな金額を失う。古典的な「蒸気ローラーの前でペニーを拾う」プロファイル。

防御策：

1. 厳格なストップロス： Zスコア > 4σでポジションをクローズ
2. レバレッジ制限： 各レッグ最大2-3倍
3. VIX/ボラティリティフィルター： インプライドボラティリティが高い場合にポジションサイズを減少
4. 分散化： 10-20ペアを同時に取引、一つに集中しない

資金要件

本格的な暗号資産統計的裁定のために：

• ベーシストレード：5万ドルから（1ペア、1取引所）
• クロス取引所裁定：10万ドルから（2つの取引所に残高）
• ペアトレードポートフォリオ（10ペア）：20万ドルから
• 機関投資家レベル：100万ドルから

それ以下の金額では、手数料と最小ポジションサイズにより戦略は非効率となる。

9. エンドツーエンドPython実装

データ取得

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def fetch_ohlcv(
    exchange_id: str,
    symbol: str,
    timeframe: str = '1h',
    days: int = 365
) -> pd.DataFrame:
    """ccxt経由でOHLCVデータを取得。"""
    exchange = getattr(ccxt, exchange_id)({
        'enableRateLimit': True,
    })

    since = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
    all_candles = []

    while True:
        candles = exchange.fetch_ohlcv(
            symbol, timeframe, since=since, limit=1000
        )
        if not candles:
            break
        all_candles.extend(candles)
        since = candles[-1][0] + 1
        if len(candles) < 1000:
            break

    df = pd.DataFrame(
        all_candles,
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df

sol = fetch_ohlcv('binance', 'SOL/USDT', '1h', 365)
avax = fetch_ohlcv('binance', 'AVAX/USDT', '1h', 365)

prices = pd.DataFrame({
    'SOL': sol['close'],
    'AVAX': avax['close']
}).dropna()

共和分検定

from statsmodels.tsa.stattools import coint, adfuller
from statsmodels.regression.linear_model import OLS
from statsmodels.tools import add_constant

def test_cointegration(y: np.ndarray, x: np.ndarray) -> dict:
    """
    診断付きの完全な共和分検定。
    """
    score, pvalue, crit_values = coint(y, x)

    x_const = add_constant(x)
    model = OLS(y, x_const).fit()
    alpha, beta = model.params
    spread = y - alpha - beta * x

    adf_stat, adf_pvalue, _, _, adf_crit, _ = adfuller(spread, maxlag=20)

    spread_lag = spread[:-1]
    spread_diff = np.diff(spread)
    spread_lag_const = add_constant(spread_lag)
    hl_model = OLS(spread_diff, spread_lag_const).fit()
    theta = -hl_model.params[1]
    half_life = np.log(2) / theta if theta > 0 else np.inf

    return {
        'coint_pvalue': pvalue,
        'cointegrated': pvalue < 0.05,
        'hedge_ratio': beta,
        'intercept': alpha,
        'adf_statistic': adf_stat,
        'adf_pvalue': adf_pvalue,
        'half_life_hours': half_life,
        'half_life_days': half_life / 24,
        'spread_mean': np.mean(spread),
        'spread_std': np.std(spread),
    }

result = test_cointegration(
    prices['SOL'].values,
    prices['AVAX'].values
)
print(f"共和分: {result['cointegrated']} "
      f"(p値: {result['coint_pvalue']:.4f})")
print(f"ヘッジ比率: {result['hedge_ratio']:.4f}")
print(f"半減期: {result['half_life_days']:.1f} 日")

カルマンフィルター + バックテスター

from filterpy.kalman import KalmanFilter

class PairsBacktester:
    """
    カルマンフィルター付きペアトレードの
    ウォークフォワードバックテスター。
    """

    def __init__(
        self,
        prices_y: np.ndarray,
        prices_x: np.ndarray,
        kalman_delta: float = 1e-4,
        obs_noise: float = 1.0,
        entry_z: float = 2.0,
        exit_z: float = 0.5,
        stop_z: float = 4.0,
        lookback: int = 60,
        fee_rate: float = 0.001,    # レッグあたり0.1%ラウンドトリップ
        slippage_rate: float = 0.0005,  # レッグあたり0.05%スリッページ
    ):
        self.prices_y = prices_y
        self.prices_x = prices_x
        self.n = len(prices_y)
        self.kalman_delta = kalman_delta
        self.obs_noise = obs_noise
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.stop_z = stop_z
        self.lookback = lookback
        self.fee_rate = fee_rate
        self.slippage_rate = slippage_rate

    def run(self) -> pd.DataFrame:
        """バックテストを実行。結果のDataFrameを返す。"""
        kf = KalmanFilter(dim_x=2, dim_z=1)
        kf.x = np.zeros((2, 1))
        kf.F = np.eye(2)
        kf.P = np.eye(2) * 1000
        kf.Q = np.eye(2) * self.kalman_delta
        kf.R = np.array([[self.obs_noise]])

        alphas = np.zeros(self.n)
        betas = np.zeros(self.n)
        spreads = np.zeros(self.n)

        for t in range(self.n):
            kf.H = np.array([[1.0, self.prices_x[t]]])
            kf.predict()
            kf.update(np.array([[self.prices_y[t]]]))
            alphas[t] = kf.x[0, 0]
            betas[t] = kf.x[1, 0]
            spreads[t] = (
                self.prices_y[t] - kf.x[0, 0]
                - kf.x[1, 0] * self.prices_x[t]
            )

        positions = np.zeros(self.n)
        z_scores = np.zeros(self.n)
        position = 0

        for t in range(self.lookback, self.n):
            window = spreads[t - self.lookback:t]
            mu = np.mean(window)
            sigma = np.std(window)
            if sigma < 1e-10:
                continue

            z = (spreads[t] - mu) / sigma
            z_scores[t] = z

            if position == 0:
                if z > self.entry_z:
                    position = -1
                elif z < -self.entry_z:
                    position = 1
            else:
                if position == 1 and z > -self.exit_z:
                    position = 0
                elif position == -1 and z < self.exit_z:
                    position = 0
                elif abs(z) > self.stop_z:
                    position = 0

            positions[t] = position

        spread_returns = np.diff(spreads) / np.abs(
            spreads[:-1] + 1e-10
        )
        pnl = np.zeros(self.n)

        for t in range(1, self.n):
            if positions[t - 1] != 0:
                raw_return = positions[t - 1] * spread_returns[t - 1]
                pnl[t] = raw_return

                if positions[t] != positions[t - 1]:
                    total_cost = 2 * (self.fee_rate + self.slippage_rate)
                    pnl[t] -= total_cost

        return pd.DataFrame({
            'price_y': self.prices_y,
            'price_x': self.prices_x,
            'alpha': alphas,
            'beta': betas,
            'spread': spreads,
            'z_score': z_scores,
            'position': positions,
            'pnl': pnl,
            'cumulative_pnl': np.cumsum(pnl),
        })

bt = PairsBacktester(
    prices_y=prices['SOL'].values,
    prices_x=prices['AVAX'].values,
    kalman_delta=1e-4,
    entry_z=2.0,
    exit_z=0.5,
    stop_z=4.0,
    lookback=60,
    fee_rate=0.001,
    slippage_rate=0.0005,
)
results = bt.run()

daily_pnl = results['pnl'].resample('D').sum() if hasattr(
    results.index, 'freq'
) else results['pnl']
metrics = calculate_metrics(daily_pnl.values)
for k, v in metrics.items():
    print(f'{k}: {v}')

ライブトレードスケルトン

import ccxt
import asyncio
import logging

logger = logging.getLogger(__name__)

class LivePairsTrader:
    """
    ライブペアトレードの最小限のスケルトン。
    本番環境向け：リトライロジック、モニタリング、
    アラート、残高照合を追加。
    """

    def __init__(
        self,
        exchange_id: str,
        symbol_y: str,
        symbol_x: str,
        api_key: str,
        secret: str,
        position_size_usd: float = 1000.0,
        entry_z: float = 2.0,
        exit_z: float = 0.5,
    ):
        self.exchange = getattr(ccxt, exchange_id)({
            'apiKey': api_key,
            'secret': secret,
            'enableRateLimit': True,
        })
        self.symbol_y = symbol_y
        self.symbol_x = symbol_x
        self.position_size = position_size_usd
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.position = 0  # +1, -1, 0

        self.kf = create_kalman_filter(delta=1e-4)
        self.spread_history = []

    async def update(self):
        """1更新サイクル。"""
        ticker_y = self.exchange.fetch_ticker(self.symbol_y)
        ticker_x = self.exchange.fetch_ticker(self.symbol_x)
        price_y = ticker_y['last']
        price_x = ticker_x['last']

        self.kf.H = np.array([[1.0, price_x]])
        self.kf.predict()
        self.kf.update(np.array([[price_y]]))

        alpha = self.kf.x[0, 0]
        beta = self.kf.x[1, 0]
        spread = price_y - alpha - beta * price_x
        self.spread_history.append(spread)

        if len(self.spread_history) < 60:
            logger.info(f"Warming up: {len(self.spread_history)}/60")
            return

        window = np.array(self.spread_history[-60:])
        z = (spread - np.mean(window)) / np.std(window)

        logger.info(
            f"β={beta:.4f} spread={spread:.4f} z={z:.2f} "
            f"pos={self.position}"
        )

        new_position = self.position

        if self.position == 0:
            if z > self.entry_z:
                new_position = -1
            elif z < -self.entry_z:
                new_position = 1
        else:
            if self.position == 1 and z > -self.exit_z:
                new_position = 0
            elif self.position == -1 and z < self.exit_z:
                new_position = 0

        if new_position != self.position:
            await self._execute_trade(
                new_position, price_y, price_x, beta
            )
            self.position = new_position

    async def _execute_trade(
        self, target: int, price_y: float, price_x: float,
        beta: float
    ):
        """ペア取引を約定。"""
        if target == 0:
            logger.info("Closing position")
        elif target == 1:
            size_y = self.position_size / price_y
            size_x = (self.position_size * beta) / price_x
            logger.info(
                f"Long spread: buy {size_y:.4f} {self.symbol_y}, "
                f"sell {size_x:.4f} {self.symbol_x}"
            )
        elif target == -1:
            size_y = self.position_size / price_y
            size_x = (self.position_size * beta) / price_x
            logger.info(
                f"Short spread: sell {size_y:.4f} {self.symbol_y}, "
                f"buy {size_x:.4f} {self.symbol_x}"
            )

    async def run_loop(self, interval_seconds: int = 60):
        """メインループ。"""
        logger.info(
            f"Starting live trading: "
            f"{self.symbol_y}/{self.symbol_x}"
        )
        while True:
            try:
                await self.update()
            except Exception as e:
                logger.error(f"Error in update: {e}")
            await asyncio.sleep(interval_seconds)

おわりに

統計的裁定取引は聖杯ではない。職人技だ。「共和分が何かを知っている」と「安定的に動作する戦略を持っている」の間には、エンジニアリングの詳細の深い溝がある：適切なデータ処理、正確なウォークフォワードバックテスト、現実的なスリッページモデル、リアルタイムモニタリング。

暗号資産市場は、従来の市場よりも統計的裁定の機会を多く提供している——分散化された流動性、未成熟な市場インフラ、そしてファンディングレート付き永久先物のようなユニークな商品が、NYSEではとっくにゼロまで裁定されてしまった非効率性を生み出している。

しかし窓は閉じつつある。機関投資家が暗号資産市場に参入し、裁定資本は成長し（推定では、2025年に暗号資産取引所の裁定資本量は215%増加した）、マージンは圧縮されている。暗号資産で統計的裁定を行うつもりなら——今始めるのが最善だ。

この記事のすべてのコードは出発点として利用できる。本格的なテストなしに本番環境で実行してはいけない。そして覚えておいてほしい：確実に機能する唯一の戦略はリスク管理だ。

主要な学術論文：

• Engle, R.F. & Granger, C.W.J. (1987). "Co-Integration and Error Correction: Representation, Estimation, and Testing". Econometrica, 55(2), 251-276.
• Gatev, E., Goetzmann, W.N. & Rouwenhorst, K.G. (2006). "Pairs Trading: Performance of a Relative-Value Arbitrage Rule". The Review of Financial Studies, 19(3), 797-827.
• Vidyamurthy, G. (2004). Pairs Trading: Quantitative Methods and Analysis. Wiley.
• Avellaneda, M. & Lee, J.H. (2010). "Statistical Arbitrage in the US Equities Market". Quantitative Finance, 10(7), 761-782.
• Frontiers (2026). "Deep learning-based pairs trading: real-time forecasting of co-integrated cryptocurrency pairs". Frontiers in Applied Mathematics and Statistics.

便利なライブラリ：

• statsmodels — 共和分、ADF、OLS
• filterpy — カルマンフィルター
• ccxt — 100以上の取引所の統一API
• arbitragelab — ペアトレード専用ライブラリ（OU、Kalman、copulas）