在上一章节《策略回测》中采用轮询的方式对整个股票池进行了回测。虽然功能已经实现，但是效率确实不高。而且生成的结果也不方便统计。
本次在上一章的基础上进行修改，实现两方面的改进：
- 采用多线程的处理方式，提高处理效率
- 将计算结果存入 TDengine 数据库，方便后期统计。
数据库初始化
数据库建模
- 数据库
backtrade专门存储回测数据。 - 表
records用于存储每次回测内容和时间，相当于回测数据的一个索引。 - 超级表
btdata用于存储回测数据。
create database backtrade vgroups 2 stt_trigger 1 wal_retention_period 0;
use backtrade;
create table 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
函数简介
多进程/线程处理
def thread_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)
def multi_thread(df,wmethod):
print('')
print('--------------------begin------------------')
threads = []
if len(df) < threadNum:
df_profit = loop_bt(df)
write_td(df_profit)
else:
listnum = int(len(df)/threadNum)
if wmethod == 'process':
for tnum in range(threadNum):
t = multiprocessing.Process(target=thread_func,args=(df,tnum,listnum))
threads.append(t)
else:
for tnum in range(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
def export_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
# 将数据写入TDengine
def write_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> select top(profit,10),fcode,fname from btdata where ts='2023-11-12 16:38:49.940' order by 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, 10 row(s) in set (0.018392s)
taos> select bottom(profit,10),fcode,fname from btdata where ts='2023-11-12 16:38:49.940' order by 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, 10 row(s) in set (0.020202s)
MACD交易策略回测程序说明
该程序用于进行MACD交易策略的回测。给定一组股票数据，程序会根据MACD指标的金叉和死叉信号进行买入和卖出操作，并计算收益率。
程序功能
- 通过RESTful请求获取股票数据
- 计算MACD指标
- 判断MACD金叉和死叉
- 计算收益率
- 将计算得到的收益率数据写入TDengine数据库
程序依赖
该程序依赖以下库：
- requests：用于发送RESTful请求
- json：用于解析RESTful请求返回的结果
- matplotlib：用于结果可视化
- pandas：用于数据处理和分析
- numpy：用于数值计算
- multiprocessing：用于多进程处理
- threading：用于多线程处理
- time：用于获取当前时间
使用方法
- 安装所需的Python库：在控制台中运行以下命令安装依赖库：
pip install requests json matplotlib pandas numpy multiprocessing
-
下载程序文件：将程序代码保存为一个Python脚本（例如
macd_all_code.py）。 -
配置数据库信息：根据实际情况，修改代码中的主机地址、端口、用户名、密码、数据库名称等。
-
运行程序：在控制台中运行以下命令来执行程序：
python macd_all_code.py
程序细节
程序主要有以下几个部分：
-
定义和初始化TDengine数据库的连接信息，包括主机地址、端口、用户名、密码、数据库名称等。
-
实现了一个
get_request(sql)函数，用于发送RESTful请求并获取查询结果。它基于提供的主机地址、端口、用户名、密码构建URL，并发送包含查询SQL的POST请求。 -
定义了两个辅助函数：
calculate_ema(prices, period)和calculate_macd(prices, short_period, long_period, signal_period)。前者用于计算指数移动平均值（EMA），后者用于计算MACD指标的值。 -
check_macd(macd_line, signal_line)函数用于判断MACD金叉和死叉的发生。它遍历MACD线和信号线的历史数据，并根据交叉点的情况进行判断。 -
calculate_profit(prices, signals)函数用于计算收益率。它根据买卖信号和股票价格，通过模拟真实的交易操作，计算出回测期间的收益情况。 -
backtrader(fcode)函数是实现回测策略的核心部分。它接收一个股票代码作为参数，使用get_request()函数获取该股票的历史价格数据，并调用辅助函数计算MACD指标。然后，根据MACD指标的金叉和死叉信号，计算回测期间的收益率。 -
程序使用
get_request()函数查询数据库，获取所有股票代码和名称。然后，它使用多线程或多进程的方式，遍历股票代码并进行回测策略。最后，将计算得到的收益率结果写入到TDengine数据库中。