PyQt6实例_pyqtgraph多曲线显示工具_代码分享

目次
概述
效果
代码 
返回结果对象
字符型横坐标
通用折线图工具
工具主界面
使用举例

---

概述

1 分析数据遇到必要一个股票多个指标对比或一个指标多个股票对比，涉及到同轴多条曲线的显示，所以开发了本工具。
2 多曲线显示部分可以当通用工具使用。
3 数据计算使用compile动态执行代码，返回固定格式的数据举行显示，尽最大可能实现工具的灵活性。
效果

代码 

返回结果对象

1. import pandas as pd
2. from dataclasses import dataclass,field
3. @dataclass
4. class MultiLineObj:
5.     title:str=''
6.     df:pd.DataFrame=pd.DataFrame()
7.     col_dict: dict = field(default_factory=dict)
8.     error_msg:str = ''
9.     status:str='ok'
10.     pass

复制代码

1 对应结果中的 title
2 字段中文名取自 col_dict = {df中的列名：中文名}
3 对应df中的数据，还用于右上角下载按钮，点击下载按钮可以将df数据下载到本地 
字符型横坐标

1. class StrAxisItem(pg.AxisItem):
2.     def __init__(self,ticks,*args,**kwargs):
3.         pg.AxisItem.__init__(self,*args,**kwargs)
4.         self.x_values = [x[0] for x in ticks]
5.         self.x_strings = [x[1] for x in ticks]
6.         pass
7.     def tickStrings(self, values, scale, spacing):
8.         strings = []
9.         for v in values:
10.             vs = v*scale
11.             if vs in self.x_values:
12.                 vstr = self.x_strings[self.x_values.index(vs)]
13.             else:
14.                 vstr = ''
15.             strings.append(vstr)
16.         return strings

复制代码

继承pg.AxisItem, 重写tickStrings方法
通用折线图工具

1. class MultiLineGraphWidget(pg.PlotWidget):
2.     def __init__(self):
3.         super().__init__()
4.         self.init_data()
5.         pass
6.     def init_data(self):
7.         self.whole_df:pd.DataFrame = pd.DataFrame()
8.         # pd的col名：显示名
9.         self.whole_col_dict:dict = {}
10.         self.color_10_list = [(30,144,255),(138,43,226),(220,20,60),(0,128,128),(0,255,255),(0,250,154),(173,255,47),(255,255,224),(255,215,0),(255,140,0)]
11.         pass
12.     def set_data(self,df:pd.DataFrame,col_dict:dict):
13.         self.clear()
14.         self.addLegend()
16.         self.whole_df = df
17.         self.whole_col_dict = col_dict
19.         x = df['x'].to_list()
20.         xTicks = df.loc[:, ['x', 'reportDate']].values
22.         i = 0
23.         for k,v in col_dict.items():
24.             one_color = self.color_10_list[i%len(self.color_10_list)]
25.             one_curve = pg.PlotCurveItem(x=np.array(x),y=np.array(df[k].to_list()),pen=pg.mkPen({'color':one_color,'width':2}),connect='finite',name=v)
26.             self.addItem(one_curve)
27.             i += 1
28.             pass
30.         horAxis = StrAxisItem(ticks=xTicks, orientation='bottom')
31.         self.setAxisItems({'bottom':horAxis})
33.         self.vLine = pg.InfiniteLine(angle=90,movable=False)
34.         self.hLine = pg.InfiniteLine(angle=0,movable=False)
35.         self.label = pg.TextItem()
37.         self.addItem(self.vLine,ignoreBounds=True)
38.         self.addItem(self.hLine,ignoreBounds=True)
39.         self.addItem(self.label,ignoreBounds=True)
41.         self.vb = self.getViewBox()
42.         self.proxy = pg.SignalProxy(self.scene().sigMouseMoved, rateLimit=60, slot=self.pw_mouseMoved)
43.         self.enableAutoRange()
44.         pass
45.     def pw_mouseMoved(self, evt):
46.         pos = evt[0]
47.         if self.sceneBoundingRect().contains(pos):
48.             mousePoint = self.vb.mapSceneToView(pos)
49.             index = int(mousePoint.x())
50.             if index>=0 and index<len(self.whole_df):
51.                 html_str = '<p style="color:white;font-size:18px;">'
52.                 html_str += f"<br/>日期：{self.whole_df.loc[self.whole_df['x'] == index].iloc[0]['reportDate']}"
53.                 for k,v in self.whole_col_dict.items():
54.                     html_str += f"<br/>{v}：{self.whole_df.loc[self.whole_df['x'] == index].iloc[0][k]:,}"
55.                     pass
56.                 html_str += '</p>'
57.                 self.label.setHtml(html_str)
58.                 self.label.setPos(mousePoint.x(),mousePoint.y())
59.                 pass
60.             self.vLine.setPos(mousePoint.x())
61.             self.hLine.setPos(mousePoint.y())
62.         pass
63.     def wheelEvent(self,ev):
64.         if len(self.whole_df) <= 0:
65.             super().wheelEvent(ev)
66.         else:
67.             delta = ev.angleDelta().x()
68.             if delta == 0:
69.                 delta = ev.angleDelta().y()
71.             s = 1.001 ** delta
73.             before_xmin, before_xmax = self.viewRange()[0]
74.             val_x = self.getViewBox().mapSceneToView(ev.position()).x()
76.             after_xmin = int(val_x - (val_x - before_xmin) // s)
77.             after_xmax = int(val_x + (before_xmax - val_x) // s)
79.             if after_xmin < 1:
80.                 after_xmin = 0
81.             if after_xmin >= len(self.whole_df):
82.                 after_xmin = max(len(self.whole_df) - 3, len(self.whole_df) - 1)
83.             if after_xmax < 1:
84.                 after_xmax = min(len(self.whole_df) - 1, 1)
85.             if after_xmax >= len(self.whole_df):
86.                 after_xmax = len(self.whole_df) - 1
88.             df00 = self.whole_df.loc[(self.whole_df['x'] >= after_xmin) & (self.whole_df['x'] <= after_xmax)].copy()
89.             min_list = []
90.             max_list = []
91.             for k in self.whole_col_dict.keys():
92.                 min_list.append(df00[k].min())
93.                 max_list.append(df00[k].max())
94.                 pass
95.             after_ymin = min(min_list)
96.             after_ymax = max(max_list)
98.             self.setXRange(after_xmin, after_xmax)
99.             self.setYRange(after_ymin, after_ymax)
100.             pass
101.     pass

复制代码

1）工具中设置了10种颜色，差别曲线将显示差别颜色，如果曲线个数凌驾10个，将循环使用颜色
2）set_data方法必要带入df 和 col_dict两个参数
2.1）df 必须要有 x 、reportDate 两个字段，x为递增整数，reportDate为横坐标要显示的字符，reportDate为字符型。
2.2）折线的y轴数据在df中的列名为 col_dict中的key值，建议列名为英文和数字组成，col_dict中的val为中文名
工具主界面

1. class PyExcuteGraphShowWidget(QWidget):
2.     def __init__(self):
3.         super().__init__()
4.         self.setWindowTitle('py文件执行并显示结果')
5.         self.setMinimumSize(QSize(1000,800))
7.         label00 = QLabel('选择py文件：')
8.         self.lineedit_file = QLineEdit()
9.         btn_choice = QPushButton('选择文件',clicked=self.btn_choice_clicked)
10.         self.btn_excute = QPushButton('执行',clicked=self.btn_excute_clicked)
11.         btn_download = QPushButton('下载数据',clicked=self.btn_download_clicked)
13.         self.label_title = QLabel('指标', alignment=Qt.AlignmentFlag.AlignHCenter)
14.         self.label_title.setStyleSheet("font-size:28px;color:#CC2EFA;")
15.         self.pw = MultiLineGraphWidget()
17.         layout00 = QHBoxLayout()
18.         layout00.addWidget(label00)
19.         layout00.addWidget(self.lineedit_file)
20.         layout00.addWidget(btn_choice)
21.         layout00.addWidget(self.btn_excute)
22.         layout00.addWidget(btn_download)
24.         layout = QVBoxLayout()
25.         layout.addLayout(layout00)
26.         layout.addWidget(self.label_title)
27.         layout.addWidget(self.pw)
28.         self.setLayout(layout)
29.         pass
30.     def open_init(self):
31.         self.whole_resObj:MultiLineObj = None
32.         pass
33.     def btn_choice_clicked(self):
34.         file_path,_ = QFileDialog.getOpenFileName(self,'选择文件')
35.         if file_path:
36.             self.lineedit_file.setText(file_path)
37.         pass
38.     def btn_excute_clicked(self):
39.         file_path = self.lineedit_file.text()
40.         if len(file_path) <= 0:
41.             QMessageBox.information(self,'提示','请选择要执行的py文件',QMessageBox.StandardButton.Ok)
42.             return
43.         with open(file_path,'r',encoding='utf-8') as fr:
44.             py_code = fr.read()
45.         namespace = {}
46.         fun_code = compile(py_code, '<string>', 'exec')
47.         exec(fun_code, namespace)
48.         res = namespace['execute_caculate']()
49.         if res.status == 'error':
50.             QMessageBox.information(self,'执行过程报错',res.error_msg,QMessageBox.StandardButton.Ok)
51.             return
52.         self.label_title.setText(res.title)
53.         self.whole_resObj = res
54.         df = res.df.copy()
55.         df['x'] = range(len(df))
56.         self.pw.set_data(df.copy(),res.col_dict)
57.         QMessageBox.information(self,'提示','执行完毕',QMessageBox.StandardButton.Ok)
58.         pass
60.     def btn_download_clicked(self):
61.         if self.whole_resObj is None or self.whole_resObj.status == 'error':
62.             QMessageBox.information(self,'提示','数据为空',QMessageBox.StandardButton.Ok)
63.             return
64.         dir_name = QFileDialog.getExistingDirectory(self,'选择保存位置')
65.         if dir_name:
66.             df = self.whole_resObj.df.copy()
67.             df.rename(columns=self.whole_resObj.col_dict,inplace=True)
68.             df.to_csv(dir_name+os.path.sep + self.whole_resObj.title +'.csv',encoding='utf-8',index=False)
69.             QMessageBox.information(self,'提示','下载完毕',QMessageBox.StandardButton.Ok)
70.             pass
71.     pass

复制代码

使用举例

必要导入的包和运行代码

1. import os,sys
2. import pandas as pd
3. import numpy as np
4. from PyQt6.QtCore import (
5. QSize,
6. Qt
7. )
8. from PyQt6.QtWidgets import (
9.     QApplication,
10.     QLabel,
11.     QPushButton,
12.     QWidget,
13.     QVBoxLayout,
14.     QHBoxLayout,
15.     QFileDialog,
16.     QMessageBox,
17.     QLineEdit
18. )
19. import pyqtgraph as pg
20. from objects import MultiLineObj
22. if __name__ == '__main__':
23.     app = QApplication(sys.argv)
24.     mw = PyExcuteGraphShowWidget()
25.     mw.show()
26.     app.exec()
27.     pass

复制代码

1）一个py文件例子，内容如下，方法名固定为 execute_caculate

1. def execute_caculate():
2.     import traceback
3.     import pandas as pd
4.     from utils import postgresql_utils
5.     from objects import MultiLineObj
6.     '''
7.     灵活py文件执行
8.     营业利润，营业外支出，营业外收入
9.     '''
10.     conn = postgresql_utils.connect_db()
11.     cur = conn.cursor()
12.     try:
13.         ticker = '000638'
14.         sql_str = f'''
15.         select reportDate,iii_operateProfit,add_nonoperateIncome,less_nonoperateExpenses from t_profit where ticker=\'{ticker}\' and reportDate like \'%-12-31\';
16.         '''
17.         cur.execute(sql_str)
18.         res = cur.fetchall()
19.         col_list = ['reportDate','a0','a1','a2']
20.         col_dict = {
21.             'a0':'营业利润',
22.             'a1':'营业外收入',
23.             'a2':'营业外支出'
24.         }
25.         df = pd.DataFrame(columns=col_list, data=res)
26.         res_obj = MultiLineObj(
27.             title=f'{ticker},营业利润、营业外收入、营业外支出',
28.             df=df,
29.             col_dict=col_dict,
30.             status='ok'
31.         )
32.         return res_obj
33.     except:
34.         res_obj = MultiLineObj(
35.             status='error',
36.             error_msg=traceback.format_exc()
37.         )
38.         return res_obj
39.     finally:
40.         cur.close()
41.         conn.close()
42.         pass
43.     pass

复制代码

保存为 test002.py
注意：例子中涉及到的postgreSQL和财报数据在往期博文中可以找到。
2）点击“选择文件”，选择 test002.py文件
3）点击“执行”，执行完毕后就能显示效果图

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。