|
C:\new_tdx\T0002\hq_cache（換成你的通達信目錄）
這些文件是通達信軟件用于組織和管理板塊信息的配置文件和數(shù)據(jù)文件。它們定義了不同類型的板塊的代碼、名稱和包含的股票代碼，幫助用戶在軟件中進行板塊相關(guān)的分析和篩選操作。
import pandas as pd
from struct import unpack PATH = 'C:/new_tdx/T0002/hq_cache/' def read_file(file_name, splits): ''' 讀取文件 :param file_name: 文件路徑 :param splits: 分隔符 | :return: ''' with open(file_name, 'r') as f: buf_lis = f.read().split('\n') return [x.split(splits) for x in buf_lis[:-1]] def get_block_file(block='gn'): ''' 獲取通達信的板塊數(shù)據(jù) :param block: gn概念板塊 fg板塊分類zs指數(shù)板塊 :return: ''' file_name = f'block_{block}.dat' with open(PATH + file_name, 'rb') as f: buff = f.read() head = unpack('<384sh', buff[:386]) blk = buff[386:] blocks = [blk[i * 2813:(i + 1) * 2813] for i in range(head[1])] bk_list = [] for bk in blocks: name = bk[:8].decode('gbk').strip('\x00') num, t = unpack('<2h', bk[9:13]) stks = bk[13:(12 + 7 * num)].decode('gbk').split('\x00') bk_list = bk_list + [[name, block, num, stks]] return pd.DataFrame(bk_list, columns=['name', 'tp', 'num', 'stocks']) def get_block_tdxzs3(block='hy'): ''' 獲取通達信的自定義板塊配置文件 :param block: :return: ''' buf_line = read_file(PATH+'tdxzs3.cfg', '|') mapping = {'hy': '2', 'dy': '3', 'gn': '4', 'fg': '5', 'sw': '12', 'zs': '6'} df = pd.DataFrame(buf_line, columns=['name', 'code', 'type', 't1', 't2', 'block']) dg = df.groupby(by='type') if (block == 'zs'): return dg temp = dg.get_group(mapping[block]).reset_index(drop=True) temp.drop(temp.columns[[2, 3, 4]], axis=1, inplace=True) return temp def get_stock_tdxhy(): ''' 通達信軟件中的行業(yè)板塊 :return: ''' buf_line = read_file(PATH+'tdxhy.cfg', '|') buf_lis = [] for x in buf_line: buf_lis.append(x) df = pd.DataFrame(buf_lis, columns=['c0', 'code', 'block', 'c1', 'c2', 'c3']) df.drop(df.columns[[0, 3, 4, 5]], axis=1, inplace=True) df = df[(df['block'] != '')] # df = df[df.code.str.startswith(('sz','sh'))] df['block5'] = df['block'].str[0:5] return df def hy_block(blk='hy'): ''' 獲取板塊數(shù)據(jù) :param blk: :return: ''' #獲取行業(yè)數(shù)據(jù) stock_list = get_stock_tdxhy() # 獲取自定義 block_list = get_block_tdxzs3(blk) block_list = block_list.drop(block_list[block_list['name'].str.contains('TDX')].index) block_list['block5'] = block_list['block'].str[0:5] block_list['num'] = 0 block_list['stocks'] = '' for i in range(len(block_list)): block_key = block_list.iat[i, 2] if (len(block_key) == 5): data_i = stock_list[stock_list['block5'] == block_key] # 根據(jù)板塊名稱過濾 else: data_i = stock_list[stock_list['block'] == block_key] # 根據(jù)板塊名稱過濾 # 板塊內(nèi)進行排序填序號 data_i = data_i.sort_values(by=['code'], ascending=[True]) code_list = data_i['code'].tolist() block_list.iat[i, 4] = len(code_list) block_list.iat[i, 5] = str(code_list) return block_list if __name__ == '__main__': # blocks = ['gn', 'fg', 'zs'] # for item in blocks: # print(get_block_file(item)) # print(get_block_tdxzs3('hy')) # print(get_stock_tdxhy()) print(hy_block('hy')) |
|
|
