目錄
描述:
效果:
代碼:?
返回結果對象
字符型橫坐標
通用散點圖工具
工具主界面
使用舉例
描述:
1 本例結合實際應用場景描述散點圖的使用。在財報分析中,需要將數值放在同行業中進行比較,從而判斷是否異常。
2 散點圖顯示部分可以當通用工具使用。
3 數據計算使用compile動態執行代碼,返回固定格式的數據進行顯示,盡最大可能實現工具的靈活性。
效果:
代碼:?
返回結果對象
@dataclass
class MultiScatterObj:title:str=''df:pd.DataFrame=pd.DataFrame()col_dict: dict = field(default_factory=dict)type_dict: dict = field(default_factory=dict)focus_target: str = ''error_msg:str = ''status:str='ok'pass
1 title 標題
2 type_dict 行業分類標準
4 col_dict 計算的指標
5 focus_target 要關注的股票
字符型橫坐標
class StrAxisItem(pg.AxisItem):def __init__(self,ticks,*args,**kwargs):pg.AxisItem.__init__(self,*args,**kwargs)self.x_values = [x[0] for x in ticks]self.x_strings = [x[1] for x in ticks]passdef tickStrings(self, values, scale, spacing):strings = []for v in values:vs = v*scaleif vs in self.x_values:vstr = self.x_strings[self.x_values.index(vs)]else:vstr = ''strings.append(vstr)return strings
通用散點圖工具
class ScatterGraphWidget(pg.PlotWidget):def __init__(self):super().__init__()self.init_data()passdef init_data(self):self.whole_df = Noneself.cur_targetItem = Noneself.focus_targetItem = Noneself.color_point = (255, 255, 0)self.color_star = (220, 20, 60)self.color_focus = (255,140,0)self.color_mean = (255, 0, 255)self.color_median = (34, 139, 34)with open(DATA_DIR+'ticker_name.json','r',encoding='utf-8') as f:self.ticker_name_dict = json.load(f)passdef set_data(self, df: pd.DataFrame,focus_ticker:str=None):self.clear()if df.empty:returnself.whole_df = dfself.xTicks = self.whole_df.loc[:, ['x', 'ticker']].valuesself.x = self.whole_df['x'].to_list()self.y = self.whole_df['target'].to_list()horAxis = StrAxisItem(ticks=self.xTicks, orientation='bottom')self.setAxisItems({'bottom': horAxis})scatters = pg.ScatterPlotItem(hoverable=True,hoverPen=pg.mkPen('g'),tip=None)spots = []for x0, y0 in zip(self.x, self.y):spots.append({'pos': (x0, y0),'size': 10,'pen': {'color': self.color_point, 'width': 2},'brush': pg.mkBrush(color=self.color_point)})scatters.addPoints(spots)self.addItem(scatters)self.label = pg.TextItem()self.addItem(self.label, ignoreBounds=True)if focus_ticker:if self.focus_targetItem:self.removeItem(self.focus_targetItem)df00 = df.loc[df['ticker']==focus_ticker]if not df00.empty:index_x = df00.iloc[0]['x']self.focus_targetItem = pg.TargetItem(pos=(index_x, self.y[index_x]),movable=False,size=20,symbol='t1',pen=self.color_focus,brush=self.color_focus)self.addItem(self.focus_targetItem)# 添加中位數線 和 平均數線mean_line = pg.InfiniteLine(pos=(0,self.whole_df['target'].mean()),movable=False,angle=0,pen=pg.mkPen({'color':self.color_mean,'width':2}),label=f'{self.whole_df["target"].mean():,} 平均數',labelOpts={'position':0.05,'color': (255, 255, 255), 'movable': True, 'fill': (self.color_mean[0], self.color_mean[1], self.color_mean[2], 100)})median_line = pg.InfiniteLine(pos=(0,self.whole_df['target'].median()),movable=False,angle=0,pen=pg.mkPen({'color':self.color_median,'width':2}),label=f'{self.whole_df["target"].median():,} 中位數',labelOpts={'position':0.05,'color': (255, 255, 255), 'movable': True, 'fill': (self.color_median[0], self.color_median[1], self.color_median[2], 100)})self.addItem(mean_line)self.addItem(median_line)scatters.sigClicked.connect(self.scatters_sigClicked)scatters.sigHovered.connect(self.scatters_sigHovered)self.enableAutoRange()passdef set_content_empty(self):self.clear()passdef set_targetItem(self,ticker:str):df = self.whole_df.copy()df = df.loc[df['ticker'] == ticker]if df.empty:returnindex_x = df.iloc[0]['x']if self.cur_targetItem:self.removeItem(self.cur_targetItem)self.cur_targetItem = pg.TargetItem(pos=(index_x, self.y[index_x]),movable=False,size=20,symbol='star',pen=self.color_star,brush=self.color_star)self.addItem(self.cur_targetItem)passdef scatters_sigClicked(self, plot, points):# 將單擊的股票代碼發送給左側圖if len(points) <= 0:returnindex_x = points[0].index()if self.cur_targetItem:self.removeItem(self.cur_targetItem)self.cur_targetItem = pg.TargetItem(pos=(index_x, self.y[index_x]),movable=False,size=20,symbol='star',pen=self.color_star,brush=self.color_star)self.addItem(self.cur_targetItem)passdef scatters_sigHovered(self, plot, points):if len(points) <= 0:returnindex_x = points[0].index()x_str = self.xTicks[index_x][1]y_val = self.y[index_x]x_str00 = self.ticker_name_dict.get(x_str,x_str)html_str = '<p style="color:white;font-size:18px;font-weight:bold;">' + x_str00 + ' ' + f'{y_val:,}' + '</p>'self.label.setHtml(html_str)self.label.setPos(points[0].pos())passdef wheelEvent(self, ev):if len(self.whole_df) <= 0:super().wheelEvent(ev)else:delta = ev.angleDelta().x()if delta == 0:delta = ev.angleDelta().y()s = 1.001 ** deltabefore_xmin, before_xmax = self.viewRange()[0]val_x = self.getViewBox().mapSceneToView(ev.position()).x()after_xmin = int(val_x - (val_x - before_xmin) // s)after_xmax = int(val_x + (before_xmax - val_x) // s)if after_xmin < 1:after_xmin = 0if after_xmin >= len(self.whole_df):after_xmin = max(len(self.whole_df) - 3, len(self.whole_df) - 1)if after_xmax < 1:after_xmax = min(len(self.whole_df) - 1, 1)if after_xmax >= len(self.whole_df):after_xmax = len(self.whole_df) - 1# print(after_xmin,after_xmax)df00 = self.whole_df.loc[(self.whole_df['x'] >= after_xmin) & (self.whole_df['x'] <= after_xmax)].copy()after_ymin = df00['target'].min()after_ymax = df00['target'].max()self.setXRange(after_xmin, after_xmax)self.setYRange(after_ymin, after_ymax)passpass
1)本例中散點圖增加了一些與實際業務相關的數據
2)set_data方法需要帶入df、focus_ticker(可為空)
2.1)df必須要有x、ticker兩個字段,x為遞增整數,ticker為橫坐標要顯示的字符
2.2)df中target字段為y軸數值
工具主界面
class PyExcuteGraphShowWidgetII(QWidget):def __init__(self):super().__init__()self.setWindowTitle('py文件執行并顯示結果(散點圖)')self.setMinimumSize(QSize(1000,800))label00 = QLabel('選擇py文件:')self.lineedit_file = QLineEdit()btn_choice = QPushButton('選擇文件',clicked=self.btn_choice_clicked)self.btn_excute = QPushButton('執行',clicked=self.btn_excute_clicked)btn_download = QPushButton('下載數據',clicked=self.btn_download_clicked)self.label_title = QLabel('指標', alignment=Qt.AlignmentFlag.AlignHCenter)self.label_title.setStyleSheet("font-size:28px;color:#CC2EFA;")label01 = QLabel('行業標準')self.combo_type = QComboBox()self.combo_type.currentIndexChanged.connect(self.combo_type_currentIndexChanged)self.label_industry = QLabel('所屬行業')self.label_industry.setStyleSheet("font-size:20px;color:#CC2EFA;")label02 = QLabel('日期')self.combo_date = QComboBox()self.combo_date.currentIndexChanged.connect(self.combo_date_currentIndexChanged)self.label_date = QLabel('日期')self.label_date.setStyleSheet("font-size:20px;color:#CC2EFA;")self.radioButtonGroup = QButtonGroup()self.radioButtonGroup.buttonClicked.connect(self.radioButtonGroup_buttonClicked)self.layout_radio = QHBoxLayout()groupBox = QGroupBox('指標')groupBox.setLayout(self.layout_radio)self.pw = ScatterGraphWidget()self.label_num = QLabel('幾個')self.table = QTableWidget()self.table.setColumnCount(2)self.table.setHorizontalHeaderLabels(['代碼','名'])self.table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)self.table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)self.table.itemClicked.connect(self.table_itemClicked)layout00 = QHBoxLayout()layout00.addWidget(label00)layout00.addWidget(self.lineedit_file)layout00.addWidget(btn_choice)layout00.addWidget(self.btn_excute)layout00.addWidget(btn_download)layout01 = QVBoxLayout()layout01.addWidget(label01)layout01.addWidget(self.combo_type)layout01.addWidget(self.label_industry)layout02 = QVBoxLayout()layout02.addWidget(label02)layout02.addWidget(self.combo_date)layout02.addWidget(self.label_date)layout03 = QHBoxLayout()layout03.addLayout(layout01)layout03.addLayout(layout02)layout04 = QVBoxLayout()layout04.addLayout(layout03)layout04.addWidget(groupBox)layout04.addWidget(self.pw)layout05 = QVBoxLayout()layout05.addWidget(self.label_num)layout05.addWidget(self.table)layout06 = QHBoxLayout()layout06.addLayout(layout04,5)layout06.addLayout(layout05,1)layout = QVBoxLayout()layout.addLayout(layout00)layout.addWidget(self.label_title)layout.addLayout(layout06)self.setLayout(layout)self.open_init()passdef open_init(self):self.whole_resObj:MultiScatterObj = Noneself.whole_current_df:pd.DataFrame = pd.DataFrame()self.whole_current_show_df:pd.DataFrame = pd.DataFrame()with open(DATA_DIR+'industry_type.json','r',encoding='utf-8') as f:self.industry_type = json.load(f)with open(DATA_DIR+'ticker_name.json','r',encoding='utf-8') as f:self.ticker_name_dict = json.load(f)self.radio_list = []self.whole_current_ticker_list = []passdef btn_choice_clicked(self):file_path,_ = QFileDialog.getOpenFileName(self,'選擇文件')if file_path:self.lineedit_file.setText(file_path)passdef btn_excute_clicked(self):file_path = self.lineedit_file.text()if len(file_path) <= 0:QMessageBox.information(self,'提示','請選擇要執行的py文件',QMessageBox.StandardButton.Ok)returnwith open(file_path,'r',encoding='utf-8') as fr:py_code = fr.read()namespace = {}fun_code = compile(py_code, '<string>', 'exec')exec(fun_code, namespace)res = namespace['execute_caculate']()if res.status == 'error':QMessageBox.information(self,'執行過程報錯',res.error_msg,QMessageBox.StandardButton.Ok)returnself.label_title.setText(res.title)self.whole_resObj = resself.reset_content()QMessageBox.information(self,'提示','執行完畢',QMessageBox.StandardButton.Ok)passdef btn_download_clicked(self):if self.whole_resObj is None or self.whole_resObj.status == 'error':QMessageBox.information(self,'提示','數據為空',QMessageBox.StandardButton.Ok)returndir_name = QFileDialog.getExistingDirectory(self,'選擇保存位置')if dir_name:df = self.whole_resObj.df.copy()df.rename(columns=self.whole_resObj.col_dict,inplace=True)df.to_csv(dir_name+os.path.sep + self.whole_resObj.title +'.csv',encoding='utf-8',index=False)QMessageBox.information(self,'提示','下載完畢',QMessageBox.StandardButton.Ok)passdef combo_type_currentIndexChanged(self,cur_i:int):cur_type = self.combo_type.currentText()self.label_industry.setText(';'.join(self.whole_resObj.type_dict[cur_type].keys()))df = self.whole_resObj.df.copy()key0 = list(self.whole_resObj.type_dict[cur_type].keys())[0]ticker_list = self.whole_resObj.type_dict[cur_type][key0]self.whole_current_df = df.loc[df['ticker'].isin(ticker_list)].copy()self.label_num.setText(f'共{len(ticker_list)}個')self.table.setRowCount(len(ticker_list))for i, item in enumerate(ticker_list):item_name = self.ticker_name_dict.get(item, item)self.table.setItem(i, 0, QTableWidgetItem(str(item)))self.table.setItem(i, 1, QTableWidgetItem(item_name))passself.table.resizeColumnsToContents()self.whole_current_ticker_list = ticker_listdate_list = list(set(self.whole_current_df['reportDate'].to_list()))date_list.sort()date_list.reverse()self.combo_date.addItems(date_list)passdef combo_date_currentIndexChanged(self,cur_i:int):cur_date = self.combo_date.currentText()self.label_date.setText(cur_date)a0 = self.radioButtonGroup.button(2)a0.setChecked(True)self.radioButtonGroup_buttonClicked(a0)passdef radioButtonGroup_buttonClicked(self,a0):indicator_name = a0.text()target_name = ''for k,v in self.whole_resObj.col_dict.items():if indicator_name == v:target_name = kbreakdf = self.whole_resObj.df.copy()df = df.loc[(df['ticker'].isin(self.whole_current_ticker_list)) & (df['reportDate']==self.combo_date.currentText())]df.rename(columns={target_name:'target'},inplace=True)df['x'] = range(len(df))self.whole_current_show_df = df.copy()self.pw.set_data(df.copy(),self.whole_resObj.focus_target)passdef table_itemClicked(self,cur_item):cur_row = cur_item.row()ticker = self.table.item(cur_row,0).text()df = self.whole_current_show_df.loc[self.whole_current_show_df['ticker']==ticker]if df.empty:QMessageBox.information(self,'提示',f'{ticker},在該日期沒有數據',QMessageBox.StandardButton.Ok)returnself.pw.set_targetItem(ticker)passdef reset_content(self):for item in self.radio_list:self.layout_radio.removeWidget(item)self.radioButtonGroup.removeButton(item)self.radio_list.clear()i = 2for item in self.whole_resObj.col_dict.values():radio = QRadioButton(item)self.layout_radio.addWidget(radio)self.radioButtonGroup.addButton(radio,i)self.radio_list.append(radio)i += 1self.pw.set_content_empty()self.combo_type.clear()self.combo_date.clear()self.table.clearContents()self.label_num.setText('--')self.combo_type.addItems(list(self.whole_resObj.type_dict.keys()))self.label_title.setText(self.whole_resObj.title)passpass
使用舉例
需要導入的包和運行代碼
import os,sys,json
import pandas as pd
import numpy as np
from PyQt6.QtCore import (
QSize,
Qt
)
from PyQt6.QtWidgets import (QApplication, QButtonGroup, QRadioButton,QMainWindow, QAbstractItemView,QLabel,QPushButton,QComboBox,QTableWidget,QTableWidgetItem,QTextEdit,QWidget,QVBoxLayout,QHBoxLayout, QGridLayout,QFileDialog,QInputDialog,QMessageBox,QLineEdit,QGroupBox, QScrollArea, QCompleter
)
import pyqtgraph as pg
from objects import MultiScatterObj
from settings import DATA_DIRif __name__ == '__main__':app = QApplication(sys.argv)mw = PyExcuteGraphShowWidgetII()mw.show()app.exec()pass
1)一個py文件例子,內容如下,方法名固定為 excute_caculate
def execute_caculate():import traceback,json,osimport pandas as pdfrom utils import postgresql_utilsfrom objects import MultiScatterObjfrom settings import DATA_DIR,INDUSTRY_DIR'''靈活py文件執行營業利潤,營業外支出,營業外收入'''conn = postgresql_utils.connect_db()cur = conn.cursor()try:ticker = '000638'file_list = os.listdir(INDUSTRY_DIR)type_dict = {}ticker_list = []for file_one in file_list:type_code = file_one[0:6]file_path = os.path.join(INDUSTRY_DIR, file_one)with open(file_path, 'r',encoding='utf-8') as f:j_obj = json.load(f)# type_dict[type_code] = {}for k,v in j_obj.items():if ticker in v:if type_dict.get(type_code,None) is None:type_dict[type_code] = {}type_dict[type_code][k] = vticker_list.extend(v)passpassticker_list = list(set(ticker_list))ticker_list_str = '\',\''.join(ticker_list)ticker_list_str = '\''+ticker_list_str+'\''sql_str = f'''select ticker,reportDate,iii_operateProfit,add_nonoperateIncome,less_nonoperateExpenses from t_profit where ticker in ({ticker_list_str}) and reportDate like \'%-12-31\';'''cur.execute(sql_str)res = cur.fetchall()col_list = ['ticker','reportDate','a0','a1','a2']col_dict = {'a0':'營業利潤','a1':'營業外收入','a2':'營業外支出'}df = pd.DataFrame(columns=col_list, data=res)res_obj = MultiScatterObj(title=f'{ticker},營業利潤、營業外收入、營業外支出',df=df,col_dict=col_dict,type_dict=type_dict,focus_target=ticker,status='ok')return res_objexcept:res_obj = MultiScatterObj(status='error',error_msg=traceback.format_exc())return res_objfinally:cur.close()conn.close()passpass
保存為test003.py
注意:例子中涉及到的postgreSQL和財報數據在往期博文中可以找到
2)點擊“選擇文件”,選擇 test002.py文件
3)點擊“執行”,選擇行業、日期、指標,就能顯示散點圖
4)右側股票列表,單擊某個股票,就會在散點圖中用紅星標注