Реализация алгоритма TWAP на Python для плавного исполнения ордеров

Суть: Алгоритм TWAP (Time-Weighted Average Price) разбивает крупный объем заявки на равные части (слайсы) и исполняет их через равные промежутки времени. Это минимизирует рыночное влияние (market impact) и скрывает присутствие крупного игрока в стакане. В статье представлена готовая асинхронная реализация TWAP-исполнителя на Python с компенсацией временного дрейфа.

При исполнении крупных институциональных ордеров отправка всей заявки по рынку (Market Order) неизбежно приводит к проскальзыванию (slippage) и неблагоприятному движению цены. Для решения этой проблемы количественные трейдеры используют алгоритмы оптимального исполнения, среди которых базовым и наиболее надежным является TWAP.

Для тестирования таких алгоритмов на исторических данных критически важно иметь качественную базу тиков, которую можно реализовать, изучив руководство Интеграция TimescaleDB для хранения тиков с Binance на Python.

Содержание

Исходный код

Ниже представлена промышленная асинхронная реализация класса TWAPExecutor на Python. Алгоритм учитывает время выполнения сетевого запроса и динамически корректирует паузу между слайсами, чтобы избежать накопления временного сдвига (drift).

import asyncio
import logging
import time
from typing import Dict, Any

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("TWAP")

class TWAPExecutor:
    def __init__(self, total_quantity: float, duration_seconds: int, interval_seconds: int, symbol: str, side: str):
        self.total_quantity = total_quantity
        self.duration_seconds = duration_seconds
        self.interval_seconds = interval_seconds
        self.symbol = symbol
        self.side = side.upper()
        
        self.num_slices = max(1, duration_seconds // interval_seconds)
        self.slice_quantity = total_quantity / self.num_slices
        self.executed_quantity = 0.0
        
    async def execute_slice(self, slice_index: int) -> Dict[str, Any]:
        # Имитация отправки ордера на биржу
        logger.info(f"Отправка слайса {slice_index + 1}/{self.num_slices}: {self.slice_quantity:.4f} {self.symbol}")
        await asyncio.sleep(0.15)  # Имитация сетевой задержки API
        return {
            "status": "FILLED",
            "price": 98500.0 + (slice_index * 12.5),  # Имитация изменения цены
            "quantity": self.slice_quantity,
            "timestamp": time.time()
        }

    async def run(self):
        logger.info(f"Старт TWAP: {self.total_quantity} {self.symbol} за {self.duration_seconds}с (интервалы по {self.interval_seconds}с)")
        
        for i in range(self.num_slices):
            start_time = time.time()
            try:
                order_result = await self.execute_slice(i)
                if order_result["status"] == "FILLED":
                    self.executed_quantity += order_result["quantity"]
                    logger.info(f"Слайс {i + 1} успешно исполнен по цене {order_result['price']:.2f}")
            except Exception as e:
                logger.error(f"Ошибка при исполнении слайса {i + 1}: {e}")
            
            elapsed = time.time() - start_time
            sleep_time = max(0.0, self.interval_seconds - elapsed)
            
            if i < self.num_slices - 1:
                logger.info(f"Ожидание следующего интервала: {sleep_time:.2f} сек.")
                await asyncio.sleep(sleep_time)
                
        logger.info(f"TWAP завершен. Исполнено: {self.executed_quantity:.4f}/{self.total_quantity} {self.symbol}")

async def main():
    # Пример: продать 1.5 BTC за 30 секунд с шагом в 10 секунд (3 слайса по 0.5 BTC)
    executor = TWAPExecutor(
        total_quantity=1.5,
        duration_seconds=30,
        interval_seconds=10,
        symbol="BTCUSDT",
        side="SELL"
    )
    await executor.run()

if __name__ == "__main__":
    asyncio.run(main())

Разбор параметров

total_quantity: Общий объем актива, который необходимо купить или продать в рамках всей сессии TWAP.
duration_seconds: Общая длительность работы алгоритма в секундах, в течение которой весь объем должен быть полностью исполнен.
interval_seconds: Интервал времени в секундах между отправками отдельных дочерних ордеров (слайсов).
symbol: Торговый инструмент (например, BTCUSDT), на котором совершаются операции.
side: Направление сделки (BUY или SELL).

Как запустить

Для запуска скрипта вам понадобится установленный интерпретатор Python версии 3.8 или выше. Скопируйте код в файл twap_executor.py и запустите его из терминала:

python twap_executor.py

В реальных торговых системах для получения актуального состояния стакана и цен перед отправкой каждого слайса рекомендуется использовать быстрые in-memory хранилища. О том, как это настроить, читайте в статье Интеграция бота-маркетмейкера на Python с Redis для обновления котировок.

После отправки каждого ордера крайне важно сохранять детали транзакции для последующего анализа качества исполнения (Transaction Cost Analysis - TCA). Для этого отлично подойдет подход, описанный в материале Асинхронное логирование торговых операций в MongoDB на Python.