PyQt5製做一個爬蟲小工具，爬取雪球網上市公司的財務數據

本文的文字及圖片來源於網絡,僅供學習、交流使用,不具備任何商業用途,若有問題請及時聯繫咱們以做處理。

如下文章來源於能夠叫我才哥 ，做者：能夠叫我才哥

 

最近有朋友須要幫忙寫個爬蟲腳本，爬取雪球網一些上市公司的財務數據。盆友但願能夠根據他本身的選擇進行自由的抓取，因此簡單給一份腳本交給盆友，盆友還須要本身搭建python環境，更須要去熟悉一些參數修改的操做，想來也是太麻煩了。

因而，結合以前作過的匯率計算器小工具，我這邊決定使用PyQt5給朋友製做一個爬蟲小工具，方便他的操做可視化。

1、效果演示

 

2、功能說明

• 能夠自由選擇證券市場類型：A股、美股和港股
• 能夠自由選擇上市公司：單選或全選
• 能夠自由選擇財務數據類型：單選或全選（主要指標、利潤表、資產負債表、現金流表）
• 能夠導出數據存儲爲excel表格文件
• 支持同一家上市公司同類型財務數據追加

3、製做過程

首先引入須要的庫

import sys
from PyQt5 import QtCore, QtGui, QtWidgets
from PyQt5.QtWidgets import QApplication, QMainWindow,QFileDialog

import os
import requests
from fake_useragent import UserAgent
import json
import  logging
import time
import pandas as pd
from openpyxl import load_workbook

 

雪球網頁拆解

這一步的目的是獲取須要爬取的數據的真正URL地址規律。

當我選中某隻股票查看財務數據某類型數據報告時，點擊下一頁，網站地址沒有變化，基本能夠知道這是動態加載的數據，對於這類數據可使用F12打開開發者模式。

 

在開發者模式下，選到Network—>XHR能夠查看到真正的數據獲取地址URL及請求方式（General裏是請求URL和請求方式說明，Request Headers有請求頭信息，如cookie，Query String Parameters就是可變參數項，通常來講數據源URL就是由基礎URL和這裏的可變參數組合而成）

 

咱們分析這段URL，能夠發現其基本結構以下：

 

基於上述結構，咱們拆分最終的組合URL地址以下

#基礎網站
base_url = f'https://stock.xueqiu.com/v5/stock/finance/{ABtype}'

#組合url地址
url = f'{base_url}/{data_type}.json?symbol={ipo_code}&type=all&is_detail=true&count={count_num}×tamp={start_time}'

 

操做界面設計

操做界面設計使用的是PyQt5，這裏不作更詳細的介紹，咱們在後續中對PyQt5的使用再專題講解。

使用QT designer對操做界面進行可視化設計，參考以下：

 

雪球網數據提取.ui中各個組件的相關設置，參考以下：

 

.ui文件可使用pyuic5指令進行編譯生成對應的.py文件，或者咱們也能夠在vscode裏直接轉譯（這裏也不作更詳細的介紹，具體見後續專題講解）。

本文沒有將操做界面定義文件單獨使用，而是將所有代碼集中在同一個.py文件，所以其轉譯後的代碼備用便可。

獲取cookie及基礎參數

獲取cookie

爲了便於小工具拿來便可使用，咱們須要自動獲取cookie地址並附加在請求頭中，而不是人爲打開網頁在開發者模式下獲取cookie後填入。

自動獲取cookie，這裏使用到的requests庫的session會話對象。

requests庫的session會話對象能夠跨請求保持某些參數，簡單來講，就是好比你使用session成功的登陸了某個網站，則在再次使用該session對象請求該網站的其餘網頁都會默認使用該session以前使用的cookie等參數

import requests
from fake_useragent import UserAgent

url = 'https://xueqiu.com'

session = requests.Session()
headers = {"User-Agent": UserAgent(verify_ssl=False).random}
 
session.get(url, headers=headers)
   
#獲取當前的Cookie
Cookie= dict(session.cookies)

 

基礎參數

基礎參數是用於財務數據請求時原始網址構成參數選擇，咱們在可視化操做工具中須要對財務數據類型進行選擇，所以這裏須要構建財務數據類型字典。

#原始網址
original_url = 'https://xueqiu.com'
#財務數據類型字典
dataType = {'全選':'all',
           '主要指標':'indicator',
           '利潤表':'income',
           '資產負債表':'balance',
           '現金流量表':'cash_flow'}

 

獲取獲取各證券市場上市名錄

由於咱們在可視化操做工具上是選定股票代碼後抓取相關數據並導出，對導出的文件名稱但願是以股票代碼+公司名稱的形式(SH600000 浦發銀行)存儲，因此咱們須要獲取股票代碼及名稱對應關係的字典表。

這其實就是一個簡單的網絡爬蟲及數據格式調整的過程，實現代碼以下：

 1import requests
 2import pandas as pd
 3import json
 4from fake_useragent import UserAgent 
 5#請求頭設置
 6headers = {"User-Agent": UserAgent(verify_ssl=False).random}
 7#股票清單列表地址解析（經過設置參數size爲9999能夠只使用1個靜態地址，所有股票數量不足5000）
 8url = 'https://xueqiu.com/service/v5/stock/screener/quote/list?page=1&size=9999&order=desc&orderby=percent&order_by=percent&market=CN&type=sh_sz'
 9#請求原始數據
10response = requests.get(url,headers = headers)
11#獲取股票列表數據
12df = response.text
13#數據格式轉化
14data = json.loads(df)
15#獲取所須要的股票代碼及股票名稱數據
16data = data['data']['list']
17#將數據轉化爲dataframe格式，並進行相關調整
18data = pd.DataFrame(data)
19data = data[['symbol','name']]
20data['name'] = data['symbol']+' '+data['name']
21data.sort_values(by = ['symbol'],inplace=True)
22data = data.set_index(data['symbol'])['name']
23#將股票列表轉化爲字典，鍵爲股票代碼，值爲股票代碼和股票名稱的組合
24ipoCodecn = data.to_dict()

 

A股股票代碼及公司名稱字典以下：

 

獲取上市公司財務數據並導出

根據在可視化操做界面選擇的 財務報告時間區間、財務報告數據類型、所選證券市場類型以及所輸入的股票代碼後，須要先根據這些參數組成咱們須要進行數據請求的網址，而後進行數據請求。

因爲請求後的數據是json格式，所以能夠直接進行轉化爲dataframe類型，而後進行導出。在數據導出的時候，咱們須要判斷該數據文件是否存在，若是存在則追加，若是不存在則新建。

獲取上市公司財務數據

經過選定的參數生成財務數據網址，而後根據是否全選決定後續數據請求的操做，所以能夠拆分爲獲取數據網址和請求詳情數據兩部分。

獲取數據網址

數據網址是根據證券市場類型、財務數據類型、股票代碼、單頁數量及起始時間戳決定，而這些參數都是經過可視化操做界面進行設置。

證券市場類型 控件 是radioButton，能夠經過你 ischecked() 方法判斷是否選中，而後用if-else進行參數設定；

財務數據類型 和 股票代碼 由於支持 全選，須要先進行全選斷定（全選條件下是須要循環獲取數據網址，不然是單一獲取便可），所以這部分須要再作拆分；

單頁數量 考慮到每一年有4份財務報告，所以這裏默認爲年份差*4；

時間戳 是 根據起始時間中的 結束時間 計算得出，因爲可視化界面輸入的 是 整數年份，咱們能夠經過 mktime() 方法獲取時間戳。

 1def Get_url(self,name,ipo_code):
 2   #獲取開始結束時間戳（開始和結束時間手動輸入）
 3   inputstartTime = str(self.start_dateEdit.date().toPyDate().year)
 4   inputendTime = str(self.end_dateEdit.date().toPyDate().year)
 5   endTime = f'{inputendTime}-12-31 00:00:00'
 6   timeArray = time.strptime(endTime, "%Y-%m-%d %H:%M:%S")
 7
 8   #獲取指定的數據類型及股票代碼
 9   filename = ipo_code
10   data_type =dataType[name]
11   #計算須要採集的數據量(一年以四個算)
12   count_num = (int(inputendTime) - int(inputstartTime) +1) * 4
13   start_time =  f'{int(time.mktime(timeArray))}001'
14
15   #證券市場類型
16   if (self.radioButtonCN.isChecked()):
17       ABtype = 'cn'
18       num = 3
19   elif (self.radioButtonUS.isChecked()):
20       ABtype = 'us'
21       num = 6
22   elif (self.radioButtonHK.isChecked()):
23       ABtype = 'hk'
24       num = 6
25   else:
26       ABtype = 'cn'
27       num = 3
28
29   #基礎網站
30   base_url = f'https://stock.xueqiu.com/v5/stock/finance/{ABtype}'
31
32   #組合url地址
33   url = f'{base_url}/{data_type}.json?symbol={ipo_code}&type=all&is_detail=true&count={count_num}×tamp={start_time}'
34
35   return url,num

 

請求詳情數據

須要根據用戶輸入決定數據採集方式，代碼中主要是根據用戶輸入作判斷而後再進行詳情數據請求。

 1#根據用戶輸入決定數據採集方式
 2def Get_data(self):
 3   #name爲財務報告數據類型（全選或單個）
 4   name = self.Typelist_comboBox.currentText()
 5   #股票代碼（全選或單個）
 6   ipo_code = self.lineEditCode.text()
 7   #判斷證券市場類型
 8   if (self.radioButtonCN.isChecked()):
 9       ipoCodex=ipoCodecn
10   elif (self.radioButtonUS.isChecked()):
11       ipoCodex=ipoCodeus
12   elif (self.radioButtonHK.isChecked()):
13       ipoCodex=ipoCodehk
14   else:
15       ipoCodex=ipoCodecn
16#根據財務報告數據類型和股票代碼類型決定數據採集的方式
17   if name == '全選' and ipo_code == '全選':
18       for ipo_code in list(ipoCodex.keys()):
19           for name in list(dataType.keys())[1:]:
20               self.re_data(name,ipo_code)
21   elif name == '全選' and ipo_code != '全選':
22           for name in list(dataType.keys())[1:]:
23               self.re_data(name,ipo_code)
24   elif ipo_code == '全選' and name != '全選':
25       for ipo_code in list(ipoCodex.keys()):
26           self.re_data(name,ipo_code)            
27   else:
28       self.re_data(name,ipo_code)
29
30#數據採集，須要調用數據網址（Get.url(name,ipo_code)    
31def re_data(self,name,ipo_code):
32   name = name
33   #獲取url和num（url爲詳情數據網址，num是詳情數據中根據不一樣證券市場類型決定的須要提取的數據起始位置）
34   url,num = self.Get_url(name,ipo_code)
35   #請求頭
36   headers = {"User-Agent": UserAgent(verify_ssl=False).random}
37   #請求數據
38   df = requests.get(url,headers = headers,cookies = cookies)
39
40   df = df.text
41try:
42      data = json.loads(df)
43  pd_df = pd.DataFrame(data['data']['list'])
44  to_xlsx(num,pd_df)
45   except KeyError:
46       log = '<font color=\"#FF0000\">該股票此類型報告不存在,請從新選擇股票代碼或數據類型</font>'
47       self.rizhi_textBrowser.append(log)  

 

財務數據處理並導出

單純的數據導出是比較簡單的操做，直接to_excel() 便可。可是考慮到同一個上市公司的財務數據類型有四種，咱們但願都保存在同一個文件下，且對於同類型的數據可能存在分批導出的狀況但願能追加。所以，須要進行特殊的處理，用pd.ExcelWriter()方法操做。

 1#數據處理並導出
 2def to_xlsx(self,num,data):
 3   pd_df = data
 4   #獲取可視化操做界面輸入的導出文件保存文件夾目錄
 5   filepath = self.filepath_lineEdit.text()
 6   #獲取文件名
 7   filename = ipoCode[ipo_code]  
 8   #組合成文件詳情（地址+文件名+文件類型）
 9   path = f'{filepath}\{filename}.xlsx'
10   #獲取原始數據列字段
11   cols = pd_df.columns.tolist()
12   #建立空dataframe類型用於存儲
13   data = pd.DataFrame()    
14   #建立報告名稱字段            
15   data['報告名稱'] = pd_df['report_name']
16   #因爲不一樣證券市場類型下各股票財務報告詳情頁數據從不一樣的列纔是須要的數據，所以須要用num做爲起點
17   for i in range(num,len(cols)):
18       col = cols[i]
19       try:
20           #每列數據中是列表形式，第一個是值，第二個是同比
21           data[col] = pd_df[col].apply(lambda x:x[0])
22       # data[f'{col}_同比'] = pd_df[col].apply(lambda x:x[1])
23       except TypeError:
24           pass
25   data = data.set_index('報告名稱')      
26   log = f'{filename}的{name}數據已經爬取成功'
27   self.rizhi_textBrowser.append(log)
28   #因爲存儲的數據行索引爲數據指標，因此須要對採集的數據進行轉T處理
29   dataT = data.T
30   dataT.rename(index = eval(f'_{name}'),inplace=True)
31   #如下爲判斷數據報告文件是否存在，若存在則追加，不存在則從新建立
32   try:
33       if os.path.exists(path):
34           #讀取文件所有頁籤
35           df_dic = pd.read_excel(path,None)
36           if name not in list(df_dic.keys()):
37               log = f'{filename}的{name}數據頁籤不存在，建立新頁籤'
38               self.rizhi_textBrowser.append(log)
39               #追加新的頁籤
40               with pd.ExcelWriter(path,mode='a') as writer:
41                   book = load_workbook(path)    
42                   writer.book = book    
43                   dataT.to_excel(writer,sheet_name=name)
44                   writer.save()
45           else:
46               log = f'{filename}的{name}數據頁籤已存在，合併中'
47               self.rizhi_textBrowser.append(log)
48               df = pd.read_excel(path,sheet_name = name,index_col=0)
49               d_ = list(set(list(dataT.columns)) - set(list(df.columns)))
50#使用merge()進行數據合併
51               dataT = pd.merge(df,dataT[d_],how='outer',left_index=True,right_index=True)
52               dataT.sort_index(axis=1,ascending=False,inplace=True)
53               #頁籤中追加數據不影響其餘頁籤
54               with pd.ExcelWriter(path,engine='openpyxl') as writer:  
55                   book = load_workbook(path)    
56                   writer.book = book
57                   idx = writer.book.sheetnames.index(name)
58                   #刪除同名的，而後從新建立一個同名的
59                   writer.book.remove(writer.book.worksheets[idx])
60                   writer.book.create_sheet(name, idx)
61                   writer.sheets = {ws.title:ws for ws in writer.book.worksheets}        
62
63                   dataT.to_excel(writer,sheet_name=name,startcol=0)
64                   writer.save()
65       else:
66           dataT.to_excel(path,sheet_name=name)
67
68       log = f'<font color=\"#00CD00\">{filename}的{name}數據已經保存成功</font>'
69       self.rizhi_textBrowser.append(log)
70
71   except FileNotFoundError:
72       log = '<font color=\"#FF0000\">未設置存儲目錄或存儲目錄不存在,請從新選擇文件夾</font>'
73       self.rizhi_textBrowser.append(log)