量化研究---索普量化通达信自选股操作算法

文章声明:本内容为个人的业余研究，和任何单位，机构没有关系，文章出现的股票代码，全部只是测试例子，不做投资参考，投资有风险，代码学习使用，不做商业用途

这几天在重新优化mini版本的板块，把大qmt的模型迁移到mini，方便本地的多线程运行

刚刚迁移好通达信的多模板块交易系统，教程可以参考网页http://14.103.193.242:9999/xms_quants.html

http://14.103.193.242:9999/xms_quants.html

迁移的代码真不少，真正的实盘开发

我把操作通达信板块的成功后模块分享给大家，包括了读取，添加删除等，原理是修改blk里面的内容

设置通达信的自定义板块的路径

点击通达信的安装文件，打开文件在的位置

点击t0002文件

点击blocknew板块文件，全部内容在这个文件夹

文件的名字就是对应板块的中文字母开头

读取板块的内容

读取删除都可以具体看代码，我全部上传了知识星球可以直接下载使用

不懂的问我就可以，加我备注入群可以加入量化群

给大家争取到的开户福利，需要的可以找我，备注开户，网页的资料免费下载

我技术支持

源代码

import pandas as pd
import numpy as np
import os
class xg_tdx:
    '''
    作者:索普量化
    索普量化通达信自选股操作模型
    作者微信xms_quants1
    '''
    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='QMTCG'):
        '''
        建立通达信自定义自选股模块
        '''
        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='QMTCG'):
        '''
        删除自定义股票池板块
        '''
        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 adjust_stock_1(self,stock='600031.SH'):
        '''
        调整代码
        '''
        if stock[-2:]=='SH' or stock[-2:]=='SZ' or stock[-2:]=='sh' or stock[-2:]=='sz':
            stock=stock.upper()
        else:
            if stock[:3] in ['600','601','603','605','688','689',
                ] or stock[:2] in ['11','51','58']:
                stock=stock+'.SH'
            else:
                stock=stock+'.SZ'
        return stock
    def read_tdx_stock(self,name='QMTCG'):
        '''
        读取通达信板块成分股
        '''
        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 read_tdx_stock_1(self,name='QMTCG'):
        '''
        读取通达信板块成分股
        外部使用
        '''
        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
            df['证券代码']=df['证券代码'].apply(lambda x:self.adjust_stock_1(str(x)[1:]))
            
        except Exception as e:
            print(e,'通达信路径有问题可能不存在',path)
            df=pd.DataFrame()
            
        return df 
    def add_tdx_stock(self,name='QMTCG',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='QMTCG',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='QMTCG',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='QMTCG',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='QMTCG'):
        '''
        清空通达信自选股股票
        '''
        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))
if __name__=='__main__':
    api=xg_tdx(path=r'C:/new_tdx/T0002/blocknew')
    df=api.read_tdx_stock_1(name='BUY1')
    print(df)