写在前面：
1 在写研究方法过程中（例如：股票研究），很多方法根据数据的更新需要重复运行获取新的结果，本工具就是固化这些需要重复运行的代码，可以直接在工具中运行得到更新的结果。
2 本文是V1.0版本，提供运行python方法的框架，结果显示的控件后续根据研究过程的需要会陆续补充
界面展示：
代码：
日志窗体
class LogShowWidget(QtWidgets.QWidget):
def __init__(self):
super().__init__()
self.init_data()
self.init_ui()
pass
def init_data(self):
self.whole_data:List = []
self.current_log: str = None
pass
def init_ui(self):
self.setWindowTitle('执行日志')
self.setMinimumWidth(600)
self.setMinimumHeight(600)
pre_btn = QtWidgets.QPushButton('上一个')
pre_btn.clicked.connect(self.pre_btn_clicked)
next_btn = QtWidgets.QPushButton('下一个')
next_btn.clicked.connect(self.next_btn_clicked)
layout_one = QtWidgets.QHBoxLayout()
layout_one.addStretch(1)
layout_one.addWidget(pre_btn)
layout_one.addWidget(next_btn)
layout_one.addStretch(1)
self.log_textedit = QtWidgets.QTextEdit()
layout = QtWidgets.QVBoxLayout()
layout.addLayout(layout_one)
layout.addWidget(self.log_textedit)
self.setLayout(layout)
pass
def set_data(self,data:List):
self.whole_data.clear()
self.whole_data = data
self.current_log = self.whole_data[-1]
self.fill_log()
pass
def pre_btn_clicked(self):
cur_i = self.whole_data.index(self.current_log)
if cur_i<=0:
cur_i = len(self.whole_data)-1
else:
cur_i = cur_i -1
self.current_log = self.whole_data[cur_i]
self.fill_log()
pass
def next_btn_clicked(self):
cur_i = self.whole_data.index(self.current_log)
if cur_i >= len(self.whole_data) - 1:
cur_i = 0
else:
cur_i = cur_i + 1
self.current_log = self.whole_data[cur_i]
self.fill_log()
pass
def fill_log(self):
self.log_textedit.clear()
self.log_textedit.setPlainText(self.current_log)
pass
pass代码执行控件（界面中蓝色框部分）
class ExcuteShowWidget(QtWidgets.QWidget):
signal_excute = QtCore.pyqtSignal(object)
def __init__(self):
super().__init__()
self.thread_caculate: Thread = None
self.init_data()
self.init_ui()
self.register_event()
self.progress_init()
pass
def init_data(self):
self.please_select_str: str = '-- 请选择 --'
self.dir_path: str = ''
self.log_list: List[str] = []
self.log_widget: QtWidgets.QWidget = None
pass
def init_ui(self):
self.caculate_progress = QtWidgets.QProgressBar()
self.caculate_status_label = QtWidgets.QLabel()
layout_progress = QtWidgets.QHBoxLayout()
layout_progress.addWidget(self.caculate_progress)
layout_progress.addWidget(self.caculate_status_label)
log_btn = QtWidgets.QPushButton('日志')
log_btn.clicked.connect(self.log_btn_clicked)
self.code_combox = QtWidgets.QComboBox()
self.code_combox.addItem(self.please_select_str)
excute_code_btn = QtWidgets.QPushButton('执行')
excute_code_btn.clicked.connect(self.excute_code_btn_clicked)
self.excute_info_label = QtWidgets.QLabel()
self.excute_info_label.setStyleSheet('QLabel{font-size:18px;color:red;font-weight:bold;}')
layout_one = QtWidgets.QHBoxLayout()
layout_one.addWidget(log_btn)
layout_one.addWidget(self.code_combox)
layout_one.addWidget(excute_code_btn)
layout_one.addWidget(self.excute_info_label)
self.scroll_layout = QtWidgets.QVBoxLayout()
self.scroll_area = QtWidgets.QScrollArea()
self.scroll_area.setWidgetResizable(True)
layout = QtWidgets.QVBoxLayout()
layout.addLayout(layout_progress)
layout.addLayout(layout_one)
layout.addWidget(self.scroll_area)
self.setLayout(layout)
pass
def register_event(self):
self.signal_excute.connect(self.process_excute_event)
pass
def process_excute_event(self,data:Dict):
mark_str = data['mark_str']
status = data['status']
if status == 'success':
self.excute_info_label.setText(f'{mark_str} 执行成功')
ret = data['ret']
target_style = ret['target_style']
if target_style == 'table':
# 表格类型结果
pre_map = {
'header':ret['target_header'],
'data':ret['target_data']
}
node_widget = TableNodeWidget()
node_widget.set_data(pre_map)
self.fill_scroll_area([node_widget])
pass
elif target_style == 'graph':
# 图类型
pre_map = {
'data':ret['target_data']
}
node_widget = GraphNodeWidget()
node_widget.set_data(pre_map)
self.fill_scroll_area([node_widget])
pass
self.thread_caculate = None
self.progress_finished()
pass
else:
self.excute_info_label.setText(f'{mark_str} 失败！失败！')
now_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_str = f'{now_str}::{mark_str}::{data["msg"]}'
self.log_list.append(log_str)
self.thread_caculate = None
self.progress_finished()
pass
pass
def excute_code_btn_clicked(self):
cur_txt = self.code_combox.currentText()
if not cur_txt or cur_txt == self.please_select_str:
QtWidgets.QMessageBox.information(
self,
'提示',
'请选择要执行的代码',
QtWidgets.QMessageBox.Yes
)
return
py_file_path = self.dir_path+ os.path.sep + cur_txt + '.py'
json_file_path = self.dir_path + os.path.sep + cur_txt + '.json'
if not os.path.exists(py_file_path):
QtWidgets.QMessageBox.information(
self,
'提示',
'不存在该py文件',
QtWidgets.QMessageBox.Yes
)
return
if not os.path.exists(json_file_path):
QtWidgets.QMessageBox.information(
self,
'提示',
'不存在该json文件',
QtWidgets.QMessageBox.Yes
)
return
with open(py_file_path,'r',encoding='utf-8') as fr:
py_code_str = fr.read()
try:
with open(json_file_path,'r',encoding='utf-8') as fr:
json_obj = json.load(fr)
except Exception as e:
QtWidgets.QMessageBox.information(
self,
'报错',
e.__str__(),
QtWidgets.QMessageBox.Yes
)
return
self.start_caculate_thread(cur_txt,{'py_code':py_code_str,'json_obj':json_obj})
pass
def set_target_combox_data(self,data:List,dir_path:str):
self.dir_path = dir_path
self.code_combox.clear()
self.code_combox.addItem(self.please_select_str)
self.code_combox.addItems(data)
pass
def log_btn_clicked(self):
if not self.log_widget:
self.log_widget = LogShowWidget()
self.log_widget.set_data(self.log_list)
self.log_widget.show()
pass
def fill_scroll_area(self,widget_list:List):
while self.scroll_layout.count():
item = self.scroll_layout.takeAt(0)
widget = item.widget()
if widget is not None:
widget.deleteLater()
pass
sc_child_widget = self.scroll_area.takeWidget()
if sc_child_widget is not None:
sc_child_widget.deleteLater()
for item in widget_list:
self.scroll_layout.addWidget(item)
one_sc_child_widget = QtWidgets.QWidget()
one_sc_child_widget.setLayout(self.scroll_layout)
self.scroll_area.setWidget(one_sc_child_widget)
pass
def start_caculate_thread(self,mark_str:str,data:Dict[str,Any]):
if self.thread_caculate:
QtWidgets.QMessageBox.information(
self,
'提示',
'线程正在执行任务，请稍后。。。',
QtWidgets.QMessageBox.Yes
)
return
self.thread_caculate = Thread(
target=self.running_caculate_thread,
args=(
mark_str, data,
)
)
self.thread_caculate.start()
self.progress_busy()
pass
def running_caculate_thread(self,mark_str:str,data:Dict[str,Any]):
py_code = data['py_code']
json_obj = data['json_obj']
try:
namespace = {}
fun_code = compile(py_code,'<string>','exec')
exec(fun_code,namespace)
ret = namespace['excute_code'](json_obj)
res_map = {
'mark_str':mark_str,
'status':'success',
'ret':ret
}
self.signal_excute.emit(res_map)
except Exception as e:
res_map = {
'mark_str':mark_str,
'status':'error',
'msg':e.__str__()
}
self.signal_excute.emit(res_map)
pass
def progress_init(self) -> None:
self.caculate_progress.setValue(0)
self.caculate_status_label.setText('无任务')
def progress_busy(self) -> None:
self.caculate_progress.setRange(0, 0)
self.caculate_status_label.setText('正在执行')
def progress_finished(self) -> None:
self.caculate_progress.setRange(0, 100)
self.caculate_progress.setValue(100)
self.caculate_status_label.setText('执行完毕')
pass
def closeEvent(self, a0: QtGui.QCloseEvent) -> None:
if self.log_widget:
self.log_widget.close()
self.close()
pass主界面
class StockAnalysisMainWidget(QtWidgets.QWidget):
def __init__(self):
super().__init__()
self.init_data()
self.init_ui()
pass
def init_data(self):
self.code_py_list:List = []
pass
def init_ui(self):
self.setWindowTitle('PyQt5运行股票研究python方法工具V1.0')
code_py_dir_btn = QtWidgets.QPushButton('代码py文件目录')
code_py_dir_btn.clicked.connect(self.code_py_dir_btn_clicked)
self.code_py_dir_lineedit = QtWidgets.QLineEdit()
tip_label = QtWidgets.QLabel('提示：json文件为参数文件，同一业务json和py文件同名')
layout_one = QtWidgets.QFormLayout()
layout_one.addRow(code_py_dir_btn,self.code_py_dir_lineedit)
layout_one.addWidget(tip_label)
copy_one_widget_btn = QtWidgets.QPushButton('复制窗体')
copy_one_widget_btn.clicked.connect(self.copy_one_widget_btn_clicked)
layout_two = QtWidgets.QHBoxLayout()
layout_two.addWidget(copy_one_widget_btn)
layout_two.addStretch(1)
layout_three = QtWidgets.QVBoxLayout()
layout_three.addLayout(layout_one)
layout_three.addLayout(layout_two)
# 分界线 s
h_line = QtWidgets.QFrame()
h_line.setFrameShape(QtWidgets.QFrame.HLine)
h_line.setFrameShadow(QtWidgets.QFrame.Sunken)
# 分界线 e
self.excute_show_widget = ExcuteShowWidget()
layout = QtWidgets.QVBoxLayout()
layout.addLayout(layout_three)
layout.addWidget(h_line)
layout.addWidget(self.excute_show_widget)
self.setLayout(layout)
pass
def code_py_dir_btn_clicked(self):
path = QtWidgets.QFileDialog.getExistingDirectory(
self,
'打开py和json文件所在目录',
'.'
)
if not path:
return
self.code_py_list.clear()
py_file_list = os.listdir(path)
for item in py_file_list:
if item.endswith('.py'):
self.code_py_list.append(item[0:-3])
self.code_py_dir_lineedit.setText(path)
self.excute_show_widget.set_target_combox_data(self.code_py_list,path)
pass
def copy_one_widget_btn_clicked(self):
pass
pass执行代码：
if __name__ == '__main__':
QtCore.QCoreApplication.setAttribute(QtCore.Qt.HighDpiScaleFactorRoundingPolicy.PassThrough)
app = QtWidgets.QApplication(sys.argv)
main_window = StockAnalysisMainWidget()
main_window.showMaximized()
app.exec()
pass
