原创文章第284篇，专注“个人成长与财富自由、世界运作的逻辑与投资"。
今天继续，把回测引擎整合到gui中。
这里最大的难点在于，由于回测过程需要的时间比较耗时，不能直接跑在gui的主线程中，需要起子线程。不过子线程的状态，要在gui里显示出来，直接跨线程操作界面会出问题。
我们使用wxpython内置的startWorker启动一个后台工作线程。
job_id = 100
startWorker(self._end_backtest, self._start_backtest, jobID=job_id)
def _append_logs(self, message):
self.panel_backtest.text_logs.AppendText(message + '\r\n')
def _start_backtest(self):
self.panel_backtest.btn_backtest.Enable(False)
try:
name = self.combo_proj.GetValue()
if name not in self.loader.proj_dict.keys():
wx.MessageBox("策略不存在")
return
self.do_backtest(name)
except:
self._append_logs('执行回测出错')
def _end_backtest(self, something):
self._append_logs('回测成功完成！')
self.panel_backtest.btn_backtest.Enable(True)
self.panel_backtest.gauge_backtest.SetValue(0)
我们看一下一个典型的策略配置文件，toml格式：
name = '静待花开的聚宝盘'
[data]
start_date = '20100101'
end_date = ''
symbols = [
'511220.SH', #城投债
'512010.SH', # 医药
'518880.SH', #黄金
'163415.SZ', #兴全商业
'159928.SZ', # 消费
'161903.SZ', # 万家行业优选
'513100.SH' # 纳指
]
fields = ['roc(close,20)']
names = ['roc_20']
benchmarks=['000300.SH']
[[algos]]
name = 'RunDays'# 运行周期与再平衡
days=5
[[algos]]
name = 'SelectBySignal'
buy_rules=['ind(roc_20)>0.02']
sell_rules=['ind(roc_20)<-0.02']
[[algos]]
name = 'WeightEqually'
toml格式与dict字典是等价的。
全局的消息转发：
class GlobalHandler:
def __init__(self):
self.observers_fns = []
def add_observer_fn(self, fn):
self.observers_fns.append(fn)
def notify(self, data: dict):
for o in self.observers_fns:
o(data)
g = GlobalHandler()
def my_logger_notify(data):
g.notify({'msg_type': 'LOGGER', 'data': data})
from loguru import logger
logger.add(my_logger_notify)
注意最后这两句，loguru可以转logger的日志传给GlobalHandler。
而界面“观察”全局处理器，进而把logger显示在gui上。
进度条更新：
同样的使用event_handler，影响on_bar事件即可，这样代码的耦合最低。
def _event_handler(self, data: dict):
# print('logger......')
if 'msg_type' in data.keys() and data['msg_type'] == 'LOGGER':
self._append_logs(data['data'])
if data['msg_type'] == 'ON_BAR':
self.panel_backtest.gauge_backtest.SetValue(data['step'])
我们使用bokeh来显现回测结果可视化。
要特别注意一点是，在计算线程里，如果调用browser的SetPage是无效的。这里“浪费”了我1个小时，应该是在call_back里调用，然后更新回测结果：
_end_backtestfrom bokeh.plotting import figure, show
from bokeh.layouts import row
from bokeh.models import HoverTool
from bokeh.models import ColumnDataSource
import numpy as np
import pandas as pd
from datetime import datetime
from engine.datafeed.dataloader import Hdf5Dataloader
import pandas_bokeh
from bokeh.models.widgets import DataTable, TableColumn
from bokeh.models import ColumnDataSource
np.random.seed(55)
class BokehMgr:
def plot_line(self, df, y_col, **kwargs):
df.plot_bokeh(kind="line", y=[y_col, 'open'], **kwargs)
def _calc_metrics(self, portfolio_df):
returns = portfolio_df['market_value'].pct_change()
loader = Hdf5Dataloader(['399006.SZ'], start_date="20100101")
bench_df = loader.load()
returns.name = '策略'
# if self.benchmarks_df is not None and len(self.benchmarks_df):
benchmark_returns = bench_df.pivot_table(index='date', columns='symbol', values='return_0')
returns = pd.concat([returns, benchmark_returns], axis=1)
from empyrical import max_drawdown, sharpe_ratio, annual_return
returns.dropna(inplace=True)
print(annual_return(returns))
print(max_drawdown(returns))
from engine.performance import PerformanceUtils
df_ratio, df_corr = PerformanceUtils().calc_rates(returns)
print(df_ratio)
df_ratio['名称'] = df_ratio.index
return df_ratio, df_corr
def _show_table(self, df):
data_table = DataTable(
columns=[TableColumn(field=Ci, title=Ci) for Ci in df.columns],
source=ColumnDataSource(df),
height=300,
)
return data_table
def show(self, backtest_h5, plot=False, return_html=True):
with pd.HDFStore(backtest_h5) as s:
portfolio_df = s['portfolio_df']
orders_df = s['orders_df']
orders_df['date'] = orders_df['date'].apply(lambda x: x.strftime('%Y-%m-%d'))
portfolio_df['equity'] = (portfolio_df['market_value'].pct_change() + 1).cumprod()
df_metrics, df_corr = self._calc_metrics(portfolio_df=portfolio_df)
table_metrics = self._show_table(df_metrics)
table_orders = self._show_table(orders_df)
table_corr = self._show_table(df_corr)
# 创建散点图:
p_equity = portfolio_df.plot_bokeh.line(
y="equity",
# category="species",
title="投资万元波动图",
show_figure=False,
rangetool=False,
)
p_market_value = portfolio_df.plot_bokeh.line(
y="market_value",
title="投资组合市值",
show_figure=False,
# rangetool=True,
)
# Combine Table and Scatterplot via grid layout:
html = pandas_bokeh.plot_grid([[p_equity, table_metrics], [table_orders, table_corr]], show_plot=plot,
return_html=return_html)
return html
if __name__ == '__main__':
from engine.datafeed.dataloader import Hdf5Dataloader
from engine.config import etfs_indexes
symbols = etfs_indexes.values()
loader = Hdf5Dataloader(['000300.SH'], start_date="20100101")
fields = ["roc(close,20)", "shift(close, -5)/shift(open, -1) - 1",
"qcut(label_c, 10)"
]
names = ["roc_20", "label_c",
'label'
]
# df = loader.load(fields=fields, names=names)
# df.dropna(inplace=True)
# df.index = pd.to_datetime(df.index)
# df['equity'] = (1 + df['return_0']).cumprod()
from engine.config import DATA_RESULTS
html = BokehMgr().show(backtest_h5=DATA_RESULTS.joinpath('等权策略.h5').resolve(), plot=True, return_html=False)
# print(html)
通过gui回测，主流程已经跑通了，后续要加上规则/模型的配置界面即可。
代码已经上传至星球，请大家请往下载。
联系客服
