feat(bridge): D-1 xtquant 可用性验证脚本 check_xtquant.py
分3步探活(import xtquant / connect miniQMT / 查询模拟账户资金持仓) 已填国金QMT模拟独立交易环境(账号66639661)
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
"""xtquant 可用性验证(D-1 bridge 开发前置)。
|
||||
|
||||
在 Windows 上跑,分 3 步确认 xtquant 能 import + 连 miniQMT + 查询账户/持仓。
|
||||
miniQMT 客户端必须先登录并保持运行(极简模式)。
|
||||
|
||||
用法:填下面三个参数,python check_xtquant.py
|
||||
若某步 API 报错,参考 miniQMT 自带 xtquant 文档 dict.thinktrader.net
|
||||
或 NAS github-repos/OSkhQuant 源码调整签名(xtquant 版本可能有差异)。
|
||||
"""
|
||||
import sys
|
||||
|
||||
# ===== 用户填写(开始)=====
|
||||
# 1. miniQMT 安装目录下的 userdata_mini 文件夹绝对路径
|
||||
MINIQMT_USERDATA = r"D:\国金QMT交易端模拟\userdata_mini"
|
||||
|
||||
# 2. miniQMT 模拟账户 ID(客户端登录后可见的数字账号)—— 这个要你填
|
||||
ACCOUNT_ID = "66639661"
|
||||
|
||||
# 3. xtquant 所在的 site-packages 路径(miniQMT 安装目录\bin.x64\Lib\site-packages)
|
||||
# 若 import 失败,在此目录找不到 xtquant,改 pip install xtquant 后填 None
|
||||
XTQUANT_SITE_PACKAGES = r"D:\国金QMT交易端模拟\bin.x64\Lib\site-packages"
|
||||
# ===== 用户填写(结束)=====
|
||||
|
||||
SESSION_ID = 20260710 # 任意 int,每个程序实例唯一
|
||||
|
||||
|
||||
def step1_import():
|
||||
"""[1/3] 验证 xtquant 能 import。"""
|
||||
print("[1/3] 验证 import xtquant ...")
|
||||
if XTQUANT_SITE_PACKAGES:
|
||||
sys.path.insert(0, XTQUANT_SITE_PACKAGES)
|
||||
try:
|
||||
from xtquant import xttrader, xtdata
|
||||
print(" ✅ import 成功")
|
||||
return xttrader, xtdata
|
||||
except Exception as e:
|
||||
print(f" ❌ import 失败: {e}")
|
||||
print(" → 检查 XTQUANT_SITE_PACKAGES 路径是否含 xtquant,或 pip install xtquant")
|
||||
return None, None
|
||||
|
||||
|
||||
def step2_connect(xttrader):
|
||||
"""[2/3] 连接 miniQMT 客户端。"""
|
||||
print("[2/3] 连接 miniQMT 客户端 ...")
|
||||
try:
|
||||
xt = xttrader.XtQuantTrader(MINIQMT_USERDATA, SESSION_ID)
|
||||
xt.start()
|
||||
ret = xt.connect()
|
||||
if ret != 0:
|
||||
print(f" ❌ connect 返回 {ret}(确认 miniQMT 客户端已登录运行 + userdata 路径正确)")
|
||||
return None
|
||||
print(" ✅ connect 成功")
|
||||
return xt
|
||||
except Exception as e:
|
||||
print(f" ❌ connect 异常: {e}")
|
||||
print(" → 确认 MINIQMT_USERDATA 路径 + miniQMT 客户端在运行")
|
||||
return None
|
||||
|
||||
|
||||
def step3_query(xt, xttrader):
|
||||
"""[3/3] 查询模拟账户资金 + 持仓。"""
|
||||
print("[3/3] 查询模拟账户 ...")
|
||||
acc_type = xttrader.STOCK_ACCOUNT
|
||||
try:
|
||||
xt.subscribe(ACCOUNT_ID, acc_type)
|
||||
except Exception as e:
|
||||
print(f" ⚠️ subscribe 异常(可忽略,继续): {e}")
|
||||
|
||||
asset = xt.query_stock_asset(ACCOUNT_ID, acc_type)
|
||||
if not asset:
|
||||
print(" ❌ query_stock_asset 返回空(账户ID/权限问题,或 miniQMT 未切到模拟账户)")
|
||||
return False
|
||||
print(f" ✅ 资金: cash={getattr(asset,'cash',None)} "
|
||||
f"frozen={getattr(asset,'frozen_cash',None)} "
|
||||
f"total={getattr(asset,'total_asset',None)}")
|
||||
|
||||
try:
|
||||
pos = xt.query_stock_positions(ACCOUNT_ID, acc_type) or []
|
||||
print(f" ✅ 持仓 {len(pos)} 只: {[getattr(p,'stock_code',None) for p in pos[:5]]}")
|
||||
except Exception as e:
|
||||
print(f" ⚠️ query_stock_positions 异常(资金通即可): {e}")
|
||||
|
||||
print("\n🎉 xtquant 全部可用 → D-1 bridge 可开发。把上面输出贴回来。")
|
||||
return True
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(f"配置: userdata={MINIQMT_USERDATA}\n account={ACCOUNT_ID}\n")
|
||||
xttrader, _ = step1_import()
|
||||
if not xttrader:
|
||||
sys.exit(1)
|
||||
xt = step2_connect(xttrader)
|
||||
if not xt:
|
||||
sys.exit(1)
|
||||
ok = step3_query(xt, xttrader)
|
||||
sys.exit(0 if ok else 1)
|
||||
Reference in New Issue
Block a user