Statistical Arbitrage และ Pairs Trading ในตลาด Crypto: จาก Cointegration สู่ Kalman Filter

ในปี 1987 กลุ่มนักฟิสิกส์ที่ Morgan Stanley สามารถทำกำไรได้ถึง 50 ล้านดอลลาร์ภายในปีเดียว ด้วยการเทรดคู่หุ้นโดยใช้อัลกอริทึมที่ไม่มีใครในทีมสามารถอธิบายให้ฝ่ายบริหารของธนาคารเข้าใจได้อย่างถ่องแท้ ฝ่ายบริหารก็ไม่ได้คัดค้านแต่อย่างใด ในปี 2026 คุณสามารถนำแนวคิดเดิมนี้ไปใช้บน crypto exchanges — ด้วย perpetual futures, ตลาดที่เปิด 24/7 และสภาพคล่องที่ Nunzio Tartaglia คงอิจฉา แต่มีข้อแม้: สิ่งที่ใช้ได้กับหุ้น Ford และ GM ในยุคก่อนอินเทอร์เน็ต ต้องการการปรับตัวอย่างจริงจังสำหรับโลกที่ BTC อาจร่วงลง 20% ในคืนเดียว และ funding rate สามารถพลิกกลับภายใน block เดียว

บทความนี้คือการวิเคราะห์เชิงลึกอย่างครบถ้วนเกี่ยวกับ statistical arbitrage และ pairs trading สำหรับตลาด crypto ตั้งแต่ทฤษฎีทางคณิตศาสตร์ (cointegration, กระบวนการ Ornstein-Uhlenbeck, Kalman filter) ไปจนถึงโค้ด Python ที่ใช้งานได้จริงซึ่งคุณสามารถรันบนข้อมูลจริงได้ สไตล์เน้นวิศวกรรม: เราอธิบายสูตร, แสดงโค้ด และไม่ซ่อนข้อผิดพลาดที่พบบ่อย

1. ประวัติย่อ: จากนักบวชเยซูอิตสู่ Quants

Statistical arbitrage ในรูปแบบสมัยใหม่เกิดขึ้นที่โต๊ะเทรดของ Morgan Stanley ในช่วงกลางทศวรรษ 1980 Nunzio Tartaglia — อดีตนักบวชเยซูอิตที่มีปริญญาเอกด้านฟิสิกส์ — รวบรวมทีมนักคณิตศาสตร์, นักฟิสิกส์ และนักวิทยาศาสตร์คอมพิวเตอร์ เป้าหมาย: ค้นหารูปแบบในราคาหุ้นที่เทรดเดอร์ทั่วไปมองไม่เห็น

แนวคิดเรียบง่ายจนน่าตกใจ หากหุ้น Coca-Cola และ Pepsi เคลื่อนไหวไปด้วยกันในประวัติศาสตร์ (ซึ่งสมเหตุสมผล — พวกเขาขายน้ำหวานเหมือนกันในขวดต่างสี) การที่ราคาแยกออกจากกันก็คือความผิดปกติชั่วคราว ซื้อตัวที่ล้าหลัง, ขาย short ตัวที่นำหน้า, รอให้ราคาบรรจบกัน, ล็อกกำไร กลยุทธ์ market-neutral: ทิศทางของตลาดโดยรวมไม่ใช่สิ่งที่เราสนใจ

ทีมของ Tartaglia มีสมาชิกที่ต่อมาจะเปลี่ยนโฉม Wall Street ทั้งใบ:

• David Shaw — ต่อมาก่อตั้ง D.E. Shaw & Co. หนึ่งในกองทุน quantitative hedge fund ที่ใหญ่ที่สุด
• Peter Muller — ก่อตั้ง PDT Partners กลุ่ม stat arb ในตำนานภายใน Morgan Stanley
• Robert Frey — ต่อมาเข้าร่วม Renaissance Technologies ภายใต้ Jim Simons

กลุ่มนี้ดำเนินการเหมือนห้องปฏิบัติการวิจัยภายในธนาคารเพื่อการลงทุน ระบบอัตโนมัติทันสมัยมาก: VAX clusters สร้าง signals และการ execute ผ่านเทอร์มินัล ในปีที่ดีที่สุด (1987-1988) กลยุทธ์ทำกำไรได้หลายสิบล้านดอลลาร์ จากนั้นก็ขาดทุนสองปีติดต่อกัน และในปี 1989 Morgan Stanley ปิดโต๊ะนั้น

แต่แนวคิดได้แพร่กระจายออกไปแล้ว ศิษย์เก่าของกลุ่มนำแนวคิด pairs trading ไปเผยแพร่ทั่ว Wall Street Gatev, Goetzmann และ Rouwenhorst ตีพิมพ์บทความวิชาการคลาสสิกในปี 2006 — "Pairs Trading: Performance of a Relative-Value Arbitrage Rule" — แสดงให้เห็นว่ากลยุทธ์ pairs trading อย่างง่ายให้ผลตอบแทนประจำปีสม่ำเสมอ ~11% ตั้งแต่ปี 1962 ถึง 2002 บนหุ้นสหรัฐ นั่นเป็นคำตอบที่น่าสนใจต่อสมมติฐานตลาดมีประสิทธิภาพ: ตลาดโดยรวมอาจมีประสิทธิภาพ แต่ คู่ ของสินทรัพย์เฉพาะเบี่ยงเบนออกจากสมดุลอย่างเป็นระบบ

ปัจจุบัน statistical arbitrage เป็นอุตสาหกรรมที่มี AUM หลายร้อยพันล้านดอลลาร์ และตลาด crypto มอบพื้นที่อันอุดมสมบูรณ์เป็นพิเศษ: สภาพคล่องที่แตกกระจาย, โครงสร้างตลาดที่ยังไม่เติบโตเต็มที่, การเทรดตลอดเวลา และ perpetual futures พร้อม funding rates — เครื่องมือที่ไม่มีอยู่ในตลาดดั้งเดิม

2. รากฐานทางคณิตศาสตร์: Correlation คือกับดัก

ทำไม Correlation ไม่ทำงาน

เริ่มกันที่ความผิดพลาดที่ quant มือใหม่ทำบ่อย: "BTC และ ETH มี correlation 0.85 ดังนั้นเราสามารถเทรดคู่นี้ได้" ไม่ ทำไม่ได้ ทำได้ก็จริง — แต่คุณจะขาดทุน

Correlation วัดความสัมพันธ์เชิงเส้นระหว่าง ผลตอบแทน ของสินทรัพย์สองตัว สองสินทรัพย์อาจ correlate กันอย่างสมบูรณ์แบบ แต่ ราคา ของพวกมันแยกออกไปตลอดกาล ตัวอย่างคลาสสิก: random walks สองตัวที่มี increments ที่ correlated — พวกมันแยกออกจากกันไม่จำกัด แม้จะมี correlation สูง คุณจะเปิดสถานะโดยคาดหวัง "การบรรจบ" ที่จะไม่มาถึงตลอดกาล

Cointegration: แนวทางที่ถูกต้อง

Cointegration คือคุณสมบัติของ series ราคา ไม่ใช่ผลตอบแทน สอง series ที่ไม่ stationary X(t) และ Y(t) เรียกว่า cointegrated หากมี linear combination:

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

ที่ stationary — หมายความว่ามันกลับสู่ค่าเฉลี่ย สัมประสิทธิ์ β เรียกว่า hedge ratio และ S(t) คือ spread

ความเข้าใจแบบง่าย: BTC และ ETH อาจพุ่งสูงสู่ดวงจันทร์หรือดิ่งลงเหว แต่ถ้า ความแตกต่าง ของพวกมัน (ปรับสเกลอย่างเหมาะสม) แกว่งรอบระดับคงที่ — นั่นคือ cointegration และนั่นคือสิ่งที่เราต้องการสำหรับการเทรด

การทดสอบ Engle-Granger (1987)

ขั้นตอนสองขั้นที่ Robert Engle และ Clive Granger ได้รับรางวัล Nobel สาขาเศรษฐศาสตร์ในปี 2003:

ขั้นที่ 1. OLS regression: Y(t) = α + β · X(t) + ε(t) เราได้ hedge ratio β และ residuals ε(t)

ขั้นที่ 2. ADF (Augmented Dickey-Fuller) test บน residuals ε(t) Null hypothesis: ε(t) มี unit root (non-stationary) หาก p-value < 0.05 เราปฏิเสธ H₀ — series เป็น cointegrated

สำคัญ: สำหรับการทดสอบ cointegration คุณ ไม่สามารถ ใช้ค่าวิกฤต ADF มาตรฐาน ค่าวิกฤต Engle-Granger ได้มาจากการจำลอง Monte Carlo และคำนึงถึงการพึ่งพาระหว่างตัวแปรใน OLS regression ใน statsmodels สิ่งนี้ถูก implement อย่างถูกต้องในฟังก์ชัน coint()

การทดสอบ Johansen

สำหรับระบบที่มีมากกว่าสองตัวแปร (เช่น BTC, ETH และ SOL พร้อมกัน) จะใช้การทดสอบ Johansen มันค้นหา ทุก ความสัมพันธ์ cointegration ในระบบและอนุญาตให้สร้างพอร์ตโฟลิโอจากสินทรัพย์หลายตัว การทดสอบอิงจากโมเดล VAR (vector autoregression) และใช้สองเกณฑ์: trace statistic และ maximum eigenvalue statistic

กระบวนการ Ornstein-Uhlenbeck

หาก spread เป็น cointegrated พลวัตของมันสามารถ model เป็นกระบวนการ Ornstein-Uhlenbeck (OU):

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

โดยที่:

• θ — ความเร็วการกลับสู่ค่าเฉลี่ย
• μ — ระดับค่าเฉลี่ยระยะยาว
• σ — ความผันผวน
• W(t) — กระบวนการ Wiener (Brownian motion)

จากพารามิเตอร์กระบวนการ OU ครึ่งชีวิตของการกลับสู่ค่าเฉลี่ย คำนวณได้จาก:

t½ = ln(2) / θ

ครึ่งชีวิตเป็นตัวชี้วัดที่สำคัญมาก หาก t½ = 5 วัน spread จะกลับสู่ค่าเฉลี่ยในประมาณ 5 วัน หาก t½ = 200 วัน คุณจะนั่งในสถานะครึ่งปีรอการบรรจบ สำหรับกลยุทธ์ crypto ครึ่งชีวิตที่เหมาะสมคือ 1-30 วัน สั้นกว่านั้น — เร็วเกินไป ค่าคอมมิชชันกินกำไร ยาวกว่านั้น — ช้าเกินไป มีความเสี่ยงของการเปลี่ยนแปลงโครงสร้าง

ในทางปฏิบัติ θ ประมาณได้ผ่าน regression:

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

โดยที่ θ = -b และ t½ = -ln(2) / b

การ Normalize ด้วย Z-Score

เพื่อสร้าง trading signals spread จะถูก normalize:

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

โดยที่ μ̂ และ σ̂ คือค่าเฉลี่ยและส่วนเบี่ยงเบนมาตรฐานแบบ rolling ของ spread z-score แสดงให้เห็นว่า spread เบี่ยงเบนออกจากค่าเฉลี่ยกี่ส่วนเบี่ยงเบนมาตรฐาน ค่า threshold เข้าเทรดทั่วไป: |z| > 2.0; ค่า threshold ออก: |z| < 0.5

3. การเลือกคู่ในตลาด Crypto

BTC-ETH: คลาสสิกที่ (บางครั้ง) ใช้ได้

BTC และ ETH เป็นคู่ที่ชัดเจนที่สุดและมีสภาพคล่องสูงที่สุด Correlation ของผลตอบแทนอยู่เหนือ 0.7 อย่างสม่ำเสมอ แต่ cointegration เป็นเรื่องอื่น มัน ปรากฏและหายไป:

• ในช่วง sideways ปี 2023 BTC/ETH เป็น cointegrated อย่างน่าเชื่อถือ (p-value < 0.01)
• ในช่วงการแยกออกปี 2024-2025 (BTC พุ่งด้วยกระแส ETF, ETH ล้าหลัง) cointegration พังทลาย
• ต้นปี 2026 หลังการเปิดตัว ETH ETF และการฟื้นตัวของอัตราส่วน ETH/BTC cointegration เสถียรอีกครั้ง

สรุป: cointegration ต้อง ติดตามอย่างต่อเนื่อง พารามิเตอร์ regression คำนวณใหม่บนหน้าต่างแบบ rolling และกลยุทธ์ปิดตัวอัตโนมัติหาก p-value ของการทดสอบ ADF เกินค่า threshold

คู่ตามเซกเตอร์

ตลาด crypto แบ่งตามเซกเตอร์อย่างสะดวก และคู่ภายในเซกเตอร์เดียวกันมักแสดง cointegration ที่เสถียร:

คู่ที่ดีที่สุดสำหรับ stat arb มีสามคุณสมบัติ: (1) cointegration ที่เสถียร บนหน้าต่างประวัติศาสตร์ >6 เดือน (2) สภาพคล่องเพียงพอ — ปริมาณรายวัน >$10M ต่อสินทรัพย์ (3) ครึ่งชีวิตที่สมเหตุสมผล — จาก 1 ถึง 30 วัน

Spot vs Perpetual Futures (Basis)

หมวดหมู่แยกของ "คู่" คือสินทรัพย์เดียวกันในตลาด spot และ futures ความแตกต่างระหว่างราคา perpetual futures และราคา spot (basis) มี stationarity โดยกำหนด: กลไก funding rate บีบมันกลับสู่ศูนย์ สิ่งนี้ทำให้ basis trading เป็นหนึ่งในรูปแบบที่น่าเชื่อถือที่สุดของ stat arb ใน crypto

4. สามแนวทางการเทรด

A. Basis Trading: Spot-Futures และ Funding Rate Carry

รูปแบบ "บริสุทธิ์" ที่สุดของ stat arb ใน crypto กลไก:

1. ซื้อ สินทรัพย์บน spot (เช่น 1 BTC)
2. เปิด short บน perpetual future (1 BTC)
3. หาก funding rate เป็นบวก (longs จ่ายให้ shorts) — คุณรับ funding ทุก 8 ชั่วโมง

ที่ funding rate เฉลี่ย 0.01% ทุก 8 ชั่วโมง นั่นคือ ~0.03% ต่อวัน หรือ ~11% ต่อปี โดยไม่มีความเสี่ยงทิศทาง ในช่วง bull market funding rate อาจสูงถึง 0.05-0.1% ทุก 8 ชั่วโมง — นั่นคือ 55-110% ต่อปี

ความเสี่ยง: funding ติดลบ (ตลาดพลิกกลับ), การ liquidation ของสถานะ short ในช่วงราคาพุ่งสูงอย่างรวดเร็ว (ต้องการ margin buffer) และค่าธรรมเนียมของ exchange

ณ มีนาคม 2026 BTC funding rate เฉลี่ยเสถียรที่ประมาณ ~0.015% ต่อ 8 ชั่วโมง — ประมาณ 50% เหนือระดับปี 2024

B. Cross-Exchange Arbitrage

เหรียญเดียวกัน, สอง exchanges, ราคาต่างกัน เหตุผล — ความแตกต่างในสภาพคล่อง, องค์ประกอบของผู้เทรด และความเร็วในการอัปเดต order book

ตัวอย่าง: BTC บน Binance: $87,150 BTC บน Bybit:$87,175 Spread: $25 (0.029%)

กลยุทธ์: ซื้อบน Binance, ขายบน Bybit ปัญหา: เมื่อถึงเวลาที่คำสั่งทั้งสองถูก execute spread อาจหายไปแล้ว วิธีแก้: รักษาดุลยภาพบนทั้งสอง exchanges และ execute พร้อมกัน

ค่าธรรมเนียมทั่วไป:

• Binance: ~0.075% taker (พร้อมส่วนลด ~0.05%)
• Bybit: ~0.03% taker (VIP)
• รวม: ~0.08%

หมายความว่า spread ต้องเกิน 0.08% สำหรับกลยุทธ์ที่ทำกำไรได้ ในปี 2026 spread ดังกล่าวเกิดขึ้น:

• บนคู่ที่มีสภาพคล่องน้อยกว่า (altcoins) — เป็นประจำ
• บนคู่หลัก (BTC, ETH) — เฉพาะในช่วงความผันผวนสูง
• ระหว่าง CEX และ DEX — บ่อยกว่า แต่มีความเสี่ยง MEV และ slippage

โดยไม่มี co-location latency ของ API คือ 10-100 ms ด้วยเครือข่ายที่ optimize แล้ว — ~1 ms นักเทรดรายย่อยส่วนใหญ่ทำงานในช่วง 100-500 ms ซึ่งเพียงพอสำหรับกลยุทธ์ arbitrage หลายอย่าง แต่ไม่เพียงพอในการแข่งขันกับสถาบัน

C. Pairs Trading ด้วย Leverage

Pairs trading คลาสสิกบนสองสินทรัพย์ต่างกันโดยใช้ leverage นี่คือกลยุทธ์ที่ซับซ้อนที่สุดในสาม — และอาจทำกำไรได้มากที่สุด

กลไกโดยใช้คู่ SOL/AVAX เป็นตัวอย่าง:

1. คำนวณ hedge ratio β (เช่น β = 1.3)
2. เมื่อ z-score > +2: short SOL, long AVAX × β
3. เมื่อ z-score < -2: long SOL, short AVAX × β
4. ออก: |z-score| < 0.5 หรือ timeout (เช่น 30 วัน)

ด้วย leverage 3x บนแต่ละ leg และการกลับตัวเฉลี่ยของ spread จาก 2σ → 3σ:

• ผลตอบแทนเป้าหมายต่อการเทรด: ~3-6%
• ความถี่เฉลี่ย: 2-4 การเทรดต่อเดือนต่อคู่
• ผลตอบแทนรายปีที่คาดหวัง: 30-60% (ก่อนค่าคอมมิชชันและ slippage)

ความเสี่ยงหลัก: correlation อาจพังทลายในช่วงเวลาที่เลวร้ายที่สุด (มักเกิดขึ้นระหว่างตลาดตก) อ่านเพิ่มเติมในหัวข้อ 8

5. Kalman Filter สำหรับ Adaptive Hedge Ratio

ทำไม Static Hedge Ratio จึงเป็นปัญหา

แนวทางคลาสสิก: ประมาณ β ผ่าน OLS บนหน้าต่างประวัติศาสตร์และตรึงมันไว้ ปัญหา: β เปลี่ยนแปลงตามเวลา ตลาด crypto non-stationary เป็นพิเศษ — การเปลี่ยน narrative (DeFi summer → NFT hype → AI tokens) เปลี่ยนความสัมพันธ์พื้นฐานระหว่างสินทรัพย์

การใช้ rolling OLS (rolling regression) เป็นมาตรการครึ่งทาง คุณต้องเลือกความยาวหน้าต่าง: สั้นเกินไป — noise; ยาวเกินไป — lag Kalman filter แก้ปัญหานี้ได้อย่างสวยงาม

State-Space Model

เราแทนความสัมพันธ์ระหว่าง Y(t) และ X(t) เป็นโมเดลเชิงเส้นที่มีสัมประสิทธิ์ ที่เปลี่ยนแปลงตามเวลา:

สมการ observation:

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

สมการ state:

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

พารามิเตอร์ α(t) และ β(t) ถือเป็น hidden state ที่ค่อยๆ เคลื่อนที่ (random walk) Kalman filter ประมาณ hidden state นี้อย่างเหมาะสมที่สุดจาก observations ที่มี noise

• R (observation noise) — variance ของ observation noise ยิ่ง R ใหญ่ filter ตอบสนองต่อข้อมูลใหม่ช้าลง
• Q (state noise) — covariance matrix ของ state noise ยิ่ง Q ใหญ่ filter ปรับตัวเร็วขึ้น

อัตราส่วน Q/R กำหนด "ความเรียบ" ของ filter — คล้ายกับการเลือกความยาวหน้าต่างใน rolling OLS แต่ไม่ตัดข้อมูลอย่างหยาบ

ข้อได้เปรียบเหนือ Rolling OLS

Spreads ที่คำนวณโดยใช้ Kalman filter มีความ stationary และ mean-reverting มากกว่าอย่างมีนัยสำคัญ กว่า spreads จาก rolling regression Kalman filter ใช้ การสังเกต ทั้งหมดในอดีตด้วยน้ำหนักที่ลดลงแบบ exponential แทนที่จะตัดข้อมูลที่ความยาวหน้าต่างคงที่ นอกจากนี้ Kalman filter ไม่ต้องการการปรับพารามิเตอร์ "ความยาวหน้าต่าง" — แต่ calibrate สมดุลระหว่าง inertia และ adaptivity โดยอัตโนมัติผ่าน matrix Q และ R

การ Implement ด้วย filterpy

import numpy as np
from filterpy.kalman import KalmanFilter

def create_kalman_filter(
    delta: float = 1e-4,
    obs_noise: float = 1.0
) -> KalmanFilter:
    """
    Creates a Kalman filter for adaptive hedge ratio estimation.

    delta: state noise variance (Q = delta * I).
           Larger delta → faster adaptation, more noise.
    obs_noise: observation noise variance (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]:
    """
    Estimates the adaptive hedge ratio using a Kalman filter.

    Returns:
        alphas: array of intercepts (α)
        betas: array of hedge ratios (β)
        spreads: array of 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 สำคัญมาก สำหรับคู่ crypto ที่มีความผันผวนสูง (memecoins, small-cap alts) ใช้ delta = 1e-3 สำหรับคู่ที่เสถียร (BTC/ETH, SOL/AVAX) — delta = 1e-5

6. Signals การเข้าและออก

Z-Score Thresholds

Logic signal พื้นฐาน:

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:
    """
    Generates trading signals based on spread z-score.

    Returns array: +1 (long spread), -1 (short spread), 0 (flat)
    """
    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  # Short spread (short Y, long X)
            elif z < -entry_z:
                position = 1   # Long spread (long Y, short 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

Momentum Filters

Signal mean-reversion บริสุทธิ์สามารถปรับปรุงด้วย filters:

1. Momentum filter: อย่าเปิดสถานะหาก spread ยังคงแยกออก รอให้ spread กลับตัว ก่อนเข้า ทางเทคนิค: z-score ผ่าน threshold แล้ว แต่การเปลี่ยนแปลง spread ปัจจุบันมุ่งสู่ค่าเฉลี่ยแล้ว

2. Volatility filter: เพิ่ม entry threshold ในช่วงความผันผวนสูง เมื่อตลาดตื่นตระหนก z-score สามารถอยู่เหนือ 3σ ได้หลายสัปดาห์

3. Cointegration filter: ก่อนเทรดทุกครั้ง ตรวจสอบว่า cointegration ยังคงอยู่ (rolling ADF test) หาก p-value > 0.1 — หยุดเทรด

การออกตามเวลา

หากสถานะเปิดนานกว่า 2× ครึ่งชีวิตและ spread ยังไม่กลับตัว — ปิดมันโดยบังคับ หาก spread ไม่กลับตัวภายใน 2× เวลาที่คาดหวัง cointegration น่าจะพังทลายแล้ว และไม่มีอะไรต้องรอ

7. Backtesting: ทำให้ถูกต้อง

Walk-Forward Analysis

Backtest มาตรฐาน (train บนข้อมูลทั้งหมด → test บนข้อมูลทั้งหมด) ไม่มีประโยชน์สำหรับ stat arb พารามิเตอร์ regression ถูก overfit กับข้อมูล และผลลัพธ์จะ optimistic

แนวทาง Walk-forward:

1. แบ่งข้อมูลเป็นช่วง: [train₁ → test₁] → [train₂ → test₂] → ...
2. บนแต่ละช่วง train: ประมาณ cointegration, คำนวณ hedge ratio, เลือก z-score thresholds
3. บนช่วง test: เทรดด้วยพารามิเตอร์คงที่
4. รวมช่วง test ทั้งหมดสำหรับการประเมินขั้นสุดท้าย

การตั้งค่าทั่วไปสำหรับ crypto: train = 180 วัน, test = 30 วัน, step = 30 วัน

Transaction Cost Model

สำหรับ crypto คุณต้องคำนึงถึง:

การเข้าและออกจากสถานะ pairs เกี่ยวข้องกับ 4 การเทรด (2 legs × เข้า + ออก) ต้นทุนรวม: ~0.3-0.5% ต่อ round trip หมายความว่ากำไรเฉลี่ยต่อการเทรดต้องเกิน 0.5% สำหรับ expected value บวก

Slippage Model

โมเดลเชิงเส้น: slippage = k × (order_size / ADV) โดยที่ ADV คือปริมาณรายวันเฉลี่ย สำหรับ crypto k ≈ 0.1 สำหรับ top-10 coins และ k ≈ 0.3-0.5 สำหรับ altcoins

โมเดลที่สมจริงกว่าคือ square-root impact: slippage = k × sqrt(order_size / ADV) มันสะท้อน microstructure ของตลาดจริงได้ดีกว่า

Metrics

def calculate_metrics(returns: np.ndarray, rf: float = 0.04) -> dict:
    """
    Calculates key strategy metrics.
    rf: risk-free rate (annual)
    """
    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}',
    }

เกณฑ์มาตรฐานสำหรับ crypto stat arb:

• Sharpe > 1.5 — กลยุทธ์ที่ดี
• Max drawdown < 15% — ความเสี่ยงที่ยอมรับได้
• Calmar > 2.0 — อัตราส่วนผลตอบแทน/drawdown ที่ยอดเยี่ยม
• Profit factor > 1.5 — edge ที่ยั่งยืน

8. ปัญหาในโลกความเป็นจริง

Slippage และสภาพคล่อง

ใน backtest คุณเข้าทันทีที่ mid-price ในความเป็นจริง — ไม่ใช่ บน altcoins ที่มีปริมาณรายวัน $5M คำสั่ง$50K สามารถเคลื่อนราคาได้ 0.2-0.5% สำหรับกลยุทธ์ pairs นั่นคือ slippage ที่เพิ่มขึ้นสองเท่า (สอง legs) และอาจกินกำไรทั้งหมด

วิธีแก้: ใช้ limit orders (maker ไม่ใช่ taker), แบ่งคำสั่งเป็นส่วน (TWAP/VWAP) และจำกัดขนาดสถานะอย่างเข้มงวดเทียบกับ ADV (สูงสุด 1-2% ของปริมาณรายวัน)

Funding Rate Risk

ด้วย basis trading คุณรับ funding rate แต่มันสามารถติดลบได้ ในตลาด bearish ของธันวาคม 2022 BTC funding rate อยู่ที่ -0.02% ทุก 8 ชั่วโมง — หากคุณนั่งอยู่ในสถานะ "long spot + short perp" คุณ จ่าย $60/วัน ต่อสถานะ$100K

การป้องกัน: ติดตาม funding rate ในแบบ real-time และปิดสถานะเมื่ออัตราพลิกกลับ แนวทางขั้นสูงกว่าคือ funding rate arbitrage ระหว่าง exchanges (long บน exchange ที่มี funding ต่ำ, short บน exchange ที่มี funding สูง)

การพังทลายของ Correlation ในวิกฤต

มีนาคม 2020, พฤษภาคม 2021, พฤศจิกายน 2022, สิงหาคม 2024 — ในทุกตลาด crypto ตก correlations พังทลาย แม่นยำกว่านั้น correlations เพิ่มขึ้น (ทุกอย่างตกพร้อมกัน) แต่ cointegration พัง — spread สามารถพุ่งไปที่ 10σ และไม่กลับมาตลอดกาล

นี่คือส้นเท้าอาคิลีสของ pairs trading กลยุทธ์ทำกำไรเล็กน้อยอย่างสม่ำเสมอ จากนั้นสูญเสียจำนวนมากในวันเดียว Profile คลาสสิกของ "การเก็บเหรียญหน้ารถไฟ"

การป้องกัน:

1. Stop-loss เข้มงวด: ปิดสถานะเมื่อ z-score > 4σ
2. จำกัด leverage: สูงสุด 2-3x บนแต่ละ leg
3. VIX/volatility filter: ลดขนาดสถานะเมื่อ implied volatility สูง
4. กระจาย: เทรด 10-20 คู่พร้อมกัน อย่าใส่ทุกอย่างในคู่เดียว

ข้อกำหนดเงินทุน

สำหรับ crypto stat arb ที่จริงจัง:

• Basis trading: จาก $50K (บนคู่เดียว, exchange เดียว)
• Cross-exchange arbitrage: จาก $100K (ดุลยภาพบนสอง exchanges)
• พอร์ตโฟลิโอ pairs trading (10 คู่): จาก $200K
• ระดับสถาบัน: จาก $1M

ด้วยเงินทุนน้อยกว่า ค่าคอมมิชชันและขนาดสถานะขั้นต่ำทำให้กลยุทธ์ไม่สามารถทำได้

9. การ Implement 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:
    """Fetch OHLCV data via ccxt."""
    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()

การทดสอบ Cointegration

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:
    """
    Full cointegration test with diagnostics.
    """
    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"Cointegration: {result['cointegrated']} "
      f"(p-value: {result['coint_pvalue']:.4f})")
print(f"Hedge ratio: {result['hedge_ratio']:.4f}")
print(f"Half-life: {result['half_life_days']:.1f} days")

Kalman Filter + Backtester

from filterpy.kalman import KalmanFilter

class PairsBacktester:
    """
    Walk-forward backtester for pairs trading
    with Kalman filter.
    """

    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% round trip per leg
        slippage_rate: float = 0.0005,  # 0.05% slippage per leg
    ):
        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:
        """Run the backtest. Returns a DataFrame with results."""
        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}')

Skeleton การเทรด Live

import ccxt
import asyncio
import logging

logger = logging.getLogger(__name__)

class LivePairsTrader:
    """
    Minimal skeleton for live pairs trading.
    For production: add retry logic, monitoring,
    alerts, balance reconciliation.
    """

    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):
        """One update cycle."""
        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
    ):
        """Execute a pairs trade."""
        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):
        """Main loop."""
        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)

แทนที่จะสรุป

Statistical arbitrage ไม่ใช่จอกศักดิ์สิทธิ์ มันคือหัตถกรรม ระหว่าง "ฉันรู้ว่า cointegration คืออะไร" และ "ฉันมีกลยุทธ์ที่ทำงานได้อย่างสม่ำเสมอ" มีความแตกต่างอย่างมหาศาลของรายละเอียดวิศวกรรม: การประมวลผลข้อมูลที่ถูกต้อง, walk-forward backtesting ที่ถูกต้อง, โมเดล slippage ที่สมจริง, การติดตาม real-time

ตลาด cryptocurrency ยังคงมีโอกาสสำหรับ stat arb มากกว่าตลาดดั้งเดิม — สภาพคล่องที่แตกกระจาย, โครงสร้างตลาดที่ยังไม่เติบโต และเครื่องมือเฉพาะอย่าง perpetual futures พร้อม funding rates สร้างความไม่มีประสิทธิภาพที่ถูก arbitrage ออกไปนานแล้วบน NYSE

แต่หน้าต่างกำลังปิดลง ผู้เล่นสถาบันกำลังเข้ามาในตลาด crypto เงินทุน arbitrage กำลังเติบโต (ตามการประมาณการ ปริมาณเงินทุน arbitrage บน crypto exchanges เติบโต 215% ในปี 2025) และ margins กำลังลดลง หากคุณจะทำ stat arb ใน crypto — ดีที่สุดที่จะเริ่มตอนนี้

โค้ดทั้งหมดในบทความนี้มีให้เป็นจุดเริ่มต้น อย่ารัน production โดยไม่มีการทดสอบอย่างจริงจัง และจำไว้: กลยุทธ์เดียวที่รับประกันว่าได้ผลคือการจัดการความเสี่ยง

ผลงานวิชาการหลัก:

• 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 — cointegration, ADF, OLS
• filterpy — Kalman filter
• ccxt — unified API สำหรับ exchanges กว่า 100 แห่ง
• arbitragelab — ไลบรารีเฉพาะสำหรับ pairs trading (OU, Kalman, copulas)