背景分析

背景概述

在当今快速发展的金融市场中，股票市场的动态变化对投资者和金融分析师至关重要。随着信息技术的进步，特别是互联网和大数据技术的发展，投资者可以实时获得大量的股票市场信息。然而，这些信息通常以原始数据形式存在，需要进一步处理和分析才能提取有价值的洞见。在这种背景下，网络股票信息爬虫与可视化技术应运而生，成为股票市场分析的重要工具。

网络股票信息爬虫

网络股票信息爬虫（Web Scraping）是一种自动化的数据收集技术，旨在从互联网上提取股票市场相关的信息。它利用程序化方法访问和抓取各种金融网站和新闻平台上的数据，例如股票价格、交易量、公司财报、新闻动态等。爬虫技术可以高效地从大规模数据源中获取最新的信息，支持实时数据更新和历史数据回溯分析。

关键技术:

  • 爬虫框架：如Scrapy、BeautifulSoup等，用于解析HTML页面和提取数据。
    • 由于本文的目标网站的股票数据并不在页面中，也就是说不需要做网页分析和提取
  • API接口：一些金融网站提供API接口，允许程序直接获取结构化的数据。
    • (例如本文使用Get方法的https请求，获取了json文件)
  • 数据存储：抓取的数据通常需要存储在数据库中，如MySQL、MongoDB等，以便后续分析和处理。
    • (本文使用作者个人服务器上MYSQL作为存储数据库)

数据可视化

数据可视化是将复杂的股票市场数据以图形化的方式呈现出来，使得数据更易于理解和分析。通过可视化技术，投资者可以直观地观察股票价格趋势、交易量变化、市场波动等重要信息，从而做出更加明智的投资决策。

常用可视化方法:

  • 时间序列图：示股票价格或交易量随时间变化的趋势。
  • 蜡烛图（CandlestickChart）:用于股票价格在一定时间内的开盘、收盘、最高和最低价格。
    • 又称K图，文本就是使用K图作为可视化方式
  • 热力图（Heatmap）:用颜色强度表示不同股票的表现或市场的热点区域。
  • 散点图和折线图：用于展示多维数据之间的关系和趋势。

小结

网络股票信息爬虫与可视化技术在金融市场分析中具有广泛的应用。它们不仅可以帮助投资者实时监控市场动态，还能够通过历史数据分析预测未来趋势。此外，这些技术还被用于量化分析、算法交易、风险管理等领域，以提高投资决策的科学性和准确性。

网络股票信息爬虫与可视化技术为现代金融市场提供了强大的数据支持和分析工具。通过自动化数据抓取和图形化呈现，投资者可以更高效地获取市场信息和洞察，有助于做出更准确的投资决策。随着技术的不断发展，这些工具将在金融分析中发挥越来越重要的作用。

业务流程图

代码实现业务功能

代码框架

  • 网络爬虫
    • 使用requests模块向网站发送定制的https请求
    • 使用json处理网站返回的json数据
    • 爬虫部分的代码封装在SnowBallCrawler.py中
  • 数据库持久化
    • 在mysql中创建项目所需的数据库和表
    • 使用pymysql模块连接数据库并执行插入和查询操作
  • GUI可视化
    • 使用dearpygui模块实现GUI可视化并管理组件
    • 使用dearpygui的统计图绘制功能绘制K图
    • 项目的主要逻辑集成在可视化代码的主要部分
    • 将自定义的复杂组件封装在类里

网络爬虫

我们首先来实现网络爬虫部分，这里的目标网站是雪球网站的茅台股票

网页分析

一般股票网站的数据不会直接放在原始的https请求的响应中，所以要打开F12分析网站的二次请求的内容，方法如图

  1. 按下f12打开网页分析器
  2. 点击选项卡中的网络(Network)切换到对应的页面,并选中下方的Fetch/XHR分类卡
  3. 刷新网页，并点击网站中的日K选项，鼠标悬停在某一天的线上
  4. 从悬停提示的数字中选一个比较有唯一性的（比如收盘价）,在搜索中填入选中的数字
  5. 一般这样就能选出唯一的响应了，里面就存着我们所需要的数据（我们这个例子就有280多条数据）

如图，我们找到了想要的响应，接下来我们分析标头header,接下来点击标头选项卡

  • 请求URL

如图，目标网站使用了GET方法请求数据，所以我们要着重分析GET方法的参数和无参URL(用于拼接请求URL)，而参数就在URL中

拆分参数并转换成python字典后如下

1
2
3
4
5
6
7
8
param_data = {
            'symbol':'SH600519',
            'begin':1726713214191,
            'period':'day',
            'type':'before',
            'count':'-284',
            'indicator':'kline,pe,pb,ps,pcf,market_capital,agt,ggt,balance'
        }

很明显，begin是时间戳，用于请求最新的数据,但是比较特别的，网站用的是整数的毫秒级时间戳,在python中使用time模块传当前的begin值时得注意

因为本项目只爬取一支股票的信息，剩下的参数不作解析

  • 标头(header)

为了防止被网站的反爬取给拦截下来，标头需要和自己在网站里查到的保持一致

转成python字典的代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
self.headers = {
            'Accept':'application/json, text/plain, */*',
            'Accept-Encoding':'gzip, deflate, br, zstd',
            'Accept-Language':'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
            'Cookie':'请使用自己的Cookie,本文作者已隐去',
            'Origin':'https://xueqiu.com',
            'Priority':'u=1, i',
            'Referer':'https://xueqiu.com/S/SH600519',
            'Sec-Ch-Ua':'"Not/A)Brand";v="8", "Chromium";v="126", "Microsoft Edge";v="126"',
            'sec-ch-ua-mobile':'?0',
            'sec-ch-ua-platform':'"Windows"',
            'Sec-Fetch-Dest':'empty',
            'Sec-Fetch-Mode':'cors',
            'Sec-Fetch-Site':'same-site',
            'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0'
        }

封装类

因为只爬取一支股票，就不封装通用性强的类了，而是只针对这一支股票，所以封装一个GZMaoTai类,并提供get_response接口，向外界传送每次爬取的数据。其中对返回的joson格式的字符串，使用Josn模块将其反序列化，并储存在Python列表中返回

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import requests
import json
import time

class GZMaoTai:
    def __init__(self):

        self.url = 'https://stock.xueqiu.com/v5/stock/chart/kline.json'
        self.headers = {
            'Accept':'application/json, text/plain, */*',
            'Accept-Encoding':'gzip, deflate, br, zstd',
            'Accept-Language':'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
            'Cookie':'请使用自己的Cookie,本文作者已隐去',
            'Origin':'https://xueqiu.com',
            'Priority':'u=1, i',
            'Referer':'https://xueqiu.com/S/SH600519',
            'Sec-Ch-Ua':'"Not/A)Brand";v="8", "Chromium";v="126", "Microsoft Edge";v="126"',
            'sec-ch-ua-mobile':'?0',
            'sec-ch-ua-platform':'"Windows"',
            'Sec-Fetch-Dest':'empty',
            'Sec-Fetch-Mode':'cors',
            'Sec-Fetch-Site':'same-site',
            'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0'
        }


    def get_response(self,date):
        nowtime = int(time.time() * 1000.0)
        param_data = {
            'symbol':'SH600519',
            'begin':str(nowtime),
            'period':'day',
            'type':'before',
            'count':'-284',
            'indicator':'kline,pe,pb,ps,pcf,market_capital,agt,ggt,balance'
        }
        print("开始爬取数据")
        response = requests.get(self.url,headers=self.headers,params=param_data,timeout=5)
        print(response.text)
        print("爬取成功，数据包如上")

        data_json = json.loads(response.text)  # Json反序列化
        items = data_json['data']['item']
        # for i in range(len(items)):
        # print(items[i])
        # json数据对应
        # [0] timestamp
        # [1] 成交量
        # [2] 开盘价
        # [3] 最高价
        # [4] 最低价
        # [5] 收盘价
        # [6] 涨跌额
        # [7] 涨跌幅
        # [8] 换手率
        # [9] 成交额(亿)
        # [10] null
        # [11] null
        # [12]
        return items

数据库持久化

这里使用远端的MYSQL管理数据库,作者使用了自己的服务器,具体的配置和安装不作介绍,我们创建一个专门管理股票数据库的远程登录账户supdiver

对于数据库的操作主要是封装几个接口，所以就直接把有关的函数声明在DBManager.py模块中，而不封装成类了。

注:host为mysql所在服务器的公有ip,本文将作者自己服务器的实际ip改成了本地路由环回的127.0.0.1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import pymysql

stockDB = pymysql.Connect(
    host='47.99.48.121',
    port=3306,
    user='supdriver',
    password='',
    db='stock',
    charset='utf8',
    autocommit=True
)


def sendData(items):
    stockDB.ping(reconnect=True)
    cursor = stockDB.cursor()
    print("开始向数据库插入数据")
    for item in items:
        sql=(f"insert into MaoTai values ({item[0]},"
             f"{item[1]},"
             f"{item[2]},"
             f"{item[3]},"
             f"{item[4]},"
             f"{item[5]},"
             f"{item[6]},"
             f"{item[7]},"
             f"{item[8]},"
             f"{item[9]});")
        # print(sql)
        cursor.execute(sql)
        # print(cursor.fetchall())
    print(f"插入数据成功，总计{len(items)}")

def getAllData(stockDB):
    stockDB.ping(reconnect=True)
    cursor = stockDB.cursor()
    sql= "select * from MaoTai"
    cursor.execute(sql)
    return cursor.fetchall()

def getLatestData():
    stockDB.ping(reconnect=True)

    cursor=stockDB.cursor()
    sql="select * from MaoTai order by timesamp desc"
    cursor.execute(sql)
    alldata = cursor.fetchall()
    resList = []
    for line in alldata:
        resList.append(line)

    return resList
def closeDB():
    stockDB.close()

def clearDB():
    stockDB.ping(reconnect=True)

    cursor = stockDB.cursor()
    sql='delete from MaoTai'
    print("开始清理数据库")
    cursor.execute(sql)
    print('清理完成')