【从零开始】：token使用指南

前言

1.本篇教程将主要使用xtquant.xtdatacenter(下简称xtdc)，也就是通过token获取数据的方式进行教学

2.本篇教程默认阅读者已经知晓python的基本语法，如变量类型，循环，条件判断

3.本篇教程编写的目的是为了帮助量化小白快速上手xtquant,并搭建自己的量化交易系统

4.作者本人并不是程序员出身，如在内容上有错误/纰漏,也欢迎各位给我进行留言或者站内信

0.安装python与xtquant包

1. 1.关于python如何安装，b站已经有了很多详细的视频教程，这里便不再赘述，需要注意的是，xtquant要求python的版本在py3.6 - py3.11之间，太旧或者太新都不行
3. 2.在使用xtdc前，我们需要将xtquant包安装到我们的python环境中，目前xtquant包可以通过在cmd中执行 pip install xtquant进行安装，或通过迅投官网下载。xtquant下载链接
5. 3.如果你是通过官网下载xtquant，那么下载完成后，需要将压缩包解压缩到本地的python目录下的 .\Lib\site-packages 文件，通过pip指令不需要这步操作
6. 在python环境中执行以下语句，如果不报错，则表示安装成功

7. from xtquant import xtdatacenter as xtdc

1.获取个人token

1. 1.个人token在用户中心的个人设置页面下能看到，直接进行复制就行
2. 2.token仅支持VIP及以上的用户权限进行获取，目前新用户注册有14天VIP试用 -> 新用户注册
3. 3.请妥善保管好自己的个人token，默认情况下，VIP仅允许单点登录，有需要多点登录的可以联系官方进行申请

2.启动xtdc数据服务

主程序代码

#主程序代码
#主程序代码
#主程序代码
from xtquant import xtdatacenter as xtdc
from xtquant import xtdata
# 设置token
xtdc.set_token("你的token")
# 开启K线全推
xtdc.set_kline_mirror_enabled(True)
# 设置start_local_service为False,使xtdc监听的端口为我们自己指定的端口
xtdc.init(start_local_service=False)
# 指定xtdc使用 58601端口
xtdc.listen(port=58601)
# 使主程序一直运行，不添加这句的话程序会直接完成运行并退出
xtdata.run()

子程序代码

# 子程序代码
# 子程序代码
# 子程序代码

from xtquant import xtdata

# 链接到主程序获取数据,链接的端口需要一致
xtdata.connect(port = 58601)
# 取一个tick看看，程序是否正常链接了
print(xtdata.get_full_tick(["000001.SZ"]))

3.使用代码进行数据更新下载

import pandas as pd
from xtquant import xtdata
from tqdm import tqdm 
import re
from myfunc import get_all_symbol_code,send_feishu_msg # myfunc可以在github上搜索xt/qmt找到，上不去的也可以加论坛圈子或者官方群找我
import sys
import json
import requests



# 链接到主程序获取数据,链接的端口需要一致
xtdata.connect(port = 58601)

# 指定VIP服务器进行链接
info={'ip': 'vipsxmd1.thinktrader.net', 'port': 55310, 'username': '', 'pwd': ''}
qs = xtdata.QuoteServer(info)
qs.connect()

# 更新基础信息
print("开始下载板块数据")
xtdata.download_sector_data() ## 更新板块信息
print("板块数据下载完成")
# 设置需要下载的周期
# period_list = ["1d","1m"] # 周期参考官方文档http://dict.thinktrader.net/nativeApi/xtdata.html?id=7zqjlm#%E4%B8%8B%E8%BD%BD%E5%8E%86%E5%8F%B2%E8%A1%8C%E6%83%85%E6%95%B0%E6%8D%AE
period_list = ["1d"] # 周期参考官方文档

# 获取要下载的标的列表
stock_list = xtdata.get_stock_list_in_sector("沪深京A股") ## 获取股票列表

ls = ['上期所', '中金所', '大商所', '广期所', '郑商所'] ## 期货交易所
future_option_list = [] # 板块里没有区分期权和期货，需要手动筛选过滤下
for i in ls:
    future_option_list += xtdata.get_stock_list_in_sector(i)
future_list = get_all_symbol_code(future_option_list) # 筛选出所有的期货标的

# 调用函数进行下载，并使用tqdm库显示进度条
print("开始下载行情数据")
for period in period_list:
    for i in tqdm(stock_list):
        xtdata.download_history_data(i,period,incrementally=True) # incrementally 参数是增量下载选项，新版本xtquant 才有的，如果这里报错了，更新xtquant包
    print(f"周期{period}股票数据下载完成")
    for i in tqdm(future_list):
        xtdata.download_history_data(i,period,incrementally=True)
    print(f"周期{period}期货数据下载完成")

# 创建一个回调函数用来看财务数据的下载进度
def on_progress(data):
    print(data)

print("开始下载财务数据")
xtdata.download_financial_data2(stock_list,callback=on_progress)
print("财务数据下载完成")

sys.exit() # 一切结束后，退出程序

# 这部分以后再出教程
# url = "f"
# send_feishu_msg(url,"收盘作业完成，程序正常退出")

4.使用任务计划，实现每日自动更新数据

任务计划是啥呢？简单说就是windows自带的一个能够自动执行特定任务或程序的功能，我们可以利用这个功能，实现在无人值守的情况下自动运行指定的py脚本，实现每日自动更新下载当日数据的功能

Step1. 打开windows任务计划管理器

Step2.点击创建基本任务,并输入任务名称,点击下一步

Step3.为需要自动启动的脚本添加一个启动任务,启动的频率和时间根据自己需要选择,注意！！！主连接程序必须在子程序之前启动，建议两个程序启动间隔 1 分钟

Step4. 【程序或脚本】处填 python.exe 或者 py.exe Step5. 【添加参数】处填py脚本的文件名 Step**6. 【起始于】处填py脚本代码的目录**

最后添加一个任务，用于停止所有脚本进程，对于无人值守的情况来说非常有必要，在交易日切换后重启并重新进行服务器链接可以避免出现不可知的错误

Step1.基本步骤和上边一致，最后的[程序或脚本]处填 taskkill, [添加参数]处填 /IM python.exe

5.利用token完成单策略实盘交易

策略介绍

本策略是通过token连接方式获取高级行情数据，并通过miniqmt进行实际下单交易，止盈止损管理

标的池选择：上证50

买入条件：当股票1h周期出现ma金叉,且1d周期出现macd位于0轴之上的情况，触发买入条件

卖出条件：当持仓的股票盈利超过设定幅度，或亏损超过设定幅度时，触发卖出条件

策略代码

# 股票池：50etf标的池
# 策略：双均线策略
# 止盈止损：比例止盈止损

import pandas as pd
import numpy as np
import time
import datetime
from xtquant import xtdata,xttrader,xtconstant
from xtquant.xttype import StockAccount


## 策略所用的指标
def ema(df:pd.DataFrame,N):   
    return df.ewm(span=N, adjust=False).mean()  

def MACD(close:pd.DataFrame, short = 12, long = 26, M = 9):
    DIF = ema(close,short) - ema(close,long)
    DEA = ema(DIF,M)
    return DIF.round(3), DEA.round(3)

def CROSSUP(a, b):
    """
    向上穿越: 表当a从下方向上穿过b, 成立返回1, 否则返回0
    Args:
        a (pandas.Series): 数据序列1

        b (pandas.Series): 数据序列2

    Returns:
        pandas.Series: 上穿标志序列
    """
    crossup_data = pd.Series(np.where((a > b) & (a.shift(1) <= b.shift(1)), 1, 0))
    return crossup_data

# 初始化部分
### 这块有很多可以自己优化的点，比如一些固定的东西可以直接写成配置文件，自己再把整个初始化的部分封装成一个函数，避免重复写

xtdata.connect(port=58601) # 连接主程序开放的数据端口，以获取历史数据和实时数据
account = StockAccount("55001435","STOCK") # 设置交易账户，这个账户必须在你界面的能看到，且能手动下单的
xt_trade = xttrader.XtQuantTrader(r"C:\Program Files\国金证券QMT交易端\userdata_mini",int(time.time())) # 连接miniqmt，如果是连接投研的话，目录是..\userdata

class Mycallback(xttrader.XtQuantTraderCallback): # 继承XtQuantTraderCallback类，覆盖定义自己的回调函数，这个类是开源的，不懂的可以点进去自己看
    # 这个策略其实用不上回调，这里只写一个演示用法
    def on_stock_order(self, order):
        """
        委托回报推送
        :param order: XtOrder对象
        :return:
        """
        print(datetime.datetime.now(), '委托回调', order.order_remark)

    def on_stock_trade(self, trade):
        """
        成交变动推送
        :param trade: XtTrade对象
        :return:
        """
        print(datetime.datetime.now(), '成交回调', trade.order_remark)
callback = Mycallback()
xt_trade.register_callback(callback) # 注册回调信息
xt_trade.start() # 启动交易线程
connect_result = xt_trade.connect() # 连接账户准备交易
subscribe_result = xt_trade.subscribe(account) # 订阅接受账户的回调信息,没这一步的话收不到

if connect_result != 0:
    raise KeyboardInterrupt("连接交易失败，请查看https://dict.thinktrader.net/nativeApi/question_function.html?id=7zqjlm按步骤进行排查")

# 设置参数
strategyName = "双均线_MACD_大小周期共振策略" # 设置策略名称
stock_list = xtdata.get_stock_list_in_sector("上证50") # 设置股票交易列表

period1 = "1d" # 大数据周期
period2 = "1h" # 小数据周期

line_1 = 10 # 短周期
line_2 = 20 # 长周期

macd_n1,macd_n2,macd_M = 12, 26, 9 # MACD指标参数

take_profit_ratio = 10 # 止盈比例，单位%
stop_loss_ratio = 5 # 止损比例，单位%

# 初始化持仓容器
holdings = {}

# 信号计算
def strategyFunc(data):
    # 获取行情数据
    kline_1d:pd.DataFrame = xtdata.get_market_data_ex([],stock_list,period=period1,count=40)
    kline_1h:pd.DataFrame = xtdata.get_market_data_ex([],stock_list,period=period2,count=300)

    # 获取账户持仓数据
    for obj in xt_trade.query_stock_positions(account):
        stock_code = obj.stock_code
        if stock_code not in stock_list:
            continue 
        holdings[stock_code] = {
            "volume":obj.volume, # 持仓量
            "can_use_volume":obj.volume, # 可用量
            "open_price":obj.open_price, # 持仓均价
            "market_value":obj.market_value # 市值
        }

    print(holdings)
    tick = xtdata.get_full_tick(list(holdings.keys()) + stock_list)
    # print(data)
    for stock in stock_list:
        data1 = kline_1h[stock] # 取这个标的的1h数据
        data2 = kline_1d[stock] # 取这个标的的1d数据

        ## 计算均线
        ma1 = data1["close"].rolling(line_1).mean() # 计算短周期均线
        ma2 = data1["close"].rolling(line_2).mean() # 计算长周期均线
        signal = CROSSUP(ma1,ma2) # 计算金叉信号序列

        ## 计算MACD
        dif,dea = MACD(data2["close"])
        diff = dif - dea

        # print(signal,diff)
        # 如果MACD是金叉状态，即DIFF在0轴之上，且1小时均线金叉，则触发买入信号
        # print(stock,signal.iloc[-1] , diff.iloc[-1])
        if signal.iloc[-1] and diff.iloc[-1] > 0:
            # 判断是否有持仓，已经有持仓就不开仓了
            if stock in holdings:
                continue
            else:
                # 按最新价买入标的1手，实际交易中还要考虑买入但未成交的情况下，怎么防止超单
                xt_trade.order_stock_async(account, stock, xtconstant.STOCK_BUY, 100, xtconstant.LATEST_PRICE, 0, strategyName, 'order_test')

    # 止盈止损部分
    for stock in holdings:
        _d = holdings[stock]
        cost = _d["open_price"] * _d["volume"] # 计算成本
        profit = _d["market_value"] - cost # 计算盈利金额
        # 如果亏损超过了设定的幅度
        if (cost / _d["market_value"] - 1) * 100 < - stop_loss_ratio and _d["volume"] > 0:
            # 卖出所有持仓
            xt_trade.order_stock_async(account, stock, xtconstant.STOCK_SELL, _d["volume"], xtconstant.LATEST_PRICE, 0, strategyName, 'stop_loss_ratio')
        # 如果盈利超过了指定的幅度
        if (cost / _d["market_value"] - 1) * 100 > take_profit_ratio and _d["volume"] > 0:
             # 卖出所有持仓
            xt_trade.order_stock_async(account, stock, xtconstant.STOCK_SELL, _d["volume"], xtconstant.LATEST_PRICE, 0, strategyName, 'take_profit_ratio')

# 行情数据订阅
for i in stock_list:
    xtdata.subscribe_quote(i,period=period1,count=-1)
    xtdata.subscribe_quote(i,period=period2,count=-1)

xtdata.subscribe_quote("000300.SH",period="tick",callback=strategyFunc) # 订阅主图tick,并用主图tick来驱动策略信号计算,来实现类似与内置python中handlebar的机制
# 非VIP用户受订阅数量限制，最大只能订阅300次，本策略目前只用了101次
xtdata.run()

6.利用token实现多策略管理交易

在第四章中，我们从服务器已经下载了所需要的 板块数据，财务数据,K线数据，接下来我们来通过token实现一个简单的双均线策略，并输出信号

策略代码

from xtquant import xtdata
import pandas as pd
import numpy as np
import json
import os
from myfunc import CROSSUP
xtdata.connect(port=58601)

strategyName = "双均线策略" # 设置策略名称
pos_file_path = r"E:\token教学\posititon" # 持仓文件路径，用于被交易端读取

strategy_pos_file = os.path.join(pos_file_path,strategyName + ".json")

try:
    with open(strategy_pos_file,"r") as f:
        strategy_pos = json.load(f)
except FileNotFoundError:
    strategy_pos = {} # 如果文件没有找到，就说明没有持仓



# stock_list = xtdata.get_stock_list_in_sector("沪深300")[:50] # 设定股票池,演示从沪深300取50只计算
stock_list = ["000001.SZ","301158.SZ"] # 本来想用五十只测，谁想到A股烂到500只都没有金叉...

period = "1d" # 数据周期
line_1 = 10 # 短周期
line_2 = 20 # 长周期

def call_back(data:dict):
    _list = list(data.keys())
    stock = _list[0]
    kline:pd.DataFrame = xtdata.get_market_data_ex([],_list,period=period,count=max(line_1,line_2)+1)[stock]
    close = kline["close"]
    ma1 = close.rolling(line_1).mean()
    ma2 = close.rolling(line_2).mean()
    signal = CROSSUP(ma1,ma2) # 计算金叉信号
    position = strategy_pos.get(stock,0) # 读取本地持仓缓存，如果没有这个标的，则认为没有持仓
    if signal.iloc[-1] and position == 0: # 如果金叉了，并且没有持仓
        strategy_pos[stock] = 100 # 则记录这个标的应该要买入100股
        with open(strategy_pos_file,"w") as f:
            json.dump(strategy_pos,f) # 把这条记录写入信号文件，交给策略交易端下单


for i in stock_list:
    xtdata.subscribe_quote(i,period=period,count=-1,callback=call_back)

xtdata.run()

运行上段代码可以发现，当交易信号触发后，我们设置的策略持仓路径下生成了一个 双均线策略.json文件，这个文件的内容就是我们目前的理论持仓

之后我们会需要一个交易端来完成真正的下单操作，并完成策略隔离。

不清楚的内容可添加下方助理微信咨询，有其他 QMT 小技巧想学习的吗？欢迎在下方留言，笔者将根据大家的留言持续更新哦！

欢迎和我一起加入迅投组建的 QMT 实战交流社群，交流群内有许多做量化交易的高手和大佬，具有良好的分享和互助氛围。且迅投官方会不定期为多次分享、乐于助人的群友申请送投研专业版的机会。

只需扫描下方的二维码，名额有限，限时加入。一起分享见解、交换信息、并共同进步，就像群友说的：“就算周末，晚上也有地方沟通交流！”

Mac虚拟机 arm win11 用不了，是不支持吗？

1. Traceback (most recent call last):
   
2.   File "C:\Users\superstar\xuntouhangqing\hangqingmain.py", line 3, in <module>
   
3.     from xtquant import xtdatacenter as xtdc
   
4.   File "C:\Users\superstar\AppData\Local\Programs\Python\Python311-arm64\Lib\site-packages\xtquant\xtdatacenter.py", line 6, in <module>
   
5.     from . import datacenter as __dc
   
6. ImportError: cannot import name 'datacenter' from 'xtquant' (C:\Users\superstar\AppData\Local\Programs\Python\Python311-arm64\Lib\site-packages\xtquant\__init__.py)

复制代码

*******5970 发表于 2024-1-29 10:44
Mac虚拟机 arm win11 用不了，是不支持吗？

不清楚虚拟环境的事

还有其他想学习的 QMT 小技巧吗？欢迎大家在下方留言，版主将根据大家的留言持续更新哦！

两个问题请教：
1、上边主-子程序的方式下，tick和k线数据都能全推了吧

2、能否贴一些具体的下单控制的代码，比如多笔下单，下出去3秒不成交就撤单后追单重下，还有持仓的风险监控、止损止盈等？谢谢

*******5601 发表于 2024-4-26 18:03
两个问题请教：
1、上边主-子程序的方式下，tick和k线数据都能全推了吧

1.对
2.有机会出教程

好的。期待