# coding:utf-8 import time, datetime, traceback, sys, json, os import redis from xtquant import xtdata from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback from xtquant.xttype import StockAccount from xtquant import xtconstant # 全局变量占位 (稍后在 main 中初始化) CONFIG = {} ORDER_CACHE = {} # ================= 配置加载模块 ================= def load_config(config_file='config.json'): """ 读取同级目录下的配置文件 """ # 获取脚本所在目录 if getattr(sys, 'frozen', False): # 如果被打包为exe base_path = os.path.dirname(sys.executable) else: # 普通脚本运行 base_path = os.path.dirname(os.path.abspath(__file__)) full_path = os.path.join(base_path, config_file) if not os.path.exists(full_path): # 尝试直接读取（兼容 QMT 内置 Python 的路径行为） if os.path.exists(config_file): full_path = config_file else: print(f"[错误] 找不到配置文件: {full_path}") sys.exit(1) try: with open(full_path, 'r', encoding='utf-8') as f: config = json.load(f) print(f"成功加载配置: {full_path}") return config except Exception as e: print(f"[错误] 配置文件格式错误: {e}") sys.exit(1) # ================= 业务逻辑类 (保持不变) ================= class PositionManager: 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) # print(f"[{strategy_name}] 乐观占位: {code}") 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) print(f"[{strategy_name}] 回滚释放槽位: {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, xt_trader, acc, pos_mgr, strategies): self.trader = xt_trader self.acc = acc self.pos_mgr = pos_mgr self.strategies = strategies self.has_settled = False def run_settlement(self): print("\n" + "="*40) print(f"开始收盘清算: {datetime.datetime.now()}") # 1. 撤单 orders = self.trader.query_stock_orders(self.acc, cancelable_only=True) if orders: for o in orders: self.trader.cancel_order_stock(self.acc, o.order_id) time.sleep(2) # 2. 获取实盘真实持仓 real_positions = self.trader.query_stock_positions(self.acc) real_pos_map = {} if real_positions: for p in real_positions: if p.volume > 0: real_pos_map[p.stock_code] = p.volume # 3. 校准 Redis for strategy in self.strategies: virtual_data = self.pos_mgr.get_all_virtual_positions(strategy) for code, v_vol_str in virtual_data.items(): if code not in real_pos_map: print(f" [修正] {strategy} 幽灵持仓 {code} -> 强制释放") self.pos_mgr.force_delete(strategy, code) print("清算完成") self.has_settled = True def reset_flag(self): self.has_settled = False class MyXtQuantTraderCallback(XtQuantTraderCallback): def __init__(self, pos_mgr): self.pos_mgr = pos_mgr def on_disconnected(self): print(">> 连接断开") def on_stock_trade(self, trade): try: order_id = trade.order_id stock_code = trade.stock_code traded_vol = trade.traded_volume cache_info = ORDER_CACHE.get(order_id) if not cache_info: return strategy_name, cached_code, action_type = cache_info print(f">>> [成交] {strategy_name} {stock_code} 成交量:{traded_vol}") if action_type == 'BUY': self.pos_mgr.update_actual_volume(strategy_name, stock_code, traded_vol) elif action_type == 'SELL': self.pos_mgr.update_actual_volume(strategy_name, stock_code, -traded_vol) except Exception: traceback.print_exc() def on_order_error(self, order_error): try: order_id = order_error.order_id print(f">>> [下单失败] ID:{order_id} Msg:{order_error.error_msg}") cache_info = ORDER_CACHE.get(order_id) if cache_info: strategy_name, stock_code, action_type = cache_info if action_type == 'BUY': self.pos_mgr.rollback_holding(strategy_name, stock_code) del ORDER_CACHE[order_id] except: pass def process_strategy_queue(strategy_name, r_client, xt_trader, acc, pos_manager): queue_key = f"{strategy_name}_real" msg_json = r_client.lpop(queue_key) if not msg_json: return try: data = json.loads(msg_json) # 校验 if data.get('is_backtest', False): return msg_ts = data.get('timestamp') if msg_ts: msg_date = datetime.datetime.strptime(msg_ts, '%Y-%m-%d %H:%M:%S').date() if msg_date != datetime.date.today(): return stock_code = data['stock_code'] action = data['action'] price = float(data['price']) target_total_slots = int(data.get('total_slots', 1)) if action == 'BUY': current_holding_count = pos_manager.get_holding_count(strategy_name) empty_slots = target_total_slots - current_holding_count if empty_slots <= 0: return asset = xt_trader.query_stock_asset(acc) if not asset: return target_amount = asset.cash / empty_slots if target_amount < 2000: return if price <= 0: price = 1.0 vol = int(target_amount / price / 100) * 100 if vol >= 100: order_id = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_BUY, vol, xtconstant.FIX_PRICE, price, strategy_name, 'PyBuy') if order_id != -1: print(f"[{strategy_name}] 买入 {stock_code} {vol}股 (1/{empty_slots})") ORDER_CACHE[order_id] = (strategy_name, stock_code, 'BUY') pos_manager.mark_holding(strategy_name, stock_code) elif action == 'SELL': virtual_vol = pos_manager.get_position(strategy_name, stock_code) if virtual_vol > 0: real_positions = xt_trader.query_stock_positions(acc) real_pos = next((p for p in real_positions if p.stock_code == stock_code), None) real_can_use = real_pos.can_use_volume if real_pos else 0 final_vol = min(virtual_vol, real_can_use) if final_vol > 0: order_id = xt_trader.order_stock(acc, stock_code, xtconstant.STOCK_SELL, final_vol, xtconstant.FIX_PRICE, price, strategy_name, 'PySell') if order_id != -1: print(f"[{strategy_name}] 卖出 {stock_code} {final_vol}股") ORDER_CACHE[order_id] = (strategy_name, stock_code, 'SELL') except Exception: traceback.print_exc() # ================= 主程序入口 ================= if __name__ == '__main__': print("正在启动...") # 1. 加载配置 CONFIG = load_config('config.json') # 从配置中提取参数 redis_cfg = CONFIG['redis'] qmt_cfg = CONFIG['qmt'] watch_list = CONFIG['strategies'] print(f"Redis目标: {redis_cfg['host']}:{redis_cfg['port']}") print(f"QMT路径: {qmt_cfg['path']}") print(f"监听策略: {watch_list}") # 2. 连接 Redis try: r = redis.Redis( host=redis_cfg['host'], port=redis_cfg['port'], password=redis_cfg['password'], db=redis_cfg['db'], decode_responses=True ) r.ping() print("Redis 连接成功") pos_manager = PositionManager(r) except Exception as e: print(f"[FATAL] Redis 连接失败: {e}") sys.exit(1) # 3. 连接 QMT try: session_id = int(time.time()) xt_trader = XtQuantTrader(qmt_cfg['path'], session_id) acc = StockAccount(qmt_cfg['account_id'], qmt_cfg['account_type']) callback = MyXtQuantTraderCallback(pos_manager) xt_trader.register_callback(callback) xt_trader.start() connect_res = xt_trader.connect() if connect_res == 0: print(f"QMT 连接成功: {qmt_cfg['account_id']}") xt_trader.subscribe(acc) else: print(f"[FATAL] QMT 连接失败，错误码: {connect_res}") sys.exit(1) except Exception as e: print(f"[FATAL] QMT 初始化异常: {e}") sys.exit(1) # 4. 初始化清算器 settler = DailySettlement(xt_trader, acc, pos_manager, watch_list) print("=== 系统就绪，开始监听 ===") try: while True: now = datetime.datetime.now() current_time_str = now.strftime('%H%M') # 交易时段 if '0900' <= current_time_str <= '1500': if settler.has_settled: settler.reset_flag() for strategy in watch_list: process_strategy_queue(strategy, r, xt_trader, acc, pos_manager) # 收盘清算时段 elif '1505' <= current_time_str <= '1510': if not settler.has_settled: settler.run_settlement() time.sleep(60) except KeyboardInterrupt: print("用户终止程序")