# 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 from dateutil.parser import parse as parse_time 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 kernel32.SetConsoleMode(kernel32.GetStdHandle(-10), 128) except: pass @dataclass class TerminalStatus: """终端实例状态封装""" qmt_id: str alias: str account_id: str is_connected: bool callback_connected: bool physical_connected: bool last_heartbeat: str # ================= 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): self.r.hsetnx(self._get_key(strategy_name), code, 0) def rollback_holding(self, strategy_name, code): 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): 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) class DailySettlement: """终端级别的日终对账""" def __init__(self, unit): self.unit = unit self.has_settled = False def run_settlement(self): trader = self.unit.xt_trader acc = self.unit.acc_obj if not trader: return real_positions = trader.query_stock_positions(acc) real_pos_map = {p.stock_code: p.volume for p in real_positions if p.volume > 0} if real_positions else {} manager = MultiEngineManager() strategies = manager.get_strategies_by_terminal(self.unit.qmt_id) for s_name in strategies: virtual = manager.pos_manager.get_all_virtual_positions(s_name) for code, v_str in virtual.items(): if code not in real_pos_map: manager.pos_manager.force_delete(s_name, code) elif int(v_str) == 0 and code in real_pos_map: manager.pos_manager.update_actual_volume(s_name, code, real_pos_map[code]) self.has_settled = True def reset_flag(self): self.has_settled = False # ================= 1.5 定时重连调度器 ================= class AutoReconnectScheduler: """每日定时自动重连调度器""" def __init__(self, manager, reconnect_time="22:00", config_file="config.json"): """ 初始化定时重连调度器。 参数： - manager: MultiEngineManager 实例 - reconnect_time: 重连时间（格式 "HH:MM"），默认 22:00 - config_file: 配置文件路径 """ self.manager = manager self.logger = logging.getLogger("QMT_Engine") self.reconnect_time = reconnect_time self.config_file = config_file self.scheduler_thread = None self.stop_event = threading.Event() self.enabled = True self._load_config() def _load_config(self): """从配置文件加载设置""" if os.path.exists(self.config_file): try: with open(self.config_file, 'r', encoding='utf-8') as f: config = json.load(f) if 'auto_reconnect' in config: self.reconnect_time = config['auto_reconnect'].get('reconnect_time', '22:00') self.enabled = config['auto_reconnect'].get('enabled', True) self.logger.info(f"加载自动重连配置: 时间={self.reconnect_time}, 启用={self.enabled}") except Exception as e: self.logger.warning(f"加载自动重连配置失败，使用默认值: {e}") def _save_config(self): """保存配置到文件""" config = {} if os.path.exists(self.config_file): try: with open(self.config_file, 'r', encoding='utf-8') as f: config = json.load(f) except: pass if 'auto_reconnect' not in config: config['auto_reconnect'] = {} config['auto_reconnect']['reconnect_time'] = self.reconnect_time config['auto_reconnect']['enabled'] = self.enabled try: with open(self.config_file, 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) self.logger.info(f"自动重连配置已保存: 时间={self.reconnect_time}, 启用={self.enabled}") except Exception as e: self.logger.error(f"保存自动重连配置失败: {e}") def _calculate_next_run_time(self): """计算下一次执行时间""" now = datetime.datetime.now() try: target_time = datetime.datetime.strptime(self.reconnect_time, "%H:%M").time() next_run = datetime.datetime.combine(now.date(), target_time) # 如果今天的时间已过，则安排到明天 if next_run <= now: next_run += datetime.timedelta(days=1) return next_run except ValueError as e: self.logger.error(f"时间格式错误 {self.reconnect_time}: {e}") # 默认返回明天的 22:00 next_run = datetime.datetime.combine(now.date() + datetime.timedelta(days=1), datetime.time(22, 0)) return next_run def _scheduler_loop(self): """调度器主循环""" while not self.stop_event.is_set(): try: if self.enabled: next_run = self._calculate_next_run_time() delay = (next_run - datetime.datetime.now()).total_seconds() if delay > 0: # 等待直到下一个执行时间，每分钟检查一次停止事件 wait_interval = 60 # 每分钟检查一次 waited = 0 while waited < delay and not self.stop_event.is_set(): sleep_time = min(wait_interval, delay - waited) time.sleep(sleep_time) waited += sleep_time if not self.stop_event.is_set() and self.enabled: self._scheduled_reconnect() else: # 如果禁用，每分钟检查一次 time.sleep(60) except Exception as e: self.logger.error(f"调度器异常: {e}") time.sleep(60) def _scheduled_reconnect(self): """执行定时重连任务""" self.logger.info(f"执行定时重连任务，时间: {self.reconnect_time}") # 1. 检测当前连接状态 statuses = self.manager.get_all_status() connected_count = sum(1 for s in statuses if s.is_connected) self.logger.info(f"当前连接状态: {connected_count}/{len(statuses)} 个终端在线") # 2. 如果有连接，先断开 if connected_count > 0: self.logger.info("正在断开所有终端连接...") for unit in self.manager.units.values(): try: if unit.xt_trader: unit.cleanup() except Exception as e: self.logger.warning(f"断开终端 {unit.alias} 失败: {e}") # 3. 等待几秒后重新连接 self.logger.info("等待 3 秒后重新连接...") time.sleep(3) # 4. 重新连接所有终端 self.logger.info("正在重新连接所有终端...") success_count = 0 for unit in self.manager.units.values(): if unit.connect(): success_count += 1 self.logger.info(f"终端 {unit.alias} 重连成功") else: self.logger.warning(f"终端 {unit.alias} 重连失败") self.logger.info(f"定时重连完成: {success_count}/{len(self.manager.units)} 个终端重连成功") def start(self): """启动定时任务""" if self.scheduler_thread and self.scheduler_thread.is_alive(): self.logger.warning("调度器已在运行中") return self.stop_event.clear() self.scheduler_thread = threading.Thread(target=self._scheduler_loop, name="AutoReconnectScheduler", daemon=True) self.scheduler_thread.start() self.logger.info(f"自动重连调度器已启动，重连时间: {self.reconnect_time}, 启用状态: {self.enabled}") def stop(self): """停止定时任务""" self.logger.info("正在停止自动重连调度器...") self.stop_event.set() if self.scheduler_thread and self.scheduler_thread.is_alive(): self.scheduler_thread.join(timeout=5) self.logger.info("自动重连调度器已停止") def set_reconnect_time(self, time_str): """ 设置重连时间。 参数： - time_str: 时间字符串，格式 "HH:MM" """ try: # 验证时间格式 datetime.datetime.strptime(time_str, "%H:%M") old_time = self.reconnect_time self.reconnect_time = time_str self._save_config() self.logger.info(f"重连时间已从 {old_time} 修改为 {time_str}") return True except ValueError as e: self.logger.error(f"时间格式错误 {time_str}: {e}") return False def set_enabled(self, enabled): """设置是否启用自动重连""" self.enabled = enabled self._save_config() self.logger.info(f"自动重连已{'启用' if enabled else '禁用'}") def get_config(self): """获取当前配置""" return { "reconnect_time": self.reconnect_time, "enabled": self.enabled } def trigger_reconnect(self): """手动触发重连（立即执行）""" self.logger.info("手动触发重连任务") threading.Thread(target=self._scheduled_reconnect, name="ManualReconnect", daemon=True).start() # ================= 2. 执行单元 (TradingUnit) ================= class UnitCallback(XtQuantTraderCallback): def __init__(self, unit): self.unit = unit self.is_connected = False def on_disconnected(self): logging.getLogger("QMT_Engine").warning(f"终端 {self.unit.alias}({self.unit.qmt_id}) 物理连接断开") self.is_connected = False def on_stock_trade(self, trade): try: cache_info = self.unit.order_cache.get(trade.order_id) if not cache_info: return s_name, _, action = cache_info manager = MultiEngineManager() if action == 'BUY': manager.pos_manager.update_actual_volume(s_name, trade.stock_code, trade.traded_volume) elif action == 'SELL': manager.pos_manager.update_actual_volume(s_name, trade.stock_code, -trade.traded_volume) except: logging.getLogger("QMT_Engine").error(traceback.format_exc()) def on_order_error(self, err): cache = self.unit.order_cache.get(err.order_id) if cache and cache[2] == 'BUY': MultiEngineManager().pos_manager.rollback_holding(cache[0], cache[1]) self.unit.order_cache.pop(err.order_id, None) class TradingUnit: """终端实例执行单元，负责管理单个 QMT 进程""" def __init__(self, t_cfg): self.qmt_id = t_cfg['qmt_id'] self.alias = t_cfg.get('alias', self.qmt_id) self.path = t_cfg['path'] self.account_id = t_cfg['account_id'] self.account_type = t_cfg['account_type'] self.xt_trader = None self.acc_obj = None self.callback = None self.settler = None self.order_cache = {} self.last_heartbeat = "N/A" def cleanup(self): """强制销毁资源，确保文件句柄释放""" if self.xt_trader: try: logging.getLogger("QMT_Engine").info(f"正在销毁终端 {self.alias} 的旧资源...") self.xt_trader.stop() self.xt_trader = None # 显式置空 self.callback = None time.sleep(1.5) # 给 C++ 引擎留出释放 down_queue 锁的时间 except: pass def connect(self): """连接 QMT 终端""" self.cleanup() # 启动前先执行清理 try: # 采用动态 Session ID 避免冲突 session_id = int(time.time()) + hash(self.qmt_id) % 1000 self.xt_trader = XtQuantTrader(self.path, session_id) self.acc_obj = StockAccount(self.account_id, self.account_type) self.callback = UnitCallback(self) self.xt_trader.register_callback(self.callback) self.xt_trader.start() res = self.xt_trader.connect() if res == 0: self.xt_trader.subscribe(self.acc_obj) self.callback.is_connected = True self.settler = DailySettlement(self) logging.getLogger("QMT_Engine").info(f"终端 {self.alias} 连接成功 (SID: {session_id})") return True return False except Exception as e: logging.getLogger("QMT_Engine").error(f"终端 {self.alias} 连接异常: {repr(e)}") return False # ================= 3. 总控中心 (MultiEngineManager) ================= class MultiEngineManager: _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.units: Dict[str, TradingUnit] = {} self.config = {} self.is_running = True self.start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') self._initialized = True def initialize(self, config_file='config.json'): self._setup_logger() with open(config_file, 'r', encoding='utf-8') as f: self.config = json.load(f) self.r = redis.Redis(**self.config['redis'], decode_responses=True) self.pos_manager = PositionManager(self.r) for t_cfg in self.config.get('qmt_terminals', []): unit = TradingUnit(t_cfg) unit.connect() self.units[unit.qmt_id] = unit 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") logger = logging.getLogger("QMT_Engine") logger.setLevel(logging.INFO) # 确保日志流为 UTF-8 fmt = logging.Formatter('[%(asctime)s] [%(threadName)s] %(message)s', '%H:%M:%S') fh = logging.FileHandler(log_file, mode='a', encoding='utf-8') fh.setFormatter(fmt) sh = logging.StreamHandler(sys.stdout) sh.setFormatter(fmt) logger.addHandler(fh) logger.addHandler(sh) def get_strategies_by_terminal(self, qmt_id): return [s for s, cfg in self.config['strategies'].items() if cfg.get('qmt_id') == qmt_id] def run_trading_loop(self): self.logger = logging.getLogger("QMT_Engine") self.logger.info(">>> 多终端交易主循环线程已启动 <<<") last_check = 0 while self.is_running: try: now_t = time.time() curr_hms = datetime.datetime.now().strftime('%H%M%S') # --- 健康检查与自动修复 --- if now_t - last_check > 25: last_check = now_t for unit in self.units.values(): is_unit_alive = False if unit.xt_trader and unit.acc_obj: try: # 物理探测：通过查资产确认连接有效性 asset = unit.xt_trader.query_stock_asset(unit.acc_obj) if asset: is_unit_alive = True unit.last_heartbeat = datetime.datetime.now().strftime('%H:%M:%S') # 状态修正：物理通但逻辑False时自动拉回 if unit.callback and not unit.callback.is_connected: unit.callback.is_connected = True self.logger.info(f"✅ 修正终端 {unit.alias} 状态为在线") except Exception as e: self.logger.error(f"健康检查失败 - 终端 {unit.alias}: {str(e)}", exc_info=True) is_unit_alive = False # 断线重连策略 if not is_unit_alive: # 避让 QMT 夜间重启高峰 (21:32 - 21:50) if not ('213200' <= curr_hms <= '215000'): self.logger.warning(f"🚫 终端 {unit.alias} 物理连接丢失，执行重连...") unit.connect() else: self.logger.info(f"⏳ 处于 QMT 重启时段 ({curr_hms})，跳过重连操作...") # --- 交易逻辑处理 --- is_trading = ('091500' <= curr_hms <= '113030') or ('130000' <= curr_hms <= '150030') if is_trading: for s_name in self.config['strategies'].keys(): self.process_route(s_name) # --- 收盘结算与标志位重置 --- elif '150500' <= curr_hms <= '151500': for unit in self.units.values(): if unit.settler and not unit.settler.has_settled: unit.settler.run_settlement() elif '153000' <= curr_hms <= '160000': for unit in self.units.values(): if unit.settler: unit.settler.reset_flag() time.sleep(1 if is_trading else 5) except: self.logger.error("主循环异常：") self.logger.error(traceback.format_exc()) time.sleep(10) def process_route(self, strategy_name): strat_cfg = self.config['strategies'].get(strategy_name) unit = self.units.get(strat_cfg.get('qmt_id')) if not unit or not unit.callback or not unit.callback.is_connected: return msg_json = self.r.lpop(f"{strategy_name}_real") if not msg_json: return try: data = json.loads(msg_json) # 严格校验消息日期 if data.get('timestamp', '').split(' ')[0] != datetime.date.today().strftime('%Y-%m-%d'): return if data['action'] == 'BUY': self._execute_buy(unit, strategy_name, data) elif data['action'] == 'SELL': self._execute_sell(unit, strategy_name, data) except: pass def _execute_buy(self, unit, strategy_name, data): strat_cfg = self.config['strategies'][strategy_name] # 1. 槽位校验 if data['total_slots'] != strat_cfg['total_slots']: self.logger.error(f"[{strategy_name}] 信号槽位({data['total_slots']})与配置({strat_cfg['total_slots']})不符") return # 2. 持仓数检查 if self.pos_manager.get_holding_count(strategy_name) >= strat_cfg['total_slots']: return try: asset = unit.xt_trader.query_stock_asset(unit.acc_obj) # 计算该终端的总槽位之和 terminal_strategies = self.get_strategies_by_terminal(unit.qmt_id) # 计算加权槽位总和（支持策略权重配置） # 权重默认为 1，支持通过 weight 字段调整资金分配比例 # 示例：strategies = {"strategy_a": {"total_slots": 5, "weight": 1}, "strategy_b": {"total_slots": 5, "weight": 2}} total_weighted_slots = sum( self.config['strategies'][s].get('total_slots', 1) * self.config['strategies'][s].get('weight', 1) for s in terminal_strategies ) if not asset or total_weighted_slots <= 0: return # 获取当前策略的权重 weight = strat_cfg.get('weight', 1) # 4. 资金加权分配 (基于该终端总资产) total_equity = asset.cash + asset.market_value target_amt = total_equity * weight / total_weighted_slots actual_amt = min(target_amt, asset.cash * 0.98) # 预留手续费滑点 if actual_amt < 2000: self.logger.warning(f"[{strategy_name}] 单笔预算 {actual_amt:.2f} 不足 2000 元，取消买入") return # 4. 价格与股数 offset = strat_cfg.get('execution', {}).get('buy_price_offset', 0.0) price = round(float(data['price']) + offset, 3) vol = int(actual_amt / (price if price > 0 else 1.0) / 100) * 100 if vol < 100: return oid = unit.xt_trader.order_stock(unit.acc_obj, data['stock_code'], xtconstant.STOCK_BUY, vol, xtconstant.FIX_PRICE, price, strategy_name, 'PyBuy') if oid != -1: unit.order_cache[oid] = (strategy_name, data['stock_code'], 'BUY') self.pos_manager.mark_holding(strategy_name, data['stock_code']) self.logger.info(f"√√√ [{unit.alias}] {strategy_name} 下单买入: {data['stock_code']} {vol}股 @ {price}") except: self.logger.error(traceback.format_exc()) def _execute_sell(self, unit, strategy_name, data): v_vol = self.pos_manager.get_position(strategy_name, data['stock_code']) if v_vol <= 0: return real_pos = unit.xt_trader.query_stock_positions(unit.acc_obj) rp = next((p for p in real_pos if p.stock_code == data['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}] 卖出拦截: {data['stock_code']} 实盘无可用持仓") return try: offset = self.config['strategies'][strategy_name].get('execution', {}).get('sell_price_offset', 0.0) price = round(float(data['price']) + offset, 3) oid = unit.xt_trader.order_stock(unit.acc_obj, data['stock_code'], xtconstant.STOCK_SELL, final_vol, xtconstant.FIX_PRICE, price, strategy_name, 'PySell') if oid != -1: unit.order_cache[oid] = (strategy_name, data['stock_code'], 'SELL') self.logger.info(f"√√√ [{unit.alias}] {strategy_name} 下单卖出: {data['stock_code']} {final_vol}股 @ {price}") except: self.logger.error(traceback.format_exc()) def verify_connection(self, timeout=5): """验证物理连接是否有效""" try: # 先检查回调状态 if not self.callback or not self.callback.is_connected: return False # 实际调用 API 进行物理探测 asset = self.xt_trader.query_stock_asset(self.acc_obj) if asset is not None: return True return False except Exception as e: logging.getLogger("QMT_Engine").warning(f"终端 {self.alias} 物理连接验证失败: {e}") return False def get_all_status(self) -> List[TerminalStatus]: """获取所有状态，包含物理连接验证""" statuses = [] for u in self.units.values(): callback_conn = u.callback.is_connected if u.callback else False # 物理探测：通过查询资产确认连接有效性（xtquant 无 verify_connection 方法） physical_conn = False if callback_conn and u.xt_trader and u.acc_obj: try: asset = u.xt_trader.query_stock_asset(u.acc_obj) physical_conn = asset is not None except: physical_conn = False is_connected = callback_conn and physical_conn statuses.append(TerminalStatus( qmt_id=u.qmt_id, alias=u.alias, account_id=u.account_id, is_connected=is_connected, callback_connected=callback_conn, physical_connected=physical_conn, last_heartbeat=u.last_heartbeat )) return statuses def get_logs(self, lines: int = 50) -> List[str]: """获取最近的系统日志 参数： lines: 返回的行数，默认50行 返回： 日志行列表 """ log_dir = "logs" log_file = os.path.join(log_dir, f"{datetime.date.today().strftime('%Y-%m-%d')}.log") if not os.path.exists(log_file): return [] try: with open(log_file, 'r', encoding='utf-8') as f: all_lines = f.readlines() # 返回最后指定行数 return all_lines[-lines:] if lines < len(all_lines) else all_lines except Exception as e: self.logger.error(f"读取日志文件失败: {e}") return [] def stop(self): self.is_running = False for u in self.units.values(): u.cleanup()