[量化投资-学习笔记013]Python+TDengine从零开始搭建量化分析平台-策略回测进阶

在上一章节《策略回测》中采用轮询的方式对整个股票池进行了回测。虽然功能已经实现，但是效率确实不高。而且生成的结果也不方便统计。

本次在上一章的基础上进行修改，实现两方面的改进：

1. 采用多线程的处理方式，提高处理效率
2. 将计算结果存入 TDengine 数据库，方便后期统计。

目录

数据库初始化

数据库建模

• 数据库 backtrade 专门存储回测数据。
• 表 records 用于存储每次回测内容和时间，相当于回测数据的一个索引。
• 超级表 btdata 用于存储回测数据。

createdatabase backtrade vgroups 2 stt_trigger 1 wal_retention_period 0;use backtrade;createtable records(ts timestamp,record nchar(200));create stable btdata(ts timestamp,profit double) tags (`fcode`binary(6),`fname`nchar(20));

创建子表

根据股票代码创建子表，因为之前

trade_data_c

已经有类似的表名称，可以借用一部分。

导出表名和标签，并构建建表语句。

taos -s"select distinct tbname,fcode,fname from trade_data_c.tdata"|grep"t_"|sed's/ //g'|awk-F'|''{print "create table " $1 " using btdata tags(\047"$2"\047,\047"$3"\047);"}'> create.sql

导入建表语句

use backtrade;
source create.sql

函数简介

多进程/线程处理

defthread_func(df_code,tnum,list_num):
    bi = tnum*list_num
    ei = bi+list_num
    if tnum <(threadNum-1):
        df = df_code.iloc[bi:ei,:]else:
        df = df_code.iloc[bi:len(df_code),:]
    df_profit = loop_bt(df) 
    write_td(df_profit)defmulti_thread(df,wmethod):print('')print('--------------------begin------------------')
    threads =[]iflen(df)< threadNum:
        df_profit = loop_bt(df)
        write_td(df_profit)else:
        listnum =int(len(df)/threadNum)if wmethod =='process':for tnum inrange(threadNum):  
                t = multiprocessing.Process(target=thread_func,args=(df,tnum,listnum))
                threads.append(t)else:for tnum inrange(threadNum):             
                t = threading.Thread(target=thread_func,args=(df,tnum,listnum))
                threads.append(t)for t in threads:  
            t.start()for t in threads:  
            t.join()print('--------------------end------------------')print('')

结果写入TDengine

defexport_sql(dbname,tbname,idata):
    exsql ='insert into 'for index,row in idata.iterrows():
        exsql = exsql + dbname +'.t_'+ idata.loc[index,"fcode"]+' values ('
        ts = current_time
        exsql = exsql +str(ts)+','
        exsql = exsql +str(idata.loc[index,"PRate"])+') '
    exsql = exsql +';'return exsql

# 将数据写入TDenginedefwrite_td(df):
    sql_trade = export_sql('backtrade','btdata',df)
    rt = get_request(sql_trade)return rt

程序主体

pfile =open(file,'a',encoding='UTF-8')
sql_code =f"select distinct tbname,fcode,fname from {db_name}.{table_name} order by fcode ;"
result = get_request(sql_code)
write_record('MACD-12_26')
df_code = pd.DataFrame(result, columns=['tbname','fcode','fname'])
multi_thread(df_code,'thread')
pfile.close()

结果写入数据库以后，可以方便的进行统计，比如：统计收益率最高和最低的前十只股票：

taos>selecttop(profit,10),fcode,fname from btdata where ts='2023-11-12 16:38:49.940'orderby profit;top(profit,10)|  fcode   |             fname              |========================================================================950.682999999999993|00353| 能源国际投资                   |993.833999999999946|08428| 国茂控股                       |1248.107999999999947|03313| 雅高控股                       |1265.869999999999891|08412| 新爱德集团                     |1266.903000000000020|09918| 丽年国际                       |1469.795000000000073|00451| 协鑫新能源                     |1746.269999999999982|01007| 龙辉国际控股                   |1828.528999999999996|00076| 谊砾控股                       |1944.867999999999938|00804| 鼎石资本                       |3372.123000000000047|00673| 中国卫生集团                   |
Query OK,10row(s)inset(0.018392s)

taos>select bottom(profit,10),fcode,fname from btdata where ts='2023-11-12 16:38:49.940'orderby profit;
     bottom(profit,10)|  fcode   |             fname              |========================================================================-95.489000000000004|08516| 广骏集团控股                   |-91.319000000000003|06878| 鼎丰集团汽车                   |-89.888000000000005|00379| 恒嘉融资租赁                   |-89.765000000000001|08305| 棠记控股                       |-87.015000000000001|08072| 罗马元宇宙集团                 |-86.963999999999999|01996| 弘阳地产                       |-84.201999999999998|02195| 盈汇企业控股                   |-83.519999999999996|08526| 荣丰集团亚洲                   |-82.787999999999997|08052| 陆庆娱乐                       |-80.664000000000001|01269| 首控集团                       |
Query OK,10row(s)inset(0.020202s)

MACD交易策略回测程序说明

该程序用于进行MACD交易策略的回测。给定一组股票数据，程序会根据MACD指标的金叉和死叉信号进行买入和卖出操作，并计算收益率。

程序功能

• 通过RESTful请求获取股票数据
• 计算MACD指标
• 判断MACD金叉和死叉
• 计算收益率
• 将计算得到的收益率数据写入TDengine数据库

程序依赖

该程序依赖以下库：

• requests：用于发送RESTful请求
• json：用于解析RESTful请求返回的结果
• matplotlib：用于结果可视化
• pandas：用于数据处理和分析
• numpy：用于数值计算
• multiprocessing：用于多进程处理
• threading：用于多线程处理
• time：用于获取当前时间

使用方法

1. 安装所需的Python库：在控制台中运行以下命令安装依赖库：

pip install requests json matplotlib pandas numpy multiprocessing

1. 下载程序文件：将程序代码保存为一个Python脚本（例如macd_all_code.py）。
2. 配置数据库信息：根据实际情况，修改代码中的主机地址、端口、用户名、密码、数据库名称等。
3. 运行程序：在控制台中运行以下命令来执行程序：

python macd_all_code.py

程序细节

程序主要有以下几个部分：

1. 定义和初始化TDengine数据库的连接信息，包括主机地址、端口、用户名、密码、数据库名称等。
2. 实现了一个get_request(sql)函数，用于发送RESTful请求并获取查询结果。它基于提供的主机地址、端口、用户名、密码构建URL，并发送包含查询SQL的POST请求。
3. 定义了两个辅助函数：calculate_ema(prices, period)和calculate_macd(prices, short_period, long_period, signal_period)。前者用于计算指数移动平均值（EMA），后者用于计算MACD指标的值。
4. check_macd(macd_line, signal_line)函数用于判断MACD金叉和死叉的发生。它遍历MACD线和信号线的历史数据，并根据交叉点的情况进行判断。
5. calculate_profit(prices, signals)函数用于计算收益率。它根据买卖信号和股票价格，通过模拟真实的交易操作，计算出回测期间的收益情况。
6. backtrader(fcode)函数是实现回测策略的核心部分。它接收一个股票代码作为参数，使用get_request()函数获取该股票的历史价格数据，并调用辅助函数计算MACD指标。然后，根据MACD指标的金叉和死叉信号，计算回测期间的收益率。
7. 程序使用get_request()函数查询数据库，获取所有股票代码和名称。然后，它使用多线程或多进程的方式，遍历股票代码并进行回测策略。最后，将计算得到的收益率结果写入到TDengine数据库中。