import csv from datetime import datetime import time import logging from typing import Any, Dict, List, Optional, Tuple from dataclasses import dataclass import mexc_spot_v3 # 配置日志 CUSTOM_DEBUG_LEVEL = 15 logging.addLevelName(CUSTOM_DEBUG_LEVEL, "CUSTOM_DEBUG") class CustomLogger(logging.getLoggerClass()): def custom_debug(self, msg, *args, **kwargs): if self.isEnabledFor(CUSTOM_DEBUG_LEVEL): self._log(CUSTOM_DEBUG_LEVEL, msg, args, **kwargs) logging.setLoggerClass(CustomLogger) logging.basicConfig( level=CUSTOM_DEBUG_LEVEL, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler("output/mexc_spot_grid_bot.log", encoding="utf-8"), logging.StreamHandler(), ], ) logger = logging.getLogger(__name__) logger.custom_debug = lambda msg, *args, **kwargs: logger._log( # pylint: disable=W0212 CUSTOM_DEBUG_LEVEL, msg, args, **kwargs ) @dataclass class Order: order_id: str price: float quantity: float side: str # 'BUY' or 'SELL' status: str # 'NEW', 'FILLED', 'CANCELED' filled_time: Optional[float] = None # 订单成交时间戳 class GridTradingBot: def __init__(self, conf: Dict): """ 初始化网格交易机器人 参数: conf (Dict): 配置字典，包含: - symbol: 交易对 (如 'BTCUSDC') - csv_symbol: CSV中映射的交易对 (如 'BTCUSDT') - csv_file: CSV记录文件 (如 'output/mexc-spot-grid-trades.csv') - grid_percentage: 每格百分比 (如 0.005 表示 0.5%) - grid_count: 单边网格数量 (如 3) - order_amount: 每单加密货币数量 (如 0.00001 BTC) - min_order_value: 最小挂单价值 (如 1 USDC) - reserve_base: 保留的基础货币数量 (如 1 BTC) - reserve_quote: 保留的报价货币数量 (如 100 USDC) """ logger.custom_debug( "[GridTradingBot.__init__] Initializing GridTradingBot with config: %r", conf, ) self.config = conf self.symbol = conf["symbol"] self.csv_symbol = conf["csv_symbol"] self.csv_file = conf["csv_file"] self.grid_percentage = conf["grid_percentage"] self.grid_count = conf["grid_count"] self.order_amount = conf["order_amount"] self.min_order_value = conf["min_order_value"] self.reserve_base = conf["reserve_base"] self.reserve_quote = conf["reserve_quote"] # 当前活跃的订单 self.active_orders: Dict[str, Order] = {} # order_id -> Order # 当前网格范围 self.current_buy_levels = 0 self.current_sell_levels = 0 # 初始化API logger.custom_debug("[GridTradingBot.__init__] Initializing MEXC API clients") self.api_trade = mexc_spot_v3.mexc_trade() self.api_market = mexc_spot_v3.mexc_market() # 运行标志 self.running = False logger.info("[GridTradingBot.__init__] GridTradingBot initialized successfully") def record_transaction(self, order_response: Dict[str, Any]) -> bool: """ 记录交易到CSV文件 Args: order_data: 订单数据字典 Returns: 是否成功记录 """ try: csv_symbol = self.csv_symbol order_id = order_response["orderId"] executed_qty = order_response["executedQty"] cummulative_quote_qty = order_response["cummulativeQuoteQty"] side = order_response["side"] trade_type = "买入" if side == "BUY" else "卖出" timestamp = datetime.fromtimestamp( order_response["updateTime"] / 1000 ).strftime("%Y-%m-%dT%H:%M") row = [ timestamp, trade_type, csv_symbol, executed_qty, cummulative_quote_qty, "资金账户", "CEX", f"MEXC API - Order ID: {order_id}", ] # 检查文件是否存在 file_exists = False try: with open(self.csv_file, "r", encoding="utf-8") as f: file_exists = True except FileNotFoundError: pass # 写入CSV with open(self.csv_file, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) if not file_exists: writer.writerow( [ "日期", "类型", "证券代码", "份额", "净额", "现金账户", "目标账户", "备注", ] ) writer.writerow(row) logger.info( "[GridTradingBot.record_transaction] Transaction recorded, order ID: %s", order_id, ) return True except Exception as e: logger.error( "[GridTradingBot.record_transaction] Transaction recording failed: %s", str(e), ) return False def api_get_price(self) -> float: """获取当前市场价格""" logger.custom_debug( "[GridTradingBot.api_get_price] Fetching current market price for symbol: %s", self.symbol, ) try: ticker = self.api_market.get_price({"symbol": self.symbol}) price = float(ticker["price"]) logger.custom_debug( "[GridTradingBot.api_get_price] Current market price: %f", price ) return price except Exception as e: logger.error( "[GridTradingBot.api_get_price] Failed to get price for %s: %s", self.symbol, str(e), ) return self.api_get_price() def api_get_balances(self) -> Tuple[float, float]: """获取基础货币和报价货币的可用余额（不包括冻结金额）""" base_currency = self.symbol[:-4] # 假设报价货币是4个字母，如USDC quote_currency = self.symbol[-4:] base_available = 0.0 quote_available = 0.0 logger.custom_debug( "[GridTradingBot.api_get_balances] Fetching balances for %s and %s", base_currency, quote_currency, ) try: balances = self.api_trade.get_account_info() for balance in balances.get("balances", []): if balance["asset"] == base_currency: base_available = float(balance["free"]) logger.custom_debug( "[GridTradingBot.api_get_balances] Base currency (%s) available: %f", base_currency, base_available, ) elif balance["asset"] == quote_currency: quote_available = float(balance["free"]) logger.custom_debug( "[GridTradingBot.api_get_balances] Quote currency (%s) available: %f", quote_currency, quote_available, ) logger.custom_debug( "[GridTradingBot.api_get_balances] Final balances - base: %f, quote: %f", base_available, quote_available, ) return base_available, quote_available except Exception as e: logger.error( "[GridTradingBot.api_get_balances] Failed to get balances: %s", str(e) ) return self.api_get_balances() def calculate_order_value(self, price: float) -> float: """计算订单价值 (价格 * 数量)""" value = price * self.order_amount logger.custom_debug( "[GridTradingBot.calculate_order_value] Calculated order value: price=%f * amount=%f = %f", price, self.order_amount, value, ) return value def is_order_value_valid(self, price: float) -> bool: """检查订单价值是否满足最小挂单要求""" value = self.calculate_order_value(price) is_valid = value >= self.min_order_value logger.custom_debug( "[GridTradingBot.is_order_value_valid] Order value validation: %f >= %f? %s", value, self.min_order_value, is_valid, ) return is_valid def calculate_grid_price(self, base_price: float, level: int, side: str) -> float: """ 计算网格价格 参数: base_price: 基础价格 level: 网格级别 (正数) side: 'BUY' 或 'SELL' 返回: 计算后的价格 """ logger.custom_debug( "[GridTradingBot.calculate_grid_price] Calculating grid price: base=%f, level=%d, side=%s", base_price, level, side, ) if side == "BUY": price = base_price / (1 + self.grid_percentage) ** level logger.custom_debug( "[GridTradingBot.calculate_grid_price] Calculated BUY price: %f / (1 + %f)^%d = %f", base_price, self.grid_percentage, level, price, ) return price elif side == "SELL": price = base_price * (1 + self.grid_percentage) ** level logger.custom_debug( "[GridTradingBot.calculate_grid_price] Calculated SELL price: %f * (1 + %f)^%d = %f", base_price, self.grid_percentage, level, price, ) return price else: logger.error( "[GridTradingBot.calculate_grid_price] Invalid side for grid price calculation: %s", side, ) raise ValueError(f"Invalid side: {side}") def api_place_order(self, price: float, side: str) -> Optional[str]: """ 下订单 参数: price: 价格 side: 'BUY' 或 'SELL' 返回: 订单ID (如果下单成功) 或 None (如果失败) """ logger.custom_debug( "[GridTradingBot.api_place_order] Attempting to place %s order at price: %f", side, price, ) if not self.is_order_value_valid(price): logger.error( "[GridTradingBot.api_place_order] Order value too small: price=%f, amount=%f, value=%f < min=%f", price, self.order_amount, price * self.order_amount, self.min_order_value, ) return None try: # logger.custom_debug( # "[GridTradingBot.api_place_order] Sending order request to exchange" # ) order_detail = self.api_trade.post_order( { "symbol": self.symbol, "side": side, "type": "LIMIT", "price": price, "quantity": self.order_amount, } ) order_id = order_detail.get("orderId") if order_id: logger.info( "[GridTradingBot.api_place_order] Successfully placed %s order - price: %f, amount: %f, order_id: %s", side, price, self.order_amount, order_id, ) return order_id else: logger.error( "[GridTradingBot.api_place_order] Failed to place %s order at %f: %r", side, price, order_detail, ) except Exception as e: logger.error( "[GridTradingBot.api_place_order] Exception occurred while placing %s order at %f: %s", side, price, str(e), ) return self.api_place_order(price, side) def api_cancel_order(self, order_id: str): """取消订单""" logger.custom_debug( "[GridTradingBot.api_cancel_order] Attempting to cancel order: %s", order_id ) try: self.api_trade.delete_order({"symbol": self.symbol, "orderId": order_id}) logger.info( "[GridTradingBot.api_cancel_order] Successfully canceled order: %s", order_id, ) except Exception as e: logger.error( "[GridTradingBot.api_cancel_order] Failed to cancel order %s: %s", order_id, str(e), ) self.api_cancel_order(order_id) def api_cancel_all_orders(self): """取消所有活跃订单""" logger.custom_debug( "[GridTradingBot.api_cancel_all_orders] Attempting to cancel all open orders for symbol: %s", self.symbol, ) try: self.api_trade.delete_openorders({"symbol": self.symbol}) self.active_orders.clear() logger.info( "[GridTradingBot.api_cancel_all_orders] Successfully canceled all open orders for %s", self.symbol, ) self.current_buy_levels = 0 self.current_sell_levels = 0 except Exception as e: logger.error( "[GridTradingBot.api_cancel_all_orders] Failed to cancel all open orders: %s", str(e), ) self.api_cancel_all_orders() def api_update_order_statuses(self): """更新所有活跃订单的状态""" logger.custom_debug( "[GridTradingBot.api_update_order_statuses] Updating status for %d active orders", len(self.active_orders), ) for order_id in list(self.active_orders.keys()): try: logger.custom_debug( "[GridTradingBot.api_update_order_statuses] Checking status for order: %s", order_id, ) order_info = self.api_trade.get_order( {"symbol": self.symbol, "orderId": order_id} ) # 更新订单状态 old_status = self.active_orders[order_id].status new_status = order_info["status"] self.active_orders[order_id].status = new_status # 记录状态变化 if old_status != new_status: logger.info( "[GridTradingBot.api_update_order_statuses] Order %s status changed from %s to %s", order_id, old_status, new_status, ) # 如果订单已完成，记录成交时间 if new_status == "FILLED": self.active_orders[order_id].filled_time = time.time() logger.custom_debug( "[GridTradingBot.api_update_order_statuses] Order %s filled at %f", order_id, self.active_orders[order_id].filled_time, ) self.record_transaction(order_info) if new_status in ["CANCELED"]: logger.custom_debug( "[GridTradingBot.api_update_order_statuses] Removing order %s from active orders (status: %s)", order_id, new_status, ) self.active_orders.pop(order_id) except Exception as e: logger.error( "[GridTradingBot.api_update_order_statuses] Failed to get status for order %s: %s", order_id, str(e), ) self.api_update_order_statuses() def get_active_orders_by_side(self, side: str) -> List[Order]: """获取指定方向的所有活跃订单""" orders = [order for order in self.active_orders.values() if order.side == side] logger.custom_debug( "[GridTradingBot.get_active_orders_by_side] Found %d active %s orders", len(orders), side, ) return orders def get_extreme_prices(self) -> Tuple[Optional[float], Optional[float]]: """ 获取当前最极端的买单和卖单价格 返回: (最低买单价格, 最高卖单价格) 如果没有订单则为 (None, None) """ buy_orders = self.get_active_orders_by_side("BUY") sell_orders = self.get_active_orders_by_side("SELL") lowest_buy = min([order.price for order in buy_orders]) if buy_orders else None highest_sell = ( max([order.price for order in sell_orders]) if sell_orders else None ) logger.custom_debug( "[GridTradingBot.get_extreme_prices] Extreme prices - lowest buy: %s, highest sell: %s", lowest_buy, highest_sell, ) return lowest_buy, highest_sell def initialize_grid(self): """初始化网格""" logger.info( "[GridTradingBot.initialize_grid] Initializing grid for symbol: %s", self.symbol, ) try: market_price = self.api_get_price() logger.info( "[GridTradingBot.initialize_grid] Current market price: %f", market_price, ) # 初始下单: 买1和卖1 buy_price = self.calculate_grid_price(market_price, 1, "BUY") sell_price = self.calculate_grid_price(market_price, 1, "SELL") logger.custom_debug( "[GridTradingBot.initialize_grid] Initial buy price: %f, sell price: %f", buy_price, sell_price, ) # 检查余额并下单 base_balance, quote_balance = self.api_get_balances() logger.info( "[GridTradingBot.initialize_grid] Current balances - base: %f, quote: %f", base_balance, quote_balance, ) # 下单买1 if quote_balance > self.reserve_quote: logger.custom_debug( "[GridTradingBot.initialize_grid] Quote balance sufficient (%f > %f), checking order value", quote_balance, self.reserve_quote, ) if self.is_order_value_valid(buy_price): logger.custom_debug( "[GridTradingBot.initialize_grid] Order value valid, placing buy order" ) buy_order_id = self.api_place_order(buy_price, "BUY") if buy_order_id: self.active_orders[buy_order_id] = Order( order_id=buy_order_id, price=buy_price, quantity=self.order_amount, side="BUY", status="NEW", ) self.current_buy_levels = 1 logger.info( "[GridTradingBot.initialize_grid] Initial buy order placed at %f", buy_price, ) else: logger.error( "[GridTradingBot.initialize_grid] Initial buy order value too small: %f < %f", buy_price * self.order_amount, self.min_order_value, ) else: logger.error( "[GridTradingBot.initialize_grid] Insufficient quote balance for initial buy: %f <= %f", quote_balance, self.reserve_quote, ) # 下单卖1 if base_balance > self.reserve_base: logger.custom_debug( "[GridTradingBot.initialize_grid] Base balance sufficient (%f > %f), checking order value", base_balance, self.reserve_base, ) if self.is_order_value_valid(sell_price): logger.custom_debug( "[GridTradingBot.initialize_grid] Order value valid, placing sell order" ) sell_order_id = self.api_place_order(sell_price, "SELL") if sell_order_id: self.active_orders[sell_order_id] = Order( order_id=sell_order_id, price=sell_price, quantity=self.order_amount, side="SELL", status="NEW", ) self.current_sell_levels = 1 logger.info( "[GridTradingBot.initialize_grid] Initial sell order placed at %f", sell_price, ) else: logger.error( "[GridTradingBot.initialize_grid] Initial sell order value too small: %f < %f", sell_price * self.order_amount, self.min_order_value, ) else: logger.error( "[GridTradingBot.initialize_grid] Insufficient base balance for initial sell: %f <= %f", base_balance, self.reserve_base, ) self.extend_grid() except Exception as e: logger.error( "[GridTradingBot.initialize_grid] Failed to initialize grid: %s", str(e) ) self.initialize_grid() def extend_grid(self): """扩展网格，保证两侧都有指定个数的挂单""" logger.custom_debug( "[GridTradingBot.extend_grid] Extending grid - current buy levels: %d/%d, sell levels: %d/%d", self.current_buy_levels, self.grid_count, self.current_sell_levels, self.grid_count, ) try: # 如果两侧都没有挂单 (如: 插针) 则重启机器人 lowest_buy, highest_sell = self.get_extreme_prices() if lowest_buy is None and highest_sell is None: self.stop() self.run() # 扩展买单网格 (向下) while self.current_buy_levels < self.grid_count: lowest_buy, highest_sell = self.get_extreme_prices() base_balance, quote_balance = self.api_get_balances() if lowest_buy is None: lowest_buy = self.calculate_grid_price( highest_sell, self.grid_count, "BUY" ) logger.custom_debug( "[GridTradingBot.extend_grid] lowest_buy was None, calculating based on highest_sell" ) new_buy_price = self.calculate_grid_price(lowest_buy, 1, "BUY") logger.custom_debug( "[GridTradingBot.extend_grid] Extending buy grid - current lowest: %f, new price: %f", lowest_buy, new_buy_price, ) if quote_balance > self.reserve_quote: logger.custom_debug( "[GridTradingBot.extend_grid] Quote balance sufficient (%f > %f), checking order value", quote_balance, self.reserve_quote, ) if self.is_order_value_valid(new_buy_price): logger.custom_debug( "[GridTradingBot.extend_grid] Order value valid, placing extended buy order" ) buy_order_id = self.api_place_order(new_buy_price, "BUY") if buy_order_id: self.active_orders[buy_order_id] = Order( order_id=buy_order_id, price=new_buy_price, quantity=self.order_amount, side="BUY", status="NEW", ) self.current_buy_levels += 1 logger.info( "[GridTradingBot.extend_grid] Extended buy grid to level %d at price %f", self.current_buy_levels, new_buy_price, ) else: logger.error( "[GridTradingBot.extend_grid] Extended buy order value too small: %f < %f", new_buy_price * self.order_amount, self.min_order_value, ) break else: logger.error( "[GridTradingBot.extend_grid] Insufficient quote balance for extended buy: %f <= %f", quote_balance, self.reserve_quote, ) break # 扩展卖单网格 (向上) while self.current_sell_levels < self.grid_count: lowest_buy, highest_sell = self.get_extreme_prices() base_balance, quote_balance = self.api_get_balances() if highest_sell is None: highest_sell = self.calculate_grid_price( lowest_buy, self.grid_count, "SELL" ) logger.custom_debug( "[GridTradingBot.extend_grid] highest_sell was None, calculating based on lowest_buy" ) new_sell_price = self.calculate_grid_price(highest_sell, 1, "SELL") logger.custom_debug( "[GridTradingBot.extend_grid] Extending sell grid - current highest: %f, new price: %f", highest_sell, new_sell_price, ) if base_balance > self.reserve_base: logger.custom_debug( "[GridTradingBot.extend_grid] Base balance sufficient (%f > %f), checking order value", base_balance, self.reserve_base, ) if self.is_order_value_valid(new_sell_price): logger.custom_debug( "[GridTradingBot.extend_grid] Order value valid, placing extended sell order" ) sell_order_id = self.api_place_order(new_sell_price, "SELL") if sell_order_id: self.active_orders[sell_order_id] = Order( order_id=sell_order_id, price=new_sell_price, quantity=self.order_amount, side="SELL", status="NEW", ) self.current_sell_levels += 1 logger.info( "[GridTradingBot.extend_grid] Extended sell grid to level %d at price %f", self.current_sell_levels, new_sell_price, ) else: logger.error( "[GridTradingBot.extend_grid] Extended sell order value too small: %f < %f", new_sell_price * self.order_amount, self.min_order_value, ) break else: logger.error( "[GridTradingBot.extend_grid] Insufficient base balance for extended sell: %f <= %f", base_balance, self.reserve_base, ) break except Exception as e: logger.error( "[GridTradingBot.extend_grid] Failed to extend grid: %s", str(e) ) self.extend_grid() def adjust_grid_for_filled(self): """调整网格 - 优化后的对称逻辑: 买单成交时: 1. 取消最高价卖单 2. 在卖单侧挂一个更低价的卖单（卖0） 3. 在买单侧挂一个更低价的买单（买n） 卖单成交时: 1. 取消最低价买单 2. 在买单侧挂一个更高价的买单（买0） 3. 在卖单侧挂一个更高价的卖单（卖n）""" logger.custom_debug( "[GridTradingBot.adjust_grid] Adjusting grid with symmetric logic" ) try: # 获取当前所有订单的快照 current_order_ids = set(self.active_orders.keys()) # 找出新成交的订单（之前存在且现在状态为FILLED） newly_filled_orders = [ order for order in self.active_orders.values() if order.order_id in current_order_ids and order.status == "FILLED" ] if newly_filled_orders: logger.custom_debug( "[GridTradingBot.adjust_grid] Newly filled orders found" ) # 处理每个新成交的订单 for filled_order in newly_filled_orders: filled_side = filled_order.side filled_price = filled_order.price filled_id = filled_order.order_id self.active_orders.pop(filled_id, None) logger.info( "[GridTradingBot.adjust_grid] Processing newly filled %s order at price %f", filled_side, filled_price, ) # 获取当前活跃订单 active_buys = [ o for o in self.active_orders.values() if o.side == "BUY" and o.status == "NEW" ] active_sells = [ o for o in self.active_orders.values() if o.side == "SELL" and o.status == "NEW" ] if filled_side == "BUY": self.current_buy_levels -= 1 elif filled_side == "SELL": self.current_sell_levels -= 1 if filled_side == "BUY" and active_sells: # 买单成交时的处理逻辑 # 1. 取消最高价卖单 highest_sell = max(active_sells, key=lambda x: x.price) self.api_cancel_order(highest_sell.order_id) self.active_orders.pop(highest_sell.order_id, None) logger.info( "[GridTradingBot.adjust_grid] Cancelled highest SELL order at %f (order_id: %s)", highest_sell.price, highest_sell.order_id, ) # 2. 在卖单侧挂一个更低价的卖单（卖0） # 使用最近成交的买单价格作为基准 new_sell_price = self.calculate_grid_price( filled_price, 1, "SELL" ) # 检查余额和订单价值 base_balance, _ = self.api_get_balances() if ( base_balance > self.reserve_base and self.is_order_value_valid(new_sell_price) ): sell_order_id = self.api_place_order(new_sell_price, "SELL") if sell_order_id: self.active_orders[sell_order_id] = Order( order_id=sell_order_id, price=new_sell_price, quantity=self.order_amount, side="SELL", status="NEW", ) logger.info( "[GridTradingBot.adjust_grid] Placed new lower SELL order at %f (order_id: %s)", new_sell_price, sell_order_id, ) elif filled_side == "SELL" and active_buys: # 卖单成交时的处理逻辑 # 1. 取消最低价买单 lowest_buy = min(active_buys, key=lambda x: x.price) self.api_cancel_order(lowest_buy.order_id) self.active_orders.pop(lowest_buy.order_id, None) logger.info( "[GridTradingBot.adjust_grid] Cancelled lowest BUY order at %f (order_id: %s)", lowest_buy.price, lowest_buy.order_id, ) # 2. 在买单侧挂一个更高价的买单（买0） # 使用最近成交的卖单价格作为基准 new_buy_price = self.calculate_grid_price( filled_price, 1, "BUY" ) # 检查余额和订单价值 _, quote_balance = self.api_get_balances() if ( quote_balance > self.reserve_quote and self.is_order_value_valid(new_buy_price) ): buy_order_id = self.api_place_order(new_buy_price, "BUY") if buy_order_id: self.active_orders[buy_order_id] = Order( order_id=buy_order_id, price=new_buy_price, quantity=self.order_amount, side="BUY", status="NEW", ) logger.info( "[GridTradingBot.adjust_grid] Placed new higher BUY order at %f (order_id: %s)", new_buy_price, buy_order_id, ) except Exception as e: logger.error( "[GridTradingBot.adjust_grid] Failed to adjust grid: %s", str(e) ) self.adjust_grid_for_filled() def adjust_grid_for_violation(self): try: market_price = self.api_get_price() lowest_buy, highest_sell = self.get_extreme_prices() if lowest_buy is None and market_price < self.calculate_grid_price( highest_sell, self.grid_count + 1, "BUY" ): logger.warning( "[GridTradingBot.adjust_grid_for_violation] Price varied too low, restarting grid." ) self.api_cancel_all_orders() self.initialize_grid() elif highest_sell is None and market_price > self.calculate_grid_price( lowest_buy, self.grid_count + 1, "SELL" ): logger.warning( "[GridTradingBot.adjust_grid_for_violation] Price varied too high, restarting grid." ) self.api_cancel_all_orders() self.initialize_grid() except Exception as e: logger.error( "[GridTradingBot.adjust_grid_for_violation] Failed to adjust grid: %s", str(e), ) self.adjust_grid_for_violation() def run(self): """运行网格交易机器人""" self.running = True logger.info( "[GridTradingBot.run] Starting grid trading bot for symbol: %s", self.symbol ) try: # 初始化网格 self.initialize_grid() # 主循环 while self.running: try: logger.custom_debug( "[GridTradingBot.run] Starting main loop iteration" ) # 更新订单状态 self.api_update_order_statuses() # 调整网格 self.adjust_grid_for_filled() self.adjust_grid_for_violation() # 尝试扩展网格 self.extend_grid() # 等待一段时间再检查 sleep_time = 0 logger.custom_debug( "[GridTradingBot.run] Sleeping for %d seconds", sleep_time ) time.sleep(sleep_time) except Exception as e: logger.error("[GridTradingBot.run] Error in main loop: %s", str(e)) sleep_time = 1 logger.custom_debug( "[GridTradingBot.run] Error occurred, sleeping for %d seconds", sleep_time, ) time.sleep(sleep_time) # 出错后等待更长时间 except KeyboardInterrupt: logger.info("[GridTradingBot.run] Received keyboard interrupt, stopping...") finally: self.stop() def stop(self): """停止机器人""" logger.info("[GridTradingBot.stop] Stopping grid trading bot...") self.running = False self.api_cancel_all_orders() logger.info("[GridTradingBot.stop] Grid trading bot stopped successfully") # 示例配置和使用 if __name__ == "__main__": config = { "symbol": "BTCUSDC", # 交易对 "csv_symbol": "BTCUSDT", # CSV记录映射交易对 "csv_file": "output/mexc_spot_grid_trades.csv", # CSV记录文件 "grid_percentage": 0.001, # 等比网格的公比 "grid_count": 3, # 单侧的挂单数，实时平衡 "order_amount": 0.00001, # BTC数量 "min_order_value": 1, # 交易所限制订单价值至少 1 USDC "reserve_base": 0, # 保留 0 BTC 不参与网格 "reserve_quote": 0, # 保留 0 USDC 不参与网格 } logger.info("[__main__] Creating GridTradingBot instance with config: %r", config) bot = GridTradingBot(config) bot.run()