TWAP 알고리즘이란?
TWAP(Time-Weighted Average Price)은 대량 주문을 일정 시간에 걸쳐 균등하게 분할 체결하는 알고리즘 주문 실행 전략입니다. 기관 투자자와 퀀트 트레이더가 시장 충격(Market Impact)을 최소화하기 위해 가장 널리 사용하는 실행 알고리즘 중 하나입니다.
핵심 원리는 단순합니다. 1000만원어치를 한 번에 매수하는 대신, 1시간 동안 5분 간격으로 12번에 나눠서 매수하면 평균 체결가가 시간가중평균가격에 수렴합니다.
TWAP vs VWAP 비교
TWAP과 함께 자주 언급되는 VWAP(거래량가중평균가격)과의 차이를 이해해야 합니다:
| 항목 | TWAP | VWAP |
|---|---|---|
| 분할 기준 | 시간 균등 | 거래량 비례 |
| 구현 난이도 | 낮음 | 중간 (거래량 예측 필요) |
| 적합한 시장 | 24시간 암호화폐, 유동성 균일 시장 | 주식 등 장중 거래량 패턴이 뚜렷한 시장 |
| 시장 충격 | 일정하게 분산 | 유동성 높은 시간대에 집중 |
| 예측 가능성 | 높음 (패턴 단순) | 낮음 (거래량 적응) |
기본 TWAP 엔진 구현
파이썬으로 구현하는 기본 TWAP 주문 실행 엔진입니다:
import time
import math
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class TWAPOrder:
symbol: str
side: str # "buy" or "sell"
total_quantity: float
duration_minutes: int
interval_seconds: int = 30
filled_quantity: float = 0
fills: list = field(default_factory=list)
@property
def num_slices(self) -> int:
return max(1, (self.duration_minutes * 60) // self.interval_seconds)
@property
def slice_quantity(self) -> float:
remaining = self.total_quantity - self.filled_quantity
remaining_slices = self.num_slices - len(self.fills)
if remaining_slices <= 0:
return remaining
return remaining / remaining_slices
@property
def progress(self) -> float:
return self.filled_quantity / self.total_quantity if self.total_quantity > 0 else 0
class TWAPEngine:
"""TWAP 알고리즘 주문 실행 엔진"""
def __init__(self, exchange):
self.exchange = exchange
self.active_orders = {}
def create_twap(self, symbol: str, side: str,
total_qty: float, duration_min: int,
interval_sec: int = 30) -> TWAPOrder:
order = TWAPOrder(
symbol=symbol,
side=side,
total_quantity=total_qty,
duration_minutes=duration_min,
interval_seconds=interval_sec,
)
self.active_orders[symbol] = order
return order
def execute_slice(self, order: TWAPOrder) -> dict:
"""단일 슬라이스 실행"""
qty = order.slice_quantity
if qty <= 0:
return {"status": "complete"}
try:
result = self.exchange.create_market_order(
symbol=order.symbol,
side=order.side,
amount=qty,
)
fill = {
"time": datetime.now().isoformat(),
"quantity": qty,
"price": result.get("average", result.get("price", 0)),
"order_id": result.get("id"),
}
order.fills.append(fill)
order.filled_quantity += qty
return {"status": "filled", "fill": fill}
except Exception as e:
return {"status": "error", "error": str(e)}
def run(self, order: TWAPOrder):
"""전체 TWAP 실행 루프"""
print(f"TWAP 시작: {order.symbol} {order.side} "
f"{order.total_quantity} / {order.duration_minutes}분")
for i in range(order.num_slices):
result = self.execute_slice(order)
print(f" 슬라이스 {i+1}/{order.num_slices}: {result['status']} "
f"({order.progress:.1%} 완료)")
if order.filled_quantity >= order.total_quantity:
break
if i < order.num_slices - 1:
time.sleep(order.interval_seconds)
avg_price = self._calc_avg_price(order)
print(f"TWAP 완료: 평균가 {avg_price:,.2f}")
return order
def _calc_avg_price(self, order: TWAPOrder) -> float:
if not order.fills:
return 0
total_cost = sum(f["quantity"] * f["price"] for f in order.fills)
total_qty = sum(f["quantity"] for f in order.fills)
return total_cost / total_qty if total_qty > 0 else 0랜덤화 TWAP: 예측 방지
기본 TWAP은 일정 간격으로 주문하기 때문에 다른 트레이더가 패턴을 예측하고 앞서 매매(front-running)할 수 있습니다. 랜덤화를 추가하면 이를 방지합니다:
import random
class RandomizedTWAP(TWAPEngine):
"""시간·수량 랜덤화 TWAP"""
def __init__(self, exchange, time_jitter: float = 0.3,
qty_jitter: float = 0.2):
super().__init__(exchange)
self.time_jitter = time_jitter # ±30% 시간 변동
self.qty_jitter = qty_jitter # ±20% 수량 변동
def run(self, order: TWAPOrder):
print(f"랜덤 TWAP 시작: {order.symbol} {order.side}")
for i in range(order.num_slices):
# 수량 랜덤화
base_qty = order.slice_quantity
jittered_qty = base_qty * (1 + random.uniform(
-self.qty_jitter, self.qty_jitter))
remaining = order.total_quantity - order.filled_quantity
actual_qty = min(jittered_qty, remaining)
if actual_qty <= 0:
break
# 임시로 슬라이스 수량 오버라이드
original_qty = order.slice_quantity
result = self.execute_slice(order)
print(f" 슬라이스 {i+1}: qty={actual_qty:.4f} "
f"({order.progress:.1%})")
if order.filled_quantity >= order.total_quantity:
break
# 시간 랜덤화
if i < order.num_slices - 1:
base_sleep = order.interval_seconds
jittered_sleep = base_sleep * (1 + random.uniform(
-self.time_jitter, self.time_jitter))
time.sleep(max(1, jittered_sleep))
return order적응형 TWAP: 시장 상황 반영
스프레드와 변동성에 따라 주문 속도를 조절하는 적응형 TWAP입니다:
import numpy as np
class AdaptiveTWAP(TWAPEngine):
"""시장 상황 적응형 TWAP"""
def __init__(self, exchange, spread_threshold: float = 0.001,
vol_window: int = 20):
super().__init__(exchange)
self.spread_threshold = spread_threshold
self.vol_window = vol_window
self.price_history = []
def assess_market(self, symbol: str) -> dict:
"""현재 시장 상태 평가"""
ob = self.exchange.fetch_order_book(symbol, limit=5)
best_bid = ob["bids"][0][0]
best_ask = ob["asks"][0][0]
mid = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid
self.price_history.append(mid)
if len(self.price_history) > self.vol_window:
self.price_history.pop(0)
volatility = 0
if len(self.price_history) >= 5:
returns = np.diff(np.log(self.price_history))
volatility = np.std(returns)
return {
"spread": spread,
"volatility": volatility,
"mid_price": mid,
"bid_depth": sum(b[1] for b in ob["bids"][:5]),
"ask_depth": sum(a[1] for a in ob["asks"][:5]),
}
def get_urgency(self, market: dict) -> float:
"""
시장 상태에 따른 긴급도 (0.5~2.0)
스프레드 좁고 변동성 낮으면 → 천천히 (< 1.0)
스프레드 넓고 변동성 높으면 → 빠르게 (> 1.0)
"""
spread_score = market["spread"] / self.spread_threshold
vol_score = 1 + market["volatility"] * 100
urgency = (spread_score + vol_score) / 2
return max(0.5, min(2.0, urgency))
def run(self, order: TWAPOrder):
print(f"적응형 TWAP 시작: {order.symbol}")
i = 0
while order.filled_quantity < order.total_quantity:
market = self.assess_market(order.symbol)
urgency = self.get_urgency(market)
result = self.execute_slice(order)
print(f" 슬라이스 {i+1}: urgency={urgency:.2f} "
f"spread={market['spread']:.4%} "
f"({order.progress:.1%})")
if order.filled_quantity >= order.total_quantity:
break
# 긴급도에 따라 대기 시간 조절
wait = order.interval_seconds / urgency
time.sleep(max(1, wait))
i += 1
return orderTWAP 성과 측정: 실행 품질 분석
TWAP 실행의 품질을 평가하는 지표들입니다:
def analyze_execution(order: TWAPOrder, market_twap: float) -> dict:
"""
TWAP 실행 품질 분석
market_twap: 같은 기간 시장의 실제 TWAP
"""
if not order.fills:
return {"error": "체결 없음"}
prices = [f["price"] for f in order.fills]
quantities = [f["quantity"] for f in order.fills]
total_cost = sum(p * q for p, q in zip(prices, quantities))
total_qty = sum(quantities)
exec_avg = total_cost / total_qty
# Implementation Shortfall (구현 부족)
arrival_price = order.fills[0]["price"]
impl_shortfall = (exec_avg - arrival_price) / arrival_price
if order.side == "sell":
impl_shortfall *= -1
# TWAP 슬리피지
twap_slippage = (exec_avg - market_twap) / market_twap
if order.side == "sell":
twap_slippage *= -1
return {
"실행평균가": f"{exec_avg:,.2f}",
"시장TWAP": f"{market_twap:,.2f}",
"도착가격": f"{arrival_price:,.2f}",
"TWAP슬리피지": f"{twap_slippage:.4%}",
"구현부족": f"{impl_shortfall:.4%}",
"체결횟수": len(order.fills),
"가격표준편차": f"{np.std(prices):,.2f}",
"총체결수량": f"{total_qty:.4f}",
}실전 적용 시 주의사항
- 최소 주문 수량: 거래소마다 최소 주문 수량이 있습니다. 슬라이스 크기가 최소 수량 미만이면 주문이 거부됩니다. 사전에 확인하고 슬라이스 수를 조절하세요.
- API 호출 제한: 대부분 거래소는 분당 API 호출 수를 제한합니다. 간격을 최소 10초 이상으로 설정하는 것이 안전합니다.
- 부분 체결 처리: 시장가 주문도 유동성 부족 시 부분 체결될 수 있습니다. 미체결분을 다음 슬라이스에 포함하는 로직이 필요합니다.
- 네트워크 장애 대응: 실행 중 연결이 끊기면 현재 진행 상태를 저장하고, 재연결 후 이어서 실행할 수 있어야 합니다.
- 슬리피지 최적화와 병행: TWAP으로 시간 분산을 하더라도 각 슬라이스의 슬리피지를 모니터링하고, 비정상적으로 높으면 실행을 일시 중단하세요.
핵심 정리
TWAP은 대량 주문의 시장 충격을 최소화하는 가장 기본적이고 효과적인 알고리즘 주문 실행 전략입니다. 기본 TWAP에 랜덤화를 추가하면 프론트러닝을 방지할 수 있고, 적응형 TWAP은 시장 상황에 따라 실행 속도를 최적화합니다. 자동매매 봇에 TWAP 엔진을 통합하면 실행 품질이 크게 향상됩니다.