PyQt5_股票策略校验工具_升级版

相对 PyQt5_股票策略校验工具 博文，图形展示方面增加多行图展示，原本只有K线图，升级版工具可以动态添加多行指标图

股票策略有效与否的确认，需要在不同股票，不同时间段，运行对比，确认在哪些条件下策略是最有效的。在整个校验过程中，有可以实时运行并实时查看结果的工具对整个效率的提升起着至关重要的作用。本工具基于此做的开发。

本文以“卖出口诀 - 顶天立地，卖出要急”为例进行讲解。

目录

效果

策略代码

工具代码

工具使用

---

效果

策略代码

前置说明：

1. 必须以 excute_strategy 为方法名

2. 策略代码保存为.py文件，将该文件保存到一个目录下（自定义），运行工具后，手动选择这个目录

 指标计算的py文件 和 指标显示用的控件py文件 要放置在项目可以导入的位置

指标计算的代码，后续有新增的指标在该py文件后面继续追加就可。指标计算的代码请查看 该博文。

指标显示用的控件，可以自行编写，也可以在本人 PyQt5 栏目下找对应的控件。

def excute_strategy(base_data,data_dir):
    '''
    卖出口诀 - 顶天立地，卖出要急
    解析：
    1. 出现上长影K线（或长实体K线），同时放出了大成交量
    自定义：
    1. 上长影K线 =》上影线是实体1倍以上
    2. 长实体K线 =》 K线实体是昨收2%以上
    3. 大成交量 =》成交量是昨日的2倍以上
    4. 卖出时点 =》 形态出现后下一交易日
    5. 胜 =》 卖出后第三个交易日收盘价下跌，为胜
    只计算最近两年的数据
    :param base_data:股票代码与股票简称 键值对
    :param data_dir:股票日数据文件所在目录
    :return:
    '''
    import pandas as pd
    import numpy as np
    import talib,os
    from datetime import datetime
    from dateutil.relativedelta import relativedelta
    from tools import stock_factor_caculate
 
    def res_pre_two_year_first_day():
        pre_year_day = (datetime.now() - relativedelta(years=2)).strftime('%Y-%m-%d')
        return pre_year_day
    caculate_start_date_str = res_pre_two_year_first_day()
 
    dailydata_file_list = os.listdir(data_dir)
 
    total_count = 0
    total_win = 0
    check_count = 0
    list_list = []
    detail_map = {}
    factor_list = ['VOL']
    for item in dailydata_file_list:
        item_arr = item.split('.')
        ticker = item_arr[0]
        secName = base_data[ticker]
        file_path = data_dir + item
        df = pd.read_csv(file_path,encoding='utf-8')
        # 删除停牌的数据
        df = df.loc[df['openPrice'] > 0].copy()
        df['o_date'] = df['tradeDate']
        df['o_date'] = pd.to_datetime(df['o_date'])
        df = df.loc[df['o_date'] >= caculate_start_date_str].copy()
        # 保存未复权收盘价数据
        df['close'] = df['closePrice']
        # 计算前复权数据
        df['openPrice'] = df['openPrice'] * df['accumAdjFactor']
        df['closePrice'] = df['closePrice'] * df['accumAdjFactor']
        df['highestPrice'] = df['highestPrice'] * df['accumAdjFactor']
        df['lowestPrice'] = df['lowestPrice'] * df['accumAdjFactor']
 
        if len(df)<=0:
            continue
 
        # 开始计算
        df.reset_index(inplace=True)
        df['i_row'] = [i for i in range(len(df))]
        df['body_length'] = abs(df['closePrice']-df['openPrice'])
        df['up_shadow'] = 0
        df.loc[df['closePrice']>df['openPrice'],'up_shadow'] = df['highestPrice'] - df['closePrice']
        df.loc[df['closePrice']'openPrice'],'up_shadow'] = df['highestPrice'] - df['openPrice']
        df['median_body'] = 0
        df.loc[(df['body_length']/df['closePrice'].shift(1)>0.02) & (df['up_shadow']/df['body_length']>=1),'median_body'] = 1
        df['vol_yeah'] = 0
        df.loc[df['turnoverVol']/df['turnoverVol'].shift(1)>=2,'vol_yeah'] = 1
 
        df['three_chg'] = round(((df['close'].shift(-3) - df['close'])/df['close'])*100,4)
        df['three_after_close'] = df['close'].shift(-3)
 
        df['target_yeah'] = 0
        df.loc[(df['median_body']==1) & (df['vol_yeah']==1),'target_yeah'] = 1
 
        df_target = df.loc[df['target_yeah']==1].copy()
 
        node_count = 0
        node_win = 0
        duration_list = []
        table_list = []
        i_row_list = df_target['i_row'].values.tolist()
        for i,row0 in enumerate(i_row_list):
            row = row0 + 1
            if row >= len(df):
                continue
            date_str = df.iloc[row]['tradeDate']
            cur_close = df.iloc[row]['close']
            three_after_close = df.iloc[row]['three_after_close']
            three_chg = df.iloc[row]['three_chg']
 
            table_list.append([
                i,date_str,cur_close,three_after_close,three_chg
            ])
            duration_list.append([row-2,row+3])
            node_count += 1
            if three_chg<0:
                node_win +=1
            pass
 
        list_list.append({
            'ticker':ticker,
            'secName':secName,
            'count':node_count,
            'win':0 if node_count<=0 else round((node_win/node_count)*100,2)
        })
        detail_map[ticker] = {
            'table_list': table_list,
            'duration_list': duration_list
        }
 
        total_count += node_count
        total_win += node_win
        check_count += 1
        pass
    df = pd.DataFrame(list_list)
 
    results_data = {
        'check_count':check_count,
        'total_count':total_count,
        'total_win':0 if total_count<=0 else round((total_win/total_count)*100,2),
        'start_date_str':caculate_start_date_str,
        'df':df,
        'detail_map':detail_map,
        'factor_list':factor_list
    }
    return results_data

工具代码

需要导入的包、pyqtgraph日期横坐标控件、pyqtgraph蜡烛图控件、分页表格控件，这些代码请查看本栏目标题有“工具”字眼的博文

K线和结果图形显示控件

class PyQtGraphScrollKWidget(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.factor_widgets = []
        self.init_data()
        self.init_ui()
        pass
    def init_data(self):
        # https://www.sioe.cn/yingyong/yanse-rgb-16/
        # self.color_line = (30, 144, 255)
        self.color_line = (255, 255, 0)
        self.color_highligh = (220,20,60)
        # 0 幽灵的白色; 1 纯黄; 2 紫红色; 3 纯绿; 4 道奇蓝
        self.color_list = [(248, 248, 255), (255, 255, 0), (255, 0, 255), (0, 128, 0), (30, 144, 255)]
        self.main_fixed_target_list = []  # 主体固定曲线，不能被删除
        self.whole_df = None
        self.whole_header = None
        self.whole_pd_header = None
        self.current_whole_data = None
        self.current_whole_df = None
        self.duration_list = None
        self.current_highligh_duration = None
        self.current_highligh_duration_list = []
        self.factor_list = None
 
        self.factor_code_widgetname_map = {
            'VOL':'VOL_PlotWidget',
            'SAR':'SAR_PlotWidget'
        }
        pass
    def init_ui(self):
        self.whole_duration_label = QtWidgets.QLabel('左边界~右边界')
        pic_download_btn = QtWidgets.QPushButton('滚动截图')
        pic_download_btn.clicked.connect(self.pic_download_btn_clicked)
        layout_top = QtWidgets.QHBoxLayout()
        layout_top.addWidget(self.whole_duration_label)
        layout_top.addStretch(1)
        layout_top.addWidget(pic_download_btn)
 
        self.title_label = QtWidgets.QLabel('执行过程查看')
        self.title_label.setAlignment(Qt.AlignCenter)
        self.title_label.setStyleSheet('QLabel{font-size:18px;font-weight:bold}')
 
        # 滚动区域开始
        self.pw_layout = QtWidgets.QVBoxLayout()
        self.scroll_area = QtWidgets.QScrollArea()
        self.scroll_area.setWidgetResizable(True)
        # self.scroll_area.setVerticalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff)
 
        layout_right = QtWidgets.QVBoxLayout()
        layout_right.addWidget(self.title_label)
        layout_right.addLayout(layout_top)
        layout_right.addWidget(self.scroll_area)
        self.setLayout(layout_right)
        pass
 
    def set_data(self, data: Dict[str, Any]):
        title_str = data['title_str']
        whole_header = data['whole_header']
        whole_df = data['whole_df']
        whole_pd_header = data['whole_pd_header']
        duration_list = data['duration_list']
        factor_list = data['factor_list']
 
        self.whole_header = whole_header
        self.whole_df = whole_df
        self.whole_pd_header = whole_pd_header
        self.duration_list = duration_list
        self.factor_list = factor_list
 
        self.title_label.setText(title_str)
        self.whole_duration_label.setText(f"{self.whole_df.iloc[0]['tradeDate']}~{self.whole_df.iloc[-1]['tradeDate']}")
 
        self.current_whole_df = self.whole_df.copy()
        self.caculate_and_show_data()
        pass
 
    def caculate_and_show_data(self):
        df = self.current_whole_df.copy()
        df.reset_index(inplace=True)
        df['i_count'] = [i for i in range(len(df))]
        tradeDate_list = df['tradeDate'].values.tolist()
        x = range(len(df))
        xTick_show = []
        x_dur = math.ceil(len(df) / 20)
        for i in range(0, len(df), x_dur):
            xTick_show.append((i, tradeDate_list[i]))
        if len(df) % 20 != 0:
            xTick_show.append((len(df) - 1, tradeDate_list[-1]))
        candle_data = []
        for i, row in df.iterrows():
            candle_data.append(
                (row['i_count'], row['openPrice'], row['closePrice'], row['lowestPrice'], row['highestPrice']))
 
        self.current_whole_data = df.loc[:, self.whole_pd_header].values.tolist()
        # 开始配置显示的内容
        self.create_candle_widget()
        self.factor_widgets.clear()
 
        xax = self.pw.getAxis('bottom')
        xax.setTicks([xTick_show])
 
        candle_fixed_target = CandlestickItem(candle_data)
        self.main_fixed_target_list.append(candle_fixed_target)
        self.pw.addItem(candle_fixed_target)
 
        # 标记技术图形 start
        if len(self.duration_list)>0:
            for item in self.duration_list:
                signal_fiexed_target = pg.LinearRegionItem([item[0], item[1]],
                                                           movable=False, brush=(
                        self.color_line[0], self.color_line[1], self.color_line[2], 50))
                self.pw.addItem(signal_fiexed_target)
            pass
        # 标记技术图形 end
 
        self.vLine = pg.InfiniteLine(angle=90, movable=False)
        self.hLine = pg.InfiniteLine(angle=0, movable=False)
        self.label = pg.TextItem()
 
        self.pw.addItem(self.vLine, ignoreBounds=True)
        self.pw.addItem(self.hLine, ignoreBounds=True)
        self.pw.addItem(self.label, ignoreBounds=True)
 
        self.vb = self.pw.getViewBox()
        self.proxy = pg.SignalProxy(self.pw.scene().sigMouseMoved, rateLimit=60, slot=self.mouseMoved)
        self.pw.enableAutoRange()
 
        # 其他项
        for item in self.factor_list:
            item_widget = eval(self.factor_code_widgetname_map[item])()
            item_widget.setMinimumHeight(300)
            item_widget.set_data({'df':self.current_whole_df.copy()})
            item_widget.setXLink(self.pw)
            # 标记技术图形 start
            if len(self.duration_list) > 0:
                for item in self.duration_list:
                    signal_fiexed_target = pg.LinearRegionItem([item[0], item[1]],
                                                               movable=False, brush=(
                            self.color_line[0], self.color_line[1], self.color_line[2], 50))
                    item_widget.addItem(signal_fiexed_target)
                pass
            # 标记技术图形 end
            self.factor_widgets.append(item_widget)
        self.fill_pw_widget()
        pass
 
    def mouseMoved(self, evt):
        pos = evt[0]
        if self.pw.sceneBoundingRect().contains(pos):
            mousePoint = self.vb.mapSceneToView(pos)
            index = int(mousePoint.x())
            if index >= 0 and index < len(self.current_whole_data):
                target_data = self.current_whole_data[index]
                html_str = ''
                for i, item in enumerate(self.whole_header):
                    html_str += f"
{item}:{target_data[i]}"
                self.label.setHtml(html_str)
                self.label.setPos(mousePoint.x(), mousePoint.y())
            self.vLine.setPos(mousePoint.x())
            self.hLine.setPos(mousePoint.y())
        pass
 
    def mouseClicked(self, evt):
        pass
 
    def updateViews(self):
        pass
 
    def set_highligh_duration(self,dur_index:int):
        highligh_dur = self.duration_list[dur_index]
        self.pw.removeItem(self.current_highligh_duration)
        signal_fiexed_target = pg.LinearRegionItem([highligh_dur[0], highligh_dur[1]],
                                                   movable=False, brush=(
                self.color_highligh[0], self.color_highligh[1], self.color_highligh[2], 50))
        self.pw.addItem(signal_fiexed_target)
        self.current_highligh_duration = signal_fiexed_target
 
        if self.current_highligh_duration_list:
            for i,item in enumerate(self.current_highligh_duration_list):
                item_w = self.factor_widgets[i]
                item_w.removeItem(item)
            pass
        self.current_highligh_duration_list.clear()
        for item in self.factor_widgets:
            signal_fiexed_target0 = pg.LinearRegionItem([highligh_dur[0], highligh_dur[1]],
                                                       movable=False, brush=(
                    self.color_highligh[0], self.color_highligh[1], self.color_highligh[2], 50))
            item.addItem(signal_fiexed_target0)
            self.current_highligh_duration_list.append(signal_fiexed_target0)
            pass
        pass
 
    def fill_pw_widget(self):
        # 清空控件
        while self.pw_layout.count():
            item = self.pw_layout.takeAt(0)
            widget = item.widget()
            if widget is not None:
                widget.deleteLater()
                pass
            pass
        sc_child_widget = self.scroll_area.takeWidget()
        if sc_child_widget is not None:
            sc_child_widget.deleteLater()
        self.pw_layout.addWidget(self.pw)
        for item in self.factor_widgets:
            self.pw_layout.addWidget(item)
        one_sc_child_widget = QtWidgets.QWidget()
        one_sc_child_widget.setLayout(self.pw_layout)
        self.scroll_area.setWidget(one_sc_child_widget)
        pass
 
    def create_candle_widget(self):
        xax = RotateAxisItem(orientation='bottom')
        xax.setHeight(h=60)
        self.pw = pg.PlotWidget(axisItems={'bottom': xax})
        self.pw.setMinimumHeight(400)
        self.pw.setMouseEnabled(x=True, y=True)
        # self.pw.enableAutoRange(x=False,y=True)
        self.pw.setAutoVisible(x=False, y=True)
        pass
 
    def pic_download_btn_clicked(self):
        now_str = datetime.now().strftime('%Y%m%d%H%M%S')
        path,_ = QtWidgets.QFileDialog.getSaveFileName(
            self,
            '选择图片保存路径',
            f"pic_{now_str}",
            'JPG(*.jpg)'
        )
        if not path:
            return
 
        widget = self.scroll_area.widget()
        pix = widget.grab()
        pix.save(path)
        pass
    pass

结果查看控件

class PyQtGraphRunningWidget(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.init_data()
        self.init_ui()
        pass
    def init_data(self):
        self.pre_output_dir = './'
        self.results_output_dir = './strategy_check_output/'
        self.json_file_name_list = []
        self.dailydata_path: str = ''
        self.total_table_header: List = ['股票数量','总次数','总胜率']
        self.list_table_header: List = ['股票代码','简称','次数','胜率']
        self.list_pd_header: List = ['ticker','secName','count','win']
        self.detail_table_header: List = ['序号','日期','收盘价','三日后收盘价','涨跌幅']
        self.please_select_str = '---请选择---'
        self.num_sort_map = {
            '升序':True,
            '降序':False
        }
        self.list_table_df = None
        self.current_list_table_df = None
        self.detail_map = None
        self.factor_list = []
        pass
    def init_ui(self):
        tip_0 = QtWidgets.QLabel('股票日数据文件夹:')
        self.dailydata_dir_lineedit = QtWidgets.QLineEdit()
        self.dailydata_dir_lineedit.setReadOnly(True)
        dailydata_choice_btn = QtWidgets.QPushButton('选择文件夹')
        dailydata_choice_btn.clicked.connect(self.dailydata_choice_btn_clicked)
 
        tip_1 = QtWidgets.QLabel('Json结果文件选择:')
        self.json_combox = QtWidgets.QComboBox()
        self.json_combox.addItem(self.please_select_str)
        self.json_combox.currentIndexChanged.connect(self.json_combox_currentIndexChanged)
        refresh_json_btn = QtWidgets.QPushButton('刷新结果下拉表')
        refresh_json_btn.clicked.connect(self.refresh_json_btn_clicked)
 
        layout_top = QtWidgets.QGridLayout()
        layout_top.addWidget(tip_0,0,0,1,1)
        layout_top.addWidget(self.dailydata_dir_lineedit,0,1,1,3)
        layout_top.addWidget(dailydata_choice_btn,0,4,1,1)
        layout_top.addWidget(tip_1,1,0,1,1)
        layout_top.addWidget(self.json_combox,1,1,1,3)
        layout_top.addWidget(refresh_json_btn,1,4,1,1)
 
        self.total_table = QtWidgets.QTableWidget()
        self.total_table.setRowCount(1)
        self.total_table.setColumnCount(len(self.total_table_header))
        self.total_table.setHorizontalHeaderLabels(self.total_table_header)
        self.total_table.setEditTriggers(QtWidgets.QAbstractItemView.NoEditTriggers)
        self.total_table.setSelectionBehavior(QtWidgets.QAbstractItemView.SelectRows)
 
        tip_3 = QtWidgets.QLabel('股票名模糊查询')
        self.query_lineedit = QtWidgets.QLineEdit()
        query_btn = QtWidgets.QPushButton('查询')
        query_btn.clicked.connect(self.query_btn_clicked)
        reset_btn = QtWidgets.QPushButton('重置')
        reset_btn.clicked.connect(self.reset_btn_clicked)
 
        tip_2 = QtWidgets.QLabel('次数:')
        self.count_combox = QtWidgets.QComboBox()
        self.count_combox.addItem(self.please_select_str)
        self.count_combox.addItems(list(self.num_sort_map.keys()))
        self.count_combox.currentIndexChanged.connect(self.count_combox_currentIndexChanged)
 
        tip_4 = QtWidgets.QLabel('胜率:')
        self.num_combox = QtWidgets.QComboBox()
        self.num_combox.addItem(self.please_select_str)
        self.num_combox.addItems(list(self.num_sort_map.keys()))
        self.num_combox.currentIndexChanged.connect(self.num_combox_currentIndexChanged)
 
        layout_query = QtWidgets.QGridLayout()
        layout_query.addWidget(tip_3,0,0,1,1)
        layout_query.addWidget(self.query_lineedit,0,1,1,3)
        layout_query.addWidget(query_btn,0,4,1,1)
        layout_query.addWidget(reset_btn,0,5,1,1)
        layout_query.addWidget(tip_2,1,0,1,1)
        layout_query.addWidget(self.count_combox,1,1,1,2)
        layout_query.addWidget(tip_4,1,3,1,1)
        layout_query.addWidget(self.num_combox,1,4,1,2)
 
        self.list_table = PageTableWidget()
        self.list_table.set_table_init_data({'headers': self.list_table_header})
        self.list_table.output_signal.connect(self.table_output_signal_emit)
 
        layout_left = QtWidgets.QVBoxLayout()
        layout_left.addWidget(self.total_table,1)
        layout_left.addLayout(layout_query,1)
        layout_left.addWidget(self.list_table,8)
 
        # self.k_widget = PyQtGraphKWidget()
        self.k_widget = PyQtGraphScrollKWidget()
        self.detail_table = PageTableWidget()
        self.detail_table.set_table_init_data({'headers': self.detail_table_header})
        self.detail_table.output_signal.connect(self.detail_table_output_signal_emit)
 
        layout_right = QtWidgets.QVBoxLayout()
        layout_right.addWidget(self.k_widget,3)
        layout_right.addWidget(self.detail_table,1)
 
        layout_down = QtWidgets.QHBoxLayout()
        layout_down.addLayout(layout_left,1)
        layout_down.addSpacing(20)
        layout_down.addLayout(layout_right,3)
 
        layout = QtWidgets.QVBoxLayout()
        layout.addLayout(layout_top)
        layout.addLayout(layout_down)
        self.setLayout(layout)
        pass
    def set_json_data(self,data:Dict[str,Any]):
        self.list_table_df = data['df']
        self.start_date_str = data['start_date_str']
        check_count = data['check_count']
        total_count = data['total_count']
        total_win = data['total_win']
        self.detail_map = data['detail_map']
        if data.get('factor_list') is not None:
            self.factor_list = data['factor_list']
        else:
            self.factor_list = []
 
        self.total_table.setItem(0,0,QtWidgets.QTableWidgetItem(str(check_count)))
        self.total_table.setItem(0,1,QtWidgets.QTableWidgetItem(str(total_count)))
        self.total_table.setItem(0,2,QtWidgets.QTableWidgetItem(str(total_win)+'%'))
 
        self.current_list_table_df = self.list_table_df.copy()
        self.fill_table_data()
        pass
    def fill_table_data(self):
        table_data = self.current_list_table_df.loc[:, self.list_pd_header].values.tolist()
        self.list_table.set_table_full_data(table_data)
        pass
 
    def table_output_signal_emit(self,data:List):
        # ticker secName count win
        dailydata_dir = self.dailydata_dir_lineedit.text()
        dailydata_dir = dailydata_dir.strip()
        if len(dailydata_dir)<=0:
            QtWidgets.QMessageBox.information(
                self,
                '提示',
                '请选择股票日数据文件件',
                QtWidgets.QMessageBox.Yes
            )
            return
        daily_file_path = dailydata_dir + '/' + data[0] + '.csv'
        df = pd.read_csv(daily_file_path,encoding='utf-8')
        # 删除停牌的数据
        df = df.loc[df['openPrice'] > 0].copy()
        df['o_date'] = df['tradeDate']
        df['o_date'] = pd.to_datetime(df['o_date'])
        df = df.loc[df['o_date'] >= self.start_date_str].copy()
        # 保存未复权收盘价数据
        df['close'] = df['closePrice']
        # 计算前复权数据
        df['openPrice'] = df['openPrice'] * df['accumAdjFactor']
        df['closePrice'] = df['closePrice'] * df['accumAdjFactor']
        df['highestPrice'] = df['highestPrice'] * df['accumAdjFactor']
        df['lowestPrice'] = df['lowestPrice'] * df['accumAdjFactor']
 
        columns_list = ['日期','收盘价','开盘价','最高价','最低价']
        columns_pd_list = ['tradeDate','closePrice','openPrice','highestPrice','lowestPrice']
 
        if self.factor_list:
            for item in self.factor_list:
                df = stock_factor_caculate.caculate_factor(df,item)
            pass
 
        df.reset_index(inplace=True)
 
        node_detail = self.detail_map[data[0]]
        table_list = node_detail['table_list']
        duration_list = node_detail['duration_list']
 
        self.detail_table.set_table_full_data(table_list)
 
        line_data = {
            'title_str': data[1],
            'whole_header': columns_list,
            'whole_df': df,
            'whole_pd_header': columns_pd_list,
            'duration_list': duration_list,
            'factor_list':self.factor_list
        }
        self.k_widget.set_data(line_data)
        pass
 
    def detail_table_output_signal_emit(self,data:List):
        self.k_widget.set_highligh_duration(int(data[0]))
        pass
 
    def dailydata_choice_btn_clicked(self):
        path = QtWidgets.QFileDialog.getExistingDirectory(
            self,
            '打开股票日数据所在文件夹',
            self.pre_output_dir
        )
        if not path:
            return
        self.dailydata_path = path
        self.dailydata_dir_lineedit.setText(path)
        pass
    def json_combox_currentIndexChanged(self,cur_i:int):
        cur_txt = self.json_combox.currentText()
        if not cur_txt or cur_txt == self.please_select_str:
            return
        current_json_file_path = self.results_output_dir + cur_txt
        with open(current_json_file_path,'r',encoding='utf-8') as fr:
            obj_json = json.load(fr)
        df = pd.DataFrame(obj_json['df_json'])
        obj_json['df'] = df
 
        self.set_json_data(obj_json)
        pass
    def count_combox_currentIndexChanged(self,cur_i:int):
        cur_txt = self.count_combox.currentText()
        if not cur_txt or cur_txt == self.please_select_str:
            return
        self.current_list_table_df.sort_values(by='count', ascending=self.num_sort_map[cur_txt], inplace=True)
        self.fill_table_data()
        pass
    def query_btn_clicked(self):
        query_str = self.query_lineedit.text()
        query_str = query_str.strip()
        if len(query_str)<=0:
            QtWidgets.QMessageBox.information(
                self,
                '提示',
                '请输入要查询的内容',
                QtWidgets.QMessageBox.Yes
            )
            return
        self.count_combox.setCurrentText(self.please_select_str)
        df = self.list_table_df.copy()
        self.current_list_table_df = df.loc[df['secName'].str.contains(query_str)].copy()
        self.fill_table_data()
        pass
    def reset_btn_clicked(self):
        self.query_lineedit.setText('')
        self.count_combox.setCurrentText(self.please_select_str)
        self.current_list_table_df = self.list_table_df.copy()
        self.fill_table_data()
        pass
    def num_combox_currentIndexChanged(self,cur_i:int):
        cur_txt = self.num_combox.currentText()
        if not cur_txt or cur_txt == self.please_select_str:
            return
        self.current_list_table_df.sort_values(by='win',ascending=self.num_sort_map[cur_txt],inplace=True)
        self.fill_table_data()
        pass
    def refresh_json_btn_clicked(self):
        # self.results_output_dir
        file_list = os.listdir(self.results_output_dir)
        json_file_list = []
        for item in file_list:
            if item.endswith('.json'):
                json_file_list.append(item)
        self.json_file_name_list.extend(json_file_list)
        json_file_set = set(self.json_file_name_list)
        self.json_file_name_list = list(json_file_set)
        self.json_combox.clear()
        self.json_combox.addItem(self.please_select_str)
        self.json_combox.addItems(self.json_file_name_list)
        pass
    pass

 主界面控件（也是运行策略代码控件）

class StrategeMainWidget(QtWidgets.QWidget):
    signal_runcode = QtCore.pyqtSignal(object)
    signal_time = QtCore.pyqtSignal(object)
    def __init__(self):
        super().__init__()
        self.thread_run: Thread = None
        self.thread_time: Thread = None
 
        self.running_graph_widget: QtWidgets.QWidget = None
 
        self.init_data()
        self.init_ui()
        self.register_event()
        pass
    def init_data(self):
        self.pre_output_dir = './'
        self.results_output_dir = './strategy_check_output/'
        self.secID_name_file_name = 'secID_name.csv'
        self.please_select_str: str = '--请选择--'
        self.stratege_name_list: List = []
        self.tip_msg_0: str = '1.选择策略所在文件夹；2.选择策略；3.点击运行。'
        self.stratege_path: str = ''
        self.stratege_run_start_time = None
        self.stratege_start = False
        self.current_stratege_py_str: str = ''
        self.dailydata_path:str = ''
        pass
    def init_ui(self):
        self.setWindowTitle('股票策略验证工具')
        tip_2 = QtWidgets.QLabel('股票日数据文件夹:')
        self.dailydata_dir_lineedit = QtWidgets.QLineEdit()
        self.dailydata_dir_lineedit.setReadOnly(True)
        dailydata_choice_btn = QtWidgets.QPushButton('选择文件夹')
        dailydata_choice_btn.clicked.connect(self.dailydata_choice_btn_clicked)
 
        tip_0 = QtWidgets.QLabel('选择策略所在文件夹:')
        self.stratege_dir_lineedit = QtWidgets.QLineEdit()
        self.stratege_dir_lineedit.setReadOnly(True)
        stratege_choice_btn = QtWidgets.QPushButton('选择文件夹')
        stratege_choice_btn.clicked.connect(self.stratege_choice_btn_clicked)
 
        tip_1 = QtWidgets.QLabel('策略:')
        self.stratege_combox = QtWidgets.QComboBox()
        self.stratege_combox.addItem(self.please_select_str)
        self.stratege_combox.currentIndexChanged.connect(self.stratege_combox_currentIndexChanged)
 
        self.run_btn = QtWidgets.QPushButton('运行')
        self.run_btn.clicked.connect(self.run_btn_clicked)
        self.force_stop_btn = QtWidgets.QPushButton('强制停止')
        self.force_stop_btn.clicked.connect(self.force_stop_btn_clicked)
 
        layout_top_left = QtWidgets.QGridLayout()
        layout_top_left.addWidget(tip_2,0,0,1,1)
        layout_top_left.addWidget(self.dailydata_dir_lineedit,0,1,1,3)
        layout_top_left.addWidget(dailydata_choice_btn,0,4,1,1)
        layout_top_left.addWidget(tip_0,1,0,1,1)
        layout_top_left.addWidget(self.stratege_dir_lineedit,1,1,1,3)
        layout_top_left.addWidget(stratege_choice_btn,1,4,1,1)
        layout_top_left.addWidget(tip_1,2,0,1,1)
        layout_top_left.addWidget(self.stratege_combox,2,1,1,2)
        layout_top_left.addWidget(self.run_btn,2,3,1,1)
        layout_top_left.addWidget(self.force_stop_btn,2,4,1,1)
 
        self.tip_msg_label = QtWidgets.QLabel()
        self.tip_msg_label.setWordWrap(True)
        self.tip_msg_label.setText(self.tip_msg_0)
        results_output_look_btn = QtWidgets.QPushButton('结果查看')
        results_output_look_btn.clicked.connect(self.results_output_look_btn_clicked)
 
        layout_top_right = QtWidgets.QHBoxLayout()
        layout_top_right.addWidget(self.tip_msg_label)
        layout_top_right.addWidget(results_output_look_btn)
 
        layout_top = QtWidgets.QHBoxLayout()
        layout_top.addLayout(layout_top_left,3)
        layout_top.addSpacing(30)
        layout_top.addLayout(layout_top_right,1)
 
        self.code_textedit = QtWidgets.QTextEdit()
        self.code_textedit.setReadOnly(True)
 
        layout = QtWidgets.QVBoxLayout()
        layout.addLayout(layout_top)
        layout.addWidget(self.code_textedit)
        self.setLayout(layout)
        pass
    def register_event(self):
        self.signal_runcode.connect(self.thread_run_excuted)
        self.signal_time.connect(self.thread_time_excuted)
        pass
    def results_output_look_btn_clicked(self):
        '''策略运行结果查看'''
        if not self.running_graph_widget:
            self.running_graph_widget = PyQtGraphRunningWidget()
        self.running_graph_widget.showMaximized()
        pass
    def dailydata_choice_btn_clicked(self):
        path = QtWidgets.QFileDialog.getExistingDirectory(
            self,
            '打开股票日数据所在文件夹',
            self.pre_output_dir
        )
        if not path:
            return
        self.dailydata_path = path+'/'
        self.dailydata_dir_lineedit.setText(path)
        pass
    def stratege_choice_btn_clicked(self):
        '''选择策略所在文件夹'''
        path = QtWidgets.QFileDialog.getExistingDirectory(
            self,
            '打开策略所在文件夹',
            self.pre_output_dir
        )
        if not path:
            return
        self.stratege_path = path
        self.stratege_dir_lineedit.setText(path)
        file_list = os.listdir(path)
        temp_file_list = set(self.stratege_name_list)
        for item in file_list:
            if item.endswith('.py'):
                temp_file_list.add(item)
        self.stratege_name_list = list(temp_file_list)
 
        self.stratege_combox.clear()
        self.stratege_combox.addItem(self.please_select_str)
        self.stratege_combox.addItems(self.stratege_name_list)
        pass
    def stratege_combox_currentIndexChanged(self,cur_i:int):
        cur_txt = self.stratege_combox.currentText()
        if not cur_txt or cur_txt == self.please_select_str:
            self.code_textedit.clear()
            return
        file_path = self.stratege_path + os.path.sep + cur_txt
        with open(file_path,'r',encoding='utf-8') as fr:
            code_txt = fr.read()
        self.code_textedit.setPlainText(code_txt)
        pass
    def run_btn_clicked(self):
        '''运行按钮'''
        # 检查股票日数据文件夹
        dailydata_dir = self.dailydata_dir_lineedit.text()
        dailydata_dir = dailydata_dir.strip()
        if len(dailydata_dir)<=0:
            QtWidgets.QMessageBox.information(
                self,
                '提示',
                '请选择股票日数据文件夹',
                QtWidgets.QMessageBox.Yes
            )
            return
        dailydata_file_list = os.listdir(dailydata_dir)
        if len(dailydata_file_list)<=0:
            QtWidgets.QMessageBox.information(
                self,
                '提示',
                '股票日数据文件夹中没有文件',
                QtWidgets.QMessageBox.Yes
            )
            return
        secID_name_file = self.results_output_dir + self.secID_name_file_name
        if not os.path.exists(secID_name_file):
            QtWidgets.QMessageBox.information(
                self,
                '提示',
                '股票码与股票名基础文件不存在',
                QtWidgets.QMessageBox.Yes
            )
            return
 
        py_str = self.code_textedit.toPlainText()
        if len(py_str)<10:
            QtWidgets.QMessageBox.information(
                self,
                '提示',
                '请选择要执行的策略',
                QtWidgets.QMessageBox.Yes
            )
            return
        self.current_stratege_py_str = py_str
        base_data = {}
        base_df = pd.read_csv(secID_name_file,encoding='utf-8')
        for i,row in base_df.iterrows():
            secID = row['secID']
            secID_arr = secID.split('.')
            base_data[secID_arr[0]] = row['secShortName']
            pass
 
        self.run_btn.setDisabled(True)
        self.stratege_combox.setDisabled(True)
        self.stratege_run_start_time = datetime.now()
        self.stratege_start = True
 
        if self.thread_run:
            QtWidgets.QMessageBox.information(
                self,
                '提示',
                '有策略正在运行',
                QtWidgets.QMessageBox.Yes
            )
            return
        pre_data = {
            'py_str':py_str,
            'base_data':base_data
        }
        self.thread_run = Thread(
            target=self.running_run_thread,
            args=(pre_data,)
        )
        self.thread_run.start()
        self.thread_time = Thread(
            target=self.running_time_thread
        )
        self.thread_time.start()
        pass
    def force_stop_btn_clicked(self):
        '''强制停止按钮'''
        self.thread_run = None
        self.thread_time = None
 
        self.run_btn.setDisabled(False)
        self.stratege_combox.setDisabled(False)
        self.stratege_start = False
        pass
    def running_run_thread(self,data:Dict[str,Any]):
        '''执行代码线程'''
        py_str = data['py_str']
        base_data = data['base_data']
 
        namespace = {}
        fun_stragegy = compile(py_str,'','exec')
        exec(fun_stragegy,namespace)
        ret = namespace['excute_strategy'](base_data,self.dailydata_path)
        self.signal_runcode.emit(ret)
        pass
    def running_time_thread(self):
        '''计时线程'''
        while self.stratege_start:
            now = datetime.now()
            interval_time = (now-self.stratege_run_start_time).seconds
            res_map = {'res':interval_time}
            self.signal_time.emit(res_map)
            time.sleep(1)
        pass
    def thread_run_excuted(self,data:Dict):
        '''策略代码执行返回结果'''
        self.run_btn.setDisabled(False)
        self.stratege_combox.setDisabled(False)
 
        # 保存结果文件
        now_datetime_str = datetime.now().strftime('%Y%m%d%H%M%S')
        df = data['df']
        df_json = df.to_dict(orient='records')
        pre_save_data = {
            'df_json':df_json,
            'check_count':data['check_count'],
            'total_count':data['total_count'],
            'total_win':data['total_win'],
            'start_date_str':data['start_date_str'],
            'detail_map':data['detail_map']
        }
        with open(self.results_output_dir + now_datetime_str + '.json','w',encoding='utf-8') as fw:
            json.dump(pre_save_data,fw)
        if not self.running_graph_widget:
            self.running_graph_widget = PyQtGraphRunningWidget()
        self.running_graph_widget.set_json_data(data)
        self.running_graph_widget.showMaximized()
        self.thread_run = None
        self.thread_time = None
        self.stratege_start = False
 
        QtWidgets.QMessageBox.information(
            self,
            '提示',
            '当前策略运行完毕',
            QtWidgets.QMessageBox.Yes
        )
        pass
    def thread_time_excuted(self,data:Dict):
        '''计时返回结果'''
        res = data['res']
        self.tip_msg_label.setText(f"{res}s")
        pass
    def closeEvent(self, a0: QtGui.QCloseEvent) -> None:
        if self.thread_time:
            self.thread_time.join()
        if self.thread_run:
            self.thread_run.join()
        if self.running_graph_widget:
            self.running_graph_widget.close()
        self.close()

工具使用

if __name__ == '__main__':
    QtCore.QCoreApplication.setAttribute(QtCore.Qt.HighDpiScaleFactorRoundingPolicy.PassThrough)
    app = QtWidgets.QApplication(sys.argv)
    t_win = StrategeMainWidget()
    t_win.showMaximized()
    app.exec()
    pass

1 在StrategeMainWidget代码的同一目录下创建“strategy_check_output” 文件夹，并把数据中的secID_name.csv文件放入这个文件夹

2. 每次运行结果会以json文件存储在strategy_check_output文件夹下

运行工具

 滚动截图