TWAP 주문 실행 전략이란?
TWAP(Time-Weighted Average Price)은 대량 주문을 일정 시간에 걸쳐 균등하게 분할 실행하는 알고리즘 트레이딩 전략입니다. 기관 투자자와 퀀트 트레이더가 시장 충격(Market Impact)을 최소화하기 위해 가장 널리 사용하는 실행 알고리즘 중 하나입니다.
자동매매 봇을 운영할 때, 한 번에 큰 수량을 매수하거나 매도하면 슬리피지가 커지고 호가창에 영향을 줍니다. TWAP은 이 문제를 해결하는 가장 기본적이면서도 효과적인 방법입니다.
TWAP vs VWAP 차이점
TWAP과 VWAP 자동매매 전략은 자주 비교됩니다. 핵심 차이를 정리하면:
| 구분 | TWAP | VWAP |
|---|---|---|
| 분할 기준 | 시간 균등 분할 | 거래량 가중 분할 |
| 구현 난이도 | 낮음 | 중간 (거래량 프로파일 필요) |
| 적합한 시장 | 24시간 암호화폐, 유동성 낮은 시장 | 주식시장 등 거래량 패턴이 뚜렷한 시장 |
| 시장 충격 | 예측 가능하게 분산 | 거래량 많은 시간에 집중 |
24시간 운영되는 암호화폐 자동매매에서는 거래량 프로파일 예측이 어렵기 때문에 TWAP이 더 실용적인 선택인 경우가 많습니다.
파이썬 TWAP 자동매매 봇 구현
아래는 ccxt 라이브러리를 활용한 TWAP 주문 실행기의 핵심 구현입니다.
import ccxt
import time
import math
class TWAPExecutor:
"""TWAP 주문 실행기: 대량 주문을 시간 균등 분할 실행"""
def __init__(self, exchange, symbol, side, total_qty, duration_min, num_slices):
self.exchange = exchange
self.symbol = symbol
self.side = side # 'buy' 또는 'sell'
self.total_qty = total_qty # 총 주문 수량
self.duration_min = duration_min # 전체 실행 시간(분)
self.num_slices = num_slices # 분할 횟수
self.slice_qty = total_qty / num_slices
self.interval = (duration_min * 60) / num_slices # 슬라이스 간격(초)
self.executed = []
def execute(self):
"""TWAP 주문 실행 루프"""
print(f"TWAP 시작: {self.total_qty} {self.symbol}")
print(f" 분할: {self.num_slices}회, 간격: {self.interval:.0f}초")
for i in range(self.num_slices):
try:
# 시장가 주문으로 슬라이스 실행
order = self.exchange.create_market_order(
symbol=self.symbol,
side=self.side,
amount=self.slice_qty
)
filled_price = order.get('average', order.get('price', 0))
self.executed.append({
'slice': i + 1,
'qty': self.slice_qty,
'price': filled_price,
'timestamp': time.time()
})
print(f" [{i+1}/{self.num_slices}] "
f"{self.slice_qty:.4f} @ {filled_price:.2f}")
except Exception as e:
print(f" [{i+1}] 주문 실패: {e}")
# 마지막 슬라이스가 아니면 대기
if i < self.num_slices - 1:
time.sleep(self.interval)
return self.get_summary()
def get_summary(self):
"""실행 결과 요약"""
if not self.executed:
return {"status": "실패", "filled": 0}
total_cost = sum(e['qty'] * e['price'] for e in self.executed)
total_filled = sum(e['qty'] for e in self.executed)
avg_price = total_cost / total_filled if total_filled else 0
return {
"status": "완료",
"total_filled": total_filled,
"avg_price": avg_price,
"num_executed": len(self.executed),
"fill_rate": len(self.executed) / self.num_slices * 100
}랜덤화로 실행 품질 높이기
기본 TWAP은 일정 간격으로 주문하므로 다른 트레이더가 패턴을 감지할 수 있습니다. 실전에서는 시간 랜덤화와 수량 랜덤화를 추가합니다.
import random
class RandomizedTWAP(TWAPExecutor):
"""랜덤화된 TWAP: 패턴 감지 방지"""
def __init__(self, *args, time_jitter=0.3, qty_jitter=0.2, **kwargs):
super().__init__(*args, **kwargs)
self.time_jitter = time_jitter # 시간 변동폭 (30%)
self.qty_jitter = qty_jitter # 수량 변동폭 (20%)
def _randomize_interval(self):
"""간격에 랜덤 지터 추가"""
jitter = self.interval * self.time_jitter
return self.interval + random.uniform(-jitter, jitter)
def _randomize_qty(self, remaining, slices_left):
"""수량 랜덤화 (총합은 유지)"""
base = remaining / slices_left
jitter = base * self.qty_jitter
qty = base + random.uniform(-jitter, jitter)
return max(qty, base * 0.5) # 최소 50%는 보장
def execute(self):
"""랜덤화된 TWAP 실행"""
remaining = self.total_qty
for i in range(self.num_slices):
slices_left = self.num_slices - i
qty = self._randomize_qty(remaining, slices_left)
qty = min(qty, remaining) # 남은 수량 초과 방지
try:
order = self.exchange.create_market_order(
self.symbol, self.side, qty
)
filled = order.get('average', 0)
remaining -= qty
self.executed.append({
'slice': i + 1, 'qty': qty,
'price': filled, 'timestamp': time.time()
})
except Exception as e:
print(f" 슬라이스 {i+1} 실패: {e}")
if i < self.num_slices - 1:
time.sleep(self._randomize_interval())
return self.get_summary()TWAP 실행 시 주의사항 5가지
TWAP 자동매매를 실전에 적용할 때 반드시 고려해야 할 사항들입니다.
1. 최소 주문 수량 확인
슬라이스를 너무 잘게 나누면 거래소의 최소 주문 수량에 미달할 수 있습니다. ccxt의 markets 정보에서 최소 수량을 확인하세요.
market = exchange.market(symbol)
min_amount = market['limits']['amount']['min']
# 슬라이스 수량이 min_amount 이상인지 검증
assert slice_qty >= min_amount, "슬라이스가 최소 주문 수량 미달"2. 수수료 누적 관리
분할 주문 횟수가 많아지면 수수료가 누적됩니다. 일반적으로 20~50회 분할이 시장 충격 감소와 수수료 사이의 적정 균형점입니다.
3. 급변 시장 대응
TWAP 실행 도중 가격이 급등하거나 급락하면 불리한 가격에 계속 매수할 수 있습니다. 가격 임계값을 설정해서 실행을 일시 중지하는 로직이 필요합니다.
# 가격 급변 감지: 시작 가격 대비 2% 이상 변동 시 중지
start_price = exchange.fetch_ticker(symbol)['last']
THRESHOLD = 0.02
for i in range(num_slices):
current = exchange.fetch_ticker(symbol)['last']
change = abs(current - start_price) / start_price
if change > THRESHOLD:
print(f"가격 급변 감지 ({change:.1%}), 실행 일시 중지")
time.sleep(60) # 1분 대기 후 재평가
continue
# ... 주문 실행4. 네트워크 장애 대비
장시간 실행되는 TWAP은 네트워크 장애를 만날 확률이 높습니다. 재시도 로직과 실행 상태 저장을 반드시 구현하세요.
5. 지정가 주문 활용
시장가 대신 최우선 호가 지정가 주문을 사용하면 슬리피지를 더 줄일 수 있습니다. 다만 미체결 처리 로직이 추가로 필요합니다.
TWAP 성과 벤치마크 측정
TWAP 실행 품질은 도착가격(Arrival Price) 대비 실행 평균가로 측정합니다. 자동매매 로그 분석과 결합하면 체계적인 성과 관리가 가능합니다.
def calculate_implementation_shortfall(arrival_price, avg_exec_price, side):
"""실행 비용(Implementation Shortfall) 계산"""
if side == 'buy':
# 매수: 실행가가 도착가보다 높으면 비용 발생
shortfall = (avg_exec_price - arrival_price) / arrival_price
else:
# 매도: 실행가가 도착가보다 낮으면 비용 발생
shortfall = (arrival_price - avg_exec_price) / arrival_price
return {
'arrival_price': arrival_price,
'avg_exec_price': avg_exec_price,
'shortfall_bps': shortfall * 10000, # bps 단위
'grade': 'Good' if shortfall < 0.001 else 'Fair' if shortfall < 0.005 else 'Poor'
}실전 적용: 1 BTC를 1시간에 걸쳐 매수
아래는 실제 바이낸스에서 TWAP으로 1 BTC를 매수하는 예시입니다.
import ccxt
# 거래소 설정
exchange = ccxt.binance({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_SECRET',
'options': {'defaultType': 'spot'}
})
# TWAP 실행: 1 BTC, 60분, 30회 분할 (2분 간격)
twap = RandomizedTWAP(
exchange=exchange,
symbol='BTC/USDT',
side='buy',
total_qty=1.0,
duration_min=60,
num_slices=30,
time_jitter=0.25,
qty_jitter=0.15
)
result = twap.execute()
print(f"평균 체결가: {result['avg_price']:.2f} USDT")
print(f"체결률: {result['fill_rate']:.0f}%")마치며
TWAP은 구현이 단순하면서도 실전에서 검증된 주문 실행 전략입니다. 특히 암호화폐 자동매매에서 대량 주문의 슬리피지를 줄이는 데 효과적입니다. 랜덤화, 급변 시장 대응, 성과 측정까지 구현하면 기관 수준의 실행 품질을 달성할 수 있습니다.
자동매매 봇의 슬리피지 최적화와 함께 적용하면 전체 트레이딩 시스템의 수익성을 크게 개선할 수 있습니다.