MexcTradingBot/mexc-spot-dca-bot
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

feat(main): load trades form json, imporve LIMIT LIMIT_MAKER order handling

Browse Source 
- Read trading instructions from JSON configuration file
- Support executing multiple trades by date (* means executing daily)
- Automatically fetch real-time price as the limit price when no price is provided
- For LIMIT, LIMIT_MAKER orders, support providing only quoteOrderQty to automatically calculate the quantity
This commit is contained in:
Zichao Lin
2025-07-15 20:38:52 +08:00
parent d918cc2d9f
commit 1c3ca56823
3 changed files with 451 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.pylintrc Normal file
View File

@@ -0,0 +1,3 @@
[MASTER]
disable=
broad-exception-caught

197
README.md Normal file
View File

@@ -0,0 +1,197 @@
# MEXC 交易机器人
## 项目简介
MEXC 交易机器人是一个自动化交易工具，用于在 MEXC 交易所执行现货交易。它通过 JSON 配置文件管理交易指令，支持多种订单类型，并自动记录交易日志。
## 主要功能
1. **配置文件管理**：从 JSON 文件读取交易指令
2. **灵活调度**：支持按日期执行交易（`*`表示每天执行）
3. **多种订单类型**：支持 LIMIT、MARKET、LIMIT_MAKER、IMMEDIATE_OR_CANCEL、FILL_OR_KILL 等订单类型
4. **智能价格获取**：当无 price 参数时自动获取实时价格作为限价
5. **自动计算数量**：LIMIT 订单支持只提供 quoteOrderQty 自动计算 quantity
6. **证券代码映射**：支持不同代码格式间的转换
7. **完整交易记录**：记录交易到 CSV 文件和日志
## 安装与使用
1. 创建配置文件`trading_config.json`
2. 运行程序：
```bash
python main.py
```
## 配置文件
### 配置文件结构
配置文件采用 JSON 格式，命名为`trading_config.json`，基本结构如下：
```json
{
"trades": [
{
"symbol": "BTCUSDT",
"order_type": "LIMIT",
"side": "BUY",
"execute_dates": ["*"],
"params": {
"quoteOrderQty": "100",
"price": "50000"
}
}
]
}
```
### 配置字段详解
#### 1. 交易列表 (`trades`)
- 类型：数组
- 描述：包含所有交易指令的列表
- 每个交易指令是一个包含以下字段的对象：
#### 2. 交易对 (`symbol`)
- 类型：字符串
- 必填：是
- 示例：`"BTCUSDT"`
- 描述：要交易的货币对，如 BTC/USDT
#### 3. 订单类型 (`order_type`)
- 类型：字符串
- 必填：是
- 可选值：
- `"LIMIT"`：限价单
- `"MARKET"`：市价单
- `"LIMIT_MAKER"`：限价做市单
- `"IMMEDIATE_OR_CANCEL"`：立即成交或取消
- `"FILL_OR_KILL"`：全部成交或取消
#### 4. 交易方向 (`side`)
- 类型：字符串
- 必填：是
- 可选值：
- `"BUY"`：买入
- `"SELL"`：卖出
#### 5. 执行日期 (`execute_dates`)
- 类型：数组
- 必填：是
- 特殊值：
- `"*"`：表示每天执行
- 示例：
- `["2023-01-01", "2023-01-15"]`：仅在指定日期执行
- `["*"]`：每天执行
- 格式：YYYY-MM-DD
#### 6. 订单参数 (`params`)
- 类型：对象
- 必填：是
- 内容根据订单类型不同而变化：
##### 参数
**必须包含 `quantity` 或 `quoteOrderQty` 之一**
- `quantity`：交易数量（字符串格式的数字）
- `quoteOrderQty`：交易金额（字符串格式的数字）
- `price`：限价价格（字符串格式的数字），如果为空则自动获取最新价
### 配置示例
#### 示例 1：每日限价买入
```json
{
"trades": [
{
"symbol": "BTCUSDT",
"order_type": "LIMIT",
"side": "BUY",
"execute_dates": ["*"],
"params": {
"quoteOrderQty": "100",
"price": "50000"
}
}
]
}
```
#### 示例 2：特定日期市价卖出
```json
{
"trades": [
{
"symbol": "ETHUSDT",
"order_type": "MARKET",
"side": "SELL",
"execute_dates": ["2023-12-25", "2023-12-31"],
"params": {
"quantity": "1.5"
}
}
]
}
```
#### 示例 3：多种订单组合
```json
{
"trades": [
{
"symbol": "BTCUSDT",
"order_type": "LIMIT",
"side": "BUY",
"execute_dates": ["2023-12-01"],
"params": {
"quantity": "0.01",
"price": "42000"
}
},
{
"symbol": "ETHUSDT",
"order_type": "MARKET",
"side": "SELL",
"execute_dates": ["*"],
"params": {
"quoteOrderQty": "200"
}
}
]
}
```
### 注意事项
1. 价格和数量都使用字符串格式而非数字，以避免浮点数精度问题
2. 对于 LIMIT 订单，可以只提供`quoteOrderQty`，系统会自动计算`quantity`
3. 当`execute_dates`包含`"*"`时，会忽略其他日期设置
4. 系统会自动创建`output`目录存放日志和交易记录
5. 所有时间均以系统时区为准
### 错误处理
如果配置文件格式错误，程序会记录错误并退出。常见错误包括：
- JSON 格式错误
- 缺少必填字段
- 日期格式不正确
- 订单参数不符合订单类型要求
## 输出文件
- `output/mexc_trading_bot.log`: 交易日志
- `output/mexc_spot_trade.csv`: 交易记录 CSV 文件
---

267
main.py
View File

@@ -5,16 +5,30 @@
MEXC 交易机器人
功能：
1. 支持多种订单类型 (LIMIT, MARKET, LIMIT_MAKER, IMMEDIATE_OR_CANCEL, FILL_OR_KILL)
2. 证券代码映射功能
3. 完整的交易记录和日志
1. 从JSON配置文件读取交易指令
2. 支持按日期执行多个交易(*表示每天执行)
3. 支持多种订单类型 (LIMIT, MARKET, LIMIT_MAKER, IMMEDIATE_OR_CANCEL, FILL_OR_KILL)
4. 当无price时自动获取实时价格作为限价
5. LIMIT订单支持只提供quoteOrderQty自动计算quantity
6. 证券代码映射功能
7. 完整的交易记录和日志
类说明：
- MexcSpotMarket: 处理市场数据查询
- MexcSpotTrade: 处理现货交易逻辑
- TradingConfig: 管理交易配置
使用示例：
python main.py
"""
import csv
import logging
import os
from datetime import datetime
from typing import Dict, Any, Optional, Tuple
import time
import json
from datetime import datetime, date
from typing import Dict, Any, Optional, Tuple, List
import mexc_spot_v3
@@ -33,7 +47,13 @@ logger = logging.getLogger(__name__)
class MexcSpotMarket:
"""MEXC 市场数据查询类"""
"""MEXC 市场数据查询类
提供获取交易对价格等功能
方法：
- get_price(symbol): 获取指定交易对的当前价格
"""
def __init__(self):
"""初始化市场数据查询接口"""
@@ -48,7 +68,11 @@ class MexcSpotMarket:
Returns:
当前价格(浮点数)或None(如果失败)
Raises:
Exception: 当API调用失败时抛出异常
"""
params = {"symbol": symbol}
try:
@@ -64,13 +88,27 @@ class MexcSpotMarket:
logger.info("获取价格成功: %s = %f", symbol, price)
return price
except Exception as e: # pylint: disable=W0703
except Exception as e:
logger.error("查询价格失败: %s", str(e))
return None
class MexcSpotTrade:
"""MEXC 现货交易类"""
"""MEXC 现货交易类
处理所有现货交易逻辑，包括订单创建、状态查询和记录
属性：
- SYMBOL_MAPPING: 证券代码映射字典
- ORDER_TYPE_REQUIREMENTS: 各订单类型必需参数
方法：
- trade(): 执行现货交易
- _api_get_order(): 查询订单状态
- _tool_map_symbol(): 映射证券代码
- _tool_validate_order_params(): 验证订单参数
- _tool_record_transaction(): 记录交易到CSV
"""
# 证券代码映射 (API代码: CSV记录代码)
SYMBOL_MAPPING = {
@@ -80,8 +118,12 @@ class MexcSpotTrade:
# 订单类型与必需参数
ORDER_TYPE_REQUIREMENTS = {
"LIMIT": ["quantity", "price"],
"MARKET": ["quantity", "quoteOrderQty"], # 任选其一
"LIMIT": [
"quantity",
"price",
"quoteOrderQty",
], # quantity和price或quoteOrderQty和price
"MARKET": ["quantity", "quoteOrderQty"], # quantity或quoteOrderQty
"LIMIT_MAKER": ["quantity", "price"],
"IMMEDIATE_OR_CANCEL": ["quantity", "price"],
"FILL_OR_KILL": ["quantity", "price"],
@@ -90,6 +132,7 @@ class MexcSpotTrade:
def __init__(self):
"""初始化交易机器人"""
self.trader = mexc_spot_v3.mexc_trade()
self.market = MexcSpotMarket()
self.csv_file = "output/mexc_spot_trade.csv"
def _api_get_order(self, symbol: str, order_id: str) -> Optional[Dict[str, Any]]:
@@ -103,6 +146,7 @@ class MexcSpotTrade:
Returns:
订单详情字典或None(如果失败)
"""
params = {
"symbol": symbol,
"orderId": order_id,
@@ -113,7 +157,7 @@ class MexcSpotTrade:
order = self.trader.get_order(params)
logger.info("订单状态: %s", order.get("status"))
return order
except Exception as e: # pylint: disable=W0703
except Exception as e:
logger.error("查询订单失败: %s", str(e))
return None
@@ -134,11 +178,23 @@ class MexcSpotTrade:
Returns:
元组(是否有效, 错误信息)
"""
required_params = self.ORDER_TYPE_REQUIREMENTS.get(order_type, [])
if not required_params:
return False, f"未知的订单类型: {order_type}"
# 特殊处理LIMIT订单
if order_type == "LIMIT":
# 需要price和(quantity或quoteOrderQty)
if "price" not in params:
return False, "LIMIT订单需要price参数"
if "quantity" not in params and "quoteOrderQty" not in params:
return False, "LIMIT订单需要quantity或quoteOrderQty参数"
return True, ""
# 特殊处理MARKET订单
if order_type == "MARKET":
if "quantity" not in params and "quoteOrderQty" not in params:
@@ -162,6 +218,7 @@ class MexcSpotTrade:
Returns:
是否成功记录
"""
try:
original_symbol = order_data["symbol"]
mapped_symbol = self._tool_map_symbol(original_symbol)
@@ -216,7 +273,7 @@ class MexcSpotTrade:
logger.info("交易记录成功, 订单ID: %s", order_id)
return True
except Exception as e: # pylint: disable=W0703
except Exception as e:
logger.error("记录交易失败: %s", str(e))
return False
@@ -234,16 +291,52 @@ class MexcSpotTrade:
Returns:
订单信息字典或None(如果失败)
Raises:
ValueError: 当参数验证失败时
Exception: 当API调用失败时
"""
# 基本参数验证
if side not in ["BUY", "SELL"]:
logger.error("无效的交易方向: %s", side)
return None
order_type = order_type.upper()
processed_kwargs = kwargs.copy()
# 处理无price的情况，获取实时价格
if order_type in ["LIMIT", "LIMIT_MAKER"] and "price" not in processed_kwargs:
current_price = self.market.get_price(symbol)
if current_price is None:
logger.error("无法获取实时价格，交易取消")
return None
processed_kwargs["price"] = current_price
logger.info("使用实时价格作为限价: %f", current_price)
# 处理LIMIT订单只有quoteOrderQty没有quantity的情况
if (
order_type in ["LIMIT", "LIMIT_MAKER"]
and "quoteOrderQty" in processed_kwargs
and "quantity" not in processed_kwargs
):
try:
quote_amount = float(processed_kwargs["quoteOrderQty"])
price = float(processed_kwargs["price"])
quantity = quote_amount / price
processed_kwargs["quantity"] = str(quantity)
logger.info("根据quoteOrderQty计算quantity: %f", quantity)
except (ValueError, KeyError) as e:
logger.error("计算quantity失败: %s", str(e))
return None
# 准备订单参数
base_params = {"symbol": symbol, "side": side, "type": order_type, **kwargs}
base_params = {
"symbol": symbol,
"side": side,
"type": order_type,
**processed_kwargs,
}
# 验证参数
is_valid, error_msg = self._tool_validate_order_params(order_type, base_params)
@@ -253,6 +346,7 @@ class MexcSpotTrade:
try:
logger.info("准备下单 %s %s 订单, 交易对: %s", side, order_type, symbol)
logger.debug("订单参数: %s", base_params)
# 测试订单
test_result = self.trader.post_order_test(base_params.copy())
@@ -275,6 +369,23 @@ class MexcSpotTrade:
logger.info("订单创建成功, 订单ID: %s", order_id)
# 查询订单详情
logger.info("等待1秒后查询订单状态...")
time.sleep(1)
order_detail = self._api_get_order(symbol, order_id)
if not order_detail:
logger.error("获取订单详情失败")
return None
# 如果不是FILLED则重复查询最多10次
retry_count = 0
while order_detail.get("status") != "FILLED" and retry_count < 10:
retry_count += 1
logger.info(
"订单未完成(状态: %s)，等待1秒后第%s次重试查询...",
order_detail.get("status", "UNKNOWN"),
retry_count,
)
time.sleep(1)
order_detail = self._api_get_order(symbol, order_id)
if not order_detail:
logger.error("获取订单详情失败")
@@ -282,32 +393,146 @@ class MexcSpotTrade:
# 记录交易
if order_detail.get("status") == "FILLED":
self._tool_record_transaction(order_detail)
if not self._tool_record_transaction(order_detail):
logger.error("交易记录失败")
else:
logger.warning(
"订单未完成(状态: %s)，未被记录到CSV。订单ID: %s",
order_detail.get("status", "UNKNOWN"),
order_id,
)
return order_detail
except Exception as e: # pylint: disable=W0703
except Exception as e:
logger.error("交易执行失败: %s", str(e))
return None
class TradingConfig:
"""交易配置管理类
负责加载和管理交易配置
方法：
- get_today_trades(): 获取今天需要执行的交易列表
- _load_config(): 加载JSON配置文件
"""
def __init__(self, config_file: str = "trading_config.json"):
"""
初始化交易配置
Args:
config_file: 配置文件路径
"""
self.config_file = config_file
self.config_data = self._load_config()
def _load_config(self) -> Dict[str, Any]:
"""加载JSON配置文件"""
try:
with open(self.config_file, "r", encoding="utf-8") as f:
config = json.load(f)
logger.info("成功加载配置文件: %s", self.config_file)
return config
except FileNotFoundError:
logger.error("配置文件不存在: %s", self.config_file)
return {}
except json.JSONDecodeError:
logger.error("配置文件格式错误，不是有效的JSON")
return {}
except Exception as e:
logger.error("加载配置文件时出错: %s", str(e))
return {}
def get_today_trades(self) -> List[Dict[str, Any]]:
"""
获取今天需要执行的交易列表
Returns:
今天需要执行的交易配置列表
Example:
>>> config = TradingConfig()
>>> today_trades = config.get_today_trades()
>>> print(len(today_trades))
"""
if not self.config_data or "trades" not in self.config_data:
return []
today = date.today().isoformat()
today_trades = []
for trade in self.config_data["trades"]:
# 检查交易是否包含执行日期
if "execute_dates" not in trade:
continue
# 检查是否设置了每天执行(*)
if "*" in trade["execute_dates"]:
today_trades.append(trade)
continue
# 检查今天是否在执行日期列表中
if today in trade["execute_dates"]:
today_trades.append(trade)
return today_trades
def main():
"""主函数"""
# 初始化交易配置
config = TradingConfig()
# 获取今天需要执行的交易
today_trades = config.get_today_trades()
if not today_trades:
logger.info("今天没有需要执行的交易")
return
logger.info("今天有 %d 个交易需要执行", len(today_trades))
# 初始化交易类
spot_trader = MexcSpotTrade()
# 执行每个交易
for trade_config in today_trades:
try:
# 提取交易参数
symbol = trade_config.get("symbol")
order_type = trade_config.get("order_type")
side = trade_config.get("side")
params = trade_config.get("params", {})
if not all([symbol, order_type, side]):
logger.error("交易配置缺少必要参数: %s", trade_config)
continue
logger.info("执行交易: %s %s %s", symbol, order_type, side)
logger.debug("交易参数: %s", params)
# 执行交易
result = spot_trader.trade(
symbol="BTCUSDC",
order_type="MARKET",
side="BUY",
quoteOrderQty="1.5",
symbol=symbol, order_type=order_type, side=side, **params
)
if result:
logger.info("交易执行成功")
logger.info("订单详情: %s", result)
logger.info("交易执行成功: %s", result)
else:
logger.error("交易执行失败")
except Exception as e:
logger.error("执行交易时出错: %s", str(e))
continue
logger.info("执行完毕")
if __name__ == "__main__":
main()