Реализация Grid Bot для Bybit на Python: Гайд Quant-разработчика

import asyncio
import time
import hmac
import hashlib
import json
import aiohttp
import logging

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

class BybitGridBot:
    def __init__(self, api_key: str, api_secret: str, symbol: str, grid_step_pct: float, grid_levels: int, qty_per_level: float, is_testnet: bool = True):
        self.api_key = api_key
        self.api_secret = api_secret
        self.symbol = symbol
        self.grid_step_pct = grid_step_pct
        self.grid_levels = grid_levels
        self.qty_per_level = qty_per_level
        self.base_url = "https://api-testnet.bybit.com" if is_testnet else "https://api.bybit.com"
        self.session = None
        self.active_orders = {}

    async def init_session(self):
        self.session = aiohttp.ClientSession()

    async def close_session(self):
        if self.session:
            await self.session.close()

    def _generate_signature(self, timestamp: str, recv_window: str, payload: str) -> str:
        param_str = timestamp + self.api_key + recv_window + payload
        return hmac.new(self.api_secret.encode('utf-8'), param_str.encode('utf-8'), hashlib.sha256).hexdigest()

    async def _request(self, method: str, path: str, data: dict = None) -> dict:
        url = f"{self.base_url}{path}"
        timestamp = str(int(time.time() * 1000))
        recv_window = "5000"
        payload = json.dumps(data) if data else ""
        
        signature = self._generate_signature(timestamp, recv_window, payload)
        
        headers = {
            "X-BAPI-API-KEY": self.api_key,
            "X-BAPI-SIGN": signature,
            "X-BAPI-SIGN-TYPE": "2",
            "X-BAPI-TIMESTAMP": timestamp,
            "X-BAPI-RECV-WINDOW": recv_window,
            "Content-Type": "application/json"
        }
        
        async with self.session.request(method, url, headers=headers, data=payload) as response:
            result = await response.json()
            if result.get("retCode") != 0:
                logging.error(f"API Error: {result.get('retMsg')} (Code: {result.get('retCode')})")
            return result

    async def get_ticker_price(self) -> float:
        path = f"/v5/market/tickers?category=linear&symbol={self.symbol}"
        res = await self._request("GET", path)
        return float(res["result"]["list"][0]["lastPrice"])

    async def cancel_all_orders(self):
        logging.info("Cancelling all active orders...")
        path = "/v5/order/cancel-all"
        payload = {"category": "linear", "symbol": self.symbol}
        await self._request("POST", path, payload)

    async def place_order(self, side: str, price: float, qty: float) -> str:
        path = "/v5/order/create"
        payload = {
            "category": "linear",
            "symbol": self.symbol,
            "side": side,
            "orderType": "Limit",
            "qty": str(qty),
            "price": f"{price:.4f}",
            "timeInForce": "GTC"
        }
        res = await self._request("POST", path, payload)
        if res.get("retCode") == 0:
            order_id = res["result"]["orderId"]
            logging.info(f"Placed {side} order at {price} | ID: {order_id}")
            return order_id
        return None

    async def setup_grid(self):
        await self.cancel_all_orders()
        self.active_orders.clear()
        
        current_price = await self.get_ticker_price()
        logging.info(f"Current price of {self.symbol}: {current_price}")
        
        for i in range(1, self.grid_levels + 1):
            price = current_price * (1 - self.grid_step_pct * i)
            order_id = await self.place_order("Buy", price, self.qty_per_level)
            if order_id:
                self.active_orders[order_id] = {"side": "Buy", "price": price, "level": i}
        
        for i in range(1, self.grid_levels + 1):
            price = current_price * (1 + self.grid_step_pct * i)
            order_id = await self.place_order("Sell", price, self.qty_per_level)
            if order_id:
                self.active_orders[order_id] = {"side": "Sell", "price": price, "level": i}

    async def monitor_grid(self):
        logging.info("Starting grid monitoring loop...")
        while True:
            try:
                await asyncio.sleep(5)
                path = f"/v5/order/realtime?category=linear&symbol={self.symbol}"
                res = await self._request("GET", path)
                
                if res.get("retCode") != 0:
                    continue
                    
                server_orders = {o["orderId"]: o["orderStatus"] for o in res["result"]["list"]}
                
                filled_orders = []
                for order_id, local_info in list(self.active_orders.items()):
                    status = server_orders.get(order_id)
                    
                    if status == "Filled" or (status is None and order_id not in server_orders):
                        logging.info(f"Order {order_id} ({local_info['side']} at {local_info['price']}) FILLED!")
                        filled_orders.append(order_id)
                        
                for order_id in filled_orders:
                    old_info = self.active_orders.pop(order_id)
                    
                    new_side = "Sell" if old_info["side"] == "Buy" else "Buy"
                    multiplier = (1 + self.grid_step_pct) if new_side == "Sell" else (1 - self.grid_step_pct)
                    new_price = old_info["price"] * multiplier
                    
                    new_order_id = await self.place_order(new_side, new_price, self.qty_per_level)
                    if new_order_id:
                        self.active_orders[new_order_id] = {
                            "side": new_side,
                            "price": new_price,
                            "level": old_info["level"]
                        }
                        
            except Exception as e:
                logging.error(f"Error in monitor loop: {e}")
                await asyncio.sleep(10)

async def main():
    API_KEY = "YOUR_API_KEY"
    API_SECRET = "YOUR_API_SECRET"
    SYMBOL = "BTCUSDT"
    
    bot = BybitGridBot(
        api_key=API_KEY,
        api_secret=API_SECRET,
        symbol=SYMBOL,
        grid_step_pct=0.002,
        grid_levels=5,
        qty_per_level=0.001,
        is_testnet=True
    )
    
    await bot.init_session()
    try:
        await bot.setup_grid()
        await bot.monitor_grid()
    finally:
        await bot.close_session()

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

pip install aiohttp

python grid_bot.py