wxpython搭建AI量化平台框架（代码已提交）

今天继续，把回测引擎整合到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_backtest

from 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回测，主流程已经跑通了，后续要加上规则/模型的配置界面即可。

代码已经上传至星球，请大家请往下载。

原创文章第277篇，专注“个人成长与财富自由、世界运作的逻辑与投资“。

我们的目标是实盘，从ETF开始。

ETF的数据维度来对股票少，基本面的数据可以算，但比较麻烦；比之可转债更少。

应用机器学习比较容易陷入过拟合，那就没有办法了嘛？不会，简单的，也可以是有效的。

动量轮动+分散=骑最快的马。

etfs = [
    '510300.SH',  # 沪深300ETF
    '159915.SZ',  # 创业板
    '518880.SH',  # 黄金ETF
    '513100.SH',  # 纳指ETF
    '159985.SZ',  # 豆柏ETF
    '511880.SH',  # 银华日利ETF
]

我选择了A股的大小盘，纳指，商品，黄金，货币这样的低相关性的资产进行动量轮动。

我的实盘逻辑：暂定10个策略组合，一个策略分配仓位10%，比如50万的本金，当前这个策略5万。

代码已经提交至星球：

dataloader作了升级——对齐交易日历，比如大类资产，不同国家交易日历不一样，在回测时可能会有日期缺失，我们使用reindex+ffill前向填充数据，确保数据可以正常计算：

def _reset_index(self, dfs: list):
    trade_calendar = []
    for df in dfs:
        trade_calendar.extend(list(df.index))
    trade_calendar = list(set(trade_calendar))
    trade_calendar.sort()

    dfs_reindex = []
    for df in dfs:
        df_new = df.reindex(trade_calendar, method='ffill')
        df_new['return_0'] = df_new['close'].pct_change()
        dfs_reindex.append(df_new)

    return dfs_reindex

我们是AI+量化的投资系统，我们也一直在关注前沿AI技术的进展。

今天介绍一个AutoML的框架——AutoGluon。

亚马逊开源的AutoML框架 – autogluon，只需要几行代码就可以轻松实现数据预处理、模型融合、择优参数以及模型选择等。autoGluon除了处理表格数据外，还可以处理图像和文本等多模态数据。

支持结构化数据、时间序列数据，这个还真挺适合量化投资的，所以我们必须来关注一下。关于模型实现、参数优化，如果不是资深的算法工程师，自己还未必能调得一手好参。

pip install autogluon

依赖的包不少，但是安装还是很顺利的。

from autogluon.tabular import TabularDataset, TabularPredictor

train_data = TabularDataset('train.csv')
test_data = TabularDataset('test.csv')

train_data.to_csv('train.csv')
test_data.to_csv('test.csv')

print(train_data)

predictor = TabularPredictor(label='class').fit(train_data=train_data)
predictions = predictor.predict(test_data)

如下代码，除了数据加载之外，就是直接指定label后使用fit直接训练。

按传统的机器学习，我们要对数据进行预处理，然后做特征工程，作数据集划分，选择适当的模型，然后对参数进行优化，最后自己对模型进行评估，这些都不需要了。

模型自动推断出这是一个“二分类”问题。

对数据自动进行预测处理：

对数据调用11个模型进行训练：

选择最优的模型是：集成学习。

代码和数据在这里，请前往星球下载：

原创文章第283篇，专注“个人成长与财富自由、世界运作的逻辑与投资“。

今天把gui的架子重新搭建了一个，还是确定使用xwpython。前面的文章有做过比较，就不细说了。

大致的框架如下：

从选股到模型到回测，到每日选股，到因子挖掘，都实现界面可操作，方便一些没有代码基础的同学。

功能没开放完成，周末赶赶进度，但大家可以先下载去看。

别上上述界面还不完整，代码量不小的。

import pandas as pd
import wx
import wx.adv
import wx.grid
from gui.widgets.datatable import DataTable
from gui.logic import g_logic

from .panel_backtest import PanelBacktest
from .panel_pickstock import PanelPickStock


class PanelAnalysis(wx.Panel):
    def __init__(self, parent):
        super(PanelAnalysis, self).__init__(parent, id=wx.ID_ANY)
        self._init_ui()
        self._init_data()

    def _init_data(self):
        return
        df = Logic().load_data_by_day('20230717')
        table = DataTable(df)
        self.grid_datas.SetTable(table, takeOwnership=True)
        self.grid_datas.AutoSizeColumns()

    def _init_ui(self):
        # 左右结构
        self.sizer = wx.BoxSizer(wx.HORIZONTAL)
        self.SetSizer(self.sizer)

        self._init_left_panel()
        self._init_right_panel()

        g_logic.reg_pickstock_panel(self.panel_pickstock)
        g_logic.reg_backtest_panel(self.panel_backtest)
        g_logic.init_data()


    def _init_left_panel(self):
        self.note_left = wx.Notebook(self, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize, wx.TAB_TRAVERSAL)
        self.sizer.Add(self.note_left, 1, wx.ALL | wx.EXPAND, 10)

        self.panel_pickstock = PanelPickStock(self.note_left, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize,
                                        wx.TAB_TRAVERSAL)
        self.note_left.AddPage(self.panel_pickstock, "择投设置", True)
        self.panel_model = wx.Panel(self.note_left, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize,
                                    wx.TAB_TRAVERSAL)
        self.note_left.AddPage(self.panel_model, "交易模型", False)
        self.panel_picktime = wx.Panel(self.note_left, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize,
                                       wx.TAB_TRAVERSAL)
        self.note_left.AddPage(self.panel_picktime, "大盘择时", False)

    def _init_right_panel(self):
        self.note_right = wx.Notebook(self, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize, wx.TAB_TRAVERSAL)
        # self.note_right.SetBackgroundColour(wx.RED)
        self.sizer.Add(self.note_right, 3, wx.ALL | wx.EXPAND, 10)

        self.panel_backtest = PanelBacktest(self.note_right, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize,
                                            wx.TAB_TRAVERSAL)

        self.note_right.AddPage(self.panel_backtest, "策略回测", True)

        self.panel_pick_day = wx.Panel(self.note_right, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize,
                                       wx.TAB_TRAVERSAL)
        self.note_right.AddPage(self.panel_pick_day, "每日选股", False)

        self.panel_rank = wx.Panel(self.note_right, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize,
                                   wx.TAB_TRAVERSAL)
        self.note_right.AddPage(self.panel_rank, "排名分析", False)



    def _init_pickstock(self):
        vbox = wx.BoxSizer(wx.VERTICAL)
        self.panel_pick_day.SetSizer(vbox)

        self.label_text = wx.StaticText(self.panel_pick_day, wx.ID_ANY, u"根据以上模型的选股设置， 在历史上任何一天选股。", wx.DefaultPosition,
                                        wx.DefaultSize, 0)
        self.label_text.Wrap(-1)

        vbox.Add(self.label_text, 0, wx.ALL, 5)

        row_sizer = wx.BoxSizer(wx.HORIZONTAL)

        self.label_time = wx.StaticText(self.panel_pick_day, wx.ID_ANY, u"选股日期", wx.DefaultPosition, wx.DefaultSize, 0)
        self.label_time.Wrap(-1)

        row_sizer.Add(self.label_time, 0, wx.ALL, 5)

        self.date_pick = wx.adv.DatePickerCtrl(self.panel_pick_day, wx.ID_ANY, wx.DefaultDateTime, wx.DefaultPosition,
                                               wx.DefaultSize, wx.adv.DP_DEFAULT)
        row_sizer.Add(self.date_pick, 0, wx.ALL, 5)

        vbox.Add(row_sizer, 1, wx.EXPAND, 5)

        self.btn_pickstock = wx.Button(self.panel_pick_day, wx.ID_ANY, u"开始选股", wx.DefaultPosition, wx.DefaultSize, 0)
        vbox.Add(self.btn_pickstock, 0, wx.ALL, 5)

        # Grid
        self.grid_datas = wx.grid.Grid(self.panel_pick_day, wx.ID_ANY, wx.DefaultPosition, wx.DefaultSize, 0)

        self.grid_datas.CreateGrid(5, 5)
        self.grid_datas.EnableEditing(False)
        self.grid_datas.EnableGridLines(True)
        self.grid_datas.EnableDragGridSize(False)
        self.grid_datas.SetMargins(0, 0)

        # Columns
        self.grid_datas.EnableDragColMove(False)
        self.grid_datas.EnableDragColSize(True)
        self.grid_datas.SetColLabelAlignment(wx.ALIGN_CENTER, wx.ALIGN_CENTER)

        # Rows
        self.grid_datas.EnableDragRowSize(True)
        self.grid_datas.SetRowLabelAlignment(wx.ALIGN_CENTER, wx.ALIGN_CENTER)

        # Label Appearance

        # Cell Defaults
        self.grid_datas.SetDefaultCellAlignment(wx.ALIGN_LEFT, wx.ALIGN_TOP)
        vbox.Add(self.grid_datas, 8, wx.ALL | wx.EXPAND, 5)

    # Virtual event handlers, override them in your derived class
    def on_btn_pickstock_clicked(self, event):
        wx.MessageBox('每日选股！')
        event.Skip()