给AI量化平台提供GUI界面（代码下载）

近期文章更新比较频繁，得到一部分同学的认可，1群加满了，后续加2群，内容都是一样的。当然后续看如何更高效连接在一起，需要看工作量和相应的机制了。

星友们有券商或私募出身，也有算法、python程序员等。有人特别懂金融，但不懂AI量化，也有人反之。这就导致众口难调，当然，我一直都沿着自己的主线在走。

这里先总结和回顾一下初心，为什么做这个“AI量化实验室”，以及知识星球社群？

AI+金融肯定是未来趋势，能带来价值，而且是为数不多的，还是“个人英雄主义”的领域。你要说自己搞大模型，可能环境都准备不了，你必须加入一个实力雄厚的团队。以前程序员是一个可以自己搞个人软件的群体，现在基本不可能，免费的模式，都是风投在支撑，体验不仅要好，服务还要极致，这留给个人的机会就不多。

量化投资是离财务自由最近的领域。之前听说很多大明星老年破产，中大奖的人过几年又赤贫的故事。这是财富管理的问题，从量化的视角望过去，如果你的财富超过一定阈值，你不作的话，是可以确保一辈子保持生活品质而自由自在生活。这个值没你想象中的高，哪怕一线的北大广深，我给过两个值，一个500万，第二个是5000万。大家第一反应是通胀，是生病了怎么办之类的问题，大家没有想到的是，这个本金要“动”起来，它为你工作，钱生钱。长期年化10%的收益率。500*10%=50万的年被动收入。第二个大家自己算。

问题就会回到10%长期年化怎么办？——要我简单回答就是大类资产配置+定期再平衡。——当然这个收入是不均衡的，尤其A股牛短熊长，可能3-5年来一次牛市，然后一直振荡横盘也未可知。但大类资产配置就同分散可以降低风险一样，是投资界少有的“免费的午餐”。

大家认同的10%之后，又问我，本金哪里来，这可为难住我的。因为这个因人而异，有人就靠本金，极简生活，也可以过上相对务自由的生活。

向来不建议大家指着“投机”高收益来积累本金，因为可能会导致本金永久损失，这种风险意识一定要在进入市场时牢记心中。

但不排除这种可能性，这就需要你的认知比别人高，工具比别人先进。

用了市场上一堆工具和平台，没有特别合适的。比如传统散户，对于炒股软件K线图一看一整天，看看F10花里胡哨的事后解说，低效、无效。怎么办呢，投研平台不应该是这样的，而且现在AI技术一日千里，理应可以为以信息主要决策依据的金融投资业。本意是开发一个系统自己用，后来交流时，发现不少同学也感兴趣，索性就搞个社群，把源代码在社群里公开，供大家讨论，学习。

综上，回到我们平台研发的思路上。不是一个大而全的商业软件，就是投自己的资，理自己的财的辅助工具。星球里我说“面向盈利去开发策略”。

策略没有所谓圣杯，它需要持续的跟踪，迭代。授人以鱼不如授人以渔。但修改代码还是有门槛的，而且金融投资更是一个槛，如何把这个门槛降下来。所以我想到“积木式”策略开发，然后又使用配置文件来生成策略。

还不够，需要一个GUI，这就变成软件操作，普通人也可以使用，研究自己的策略了。

我的选型是本地桌面软件，web版本、三方平台主要是太慢，不灵活，影响投研效率。当然之于高手这里有策略安全的问题。我是代码全开源，且是本地软件。我开发一个C/S结构平台，从服务器取数，策略构建均在本地。

目前技术选型，GUI的框架是wxpython，绘图的框架是bokeh。

main.py是统一入口，gui的代码在gui目录下：

import wx
from gui.mainframe import MainFrame
import subprocess

# '--server.port', '5002'
# p_restart = subprocess.Popen(['streamlit', 'run', 'main_streamlit.py'])
if __name__ == '__main__':
    app = wx.App()
    frm = MainFrame(None, title='AI智能量化投研平台')
    frm.Show()
    app.MainLoop()
    # p_restart.kill()

wxpython的界面是高度模块化的，使用winform builder反倒没有这么便利。

self.m_notebook.AddPage(WidgetBacktest(self.m_notebook), "可视化AI量化策略构建系统", True)

功能计划：向导式策略搭建，回测，调参，因子分析。

代码如下：

import wx
from gui.proj_utils import ConfigLoader


class WidgetBacktest(wx.Panel):
    def __init__(self, parent):
        super(WidgetBacktest, self).__init__(parent, id=wx.ID_ANY)
        self.loader = ConfigLoader()

        self._init_ui()
        self._init_datas()

    def _init_ui(self):
        # panel内的垂直布局
        sizer = wx.BoxSizer(wx.VERTICAL)
        self.SetSizer(sizer)

        # 垂直布局里的水平布局
        pick_sizer = wx.BoxSizer(wx.HORIZONTAL)
        sizer.Add(pick_sizer)

        title = wx.StaticText(self, wx.ID_ANY, "请选择策略：")
        self.combo_proj = wx.ComboBox(self, wx.ID_ANY, style=wx.CB_READONLY)
        self.btn_backtest = wx.Button(self, wx.ID_ANY, "开始回测")
        self.btn_backtest.Bind(wx.EVT_BUTTON, self._on_click_backtest)

        pick_sizer.Add(title, 0, wx.ALL, 5)
        pick_sizer.Add(self.combo_proj, 0, wx.ALL, 5)
        pick_sizer.Add(self.btn_backtest, 0, wx.ALL, 5)

    def _init_datas(self):
        names = list(self.loader.proj_dict.keys())
        self.combo_proj.AppendItems(names)

    def _on_click_backtest(self, e):
        wx.MessageBox('回测' + self.combo_proj.GetValue())

线程安全与任务回测

回测一般是比较耗时的事情，而且在子线程里调用主线程里的GUI会引入异常，这时需要线程安全的操作。

这里核心使用了wxpython的线程安全机制：

wx.lib.delayedresult里的startWorker，加一个观察者模式：

from wx.lib.delayedresult import startWorker

def _on_click_backtest(self, e):
    name = self.combo_proj.GetValue()
    if name == '' or name not in self.loader.proj_dict.keys():
        wx.MessageBox('请选择策略。')
        return
    job_id = 100
    startWorker(self._end_backtest, self._start_backtest, jobID=job_id)

def _start_backtest(self):
    self.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:
        print('执行回测出错')


def _end_backtest(self, something):
    print('_end')
    self.btn_backtest.Enable(True)
    self.gauge.SetValue(0)

def update(self, data: dict):  # env观察者需要用的更新
    if 'msg_type' in data.keys():
        msg_type = data['msg_type']
        if msg_type == 'ON_BAR':
            wx.CallAfter(self.gauge.SetValue, data['step'])

def do_backtest(self, name):
    print('执行回测, Thread=', threading.currentThread().name)
    config = self.loader.proj_dict[name]
    e = Toml2Env(config)
    e.add_observers(self)
    e.backtest_loop()
    print('回测完成！')
    # e.show_results()