314 lines
9.2 KiB
Python
314 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
MEXC 交易机器人
|
|
|
|
功能：
|
|
1. 支持多种订单类型 (LIMIT, MARKET, LIMIT_MAKER, IMMEDIATE_OR_CANCEL, FILL_OR_KILL)
|
|
2. 证券代码映射功能
|
|
3. 完整的交易记录和日志
|
|
"""
|
|
|
|
import csv
|
|
import logging
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Optional, Tuple
|
|
|
|
import mexc_spot_v3
|
|
|
|
os.makedirs("output", exist_ok=True)
|
|
|
|
# 配置日志
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
handlers=[
|
|
logging.FileHandler("output/mexc_trading_bot.log", encoding="utf-8"),
|
|
logging.StreamHandler(),
|
|
],
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MexcSpotMarket:
|
|
"""MEXC 市场数据查询类"""
|
|
|
|
def __init__(self):
|
|
"""初始化市场数据查询接口"""
|
|
self.market = mexc_spot_v3.mexc_market()
|
|
|
|
def get_price(self, symbol: str) -> Optional[float]:
|
|
"""
|
|
获取指定交易对的当前价格
|
|
|
|
Args:
|
|
symbol: 交易对，如 "BTCUSDT"
|
|
|
|
Returns:
|
|
当前价格(浮点数)或None(如果失败)
|
|
"""
|
|
params = {"symbol": symbol}
|
|
|
|
try:
|
|
logger.info("查询交易对价格: %s", symbol)
|
|
price_data = self.market.get_price(params)
|
|
|
|
if not price_data or "price" not in price_data:
|
|
logger.error("获取价格数据失败: %s", price_data)
|
|
return None
|
|
|
|
price_str = price_data["price"]
|
|
price = float(price_str)
|
|
logger.info("获取价格成功: %s = %f", symbol, price)
|
|
return price
|
|
|
|
except Exception as e: # pylint: disable=W0703
|
|
logger.error("查询价格失败: %s", str(e))
|
|
return None
|
|
|
|
|
|
class MexcSpotTrade:
|
|
"""MEXC 现货交易类"""
|
|
|
|
# 证券代码映射 (API代码: CSV记录代码)
|
|
SYMBOL_MAPPING = {
|
|
"BTCUSDC": "BTCUSDT",
|
|
# 可以在此添加更多映射
|
|
}
|
|
|
|
# 订单类型与必需参数
|
|
ORDER_TYPE_REQUIREMENTS = {
|
|
"LIMIT": ["quantity", "price"],
|
|
"MARKET": ["quantity", "quoteOrderQty"], # 任选其一
|
|
"LIMIT_MAKER": ["quantity", "price"],
|
|
"IMMEDIATE_OR_CANCEL": ["quantity", "price"],
|
|
"FILL_OR_KILL": ["quantity", "price"],
|
|
}
|
|
|
|
def __init__(self):
|
|
"""初始化交易机器人"""
|
|
self.trader = mexc_spot_v3.mexc_trade()
|
|
self.csv_file = "output/mexc_spot_trade.csv"
|
|
|
|
def _api_get_order(self, symbol: str, order_id: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
查询订单状态
|
|
|
|
Args:
|
|
symbol: 交易对
|
|
order_id: 订单ID
|
|
|
|
Returns:
|
|
订单详情字典或None(如果失败)
|
|
"""
|
|
params = {
|
|
"symbol": symbol,
|
|
"orderId": order_id,
|
|
}
|
|
|
|
try:
|
|
logger.info("查询订单状态, 订单ID: %s", order_id)
|
|
order = self.trader.get_order(params)
|
|
logger.info("订单状态: %s", order.get("status"))
|
|
return order
|
|
except Exception as e: # pylint: disable=W0703
|
|
logger.error("查询订单失败: %s", str(e))
|
|
return None
|
|
|
|
def _tool_map_symbol(self, symbol: str) -> str:
|
|
"""映射证券代码用于记录"""
|
|
return self.SYMBOL_MAPPING.get(symbol, symbol)
|
|
|
|
def _tool_validate_order_params(
|
|
self, order_type: str, params: Dict[str, Any]
|
|
) -> Tuple[bool, str]:
|
|
"""
|
|
验证订单参数是否符合要求
|
|
|
|
Args:
|
|
order_type: 订单类型
|
|
params: 订单参数
|
|
|
|
Returns:
|
|
元组(是否有效, 错误信息)
|
|
"""
|
|
required_params = self.ORDER_TYPE_REQUIREMENTS.get(order_type, [])
|
|
|
|
if not required_params:
|
|
return False, f"未知的订单类型: {order_type}"
|
|
|
|
# 特殊处理MARKET订单
|
|
if order_type == "MARKET":
|
|
if "quantity" not in params and "quoteOrderQty" not in params:
|
|
return False, "MARKET订单需要quantity或quoteOrderQty参数"
|
|
return True, ""
|
|
|
|
# 检查其他订单类型的必需参数
|
|
missing_params = [p for p in required_params if p not in params]
|
|
if missing_params:
|
|
return False, f"{order_type}订单缺少必需参数: {', '.join(missing_params)}"
|
|
|
|
return True, ""
|
|
|
|
def _tool_record_transaction(self, order_data: Dict[str, Any]) -> bool:
|
|
"""
|
|
记录交易到CSV文件
|
|
|
|
Args:
|
|
order_data: 订单数据字典
|
|
|
|
Returns:
|
|
是否成功记录
|
|
"""
|
|
try:
|
|
original_symbol = order_data["symbol"]
|
|
mapped_symbol = self._tool_map_symbol(original_symbol)
|
|
order_id = order_data["orderId"]
|
|
executed_qty = order_data["executedQty"]
|
|
cummulative_quote_qty = order_data["cummulativeQuoteQty"]
|
|
side = order_data["side"]
|
|
|
|
# 确定交易类型显示
|
|
trade_type = "买入" if side == "BUY" else "卖出"
|
|
|
|
timestamp = datetime.fromtimestamp(order_data["time"] / 1000).strftime(
|
|
"%Y-%m-%dT%H:%M"
|
|
)
|
|
|
|
row = [
|
|
timestamp,
|
|
trade_type,
|
|
mapped_symbol,
|
|
executed_qty,
|
|
cummulative_quote_qty,
|
|
"资金账户",
|
|
"FW#id",
|
|
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("交易记录成功, 订单ID: %s", order_id)
|
|
return True
|
|
except Exception as e: # pylint: disable=W0703
|
|
logger.error("记录交易失败: %s", str(e))
|
|
return False
|
|
|
|
def trade(
|
|
self, symbol: str, order_type: str, side: str, **kwargs
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
执行现货交易
|
|
|
|
Args:
|
|
symbol: 交易对 (如 BTCUSDC)
|
|
order_type: 订单类型 (LIMIT, MARKET等)
|
|
side: 交易方向 (BUY, SELL)
|
|
**kwargs: 其他订单参数
|
|
|
|
Returns:
|
|
订单信息字典或None(如果失败)
|
|
"""
|
|
# 基本参数验证
|
|
if side not in ["BUY", "SELL"]:
|
|
logger.error("无效的交易方向: %s", side)
|
|
return None
|
|
|
|
order_type = order_type.upper()
|
|
|
|
# 准备订单参数
|
|
base_params = {"symbol": symbol, "side": side, "type": order_type, **kwargs}
|
|
|
|
# 验证参数
|
|
is_valid, error_msg = self._tool_validate_order_params(order_type, base_params)
|
|
if not is_valid:
|
|
logger.error("订单参数验证失败: %s", error_msg)
|
|
return None
|
|
|
|
try:
|
|
logger.info("准备下单 %s %s 订单, 交易对: %s", side, order_type, symbol)
|
|
|
|
# 测试订单
|
|
test_result = self.trader.post_order_test(base_params.copy())
|
|
if test_result != {}:
|
|
logger.error("订单测试失败，参数有误: %s", test_result)
|
|
return None
|
|
|
|
logger.info("订单参数测试通过，准备正式下单")
|
|
|
|
# 正式下单
|
|
order = self.trader.post_order(base_params.copy())
|
|
order_id = order.get("orderId")
|
|
|
|
if not order_id:
|
|
logger.error("下单失败: 未获取到订单ID")
|
|
logger.error(base_params.copy())
|
|
logger.error(order)
|
|
return None
|
|
|
|
logger.info("订单创建成功, 订单ID: %s", order_id)
|
|
|
|
# 查询订单详情
|
|
order_detail = self._api_get_order(symbol, order_id)
|
|
if not order_detail:
|
|
logger.error("获取订单详情失败")
|
|
return None
|
|
|
|
# 记录交易
|
|
if order_detail.get("status") == "FILLED":
|
|
self._tool_record_transaction(order_detail)
|
|
|
|
return order_detail
|
|
|
|
except Exception as e: # pylint: disable=W0703
|
|
logger.error("交易执行失败: %s", str(e))
|
|
return None
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
spot_trader = MexcSpotTrade()
|
|
|
|
result = spot_trader.trade(
|
|
symbol="BTCUSDC",
|
|
order_type="MARKET",
|
|
side="BUY",
|
|
quoteOrderQty="1.5",
|
|
)
|
|
|
|
if result:
|
|
logger.info("交易执行成功")
|
|
logger.info("订单详情: %s", result)
|
|
else:
|
|
logger.error("交易执行失败")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|