文章声明:本内容为个人的业余研究，和任何单位，机构没有关系，文章出现的股票代码，全部只是测试例子，不做投资参考，投资有风险，代码学习使用，不做商业用途
这几天在重新优化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)
