Расчет Taker Buy/Sell Ratio с Binance AggTrades на Python

import asyncio
import websockets
import json
import pandas as pd
from collections import deque

class TakerRatioCalculator:
    def __init__(self, symbol: str, window_size: int = 100):
        self.symbol = symbol.upper()
        self.window_size = window_size
        self.trades_buffer = deque(maxlen=window_size)
        self.taker_buy_volume = 0.0
        self.taker_sell_volume = 0.0
        print(f"[INFO] Инициализирован калькулятор для {self.symbol} с окном {self.window_size} сделок.")

    def _process_trade(self, trade_data: dict):
        """
        Обрабатывает одну агрегированную сделку, определяя ее как Taker Buy или Taker Sell.
        'm': true - buyer is maker, means seller is taker (Taker Sell)
        'm': false - seller is maker, means buyer is taker (Taker Buy)
        """
        quantity = float(trade_data['q']) # Количество
        price = float(trade_data['p'])   # Цена
        is_buyer_maker = trade_data['m'] # True, если покупатель - мейкер
        trade_volume = quantity * price

        if is_buyer_maker: # Покупатель - мейкер -> продавец - тейкер -> Taker Sell
            self.taker_sell_volume += trade_volume
            return 'sell', trade_volume
        else: # Продавец - мейкер -> покупатель - тейкер -> Taker Buy
            self.taker_buy_volume += trade_volume
            return 'buy', trade_volume

    def _update_buffer(self, trade_type: str, trade_volume: float):
        """
        Добавляет новую сделку в буфер и удаляет старейшую, если буфер полон.
        Обновляет суммарные объемы Taker Buy/Sell.
        """
        if len(self.trades_buffer) == self.window_size:
            old_trade_type, old_trade_volume = self.trades_buffer.popleft()
            if old_trade_type == 'buy':
                self.taker_buy_volume -= old_trade_volume
            else:
                self.taker_sell_volume -= old_trade_volume
        
        self.trades_buffer.append((trade_type, trade_volume))

    def _calculate_ratio(self) -> float:
        """
        Рассчитывает Taker Buy/Sell Ratio.
        Ratio = Taker Buy Volume / (Taker Buy Volume + Taker Sell Volume)
        """
        total_taker_volume = self.taker_buy_volume + self.taker_sell_volume
        if total_taker_volume == 0:
            return 0.5 # Нейтральное значение, если нет объема
        return self.taker_buy_volume / total_taker_volume

    async def _handle_message(self, message: str):
        """
        Обрабатывает входящее сообщение WebSocket.
        """
        data = json.loads(message)
        if data['e'] == 'aggTrade': # Проверяем, что это агрегированная сделка
            trade_type, trade_volume = self._process_trade(data)
            self._update_buffer(trade_type, trade_volume)
            
            if len(self.trades_buffer) == self.window_size: # Рассчитываем только когда буфер заполнен
                ratio = self._calculate_ratio()
                print(f"[{pd.to_datetime(data['E'], unit='ms')}] {self.symbol} Taker Buy/Sell Ratio: {ratio:.4f} (Buy: {self.taker_buy_volume:.2f}, Sell: {self.taker_sell_volume:.2f})")

async def main():
    symbol = "BTCUSDT"
    window_size = 200 # Количество последних сделок для расчета соотношения
    calculator = TakerRatioCalculator(symbol, window_size)

    uri = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@aggTrade"
    print(f"[INFO] Подключение к WebSocket: {uri}")

    async with websockets.connect(uri) as websocket:
        print("[INFO] Подключено. Ожидание данных...")
        while True:
            try:
                message = await websocket.recv()
                await calculator._handle_message(message)
            except websockets.exceptions.ConnectionClosedOK:
                print("[INFO] Соединение закрыто.")
                break
            except Exception as e:
                print(f"[ERROR] Произошла ошибка: {e}")
                await asyncio.sleep(1) # Пауза перед повторной попыткой

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