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 logging.basicConfig( level=logging.DEBUG, 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__) @dataclass class Order: order_id: str price: float quantity: float side: str status: str filled_time: Optional[float] = None class GridTradingBot: def __init__(self, conf: Dict): 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] = {} self.current_buy_levels = 0 self.current_sell_levels = 0 self.api_trade = mexc_spot_v3.mexc_trade() self.api_market = mexc_spot_v3.mexc_market() self.running = False def record_transaction(self, order_response: Dict[str, Any]) -> bool: 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 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) return True except Exception as e: return False def api_get_price(self) -> float: try: ticker = self.api_market.get_price({"symbol": self.symbol}) price = float(ticker["price"]) return price except Exception as e: return self.api_get_price() def api_get_balances(self) -> Tuple[float, float]: base_currency = self.symbol[:-4] quote_currency = self.symbol[-4:] base_available = 0.0 quote_available = 0.0 try: balances = self.api_trade.get_account_info() for balance in balances.get("balances", []): if balance["asset"] == base_currency: base_available = float(balance["free"]) elif balance["asset"] == quote_currency: quote_available = float(balance["free"]) return base_available, quote_available except Exception as e: return self.api_get_balances() def calculate_order_value(self, price: float) -> float: value = price * self.order_amount return value def is_order_value_valid(self, price: float) -> bool: value = self.calculate_order_value(price) is_valid = value >= self.min_order_value return is_valid def calculate_grid_price(self, base_price: float, level: int, side: str) -> float: if side == "BUY": price = base_price / (1 + self.grid_percentage) ** level return price elif side == "SELL": price = base_price * (1 + self.grid_percentage) ** level return price else: raise ValueError(f"Invalid side: {side}") def api_place_order(self, price: float, side: str) -> Optional[str]: if not self.is_order_value_valid(price): return None try: 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: return order_id else: return None except Exception as e: return self.api_place_order(price, side) def api_cancel_order(self, order_id: str): try: self.api_trade.delete_order({"symbol": self.symbol, "orderId": order_id}) except Exception as e: self.api_cancel_order(order_id) def api_cancel_all_orders(self): try: self.api_trade.delete_openorders({"symbol": self.symbol}) self.active_orders.clear() self.current_buy_levels = 0 self.current_sell_levels = 0 except Exception as e: self.api_cancel_all_orders() def api_update_order_statuses(self): for side in ["BUY", "SELL"]: orders = self.get_active_orders_by_side(side) if side == "BUY": orders.sort(key=lambda x: float(x.price), reverse=True) else: orders.sort(key=lambda x: float(x.price)) for order in orders: order_id = order.order_id try: 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 new_status == "FILLED": self.active_orders[order_id].filled_time = time.time() self.record_transaction(order_info) if new_status in ["CANCELED"]: self.active_orders.pop(order_id) if new_status == "NEW": break except Exception as 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] return orders def get_extreme_prices(self) -> Tuple[Optional[float], Optional[float]]: 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 ) return lowest_buy, highest_sell def initialize_grid(self): try: market_price = self.api_get_price() buy_price = self.calculate_grid_price(market_price, 1, "BUY") sell_price = self.calculate_grid_price(market_price, 1, "SELL") base_balance, quote_balance = self.api_get_balances() if quote_balance > self.reserve_quote: if self.is_order_value_valid(buy_price): 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 if base_balance > self.reserve_base: if self.is_order_value_valid(sell_price): 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 self.extend_grid() except Exception as e: self.initialize_grid() def extend_grid(self): 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" ) new_buy_price = self.calculate_grid_price(lowest_buy, 1, "BUY") if ( quote_balance - self.order_amount * new_buy_price > self.reserve_quote ): if 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", ) self.current_buy_levels += 1 else: break else: 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" ) new_sell_price = self.calculate_grid_price(highest_sell, 1, "SELL") if base_balance > self.reserve_base: if 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", ) self.current_sell_levels += 1 else: break else: break except Exception as e: self.extend_grid() def adjust_grid_for_filled(self): try: current_order_ids = set(self.active_orders.keys()) 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: 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) 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: 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) 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", ) elif filled_side == "SELL" and active_buys: 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) 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", ) except Exception as 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" ): 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" ): self.api_cancel_all_orders() self.initialize_grid() except Exception as e: self.adjust_grid_for_violation() def run(self): self.running = True try: self.initialize_grid() while self.running: try: self.api_update_order_statuses() self.adjust_grid_for_filled() self.adjust_grid_for_violation() self.extend_grid() time.sleep(1) except Exception as e: time.sleep(3) except KeyboardInterrupt: pass finally: self.stop() def stop(self): self.running = False self.api_cancel_all_orders() if __name__ == "__main__": config = { "symbol": "BTCUSDC", "csv_symbol": "BTCUSDT", "csv_file": "output/mexc_spot_grid_trades.csv", "grid_percentage": 0.001, "grid_count": 3, "order_amount": 0.00001, "min_order_value": 1, "reserve_base": 0, "reserve_quote": 0, } bot = GridTradingBot(config) bot.run()