607 lines
25 KiB
Python
607 lines
25 KiB
Python
"""
|
||
QMT 真实交易实现 - 封装 xtquant SDK
|
||
"""
|
||
import datetime
|
||
import os
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
import config
|
||
import core.eventbus as eBus
|
||
from core.logger import LogLevel, PrintLog
|
||
|
||
|
||
class RealQmtV:
|
||
"""
|
||
真实 QMT 交易器
|
||
封装 xtquant 的 XtQuantTrader，提供与模拟器一致的接口
|
||
"""
|
||
|
||
# miniQMT 进程名关键字（GUI 壳: XtMiniQmt.exe，交易引擎: miniquote.exe）
|
||
_QMT_PROCESS_KEYWORDS = ['Qmt', 'qmt', 'QMT', 'miniquote', 'MiniQuote']
|
||
|
||
@staticmethod
|
||
def _discover_qmt_port() -> int:
|
||
"""
|
||
自动探测 miniQMT 监听端口。
|
||
方法1: SDK 内部扫描 (读取配置)
|
||
方法2: netstat 找 LISTENING 端口 → 反查所属进程名 → 匹配 QMT 关键字
|
||
返回端口号，未找到返回 0。
|
||
"""
|
||
# ---- 方法1: SDK 内部扫描 ----
|
||
try:
|
||
from xtquant import xtconn
|
||
addrs = xtconn.scan_available_server_addr()
|
||
for addr in addrs:
|
||
try:
|
||
port = int(addr.split(':')[1])
|
||
if port:
|
||
PrintLog(LogLevel.INFO, f'[端口探测] SDK 扫描发现端口: {port}')
|
||
return port
|
||
except (ValueError, IndexError):
|
||
continue
|
||
except Exception as e:
|
||
PrintLog(LogLevel.DEBUG, f'[端口探测] SDK 扫描异常: {e}')
|
||
|
||
# ---- 方法2: netstat → 反向查进程名 ----
|
||
try:
|
||
# 2a. netstat 找出所有 LISTENING 端口的 PID
|
||
pid_ports = {} # pid -> [port, ...]
|
||
netstat = subprocess.run(
|
||
['netstat', '-ano'],
|
||
capture_output=True, text=True, timeout=10
|
||
)
|
||
for line in netstat.stdout.splitlines():
|
||
if 'LISTENING' not in line and 'LISTEN' not in line:
|
||
continue
|
||
parts = line.split()
|
||
if len(parts) < 5:
|
||
continue
|
||
try:
|
||
local_addr = parts[1]
|
||
port = int(local_addr.rsplit(':', 1)[-1])
|
||
pid = int(parts[-1])
|
||
if port > 0:
|
||
pid_ports.setdefault(pid, []).append(port)
|
||
except (ValueError, IndexError):
|
||
continue
|
||
|
||
if not pid_ports:
|
||
PrintLog(LogLevel.DEBUG, '[端口探测] netstat 未找到任何 LISTENING 端口')
|
||
return 0
|
||
|
||
# 2b. 对每个有监听端口的 PID，查进程名是否匹配 QMT
|
||
for pid, ports in pid_ports.items():
|
||
name = RealQmtV._get_process_name(pid)
|
||
if name and any(kw in name for kw in RealQmtV._QMT_PROCESS_KEYWORDS):
|
||
port = ports[0]
|
||
PrintLog(LogLevel.INFO, f'[端口探测] 发现 QMT 进程: {name} (PID={pid}), 端口: {port}')
|
||
|
||
# 同时探测 userdata_mini 路径
|
||
exe_path = RealQmtV._get_process_exe_path(pid)
|
||
if exe_path:
|
||
PrintLog(LogLevel.INFO, f'[路径探测] 进程路径: {exe_path}')
|
||
found_path = RealQmtV._find_userdata_mini(exe_path)
|
||
if found_path:
|
||
PrintLog(LogLevel.INFO, f'[路径探测] 发现 userdata_mini: {found_path}')
|
||
if found_path != config.miniQMTPath:
|
||
PrintLog(LogLevel.INFO, f'[路径探测] 自动修正 miniQMTPath: {config.miniQMTPath} -> {found_path}')
|
||
config.miniQMTPath = found_path
|
||
|
||
# 同时从窗口标题提取资金账号
|
||
account = RealQmtV._discover_account()
|
||
if account:
|
||
if account != config.account_no:
|
||
PrintLog(LogLevel.INFO, f'[账号探测] 自动修正 account_no: {config.account_no[-4:]}**** -> {account[-4:]}****')
|
||
config.account_no = account
|
||
else:
|
||
PrintLog(LogLevel.INFO, f'[账号探测] 确认账号: {account[-4:]}****')
|
||
|
||
return port
|
||
|
||
except Exception as e:
|
||
PrintLog(LogLevel.INFO, f'[端口探测] 进程扫描异常: {e}')
|
||
|
||
PrintLog(LogLevel.WARNING, '[端口探测] 未能自动发现 miniQMT 端口')
|
||
return 0
|
||
|
||
@staticmethod
|
||
def _get_process_name(pid: int) -> str:
|
||
"""通过 PID 获取进程名（单个查询，不用扫全量 tasklist）"""
|
||
try:
|
||
result = subprocess.run(
|
||
['tasklist', '/fi', f'PID eq {pid}', '/fo', 'csv', '/nh'],
|
||
capture_output=True, text=True, timeout=5
|
||
)
|
||
for line in result.stdout.splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith('INFO:'):
|
||
continue
|
||
parts = [p.strip('"').strip() for p in line.split('","')]
|
||
if len(parts) >= 2:
|
||
return parts[0]
|
||
except Exception:
|
||
pass
|
||
return ''
|
||
|
||
@staticmethod
|
||
def _get_process_exe_path(pid: int) -> str:
|
||
"""通过 PID 获取进程的可执行文件完整路径"""
|
||
try:
|
||
result = subprocess.run(
|
||
['powershell', '-NoProfile', '-Command',
|
||
f'(Get-Process -Id {pid}).Path'],
|
||
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5
|
||
)
|
||
path = result.stdout.strip()
|
||
if path and os.path.isfile(path):
|
||
return path
|
||
except Exception:
|
||
pass
|
||
return ''
|
||
|
||
@staticmethod
|
||
def _find_userdata_mini(exe_path: str) -> str:
|
||
"""从 QMT 可执行文件路径向上查找 userdata_mini 目录"""
|
||
exe_dir = os.path.dirname(exe_path)
|
||
# 从 exe 所在目录开始，向上最多 3 层
|
||
for _ in range(4):
|
||
candidate = os.path.join(exe_dir, 'userdata_mini')
|
||
if os.path.isdir(candidate):
|
||
return candidate
|
||
parent = os.path.dirname(exe_dir)
|
||
if parent == exe_dir:
|
||
break
|
||
exe_dir = parent
|
||
return ''
|
||
|
||
@staticmethod
|
||
def _discover_account() -> str:
|
||
"""
|
||
从 XtMiniQmt.exe 的窗口标题中提取资金账号。
|
||
标题格式: "8882874667 - 国金证券QMT交易端 2.0.8.300"
|
||
返回账号字符串，失败返回空字符串。
|
||
"""
|
||
try:
|
||
# 找到 XtMiniQmt.exe 的 PID
|
||
tasklist = subprocess.run(
|
||
['tasklist', '/fo', 'csv', '/nh'],
|
||
capture_output=True, text=True, timeout=10
|
||
)
|
||
gui_pid = 0
|
||
for line in tasklist.stdout.splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
parts = [p.strip('"').strip() for p in line.split('","')]
|
||
if len(parts) >= 2 and 'XtMiniQmt' in parts[0]:
|
||
gui_pid = int(parts[1])
|
||
break
|
||
|
||
if not gui_pid:
|
||
return ''
|
||
|
||
# 获取窗口标题（PowerShell 输出可能含中文，用 utf-8）
|
||
result = subprocess.run(
|
||
['powershell', '-NoProfile', '-Command',
|
||
f'(Get-Process -Id {gui_pid}).MainWindowTitle'],
|
||
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5
|
||
)
|
||
title = result.stdout.strip()
|
||
if title and ' - ' in title:
|
||
account = title.split(' - ')[0].strip()
|
||
if account.isdigit():
|
||
return account
|
||
except Exception:
|
||
pass
|
||
return ''
|
||
|
||
@staticmethod
|
||
def _to_plain_code(stock_code: str) -> str:
|
||
"""将 xtquant 格式 '600519.SH' 转换为数据库格式 '600519'"""
|
||
return stock_code.split('.')[0] if '.' in stock_code else stock_code
|
||
|
||
@staticmethod
|
||
def _to_full_code(stock_code: str) -> str:
|
||
"""将数据库格式 '600519' 转换为 xtquant 格式 '600519.SH'"""
|
||
if '.' in stock_code:
|
||
return stock_code # already has suffix
|
||
code = stock_code
|
||
if code.startswith(('6', '5', '9')):
|
||
return f'{code}.SH'
|
||
elif code.startswith(('0', '3', '2')):
|
||
return f'{code}.SZ'
|
||
# fallback: try both, prefer SH
|
||
return f'{code}.SH'
|
||
|
||
def __init__(self) -> None:
|
||
self.inited = False
|
||
self.connected = False
|
||
self.account = None
|
||
self.xt_trader = None
|
||
self.mini_qmt_path = ""
|
||
self._positions = {}
|
||
self._pending_orders = []
|
||
self._market_data_thread = None
|
||
self.isMarketActive = False
|
||
self.lastMarketDataUpdateTimestamp = time.time()
|
||
self.details = {}
|
||
|
||
def getTrader(self):
|
||
return self
|
||
|
||
def init_qmtv(self):
|
||
"""初始化 QMT 交易器"""
|
||
try:
|
||
from xtquant.xttrader import XtQuantTrader
|
||
from xtquant.xttype import StockAccount
|
||
|
||
self.mini_qmt_path = config.miniQMTPath
|
||
self.account = StockAccount(config.account_no, 'STOCK')
|
||
PrintLog(LogLevel.INFO, f'[QMT] 初始化: path={self.mini_qmt_path}, account={config.account_no[-4:]}****')
|
||
|
||
# 创建 XtQuantTrader 实例
|
||
session_id = int(time.time()) % 10000
|
||
PrintLog(LogLevel.INFO, f'[QMT] 创建 XtQuantTrader, session={session_id}')
|
||
self.xt_trader = XtQuantTrader(self.mini_qmt_path, session_id)
|
||
|
||
# 注册回调 — xtquant 只接受一个回调对象，会在上面调用 on_xxx 方法
|
||
self.xt_trader.register_callback(self)
|
||
|
||
self.inited = True
|
||
PrintLog(LogLevel.INFO, f'- [真实] QMT 交易器初始化成功')
|
||
except Exception as e:
|
||
self.inited = False
|
||
PrintLog(LogLevel.ERROR, f'- [失败] QMT 初始化: {e}')
|
||
|
||
def connect(self) -> bool:
|
||
"""连接 MiniQMT，失败自动探测端口并重试"""
|
||
if not self.inited:
|
||
PrintLog(LogLevel.ERROR, '[QMT] 连接失败: 未初始化')
|
||
return False
|
||
|
||
_connect_errors = {
|
||
0: '成功',
|
||
-1: '一般错误（miniQMT 可能未启动）',
|
||
-2: 'miniQMT 未运行（请先启动极简QMT）',
|
||
-3: '连接超时',
|
||
}
|
||
|
||
def _do_connect() -> int:
|
||
self.xt_trader.start()
|
||
PrintLog(LogLevel.INFO, '[QMT] xt_trader.start() 完成')
|
||
PrintLog(LogLevel.INFO, '[QMT] 正在连接 miniQMT...')
|
||
return self.xt_trader.connect()
|
||
|
||
try:
|
||
# 尝试默认连接
|
||
PrintLog(LogLevel.INFO, '[QMT] 尝试默认方式连接...')
|
||
connect_result = _do_connect()
|
||
|
||
# 失败则自动探测端口并重试
|
||
if connect_result != 0:
|
||
PrintLog(LogLevel.INFO, '[QMT] 默认连接失败，启动端口自动探测...')
|
||
discovered_port = self._discover_qmt_port()
|
||
if discovered_port > 0:
|
||
PrintLog(LogLevel.INFO, f'[QMT] 探测到端口 {discovered_port}，尝试连接...')
|
||
try:
|
||
from xtquant import xtdata
|
||
xtdata.connect(ip='127.0.0.1', port=discovered_port)
|
||
PrintLog(LogLevel.INFO, f'[QMT] xtdata 连接成功 (端口: {discovered_port})')
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'[QMT] xtdata 连接失败 (端口: {discovered_port}): {e}')
|
||
return False
|
||
connect_result = _do_connect()
|
||
else:
|
||
PrintLog(LogLevel.WARNING, '[QMT] 端口自动探测未找到 miniQMT 进程')
|
||
|
||
result_desc = _connect_errors.get(connect_result, f'未知({connect_result})')
|
||
PrintLog(LogLevel.INFO, f'[QMT] connect() 返回: {connect_result} ({result_desc})')
|
||
|
||
if connect_result == 0:
|
||
PrintLog(LogLevel.INFO, f'[QMT] 订阅账户...')
|
||
self.xt_trader.subscribe(self.account)
|
||
PrintLog(LogLevel.INFO, '[QMT] 订阅完成')
|
||
self.connected = True
|
||
self.startMarketDataSubscription()
|
||
PrintLog(LogLevel.INFO, f'[QMT] 连接成功 (账号: {config.account_no[-4:]}****)')
|
||
return True
|
||
else:
|
||
PrintLog(LogLevel.ERROR, f'[QMT] 连接失败: {result_desc}')
|
||
return False
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'[QMT] 连接异常: {e}')
|
||
return False
|
||
|
||
def getAllPositions(self) -> dict:
|
||
"""获取全部持仓，返回 {plain_code: position_object}"""
|
||
if not self.connected:
|
||
return {}
|
||
try:
|
||
positions = self.xt_trader.query_stock_positions(self.account)
|
||
result = {}
|
||
for pos in positions:
|
||
code = self._to_plain_code(getattr(pos, 'stock_code', ''))
|
||
result[code] = pos
|
||
# 缓存以供 getStockPosition 使用
|
||
self._position_cache = result
|
||
return result
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'- [获取全部持仓失败]: {e}')
|
||
return {}
|
||
|
||
def getStockPosition(self, stock_code: str):
|
||
"""获取单只股票持仓（优先使用缓存）"""
|
||
if not self.connected:
|
||
return None
|
||
try:
|
||
# 优先查缓存
|
||
if hasattr(self, '_position_cache') and stock_code in self._position_cache:
|
||
return self._position_cache[stock_code]
|
||
# 回退查询
|
||
positions = self.xt_trader.query_stock_positions(self.account)
|
||
for pos in positions:
|
||
pos_code = self._to_plain_code(getattr(pos, 'stock_code', ''))
|
||
if pos_code == stock_code:
|
||
return pos
|
||
return None
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'- [持仓查询失败] {stock_code}: {e}')
|
||
return None
|
||
|
||
def queryPendingOrder(self, stock_code: str, tag: str) -> list:
|
||
"""查询挂单（过滤已撤/废单）"""
|
||
if not self.connected:
|
||
return []
|
||
try:
|
||
orders = self.xt_trader.query_stock_orders(self.account)
|
||
# 过滤已撤(54)和废单(57)，避免策略误判"已有挂单"跳过下单
|
||
_CANCELED = {54, 57}
|
||
return [o for o in orders
|
||
if self._to_plain_code(getattr(o, 'stock_code', '')) == stock_code and
|
||
(tag is None or getattr(o, 'strategy_name', None) == tag) and
|
||
getattr(o, 'order_status', 0) not in _CANCELED]
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'- [查询挂单失败] {e}')
|
||
return []
|
||
|
||
def queryTodayOrders(self) -> list:
|
||
"""查询当日所有委托"""
|
||
if not self.connected:
|
||
return []
|
||
try:
|
||
return list(self.xt_trader.query_stock_orders(self.account))
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'- [查询委托失败] {e}')
|
||
return []
|
||
|
||
def queryTodayTrades(self) -> list:
|
||
"""查询当日所有成交"""
|
||
if not self.connected:
|
||
return []
|
||
try:
|
||
return list(self.xt_trader.query_stock_trades(self.account))
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'- [查询成交失败] {e}')
|
||
return []
|
||
|
||
def orderAsync(self, stock_code, orderVolume, orderType, orderPrice, priceType, orderRemark, strategy_name):
|
||
"""异步下单"""
|
||
if not self.connected:
|
||
PrintLog(LogLevel.ERROR, '- [下单失败] 未连接')
|
||
return -1
|
||
|
||
try:
|
||
full_code = self._to_full_code(stock_code)
|
||
seq = self.xt_trader.order_stock_async(
|
||
account=self.account,
|
||
stock_code=full_code,
|
||
order_volume=orderVolume,
|
||
order_type=orderType,
|
||
price=orderPrice,
|
||
price_type=priceType,
|
||
order_remark=orderRemark,
|
||
strategy_name=strategy_name
|
||
)
|
||
PrintLog(LogLevel.INFO,
|
||
f'- [下单] {stock_code} 数量:{orderVolume} 价格:{orderPrice} 类型:{orderType} seq:{seq}')
|
||
return 0
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'- [下单失败] {stock_code}: {e}')
|
||
return -1
|
||
|
||
def cacheStockDetail(self, stock_code: str):
|
||
"""获取股票详情"""
|
||
if stock_code not in self.details:
|
||
try:
|
||
from xtquant import xtdata
|
||
# xtquant 需要带后缀的完整代码
|
||
full_code = self._to_full_code(stock_code)
|
||
detail = xtdata.get_instrument_detail(full_code)
|
||
if detail:
|
||
# xtquant 返回 dict，使用 .get() 读取
|
||
self.details[stock_code] = {
|
||
'InstrumentName': detail.get('InstrumentName', stock_code) if isinstance(detail, dict) else getattr(detail, 'InstrumentName', stock_code),
|
||
'UpStopPrice': detail.get('UpStopPrice', 0) if isinstance(detail, dict) else getattr(detail, 'UpStopPrice', 0),
|
||
'DownStopPrice': detail.get('DownStopPrice', 0) if isinstance(detail, dict) else getattr(detail, 'DownStopPrice', 0)
|
||
}
|
||
else:
|
||
self.details[stock_code] = {
|
||
'InstrumentName': stock_code,
|
||
'UpStopPrice': 0,
|
||
'DownStopPrice': 0
|
||
}
|
||
except Exception:
|
||
self.details[stock_code] = {
|
||
'InstrumentName': stock_code,
|
||
'UpStopPrice': 0,
|
||
'DownStopPrice': 0
|
||
}
|
||
return self.details[stock_code]
|
||
|
||
def getInstrumentName(self, stock_code: str) -> str:
|
||
"""获取股票名称"""
|
||
return self.cacheStockDetail(stock_code)['InstrumentName']
|
||
|
||
def dailyUpStop(self, stock_code: str):
|
||
"""获取涨停价"""
|
||
detail = self.cacheStockDetail(stock_code)
|
||
up_stop = detail.get('UpStopPrice', 0)
|
||
PrintLog(LogLevel.DEBUG, f'- [详情] {stock_code} {detail["InstrumentName"]} 涨停价: {up_stop}')
|
||
return up_stop or 0.0
|
||
|
||
def dailyDownStop(self, stock_code: str):
|
||
"""获取跌停价"""
|
||
detail = self.cacheStockDetail(stock_code)
|
||
down_stop = detail.get('DownStopPrice', 0)
|
||
return down_stop or 0.0
|
||
|
||
def getLastPrice(self, stock_code: str) -> float:
|
||
"""主动获取最新市价（拉取模式，作为推送的兜底）"""
|
||
try:
|
||
from xtquant import xtdata
|
||
import json
|
||
full_code = self._to_full_code(stock_code)
|
||
|
||
# 方式1: 尝试 get_full_tick（参数是 list[str]，返回 dict {code: {...}}）
|
||
raw = xtdata.get_full_tick([full_code])
|
||
if raw:
|
||
tick = json.loads(raw) if isinstance(raw, str) else raw
|
||
if isinstance(tick, dict):
|
||
# 格式: {'600519.SH': {'lastPrice': 8.97, ...}}
|
||
for code, info in tick.items():
|
||
if isinstance(info, dict) and info.get('lastPrice', 0) > 0:
|
||
PrintLog(LogLevel.DEBUG, f'[getLastPrice] {stock_code} → tick: {info["lastPrice"]:.3f}')
|
||
return float(info['lastPrice'])
|
||
|
||
# 方式2: get_market_data 取最新1分钟K线收盘价
|
||
data = xtdata.get_market_data(
|
||
field_list=['close'],
|
||
stock_list=[full_code],
|
||
period='1m',
|
||
count=1
|
||
)
|
||
if data:
|
||
vals = None
|
||
if full_code in data:
|
||
row = data[full_code]
|
||
if hasattr(row, '__iter__') and not isinstance(row, str):
|
||
row = list(row)
|
||
if row:
|
||
vals = row
|
||
if not vals and 'close' in data:
|
||
field_data = data['close']
|
||
if full_code in field_data:
|
||
vals = list(field_data[full_code])
|
||
if vals and len(vals) > 0 and float(vals[0]) > 0:
|
||
PrintLog(LogLevel.DEBUG, f'[getLastPrice] {stock_code} → kline: {float(vals[0]):.3f}')
|
||
return float(vals[0])
|
||
|
||
# 方式3: 下载历史数据后再试
|
||
xtdata.download_history_data(full_code, '1m', '')
|
||
data = xtdata.get_market_data(
|
||
field_list=['close'],
|
||
stock_list=[full_code],
|
||
period='1m',
|
||
count=1
|
||
)
|
||
if data:
|
||
vals = None
|
||
if full_code in data:
|
||
row = data[full_code]
|
||
if hasattr(row, '__iter__') and not isinstance(row, str):
|
||
row = list(row)
|
||
if row:
|
||
vals = row
|
||
if not vals and 'close' in data:
|
||
field_data = data['close']
|
||
if full_code in field_data:
|
||
vals = list(field_data[full_code])
|
||
if vals and len(vals) > 0 and float(vals[0]) > 0:
|
||
PrintLog(LogLevel.DEBUG, f'[getLastPrice] {stock_code} → download+kline: {float(vals[0]):.3f}')
|
||
return float(vals[0])
|
||
|
||
PrintLog(LogLevel.DEBUG, f'[getLastPrice] {stock_code} → 失败: 所有方式均无数据, raw={raw}')
|
||
|
||
except Exception as e:
|
||
PrintLog(LogLevel.DEBUG, f'[getLastPrice] {stock_code} → 异常: {e}')
|
||
return 0.0
|
||
|
||
def startMarketDataSubscription(self):
|
||
"""启动市场数据订阅"""
|
||
try:
|
||
from xtquant import xtdata
|
||
|
||
# 订阅沪深全市场实时行情
|
||
seq = xtdata.subscribe_whole_quote(['SH', 'SZ'], self._on_market_data)
|
||
PrintLog(LogLevel.INFO, f'- [市场数据订阅成功-真实] seq={seq}')
|
||
|
||
# 订阅成功即标记市场活跃，避免策略初始化时因等待首条数据被误判为休市
|
||
self.isMarketActive = True
|
||
eBus.event_bus.publish(eBus.EventMarketActiveSwitch, True)
|
||
|
||
# 启动行情活跃监控线程
|
||
self._market_data_thread = threading.Thread(
|
||
target=self._market_data_watchdog, daemon=True
|
||
)
|
||
self._market_data_thread.start()
|
||
except Exception as e:
|
||
PrintLog(LogLevel.ERROR, f'- [市场数据订阅失败-{e}]')
|
||
|
||
def _on_market_data(self, datas: dict):
|
||
"""xtquant 行情回调 — 将数据转换为事件总线格式"""
|
||
self.lastMarketDataUpdateTimestamp = time.time()
|
||
if not self.isMarketActive:
|
||
self.isMarketActive = True
|
||
eBus.event_bus.publish(eBus.EventMarketActiveSwitch, True)
|
||
|
||
# xtquant 返回 "600519.SH" 格式 key，UI 使用纯代码 "600519"
|
||
# 构建同时包含两种 key 的数据确保匹配
|
||
# 直接发布 xtquant 原始数据（代码带 .SH/.SZ 后缀）
|
||
eBus.event_bus.publish(eBus.MarketDataUpdate, datas)
|
||
|
||
def _market_data_watchdog(self):
|
||
"""行情活跃监控 — 超过 30 秒无数据则标记市场不活跃"""
|
||
while True:
|
||
time.sleep(10)
|
||
if self.isMarketActive:
|
||
elapsed = time.time() - self.lastMarketDataUpdateTimestamp
|
||
if elapsed > 30:
|
||
self.isMarketActive = False
|
||
eBus.event_bus.publish(eBus.EventMarketActiveSwitch, False)
|
||
PrintLog(LogLevel.WARNING, f'- [行情] 超过 {elapsed:.0f} 秒无更新，市场标记为不活跃')
|
||
|
||
def stopMarketDataSubscription(self):
|
||
"""停止市场数据订阅"""
|
||
self.isMarketActive = False
|
||
PrintLog(LogLevel.INFO, '- [市场数据订阅已停止]')
|
||
|
||
# ---- xtquant 回调处理 (xtquant 通过回调对象调用 on_xxx 方法) ----
|
||
|
||
def on_connected(self):
|
||
PrintLog(LogLevel.INFO, f'[QMT] on_connected: 真实 QMT 连接成功 {datetime.datetime.now()}')
|
||
|
||
def on_disconnected(self):
|
||
PrintLog(LogLevel.WARNING, f'[QMT] on_disconnected: 真实 QMT 连接断开 {datetime.datetime.now()}')
|
||
|
||
def on_stock_order(self, order):
|
||
self._pending_orders.append(order)
|
||
|
||
def on_stock_trade(self, trade):
|
||
eBus.event_bus.publish(eBus.MarketOrderTraded, trade)
|
||
|
||
def on_order_stock_async_response(self, response):
|
||
eBus.event_bus.publish(eBus.MarketOrderCreated, response)
|
||
|
||
def on_order_error(self, order_error):
|
||
PrintLog(LogLevel.ERROR,
|
||
f'[QMT] 委托报错: order_id={order_error.order_id}, error_id={order_error.error_id}, '
|
||
f'error_msg={order_error.error_msg}, remark={order_error.order_remark}')
|
||
eBus.event_bus.publish(eBus.MarketOrderError, order_error)
|
||
|
||
def on_account_status(self, status):
|
||
PrintLog(LogLevel.INFO, f'[QMT] on_account_status: {datetime.datetime.now()} {status}')
|
||
|
||
|
||
qmtv = RealQmtV()
|