liaozhaorun
e407225d29
- 添加 is_scheduled_reconnecting 标志位协调重连逻辑 - 增强定时重连任务的日志前缀便于追踪 - 改进异常处理和资源清理日志 - 优化代码格式和注释
673 lines
28 KiB
Python
673 lines
28 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
|
||
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"[AutoReconnectScheduler] 执行定时重连任务，时间: {self.reconnect_time}")
|
||
|
||
# 设置重连中标志位，通知主循环暂停健康检查重连
|
||
self.manager.is_scheduled_reconnecting = True
|
||
|
||
try:
|
||
# 强制断开所有终端连接（无论当前是否在线）
|
||
self.logger.info("[AutoReconnectScheduler] 正在断开所有终端连接...")
|
||
for unit in self.manager.units.values():
|
||
try:
|
||
if unit.xt_trader:
|
||
unit.cleanup()
|
||
self.logger.info(f"[AutoReconnectScheduler] 已断开终端 {unit.alias} 的连接")
|
||
except Exception as e:
|
||
self.logger.warning(f"[AutoReconnectScheduler] 断开终端 {unit.alias} 失败: {e}")
|
||
|
||
# 等待几秒后重新连接（固定等待时间）
|
||
self.logger.info("[AutoReconnectScheduler] 等待 3 秒后重新连接...")
|
||
time.sleep(3)
|
||
|
||
# 重新连接所有终端
|
||
self.logger.info("[AutoReconnectScheduler] 正在重新连接所有终端...")
|
||
success_count = 0
|
||
for unit in self.manager.units.values():
|
||
if unit.connect():
|
||
success_count += 1
|
||
self.logger.info(f"[AutoReconnectScheduler] 终端 {unit.alias} 重连成功")
|
||
else:
|
||
self.logger.warning(f"[AutoReconnectScheduler] 终端 {unit.alias} 重连失败")
|
||
|
||
self.logger.info(f"[AutoReconnectScheduler] 定时重连完成: {success_count}/{len(self.manager.units)} 个终端重连成功")
|
||
finally:
|
||
# 确保无论成功与否都重置标志位
|
||
self.manager.is_scheduled_reconnecting = False
|
||
|
||
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 Exception as e:
|
||
logging.getLogger("QMT_Engine").warning(f"销毁终端 {self.alias} 资源时出现异常: {e}")
|
||
|
||
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.is_scheduled_reconnecting = False # 定时重连调度器正在执行标志
|
||
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'):
|
||
# 检查是否正在执行定时重连调度
|
||
if self.is_scheduled_reconnecting:
|
||
self.logger.info(f"⏳ 定时重连调度器正在执行，跳过健康检查重连...")
|
||
else:
|
||
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() |