近两年年化是177.6%，wxpython+backtrader+quantstats的智能投研平台（系统源码+策略下载）

原创内容第852篇，专注智能量化投资、个人成长与财富自由。

明天我们会如期发布aitrader v6.0.1的源代码：

带了4个策略过来，其余会陆续迁移：

import platformimport subprocess
import wximport wx.html2 as webviewimport osfrom datetime import datetimeimport threadingimport timeimport wx.lib.newevent as neimport wx.adv
from backtrader_extends.engine import run_taskfrom config import DATA_REPORTS, DATA_TASKSfrom core.backtrader_extends.task import local_tasksfrom gui.backtest_visualizer import BacktestVisualizer
# 自定义事件，用于线程与主线程通信BacktestProgressEvent, EVT_BACKTEST_PROGRESS = ne.NewEvent()BacktestCompleteEvent, EVT_BACKTEST_COMPLETE = ne.NewEvent()
class BacktestThread(threading.Thread):    """回测线程"""    def __init__(self, panel, strategy, start_date, end_date, benchmark):        super().__init__()        self.panel = panel        self.strategy = strategy        self.start_date = start_date        self.end_date = end_date        self.benchmark = benchmark        self._stop_event = threading.Event()
    def run(self):        """执行回测逻辑"""        try:            task = local_tasks[self.strategy]            task.benchmark = self.benchmark            task.date.start_date = self.start_date            task.date.end_date = self.end_date            returns,positions, transactions = run_task(task)

            # 发送完成事件            wx.PostEvent(self.panel, BacktestCompleteEvent(strategy=self.strategy))
        except Exception as e:            wx.CallAfter(wx.MessageBox, f"回测出错: {str(e)}", "错误", wx.OK|wx.ICON_ERROR)
    def stop(self):        """停止回测"""        self._stop_event.set()
class BacktestPanel(wx.Panel):    def __init__(self, parent):        super().__init__(parent)
        # 当前回测线程        self.backtest_thread = None
        # 主布局        self.main_sizer = wx.BoxSizer(wx.HORIZONTAL)
        # 左侧面板 - 回测条件        self.left_panel = wx.Panel(self)        left_sizer = wx.BoxSizer(wx.VERTICAL)
        # 策略选择器        strategy_box = wx.StaticBox(self.left_panel, label="策略选择")        strategy_sizer = wx.StaticBoxSizer(strategy_box, wx.VERTICAL)        self.strategy_choice = wx.Choice(strategy_box, choices=list(local_tasks.keys()))        # 默认选择第一项（索引为0）

        # 绑定策略选择事件        self.strategy_choice.Bind(wx.EVT_CHOICE, self.on_strategy_changed)
        strategy_sizer.Add(self.strategy_choice, 0, wx.EXPAND|wx.ALL, 5)
        # 策略操作按钮        btn_sizer = wx.BoxSizer(wx.HORIZONTAL)        self.open_strategy_btn = wx.Button(strategy_box, label="策略目录")        self.open_strategy_btn.Bind(wx.EVT_BUTTON, self.on_open_strategy)        btn_sizer.Add(self.open_strategy_btn, 1, wx.EXPAND|wx.RIGHT, 5)
        self.edit_strategy_btn = wx.Button(strategy_box, label="编辑策略")        btn_sizer.Add(self.edit_strategy_btn, 1, wx.EXPAND)        self.edit_strategy_btn.Bind(wx.EVT_BUTTON,self.on_edit_strategy)        strategy_sizer.Add(btn_sizer, 0, wx.EXPAND|wx.ALL, 5)
        # 日期选择        date_box = wx.StaticBox(self.left_panel, label="回测日期范围")        date_sizer = wx.StaticBoxSizer(date_box, wx.VERTICAL)
        start_date_sizer = wx.BoxSizer(wx.HORIZONTAL)        start_date_sizer.Add(wx.StaticText(date_box, label="开始日期:"), 0, wx.ALIGN_CENTER_VERTICAL|wx.RIGHT, 5)        self.start_date = wx.adv.DatePickerCtrl(date_box, style=wx.adv.DP_DROPDOWN|wx.adv.DP_SHOWCENTURY)        start_date_sizer.Add(self.start_date, 1, wx.EXPAND)        date_sizer.Add(start_date_sizer, 0, wx.EXPAND|wx.ALL, 5)
        end_date_sizer = wx.BoxSizer(wx.HORIZONTAL)        end_date_sizer.Add(wx.StaticText(date_box, label="结束日期:"), 0, wx.ALIGN_CENTER_VERTICAL|wx.RIGHT, 5)        self.end_date = wx.adv.DatePickerCtrl(date_box, style=wx.adv.DP_DROPDOWN|wx.adv.DP_SHOWCENTURY)        end_date_sizer.Add(self.end_date, 1, wx.EXPAND)        date_sizer.Add(end_date_sizer, 0, wx.EXPAND|wx.ALL, 5)
        # Benchmark选择        benchmark_box = wx.StaticBox(self.left_panel, label="基准指标")        benchmark_sizer = wx.StaticBoxSizer(benchmark_box, wx.VERTICAL)
        # 基准代码映射表        self.benchmark_map = {            "沪深300": "510300.SH",            "中证500": "510500.SH",            "创业板指": "159915.SZ"        }
        self.benchmark_choice = wx.Choice(benchmark_box, choices=list(self.benchmark_map.keys()))        #self.benchmark_choice.Bind(wx.EVT_CHOICE, self.on_choice_selected)        benchmark_sizer.Add(self.benchmark_choice, 0, wx.EXPAND|wx.ALL, 5)        self.benchmark_choice.SetSelection(0)  # 默认选中第一个选项
        # 回测进度条        self.progress = wx.Gauge(self.left_panel, range=100)
        # 启动回测按钮        self.start_btn = wx.Button(self.left_panel, label="启动回测")        self.start_btn.Bind(wx.EVT_BUTTON, self.on_start_backtest)
        # 停止回测按钮        self.stop_btn = wx.Button(self.left_panel, label="停止回测")        self.stop_btn.Bind(wx.EVT_BUTTON, self.on_stop_backtest)        self.stop_btn.Disable()
        # 按钮布局        btn_sizer = wx.BoxSizer(wx.HORIZONTAL)        btn_sizer.Add(self.start_btn, 1, wx.EXPAND|wx.RIGHT, 5)        btn_sizer.Add(self.stop_btn, 1, wx.EXPAND)
        # 添加左侧组件        left_sizer.Add(strategy_sizer, 0, wx.EXPAND|wx.ALL, 5)        left_sizer.Add(date_sizer, 0, wx.EXPAND|wx.ALL, 5)        left_sizer.Add(benchmark_sizer, 0, wx.EXPAND|wx.ALL, 5)        left_sizer.Add(self.progress, 0, wx.EXPAND|wx.ALL, 5)        left_sizer.Add(btn_sizer, 0, wx.EXPAND|wx.ALL, 5)
        self.left_panel.SetSizer(left_sizer)
        # 右侧面板 - WebView显示结果        self.right_panel = wx.Panel(self)        right_sizer = wx.BoxSizer(wx.VERTICAL)
        # 创建WebView        self.webview = webview.WebView.New(self.right_panel)        right_sizer.Add(self.webview, 1, wx.EXPAND)
        self.right_panel.SetSizer(right_sizer)
        # 设置主布局比例        self.main_sizer.Add(self.left_panel, 1, wx.EXPAND)        self.main_sizer.Add(self.right_panel, 4, wx.EXPAND)        self.SetSizer(self.main_sizer)
        # 获取当前日期        today = datetime.today()
        # 设置开始日期（1年前）        start_date = wx.DateTime()        start_date.Set(            day=today.day,            month=today.month - 1,  # wx.DateTime 月份是 0-11            year=today.year - 1        )
        # 设置结束日期（当前日期）        end_date = wx.DateTime()        end_date.Set(            day=today.day,            month=today.month - 1,  # wx.DateTime 月份是 0-11            year=today.year        )
        # 应用到控件        self.start_date.SetValue(start_date)        self.end_date.SetValue(end_date)
        # 绑定自定义事件        self.Bind(EVT_BACKTEST_PROGRESS, self.on_backtest_progress)        self.Bind(EVT_BACKTEST_COMPLETE, self.on_backtest_complete)
        # 默认加载策略的HTML        self.load_default_html()
        if self.strategy_choice.GetCount() > 0:  # 确保有选项            self.strategy_choice.SetSelection(0)            self.update_dates_from_task(self.strategy_choice.GetStringSelection())
    def on_strategy_changed(self, event):        """当策略选择变化时更新日期"""        selected_strategy = self.strategy_choice.GetStringSelection()        self.load_default_html()        self.update_dates_from_task(selected_strategy)        event.Skip()
    def on_edit_strategy(self, event):        strategy_name = self.strategy_choice.GetStringSelection()        if not strategy_name:            strategy_name = self.strategy_choice.GetStrings()[0]
        file_path = f"{strategy_name}.toml"        file_path = str(DATA_TASKS.joinpath(file_path).resolve())
        self.open_file_with_default_editor(file_path)
    def open_file_with_default_editor(self, file_path):        """用系统默认编辑器打开文件"""        if not os.path.exists(file_path):            wx.MessageBox(f"文件不存在: {file_path}", "错误", wx.OK | wx.ICON_ERROR)            return
        try:            if platform.system() == "Windows":                os.startfile(file_path)            elif platform.system() == "Darwin":  # macOS                subprocess.run(["open", file_path])            else:  # Linux                subprocess.run(["xdg-open", file_path])        except Exception as e:            wx.MessageBox(f"无法打开文件: {str(e)}", "错误", wx.OK | wx.ICON_ERROR)

    def update_dates_from_task(self, strategy_name):        """根据策略名称更新日期控件"""        if strategy_name in local_tasks.keys():            task = local_tasks[strategy_name]
            # 解析开始日期            start_date_str = task.date.start_date  # 格式: "20100101"            start_date = datetime.strptime(start_date_str, "%Y%m%d")            wx_start = wx.DateTime()            wx_start.Set(                day=start_date.day,                month=start_date.month - 1,  # wx月份是0-11                year=start_date.year            )            self.start_date.SetValue(wx_start)
            # 解析结束日期            end_date_str = task.date.end_date  # 格式: "20201231"            if end_date_str == '':                end_date_str = datetime.now().strftime('%Y%m%d')            end_date = datetime.strptime(end_date_str, "%Y%m%d")            wx_end = wx.DateTime()            wx_end.Set(                day=end_date.day,                month=end_date.month - 1,  # wx月份是0-11                year=end_date.year            )            self.end_date.SetValue(wx_end)
    def load_default_html(self):        """加载默认的策略HTML文件"""        strategy_name = self.strategy_choice.GetStringSelection()        if not strategy_name:            strategy_name = self.strategy_choice.GetStrings()[0]
        html_file = f"{strategy_name}.html"        html_file = str(DATA_REPORTS.joinpath(html_file).resolve())
        #print(html_file)
        if os.path.exists(html_file):            self.webview.LoadURL(f"file://{os.path.abspath(html_file)}")        else:            # 如果文件不存在，显示空白页面或默认页面            self.webview.SetPage("<html><body><h1>回测结果将显示在这里</h1></body></html>", "")
    def on_open_strategy(self, event):        """打开策略按钮事件处理"""        strategy = self.strategy_choice.GetStringSelection()        if not strategy:            wx.MessageBox("请先选择一个策略", "提示", wx.OK|wx.ICON_INFORMATION)            return
        # 这里应该是打开策略文件的逻辑        from config import DATA_TASKS
        # 检查目录是否存在        target_dir = str(DATA_TASKS.resolve())        if not os.path.exists(str(DATA_TASKS.resolve())):            wx.MessageBox(f"目录不存在: {target_dir}", "错误", wx.OK | wx.ICON_ERROR)            return
        # 根据不同操作系统打开目录        system = platform.system()        try:            if system == "Windows":                os.startfile(target_dir)            elif system == "Darwin":  # macOS                os.system(f'open "{target_dir}"')            else:  # Linux                os.system(f'xdg-open "{target_dir}"')        except Exception as e:            wx.MessageBox(f"无法打开目录: {str(e)}", "错误", wx.OK | wx.ICON_ERROR)
        # 例如：打开策略代码文件或策略配置对话框        #wx.MessageBox(f"策略目录: {strategy}", "提示", wx.OK|wx.ICON_INFORMATION)
    def on_start_backtest(self, event):        """启动回测按钮事件处理"""        if self.backtest_thread and self.backtest_thread.is_alive():            wx.MessageBox("已有回测正在运行", "提示", wx.OK|wx.ICON_INFORMATION)            return
        # 获取选择的参数        strategy = self.strategy_choice.GetStringSelection()        if not strategy:            wx.MessageBox("请选择一个策略", "错误", wx.OK|wx.ICON_ERROR)            return
        start_date = self.start_date.GetValue().FormatISODate().replace('-','')        end_date = self.end_date.GetValue().FormatISODate().replace('-','')        benchmark = self.benchmark_map[self.benchmark_choice.GetStringSelection()]
        # 重置进度条        self.progress.SetValue(0)        self.start_btn.Disable()        self.stop_btn.Enable()
        # 创建并启动回测线程        self.backtest_thread = BacktestThread(self, strategy, start_date, end_date, benchmark)        self.backtest_thread.start()
    def on_stop_backtest(self, event):        """停止回测按钮事件处理"""        if self.backtest_thread and self.backtest_thread.is_alive():            self.backtest_thread.stop()            self.stop_btn.Disable()            wx.MessageBox("回测已停止", "提示", wx.OK|wx.ICON_INFORMATION)
    def on_backtest_progress(self, event):        """回测进度更新事件处理"""        self.progress.SetValue(event.progress)
    def on_backtest_complete(self, event):        """回测完成事件处理"""        self.progress.SetValue(100)        self.start_btn.Enable()        self.stop_btn.Disable()
        # 加载回测结果        html_file = f"{event.strategy}.html"        if os.path.exists(html_file):            self.webview.LoadURL(f"file://{os.path.abspath(html_file)}")        else:            wx.MessageBox(f"回测完成，但未找到结果文件: {html_file}", "提示", wx.OK|wx.ICON_INFORMATION)
# 使用示例if __name__ == "__main__":    app = wx.App(False)    frame = wx.Frame(None, title="量化回测工具", size=(1000, 600))    panel = BacktestPanel(frame)    frame.Show()    app.MainLoop()

吾日三省吾身

其实都知道，焦虑终将会过去。

事缓则圆，时间将带走一切，如此而已。

一个个具体的所谓“事件”，其实是解决不完了，无穷无尽，没完没了。

如果什么都不做，老了会不会后悔，说年轻时为什么不勇敢一点点。

允许一切发生。

可是，真心去做了发现，又懊悔自寻烦恼。

这需要决心和勇气，以及正常的方法论。

经历过周期最长的，也经历过强度最大的。

战胜过，多数是配合去完成。

当然仍然没有习惯，也无法习惯。

真实的情况，就是什么也没有发生，什么也不会发生。一切缘起于一个想象构建的虚幻。

但足矣压倒一切。

中岁颇好道，晚家南山陲。

兴来每独往，胜事空自知。

行到水穷处，坐看云起时。

偶然值林叟，谈笑无还期。

代码和数据下载：AI量化实验室——2025量化投资的星辰大海

扩展  •  历史文章   

EarnMore（赚得更多）基于RL的投资组合管理框架：一致的股票表示，可定制股票池管理。（附论文+代码）

年化收益200%+的策略集 | 实时板块资金热力图 「aitrader 5.0系统代码发布」

年化19.66%，回撤12%的稳健策略|manus的启发：基于大模型多智能体的智能投研系统（python代码+数据）

年化30.24%，最大回撤19%，综合动量多因子评分策略再升级（python代码+数据）

年化18%-39.3%的策略集 | backtrader通过xtquant连接qmt实战

stock-pandas，一个易用的talib的替代开源库。