426 lines
17 KiB
Python
426 lines
17 KiB
Python
# coding:utf-8
|
||
import time
|
||
import datetime
|
||
import traceback
|
||
import sys
|
||
import json
|
||
import os
|
||
import threading
|
||
import logging
|
||
from typing import Optional, Dict, Any, List
|
||
from dataclasses import dataclass
|
||
|
||
import redis
|
||
from xtquant import xtdata
|
||
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
|
||
from xtquant.xttype import StockAccount
|
||
from xtquant import xtconstant
|
||
|
||
# ================= 0. Windows 控制台防卡死补丁 =================
|
||
try:
|
||
import ctypes
|
||
|
||
kernel32 = ctypes.windll.kernel32
|
||
# 禁用快速编辑模式 (0x0040)，防止鼠标点击终端导致程序挂起
|
||
kernel32.SetConsoleMode(kernel32.GetStdHandle(-10), 128)
|
||
except:
|
||
pass
|
||
|
||
|
||
@dataclass
|
||
class QMTStatus:
|
||
"""系统状态封装类"""
|
||
is_connected: bool
|
||
start_time: str
|
||
last_heartbeat: str
|
||
account_id: str
|
||
is_running: bool
|
||
|
||
|
||
# ================= 1. 虚拟持仓与对账辅助类 =================
|
||
|
||
class PositionManager:
|
||
"""Redis 持仓管理器：负责维护每个子策略的虚拟仓位"""
|
||
|
||
def __init__(self, r_client):
|
||
self.r = r_client
|
||
|
||
def _get_key(self, strategy_name):
|
||
return f"POS:{strategy_name}"
|
||
|
||
def mark_holding(self, strategy_name, code):
|
||
"""下单时先在 Redis 占位（0股），占用一个槽位"""
|
||
self.r.hsetnx(self._get_key(strategy_name), code, 0)
|
||
|
||
def rollback_holding(self, strategy_name, code):
|
||
"""报单失败时回滚，释放 Redis 占位"""
|
||
key = self._get_key(strategy_name)
|
||
val = self.r.hget(key, code)
|
||
if val is not None and int(val) == 0:
|
||
self.r.hdel(key, code)
|
||
|
||
def update_actual_volume(self, strategy_name, code, delta_vol):
|
||
"""成交回调时更新 Redis 实际股数"""
|
||
key = self._get_key(strategy_name)
|
||
new_vol = self.r.hincrby(key, code, int(delta_vol))
|
||
if new_vol <= 0:
|
||
self.r.hdel(key, code)
|
||
new_vol = 0
|
||
return new_vol
|
||
|
||
def get_position(self, strategy_name, code):
|
||
"""获取某个策略下某只股票的虚拟持仓"""
|
||
vol = self.r.hget(self._get_key(strategy_name), code)
|
||
return int(vol) if vol else 0
|
||
|
||
def get_holding_count(self, strategy_name):
|
||
"""获取当前策略已占用的槽位总数"""
|
||
return self.r.hlen(self._get_key(strategy_name))
|
||
|
||
def get_all_virtual_positions(self, strategy_name):
|
||
return self.r.hgetall(self._get_key(strategy_name))
|
||
|
||
def force_delete(self, strategy_name, code):
|
||
self.r.hdel(self._get_key(strategy_name), code)
|
||
|
||
def clean_stale_placeholders(self, strategy_name, xt_trader, acc):
|
||
"""清理长时间未成交且实盘无持仓的占位符"""
|
||
try:
|
||
key = self._get_key(strategy_name)
|
||
all_pos = self.r.hgetall(key)
|
||
if not all_pos: return
|
||
|
||
active_orders = xt_trader.query_stock_orders(acc, cancelable_only=True)
|
||
active_codes = [o.stock_code for o in active_orders] if active_orders else []
|
||
real_positions = xt_trader.query_stock_positions(acc)
|
||
real_holdings = [p.stock_code for p in real_positions if p.volume > 0] if real_positions else []
|
||
|
||
for code, vol_str in all_pos.items():
|
||
if int(vol_str) == 0:
|
||
if (code not in real_holdings) and (code not in active_codes):
|
||
self.r.hdel(key, code)
|
||
except Exception as e:
|
||
logging.getLogger("QMT_Engine").error(f"清理占位异常: {e}")
|
||
|
||
|
||
class DailySettlement:
|
||
"""收盘对账逻辑"""
|
||
|
||
def __init__(self, xt_trader, acc, pos_mgr, strategies_config):
|
||
self.trader = xt_trader
|
||
self.acc = acc
|
||
self.pos_mgr = pos_mgr
|
||
self.strategies_config = strategies_config
|
||
self.has_settled = False
|
||
|
||
def run_settlement(self):
|
||
"""收盘后强制同步 Redis 和实盘持仓"""
|
||
real_positions = self.trader.query_stock_positions(self.acc)
|
||
real_pos_map = {p.stock_code: p.volume for p in real_positions if p.volume > 0} if real_positions else {}
|
||
|
||
for strategy in self.strategies_config.keys():
|
||
virtual = self.pos_mgr.get_all_virtual_positions(strategy)
|
||
for code, v_str in virtual.items():
|
||
v = int(v_str)
|
||
if code not in real_pos_map:
|
||
self.pos_mgr.force_delete(strategy, code)
|
||
elif v == 0 and code in real_pos_map:
|
||
self.pos_mgr.update_actual_volume(strategy, code, real_pos_map[code])
|
||
self.has_settled = True
|
||
|
||
|
||
# ================= 2. QMT 核心引擎 =================
|
||
|
||
class MyXtQuantTraderCallback(XtQuantTraderCallback):
|
||
"""交易回调事件监听"""
|
||
|
||
def __init__(self, pos_mgr):
|
||
self.pos_mgr = pos_mgr
|
||
self.is_connected = False
|
||
self.logger = logging.getLogger("QMT_Engine")
|
||
|
||
def on_disconnected(self):
|
||
self.logger.warning(">> 回调通知: 交易端连接断开")
|
||
self.is_connected = False
|
||
|
||
def on_stock_trade(self, trade):
|
||
try:
|
||
# QMTEngine 是单例，可直接通过类访问
|
||
cache_info = QMTEngine().order_cache.get(trade.order_id)
|
||
if not cache_info: return
|
||
strategy, _, action = cache_info
|
||
self.logger.info(f">>> [成交] {strategy} | {trade.stock_code} | {trade.traded_volume}股")
|
||
if action == 'BUY':
|
||
self.pos_mgr.update_actual_volume(strategy, trade.stock_code, trade.traded_volume)
|
||
elif action == 'SELL':
|
||
self.pos_mgr.update_actual_volume(strategy, trade.stock_code, -trade.traded_volume)
|
||
except:
|
||
traceback.print_exc()
|
||
|
||
def on_order_error(self, err):
|
||
try:
|
||
self.logger.error(f"下单失败回调: {err.error_msg} ID:{err.order_id}")
|
||
cache = QMTEngine().order_cache.get(err.order_id)
|
||
if cache and cache[2] == 'BUY':
|
||
self.pos_mgr.rollback_holding(cache[0], cache[1])
|
||
if err.order_id in QMTEngine().order_cache:
|
||
del QMTEngine().order_cache[err.order_id]
|
||
except:
|
||
pass
|
||
|
||
|
||
class QMTEngine:
|
||
"""QMT 交易引擎单例"""
|
||
_instance = None
|
||
_lock = threading.Lock()
|
||
|
||
def __new__(cls):
|
||
if cls._instance is None:
|
||
with cls._lock:
|
||
if cls._instance is None:
|
||
cls._instance = super().__new__(cls)
|
||
return cls._instance
|
||
|
||
def __init__(self):
|
||
if hasattr(self, '_initialized'): return
|
||
self.logger = None
|
||
self.config = {}
|
||
self.xt_trader = None
|
||
self.acc = None
|
||
self.pos_manager = None
|
||
self.callback = None
|
||
self.is_running = True
|
||
self.start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
self.last_heartbeat = "Initializing..."
|
||
self.order_cache = {} # OrderID -> (Strategy, Code, Action)
|
||
self.settler = None
|
||
self._initialized = True
|
||
|
||
def initialize(self, config_file='config.json'):
|
||
self._setup_logger()
|
||
self.config = self._load_config(config_file)
|
||
# 初始化 Redis
|
||
try:
|
||
self.redis_client = redis.Redis(**self.config['redis'], decode_responses=True)
|
||
self.redis_client.ping()
|
||
self.pos_manager = PositionManager(self.redis_client)
|
||
self.logger.info("Redis 建立连接成功")
|
||
except Exception as e:
|
||
self.logger.critical(f"Redis 连接失败: {e}")
|
||
raise
|
||
self._reconnect_qmt()
|
||
|
||
def _setup_logger(self):
|
||
log_dir = "logs"
|
||
if not os.path.exists(log_dir): os.makedirs(log_dir)
|
||
log_file = os.path.join(log_dir, f"{datetime.date.today().strftime('%Y-%m-%d')}.log")
|
||
self.logger = logging.getLogger("QMT_Engine")
|
||
self.logger.setLevel(logging.INFO)
|
||
if self.logger.handlers:
|
||
for h in self.logger.handlers[:]: h.close(); self.logger.removeHandler(h)
|
||
fmt = logging.Formatter('[%(asctime)s] [%(levelname)s] [%(threadName)s] %(message)s', '%Y-%m-%d %H:%M:%S')
|
||
fh = logging.FileHandler(log_file, mode='a', encoding='utf-8')
|
||
fh.setFormatter(fmt)
|
||
sh = logging.StreamHandler(sys.stdout)
|
||
sh.setFormatter(fmt)
|
||
self.logger.addHandler(fh);
|
||
self.logger.addHandler(sh)
|
||
|
||
def _load_config(self, config_file):
|
||
base = os.path.dirname(sys.executable) if getattr(sys, 'frozen', False) else os.path.dirname(
|
||
os.path.abspath(__file__))
|
||
path = os.path.join(base, config_file)
|
||
with open(path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
|
||
def _get_global_total_slots(self):
|
||
"""本地引擎计算：所有策略总共分配了多少个槽位"""
|
||
return sum(info.get('total_slots', 0) for info in self.config.get('strategies', {}).values())
|
||
|
||
def _get_execution_setting(self, strategy_name, key, default=None):
|
||
"""扩展性配置读取：读取 execution 字典中的参数"""
|
||
strat_cfg = self.config.get('strategies', {}).get(strategy_name, {})
|
||
exec_cfg = strat_cfg.get('execution', {})
|
||
return exec_cfg.get(key, default)
|
||
|
||
def _reconnect_qmt(self):
|
||
q = self.config['qmt']
|
||
if self.xt_trader:
|
||
try:
|
||
self.xt_trader.stop()
|
||
except:
|
||
pass
|
||
self.xt_trader = XtQuantTrader(q['path'], int(time.time()))
|
||
self.acc = StockAccount(q['account_id'], q['account_type'])
|
||
self.callback = MyXtQuantTraderCallback(self.pos_manager)
|
||
self.xt_trader.register_callback(self.callback)
|
||
self.xt_trader.start()
|
||
if self.xt_trader.connect() == 0:
|
||
self.xt_trader.subscribe(self.acc)
|
||
self.callback.is_connected = True
|
||
self.settler = DailySettlement(self.xt_trader, self.acc, self.pos_manager, self.config['strategies'])
|
||
for s in self.config['strategies'].keys():
|
||
self.pos_manager.clean_stale_placeholders(s, self.xt_trader, self.acc)
|
||
self.logger.info("✅ QMT 终端连接成功")
|
||
return True
|
||
return False
|
||
|
||
def process_strategy_queue(self, strategy_name):
|
||
"""处理 Redis 中的策略信号"""
|
||
queue_key = f"{strategy_name}_real"
|
||
msg_json = self.redis_client.lpop(queue_key)
|
||
if not msg_json: return
|
||
try:
|
||
self.redis_client.rpush(f"{queue_key}:history", msg_json)
|
||
data = json.loads(msg_json)
|
||
if data.get('is_backtest'): return
|
||
today = datetime.date.today().strftime('%Y-%m-%d')
|
||
if data.get('timestamp', '').split(' ')[0] != today: return
|
||
|
||
action = data.get('action')
|
||
stock = data.get('stock_code')
|
||
price = float(data.get('price', 0))
|
||
msg_slots = int(data.get('total_slots', 0))
|
||
|
||
if action == 'BUY':
|
||
self._process_buy(strategy_name, stock, price, msg_slots)
|
||
elif action == 'SELL':
|
||
self._process_sell(strategy_name, stock, price)
|
||
except Exception as e:
|
||
self.logger.error(f"消息解析异常: {e}")
|
||
|
||
def _process_buy(self, strategy_name, stock_code, price, msg_slots):
|
||
"""核心开仓逻辑"""
|
||
# 1. 验证配置
|
||
strat_cfg = self.config.get('strategies', {}).get(strategy_name)
|
||
if not strat_cfg: return
|
||
local_slots = strat_cfg.get('total_slots', 0)
|
||
|
||
# 2. 安全校验：信号槽位与本地实盘配置必须严格一致
|
||
if msg_slots != local_slots:
|
||
self.logger.error(f"⚠️ [{strategy_name}] 槽位不匹配！拒绝下单。信号预期:{msg_slots} | 本地配置:{local_slots}")
|
||
return
|
||
|
||
# 3. 检查子策略占用
|
||
if self.pos_manager.get_holding_count(strategy_name) >= local_slots:
|
||
self.logger.warning(f"[{strategy_name}] 槽位已满，拦截买入 {stock_code}")
|
||
return
|
||
|
||
# 4. 资金计算（由本地引擎统筹全局）
|
||
try:
|
||
asset = self.xt_trader.query_stock_asset(self.acc)
|
||
global_total = self._get_global_total_slots()
|
||
if not asset or global_total <= 0: return
|
||
|
||
# 单笔预算 = (总资产现金 + 持仓市值) / 全局总槽位
|
||
total_equity = asset.cash + asset.market_value
|
||
target_amt = total_equity / global_total
|
||
# 实际可用金额不超过现金的 98%（预留滑点/手续费）
|
||
actual_amt = min(target_amt, asset.cash * 0.98)
|
||
|
||
if actual_amt < 2000:
|
||
self.logger.warning(f"[{strategy_name}] 可用金额不足2000，取消买入 {stock_code}")
|
||
return
|
||
|
||
# --- 价格偏移处理 ---
|
||
offset = self._get_execution_setting(strategy_name, 'buy_price_offset', 0.0)
|
||
final_price = round(price + offset, 3)
|
||
|
||
vol = int(actual_amt / (final_price if final_price > 0 else 1.0) / 100) * 100
|
||
if vol < 100: return
|
||
|
||
oid = self.xt_trader.order_stock(self.acc, stock_code, xtconstant.STOCK_BUY, vol, xtconstant.FIX_PRICE,
|
||
final_price, strategy_name, 'PyBuy')
|
||
if oid != -1:
|
||
self.logger.info(
|
||
f"√√√ [{strategy_name}] 开仓下单: {stock_code} | 价格:{final_price}(加价:{offset}) | 数量:{vol}")
|
||
self.order_cache[oid] = (strategy_name, stock_code, 'BUY')
|
||
self.pos_manager.mark_holding(strategy_name, stock_code)
|
||
else:
|
||
self.logger.error(f"XXX [{strategy_name}] 开仓发单拒绝")
|
||
except Exception as e:
|
||
self.logger.error(f"买入异常: {e}", exc_info=True)
|
||
|
||
def _process_sell(self, strategy_name, stock_code, price):
|
||
"""核心平仓逻辑"""
|
||
v_vol = self.pos_manager.get_position(strategy_name, stock_code)
|
||
if v_vol <= 0: return
|
||
|
||
real_pos = self.xt_trader.query_stock_positions(self.acc)
|
||
rp = next((p for p in real_pos if p.stock_code == stock_code), None) if real_pos else None
|
||
can_use = rp.can_use_volume if rp else 0
|
||
|
||
final_vol = min(v_vol, can_use)
|
||
if final_vol <= 0:
|
||
self.logger.warning(f"[{strategy_name}] {stock_code} 无可用平仓额度 (Redis:{v_vol}, 实盘:{can_use})")
|
||
return
|
||
|
||
# --- 价格偏移处理 ---
|
||
offset = self._get_execution_setting(strategy_name, 'sell_price_offset', 0.0)
|
||
final_price = round(price + offset, 3)
|
||
|
||
oid = self.xt_trader.order_stock(self.acc, stock_code, xtconstant.STOCK_SELL, final_vol, xtconstant.FIX_PRICE,
|
||
final_price, strategy_name, 'PySell')
|
||
if oid != -1:
|
||
self.logger.info(
|
||
f"√√√ [{strategy_name}] 平仓下单: {stock_code} | 价格:{final_price}(偏移:{offset}) | 数量:{final_vol}")
|
||
self.order_cache[oid] = (strategy_name, stock_code, 'SELL')
|
||
|
||
def run_trading_loop(self):
|
||
"""交易主线程循环"""
|
||
self.logger.info(">>> 交易主循环子线程已启动 <<<")
|
||
last_check = 0
|
||
while self.is_running:
|
||
try:
|
||
self.last_heartbeat = datetime.datetime.now().strftime('%H:%M:%S')
|
||
# 健康检查
|
||
if time.time() - last_check > 15:
|
||
last_check = time.time()
|
||
try:
|
||
if not (self.xt_trader and self.acc and self.xt_trader.query_stock_asset(self.acc)):
|
||
self._reconnect_qmt()
|
||
except:
|
||
self._reconnect_qmt()
|
||
|
||
# 交易时间判断
|
||
curr = datetime.datetime.now().strftime('%H%M%S')
|
||
is_trading = ('091500' <= curr <= '113000') or ('130000' <= curr <= '150000')
|
||
|
||
if is_trading and self.callback and self.callback.is_connected:
|
||
if self.settler: self.settler.reset_flag()
|
||
for s in self.config.get('strategies', {}).keys():
|
||
self.process_strategy_queue(s)
|
||
elif '150500' <= curr <= '151500' and self.settler and not self.settler.has_settled:
|
||
self.settler.run_settlement()
|
||
|
||
time.sleep(1 if is_trading else 5)
|
||
except Exception as e:
|
||
self.logger.error(f"主循环异常: {e}")
|
||
time.sleep(10)
|
||
|
||
# ================= 外部接口 =================
|
||
|
||
def get_status(self) -> QMTStatus:
|
||
conn = self.callback.is_connected if self.callback else False
|
||
return QMTStatus(conn, self.start_time, self.last_heartbeat,
|
||
self.acc.account_id if self.acc else "Unknown", self.is_running)
|
||
|
||
def get_positions(self) -> Dict[str, Any]:
|
||
real = []
|
||
if self.callback and self.callback.is_connected:
|
||
pos = self.xt_trader.query_stock_positions(self.acc)
|
||
if pos:
|
||
real = [{"code": p.stock_code, "volume": p.volume, "can_use": p.can_use_volume, "value": p.market_value}
|
||
for p in pos if p.volume > 0]
|
||
virtual = {s: self.pos_manager.get_all_virtual_positions(s) for s in self.config.get('strategies', {}).keys()}
|
||
return {"real_positions": real, "virtual_positions": virtual}
|
||
|
||
def get_logs(self, lines=50):
|
||
log_path = os.path.join("logs", f"{datetime.date.today().strftime('%Y-%m-%d')}.log")
|
||
if not os.path.exists(log_path): return ["今日暂无日志"]
|
||
with open(log_path, 'r', encoding='utf-8') as f:
|
||
return [l.strip() for l in f.readlines()[-lines:]]
|
||
|
||
def stop(self):
|
||
self.is_running = False
|
||
self.logger.info("收到引擎停止指令") |