페어 트레이딩이란?
페어 트레이딩(Pairs Trading)은 통계적 차익거래(Statistical Arbitrage)의 대표적인 전략입니다. 상관관계가 높은 두 자산의 가격 스프레드가 평균에서 벗어날 때, 고평가된 자산을 매도하고 저평가된 자산을 매수하여 스프레드가 평균으로 회귀할 때 수익을 얻는 방식입니다.
시장 방향에 중립적(Market Neutral)이기 때문에 상승장·하락장 모두에서 수익 기회를 가질 수 있어, 퀀트 트레이더들이 가장 먼저 배우는 전략 중 하나입니다. 이 글에서는 파이썬으로 페어 트레이딩 자동매매 시스템을 구현하는 전체 과정을 다룹니다.
페어 선정: 공적분 검정
페어 트레이딩의 핵심은 단순 상관관계가 아닌 공적분(Cointegration) 관계를 가진 종목 쌍을 찾는 것입니다. 상관관계는 시간에 따라 변하지만, 공적분 관계는 장기적으로 안정적인 균형을 유지합니다.
import numpy as np
import statsmodels.api as sm
from statsmodels.tsa.stattools import coint, adfuller
def find_cointegrated_pairs(price_data, significance=0.05):
"""공적분 관계가 있는 종목 쌍 탐색"""
n = price_data.shape[1]
symbols = price_data.columns
pairs = []
for i in range(n):
for j in range(i + 1, n):
s1 = price_data.iloc[:, i]
s2 = price_data.iloc[:, j]
# Engle-Granger 공적분 검정
score, p_value, _ = coint(s1, s2)
if p_value < significance:
# 헤지 비율 계산 (OLS 회귀)
model = sm.OLS(s1, sm.add_constant(s2)).fit()
hedge_ratio = model.params[1]
pairs.append({
'asset_1': symbols[i],
'asset_2': symbols[j],
'p_value': round(p_value, 4),
'hedge_ratio': round(hedge_ratio, 4)
})
return sorted(pairs, key=lambda x: x['p_value'])Engle-Granger 검정의 p-value가 0.05 미만이면 공적분 관계가 통계적으로 유의합니다. 헤지 비율(Hedge Ratio)은 두 자산의 포지션 크기 비율을 결정하며, OLS 회귀 계수로 추정합니다.
스프레드 계산과 Z-Score
공적분 쌍을 찾았으면, 두 자산 간의 스프레드(Spread)를 계산하고 이를 표준화한 Z-Score로 매매 신호를 생성합니다.
import pandas as pd
class SpreadCalculator:
def __init__(self, lookback=60):
self.lookback = lookback
def calculate_spread(self, price_a, price_b, hedge_ratio):
"""스프레드 = 자산A - (헤지비율 × 자산B)"""
spread = price_a - hedge_ratio * price_b
return spread
def calculate_zscore(self, spread):
"""이동평균 기반 Z-Score 계산"""
mean = spread.rolling(window=self.lookback).mean()
std = spread.rolling(window=self.lookback).std()
zscore = (spread - mean) / std
return zscore
def generate_signals(self, zscore, entry=2.0, exit=0.5):
"""
Z-Score 기반 매매 신호 생성
- |Z| > entry: 진입 (스프레드 확대)
- |Z| < exit: 청산 (스프레드 수렴)
"""
signals = pd.Series(0, index=zscore.index)
# Z > entry_threshold → 스프레드 과대 → A 매도, B 매수
signals[zscore > entry] = -1
# Z < -entry_threshold → 스프레드 과소 → A 매수, B 매도
signals[zscore < -entry] = 1
# 청산 구간
signals[abs(zscore) < exit] = 0
return signalsZ-Score가 ±2 이상이면 스프레드가 평균에서 2 표준편차 이상 벗어난 것이므로 진입하고, ±0.5 이내로 돌아오면 청산합니다. 이 임계값은 백테스트를 통해 최적화할 수 있습니다.
자동매매 엔진 구현
실시간으로 스프레드를 모니터링하고 자동으로 주문을 실행하는 트레이딩 엔진을 구현합니다.
import asyncio
from dataclasses import dataclass
from typing import Optional
from datetime import datetime
@dataclass
class Position:
asset_a_qty: float = 0
asset_b_qty: float = 0
entry_zscore: float = 0
entry_time: Optional[datetime] = None
pnl: float = 0
class PairsTradingEngine:
def __init__(self, exchange_client, config):
self.exchange = exchange_client
self.config = config
self.position = Position()
self.spread_calc = SpreadCalculator(
lookback=config['lookback']
)
async def run(self):
"""메인 트레이딩 루프"""
while True:
try:
prices = await self.fetch_latest_prices()
spread = self.spread_calc.calculate_spread(
prices['asset_a'],
prices['asset_b'],
self.config['hedge_ratio']
)
zscore = self.spread_calc.calculate_zscore(spread)
current_z = zscore.iloc[-1]
await self.evaluate_and_execute(current_z, prices)
await asyncio.sleep(self.config['interval_sec'])
except Exception as e:
print(f"[ERROR] {e}")
await asyncio.sleep(60)
async def evaluate_and_execute(self, zscore, prices):
"""Z-Score 기반 포지션 평가 및 주문 실행"""
entry_th = self.config['entry_threshold']
exit_th = self.config['exit_threshold']
stop_th = self.config.get('stop_threshold', 4.0)
# 포지션이 없을 때 → 진입 검토
if self.position.asset_a_qty == 0:
if zscore > entry_th:
await self.open_position('short_spread', prices)
elif zscore < -entry_th:
await self.open_position('long_spread', prices)
# 포지션이 있을 때 → 청산 검토
else:
if abs(zscore) < exit_th:
await self.close_position(prices, reason='mean_revert')
elif abs(zscore) > stop_th:
await self.close_position(prices, reason='stop_loss')
async def open_position(self, direction, prices):
"""포지션 진입"""
qty_a = self.config['position_size']
qty_b = qty_a * self.config['hedge_ratio']
if direction == 'short_spread':
# A 매도, B 매수
await self.exchange.sell(self.config['asset_a'], qty_a)
await self.exchange.buy(self.config['asset_b'], qty_b)
self.position = Position(-qty_a, qty_b)
else:
# A 매수, B 매도
await self.exchange.buy(self.config['asset_a'], qty_a)
await self.exchange.sell(self.config['asset_b'], qty_b)
self.position = Position(qty_a, -qty_b)
self.position.entry_time = datetime.now()
print(f"[OPEN] {direction} | A: {self.position.asset_a_qty}, B: {self.position.asset_b_qty}")
async def close_position(self, prices, reason):
"""포지션 청산"""
if self.position.asset_a_qty > 0:
await self.exchange.sell(self.config['asset_a'],
abs(self.position.asset_a_qty))
else:
await self.exchange.buy(self.config['asset_a'],
abs(self.position.asset_a_qty))
if self.position.asset_b_qty > 0:
await self.exchange.sell(self.config['asset_b'],
abs(self.position.asset_b_qty))
else:
await self.exchange.buy(self.config['asset_b'],
abs(self.position.asset_b_qty))
print(f"[CLOSE] reason={reason}")
self.position = Position()핵심 포인트는 손절(Stop Loss) 로직입니다. Z-Score가 ±4를 초과하면 공적분 관계가 붕괴될 수 있으므로 즉시 청산합니다. 이는 리스크 관리 시스템의 필수 요소입니다.
롤링 공적분 검정으로 안정성 확보
공적분 관계는 시간이 지나면 약해질 수 있습니다. 롤링 윈도우로 주기적으로 공적분을 재검정하여 전략의 유효성을 확인해야 합니다.
class RollingCointegrationMonitor:
def __init__(self, window=120, recheck_interval=20):
self.window = window
self.recheck_interval = recheck_interval
def check_validity(self, price_a, price_b):
"""롤링 윈도우 공적분 검정"""
recent_a = price_a[-self.window:]
recent_b = price_b[-self.window:]
_, p_value, _ = coint(recent_a, recent_b)
# ADF 검정으로 스프레드 정상성 확인
spread = recent_a - self.estimate_hedge(recent_a, recent_b) * recent_b
adf_stat, adf_p, *_ = adfuller(spread)
is_valid = p_value < 0.05 and adf_p < 0.05
return {
'cointegration_pvalue': round(p_value, 4),
'adf_pvalue': round(adf_p, 4),
'is_valid': is_valid,
'hedge_ratio': self.estimate_hedge(recent_a, recent_b)
}
def estimate_hedge(self, a, b):
"""헤지 비율 재추정"""
model = sm.OLS(a, sm.add_constant(b)).fit()
return model.params[1]공적분 p-value가 0.05를 초과하면 해당 페어의 거래를 중단하고 새로운 페어를 탐색해야 합니다. 이 모니터링을 자동화하면 전략 붕괴 전에 미리 대응할 수 있습니다.
반감기(Half-Life) 계산
스프레드가 평균으로 회귀하는 속도를 반감기(Half-Life)로 측정합니다. 반감기가 너무 길면 자본이 오래 묶이고, 너무 짧으면 거래 비용 대비 수익이 적습니다.
def calculate_half_life(spread):
"""
Ornstein-Uhlenbeck 과정 기반 반감기 계산
- 이상적 범위: 5~60일 (일봉 기준)
"""
spread_lag = spread.shift(1).dropna()
spread_diff = spread.diff().dropna()
# 정렬
spread_lag = spread_lag.iloc[1:]
spread_diff = spread_diff.iloc[1:]
model = sm.OLS(spread_diff, sm.add_constant(spread_lag)).fit()
theta = model.params[1] # 평균회귀 속도
if theta >= 0:
return float('inf') # 평균회귀 없음
half_life = -np.log(2) / theta
return round(half_life, 1)
# 사용 예시
# half_life = calculate_half_life(spread)
# if 5 <= half_life <= 60:
# print(f"적합한 페어 (반감기: {half_life}일)")
# else:
# print("부적합: 반감기가 범위 밖")일반적으로 반감기 5~60일이 실전에서 적합한 범위입니다. 이 값은 룩백 기간(lookback window) 설정에도 활용됩니다.
실전 적용 시 주의사항
| 항목 | 권장 사항 |
|---|---|
| 거래 비용 | 양방향 매매이므로 수수료 2배 — 스프레드 수익이 비용을 초과하는지 확인 |
| 슬리피지 | 두 자산 동시 주문 시 체결 시차 발생 → 시장가보다 지정가 주문 권장 |
| 구조적 변화 | 합병, 상장폐지 등으로 공적분 붕괴 가능 → 롤링 검정 필수 |
| 레버리지 | 마켓 뉴트럴이라도 과도한 레버리지는 위험 — 최대 2배 권장 |
| 유동성 | 저유동성 자산은 진입/청산이 어려움 → 일 거래량 상위 종목 선택 |
암호화폐 페어 트레이딩 예시
암호화폐 시장에서 자주 활용되는 페어 예시입니다:
- BTC/ETH: 가장 대표적인 크립토 페어, 장기 공적분 관계 확인됨
- SOL/AVAX: L1 경쟁 체인 간 높은 상관성
- LINK/UNI: DeFi 섹터 내 대형 토큰 쌍
- 거래소 토큰(BNB/OKB): 유사 비즈니스 모델 기반 쌍
다만 암호화폐는 전통 자산보다 공적분 관계가 불안정할 수 있으므로, 롤링 검정 주기를 더 짧게(10~20일) 설정하는 것이 안전합니다.
성과 측정 지표
페어 트레이딩 전략의 성과는 다음 지표로 평가합니다:
- 샤프 비율: 마켓 뉴트럴 전략은 1.5 이상이 목표
- 승률: 평균회귀 전략 특성상 60~70% 이상 기대
- 최대 낙폭(MDD): 공적분 붕괴 시 대규모 손실 가능 → 10% 이내 관리
- 평균 보유 기간: 반감기와 유사해야 정상
- 수익 팩터: 총이익 / 총손실 → 1.5 이상 권장
마무리
페어 트레이딩은 시장 방향에 베팅하지 않고도 통계적 우위를 확보할 수 있는 강력한 퀀트 전략입니다. 핵심은 공적분 검정으로 유효한 페어를 찾고, Z-Score 기반으로 진입/청산 타이밍을 잡으며, 롤링 검정으로 전략의 유효성을 지속 모니터링하는 것입니다.
특히 암호화폐 시장에서는 24시간 운영, 높은 변동성, 다양한 섹터 페어 등의 특성 덕분에 페어 트레이딩이 활발히 활용되고 있습니다. 포트폴리오 리밸런싱과 결합하면 더욱 안정적인 퀀트 시스템을 구축할 수 있습니다.