量化研究---小果通达信QMT实时同步持股系统

*******0079 · 发表于 2025-3-26 09:30:04

小果通达信QMT实时同步持股系统

介绍

小果通达信QMT实时同步持股系统介绍小果通达信QMT实时同步持股系统是一款基于Python开发的自动化交易解决方案，通过将通达信软件与券商QMT交易系统深度结合，实现股票持仓信息的实时同步与自动化交易管理。该系统专为量化交易者设计，能够有效解决传统通达信软件在实时交易中的局限性，提供更高效、更精准的交易执行能力。

系统核心功能小果通达信QMT实时同步持股系统具备以下核心功能特点：

实时持仓同步：系统能够实时监控并同步通达信软件中的持仓股票信息，确保交易账户与策略信号保持高度一致58。通过持续跟踪持仓变化，系统可以自动触发相应的交易操作，减少人工干预带来的延迟。

自动化交易执行：基于xtquant库连接券商交易接口，系统可以自动向券商服务器发送买入或卖出指令，支持多种委托类型，包括限价单、最优五档、对手方最优和本方最优等13。这种自动化执行大大提高了交易效率，特别适合需要快速响应的交易策略。

突破价格限制：系统特别设计了绕过通达信软件2%价格笼子限制的功能，使交易者能够以更理想的价格成交，尤其适用于打板策略3。这一特性解决了传统通达信在快速变动市场中无法以理想价格成交的痛点。

智能监控与预警：系统持续监控通达信板块文件或特定输出文件的变化，一旦检测到符合交易条件的信号，立即触发交易逻辑13。同时，系统还集成了错误处理和邮件通知功能，确保用户能够及时了解系统运行状态和交易执行情况。

多账户管理与风控：系统支持多账户管理，能够在买入前检查可用资金，在卖出前检查可用持仓，确保交易的合法性3。此外，系统还提供基本的风险管理功能，如单笔和总金额限制设置，防止意外大额交易。系统优势相比传统通达信使用方式，小果通达信QMT实时同步持股系统具有以下显著优势：

执行速度优势：通过直接对接券商交易接口，绕过通达信前端界面，交易指令执行速度大幅提升3。测试表明，系统从信号检测到订单提交的平均延迟低于100毫秒。

策略灵活性：系统支持用户自定义交易策略参数，包括交易时间段、单笔金额限制、总金额限制、价格调整幅度等1。这种灵活性使系统能够适应多种交易风格和市场环境。

稳定性保障：系统设计了完善的错误处理机制和自动重连功能，当连接断开时会自动尝试重新连接，确保交易连续性1。同时，系统会记录详细的操作日志，便于问题排查和策略优化。

用户体验优化：系统提供了图形用户界面(GUI)，用户可以方便地配置参数和监控交易状态1。界面设计简洁直观，降低了量化交易的技术门槛。

风险管理增强：系统在传统通达信功能基础上增加了资金和持仓检查机制，避免因资金或持仓不足导致的交易失败3。同时，系统支持设置多种交易限制参数，有效控制风险敞口。

应用场景小果通达信QMT实时同步持股系统适用于多种交易场景：

量化策略执行：适合执行基于技术指标或量化模型的自动化交易策略，系统能够快速准确地执行大量交易指令19。

打板策略实施：特别优化的价格突破功能使其成为实施打板策略的理想工具，能够有效解决2%价格笼子限制问题3。

组合再平衡：对于需要进行定期持仓调整的投资组合，系统可以自动监控持仓偏差并执行再平衡交易58。

套利交易：在需要快速执行配对交易或统计套利策略时，系统的快速响应特性尤为重要1。

多账户管理：机构投资者或管理多个账户的交易者可以利用系统实现多账户的同步交易和集中管理3。

软件架构

软件架构说明

安装教程

导入大qmt就可以作者微信15117320079

使用说明

一打开qnt

打开qmt，不选择独立默认进入大qmt

二打开通达信

可以点击看自选股板块

三建立通达信自选股

1点击自选股管理

不要选择板块缓存，方便实时读取数据

建立板块比如QMTTDX

四导入策略模型

点击模型研究导入策略

五qmt设置

1设置qmt账户

2设置通达信板块路径，默认的差不多应该是一样的

3设置我们建立的板块比如QMTTDX

六点击测试一下

点击运行测试一下

七挂模型交易

挂模型交易可以实时运行，支持稍等交易的持股

点击运行看策略日志

八策略效果

点击通达信自选股

持股全部同步到了通达信

策略源代码

#encoding:gbk
'''
小果通达信qmt实时同步持股系统
作者:小果
微信:15117320079
策略原理实时把qmt的持股同步到通达信自选股
'''
import pandas as pd
import numpy as np
import talib
import os
import time
from datetime import datetime
text={
    "账户":"55001948",
    "账户支持融资融券":"账户支持融资融券,账户类型STOCK/CREDIT",
    "账户类型":"STOCK",
    "通达信路径":"E:/tdx/T0002/blocknew",
    "自选股板块":"QMTTDX",
    "时间设置":"时间设置********",
    "交易时间段":8,
    "交易开始时间":0,
    "交易结束时间":24,
    "是否参加集合竞价":"否",
    "开始交易分钟":0,

}
def init(c):
    #账户
    c.account=text['账户']
    #账户类型
    c.account_type=text['账户类型']
    if c.account_type=='stock' or c.account_type=='STOCK':
        c.buy_code=23
        c.sell_code=24
    else:
        #融资融券
        c.buy_code=33
        c.sell_code=34
    c.path=text['通达信路径']
    c.name=text['自选股板块']
    c.xg_tdx=xg_tdx(path=r'{}'.format(c.path))
    #循环模式3秒
    c.run_time("update_all_data","3nSecond","2024-07-25 13:20:00")
    print(get_account(c,c.account,c.account_type))
    print(get_position(c,c.account,c.account_type))
    update_all_data(c)
def update_all_data(c):
    '''
    更新数据
    通达信板块不在持股里面的删除，持股没有在里面的添加
    '''
    if check_is_trader_date_1():
        print('********************************************************')
        print('********************************************************')
        print('********************************************************')
        #读取持股
        hold_stock=get_position(c,c.account,c.account_type)
        if hold_stock.shape[0]>0:
            hold_stock['证券代码']=hold_stock['证券代码'].apply(lambda x: str(x).split('.')[0])
            hold_stock=hold_stock[hold_stock['持仓量']>=10]
            if hold_stock.shape[0]>0:
                hold_stock_list=hold_stock['证券代码'].tolist()
            else:
                hold_stock_list=[]
        else:
            hold_stock_list=[]
        #读取通达信自选股
        df=c.xg_tdx.read_tdx_stock(name=c.name)
        if df.shape[0]>0:
            df['证券代码']=df['证券代码'].apply(lambda x: str(x)[1:])
            tdx_stock_list=df['证券代码'].tolist()

        else:
            tdx_stock_list=[]
        #卖出，通达信自选股不在持股剔除
        sell_stock_list=[]
        for stock in tdx_stock_list:
            if stock not in hold_stock_list:
                print('{}通达信{}自选股不在qmt持股剔除'.format(stock,c.name))
                sell_stock_list.append(stock)
            else:
                print('{}通达信{}自选股在qmt持股,持有'.format(stock,c.name))
        #添加股票池，持股不在通达信自选股添加
        buy_stock_list=[]
        for stock in hold_stock_list:
            if stock not in tdx_stock_list:
                print('{}qmt持股不在通达信{}自选股，添加到通达信自选股'.format(stock,c.name))
                buy_stock_list.append(stock)
            else:
                print('{}qmt持股在通达信{}自选股，不添加到通达信自选股'.format(stock,c.name))
        #添加
        if len(buy_stock_list)>0:
            c.xg_tdx.add_tdx_stock_list(name=c.name,user_stock_list=buy_stock_list)
        else:
            print('没有符合添加到qmt的持股')
        #剔除通达信
        if len(sell_stock_list)>0:
            c.xg_tdx.del_tdx_stock_list(name=c.name,user_stock_list=sell_stock_list)
        else:
            print('没有通达信剔除的持股')
    else:
        print(datetime.now(),'目前不是交易时间等待')
class xg_tdx:
    '''
    小果通达信自选股操作模型
    作者微信15117320079
    '''
    def __init__(self,path=r'E:\tdx\T0002\blocknew'):
            '''
            小果通达信自选股操作模型
            '''
            self.path=path

    def read_all_tdx_stock(self):
        '''
        读取全部的通达信板块
        '''
        try:
            all_path=os.listdir(r'{}'.format(self.path))
        except Exception as e:
            print(e,'通达信板块文件不存在')
            all_path=[]
        return all_path
    def creat_tdx_user_def_stock(self,name='CS_1'):
        '''
        建立通达信自定义自选股模块
        '''
        name_1='{}.blk'.format(name)
        path="{}\{}.blk".format(self.path,name)
        all_path=self.read_all_tdx_stock()
        if name_1 in all_path:
            print('{} 通达信自选股模块已经存在不建立'.format(name))
        else:
            with open(path, 'w', encoding='gbk') as file:
                file.writelines('')
                print('{} 通达信自选股板块建立成功'.format(name))
    def del_tdx_user_def_stock(self,name='CS'):
        '''
        删除自定义股票池板块
        '''
        name_1='{}.blk'.format(name)
        path="{}\{}.blk".format(self.path,name)
        all_path=self.read_all_tdx_stock()
        if name_1 in all_path:
            os.remove(path=path)
            print('自定义模块{}删除成功'.format(name))
        else:
            print(name_1,'不存在')
    def adjust_stock(self,stock='600031.SH'):
        '''
        调整代码
        '''
        if stock[:3] in ['600','601','603','605','688','689',
                ] or stock[:2] in ['11','51','58']:
            stock="1{}".format(stock)
        else:
            stock="0{}".format(stock)
        return stock
    def read_tdx_stock(self,name='CS'):
        '''
        读取通达信板块成分股
        '''
        path="{}\{}.blk".format(self.path,name)
        try:
            stock_list=[]
            with open(r'{}'.format(path)) as p:
                com=p.readlines()
                for stock in com:
                    if len(stock)>=6:
                        stock=stock.replace("\n", "")
                        stock_list.append(stock)
            df=pd.DataFrame()
            df['证券代码']=stock_list

        except Exception as e:
            print(e,'通达信路径有问题可能不存在',path)
            df=pd.DataFrame()
        return df 
    def add_tdx_stock(self,name='CS',stock='000001'):
        '''
        把股票添加到通达信自选股
        '''
        stock=self.adjust_stock(stock)
        name_1='{}.blk'.format(name)
        path="{}\{}.blk".format(self.path,name)
        all_path=self.read_all_tdx_stock()
        if name_1 in all_path:
            pass
        else:
            self.creat_tdx_user_def_stock(name=name)
            print(name,'自选股不存在建立')
        df=self.read_tdx_stock(name=name)
        if df.shape[0]>0:
            stock_list=df['证券代码'].tolist()
        else:
            print('{}自定义没有数据'.format(name))
            stock_list=[]
        if stock in stock_list:
            print('{} 在自选股{} 不添加'.format(stock,name))
        else:
            stock_list.append(stock)
            with open(path, 'w', encoding='gbk') as file:
                for stock in stock_list:
                    file.writelines(str(stock)+'\n')
                    print('{} 添加到自选股{}成功'.format(stock,name))


    def add_tdx_stock_list(self,name='CS',user_stock_list=['000001']):
        '''
        批量添加股票池到自选股
        '''
        name_1='{}.blk'.format(name)
        path="{}\{}.blk".format(self.path,name)
        all_path=self.read_all_tdx_stock()
        if name_1 in all_path:
            pass
        else:
            self.creat_tdx_user_def_stock(name=name)
            print(name,'自选股不存在建立')
        df=self.read_tdx_stock(name=name)
        if df.shape[0]>0:
            stock_list=df['证券代码'].tolist()
        else:
            print('{}自定义没有数据'.format(name))
            stock_list=[]
        for stock in user_stock_list:
            stock=self.adjust_stock(stock)
            if stock in stock_list:
                print('{} 在自选股{} 不添加'.format(stock,name))
            else:
                stock_list.append(stock)
        with open(path, 'w', encoding='gbk') as file:
            for stock in stock_list:
                file.writelines(str(stock)+'\n')
                print('{} 添加到自选股{}成功'.format(stock,name))
    def del_tdx_stock(self,name='CS',stock='000001'):
        '''
        删除通达信自选股成分股
        '''
        stock=self.adjust_stock(stock)
        name_1='{}.blk'.format(name)
        path="{}\{}.blk".format(self.path,name)
        all_path=self.read_all_tdx_stock()
        if name_1 in all_path:
            df=self.read_tdx_stock(name=name)
            if df.shape[0]>0:
                stock_list=df['证券代码'].tolist()
            else:
                print('{}自定义没有数据'.format(name))
                stock_list=[]
            if len(stock_list)>0:
                if stock in stock_list:
                    stock_list.remove(stock)
                    print('{} 删除自选股{}成功'.format(stock,name))
                else:
                    print('{}不在{}自选股不能删除'.format(stock,name))
            else:
                print('{} 自选股没有数据'.format(name))
            with open(path, 'w', encoding='gbk') as file:
                for stock in stock_list:
                    file.writelines(str(stock)+'\n')
        else:
            print('{} 自选股不存在'.format(name))
    def del_tdx_stock_list(self,name='CS',user_stock_list=['000001']):
        '''
        批量删除通达信自选股成分股
        '''
        name_1='{}.blk'.format(name)
        path="{}\{}.blk".format(self.path,name)
        all_path=self.read_all_tdx_stock()
        if name_1 in all_path:
            df=self.read_tdx_stock(name=name)
            if df.shape[0]>0:
                stock_list=df['证券代码'].tolist()
            else:
                print('{}自定义没有数据'.format(name))
                stock_list=[]
            if len(stock_list)>0:
                for stock in user_stock_list:
                    stock=self.adjust_stock(stock)
                    if stock in stock_list:
                        stock_list.remove(stock)
                        print('{} 删除自选股{}成功'.format(stock,name))
                    else:
                        print('{}不在{}自选股不能删除'.format(stock,name))
            else:
                print('{} 自选股没有数据'.format(name))
            with open(path, 'w', encoding='gbk') as file:
                for stock in stock_list:
                    file.writelines(str(stock)+'\n')
                    print('{} 从板块{}删除成功'.format(stock,name))
        else:
            print('{} 自选股不存在'.format(name))

    def del_all_tdx_stock(self,name='CS'):
        '''
        清空通达信自选股股票
        '''
        name_1='{}.blk'.format(name)
        path="{}\{}.blk".format(self.path,name)
        all_path=self.read_all_tdx_stock()
        if name_1 in all_path:
            stock_list=[]
            with open(path, 'w', encoding='gbk') as file:
                for stock in stock_list:
                    file.writelines(str(stock)+'\n')
            print('{}全部板块内容清空成功'.format(name))
        else:
            print('{} 自选股不存在'.format(name))
def check_is_trader_date_1():
    '''
    检测是不是交易时间
    '''
    trader_time=text['交易时间段']
    start_date=text['交易开始时间']
    end_date=text['交易结束时间']
    start_mi=text['开始交易分钟']
    jhjj=text['是否参加集合竞价']
    if jhjj=='是':
        jhjj_time=15
    else:
        jhjj_time=30
    loc=time.localtime()
    tm_hour=loc.tm_hour
    tm_min=loc.tm_min
    wo=loc.tm_wday
    if wo<=trader_time:
        if tm_hour>=start_date and tm_hour<=end_date:
            if tm_hour==9 and tm_min<jhjj_time:
                return False

            elif tm_min>=start_mi:
                return True
            else:
                return False
        else:
            return False  
    else:
        print('周末')
        return False
#获取账户总权益m_dBalance
def get_account(c,accountid,datatype):
    '''
    获取账户数据
    '''
    accounts = get_trade_detail_data(accountid, datatype, 'account')
    result={}
    for dt in accounts:
        result['总资产']=dt.m_dBalance
        result['净资产']=dt.m_dAssureAsset
        result['总市值']=dt.m_dInstrumentValue
        result['总负债']=dt.m_dTotalDebit
        result['可用金额']=dt.m_dAvailable
        result['盈亏']=dt.m_dPositionProfit
    return result
#获取持仓信息{code.market:手数}
def get_position(c,accountid,datatype):
    '''
    获取持股数据
    '''
    positions = get_trade_detail_data(accountid,datatype, 'position')
    data=pd.DataFrame()
    print('持股数量{}'.format(len(positions)))
    if len(positions)>0:
        df=pd.DataFrame()
        try:
            for dt in positions:
                df['股票代码']=[dt.m_strInstrumentID]
                df['市场类型']=[dt.m_strExchangeID]
                df['证券代码']=df['股票代码']+'.'+df['市场类型']
                df['证券名称']=[dt.m_strInstrumentName]
                df['持仓量']=[dt.m_nVolume]
                df['可用数量']=[dt.m_nCanUseVolume]
                df['成本价']=[dt.m_dOpenPrice]
                df['市值']=[dt.m_dInstrumentValue]
                df['持仓成本']=[dt.m_dPositionCost]
                df['盈亏']=[dt.m_dPositionProfit]
                data=pd.concat([data,df],ignore_index=True)

        except Exception as e:
            print('获取持股隔离股票池有问题')
            data=pd.DataFrame()
    else:
        data=pd.DataFrame()
    return data 
def get_order(c,accountid,datatype):
    '''
    获取委托
    '''
    data=pd.DataFrame()
    orders = get_trade_detail_data(accountid,datatype, 'order')
    print('委托数量{}'.format(len(orders)))
    if len(orders)>0:
        df=pd.DataFrame()
        for o in orders:
            df['股票代码']=[o.m_strInstrumentID]
            df['市场类型']=[o.m_strExchangeID]
            df['证券代码']=df['股票代码']+'.'+df['市场类型']
            df['买卖方向']=[o.m_nOffsetFlag]
            df['委托数量']=[o.m_nVolumeTotalOriginal]
            df['成交均价']=[o.m_dTradedPrice]
            df['成交数量']=[o.m_nVolumeTraded]
            df['成交金额']=[o.m_dTradeAmount]
            df['投资备注']=[o.m_strRemark]
            df['委托状态']=[o.m_nOrderStatus]
            data=pd.concat([data,df],ignore_index=True)
    else:
        data=pd.DataFrame()
    return data
def get_deal(c,accountid,datatype):
    '''
    获取成交
    '''
    data=pd.DataFrame()
    deals = get_trade_detail_data(account, 'stock', 'deal')
    print('成交数量{}'.format(len(deals)))
    if len(deals):
        df=pd.DataFrame()
        for dt in deals:
            df['股票代码']=[dt.m_strInstrumentID]
            df['市场类型']=[dt.m_strExchangeID]
            df['证券代码']=df['股票代码']+'.'+df['市场类型']
            df['证券名称']=[dt.m_strInstrumentName]
            df['买卖方向']=[dt.m_nOffsetFlag]
            df['成交价格']=[dt.m_dPrice]
            df['成交数量']=[dt.m_nVolume]
            df['成交金额']=[dt.m_dTradeAmount]
            data=pd.concat([data,df],ignore_index=True)
    else:
        data=pd.DataFrame()

def handlebar(c):
    pass

[groupid=42]量化交易研究[/groupid]

*******0079 · 发表于 2025-3-26 09:32:42

作者微信15117320079，不懂的问我就可以

量化研究---小果通达信QMT实时同步持股系统