Python 爬虫抓取金融/股票数据并实时分析实战

本文介绍了使用Python爬虫抓取股票金融数据并进行实时分析的完整流程。首先讲解了技术选型，包括数据源选择、爬虫库和可视化工具。随后详细演示了如何通过requests和aiohttp获取新浪股票数据，并实现批量抓取和异步处理。文章还涵盖了实时数据分析方法，如涨跌幅计算、移动均线和波动率分析，以及使用matplotlib、plotly和streamlit进行数据可视化。最后提供了优化大规模金融数据抓

shanwei_spider

359人浏览 · 2025-09-05 08:19:05

shanwei_spider · 2025-09-05 08:19:05 发布

在金融领域，实时数据对于交易决策、行情分析和量化策略至关重要。Python 拥有丰富的数据抓取、分析和可视化工具，使得金融数据实时处理变得可行。本文将带你从零到一，完成 Python 爬虫抓取股票/金融数据并实时分析 的完整实战流程。

一、技术选型

数据源
- 新浪财经、东方财富、雅虎财经等公开 API 或网页
- 重点关注股票实时行情、历史行情、财务指标等
Python 爬虫库
- requests / aiohttp：同步与异步 HTTP 请求
- BeautifulSoup / lxml：网页解析
- pandas：数据清洗与分析
实时分析与可视化
- pandas：计算涨跌幅、均线、波动率
- matplotlib / plotly：图表可视化
- streamlit 或 dash：构建实时数据仪表盘

二、抓取股票实时行情

1. 使用 requests 抓取新浪股票数据

import requests
import pandas as pd

url = "http://hq.sinajs.cn/list=sh600000,sz000001"
res = requests.get(url)
res.encoding = "gbk"
data = res.text.split(',')

stock_info = {
    "symbol": "sh600000",
    "name": data[0].split('"')[1],
    "open": float(data[1]),
    "prev_close": float(data[2]),
    "price": float(data[3]),
    "high": float(data[4]),
    "low": float(data[5]),
    "volume": int(data[8])
}

df = pd.DataFrame([stock_info])
print(df)

说明：

URL 中的 sh600000 和 sz000001 分别为沪深两市股票代码
数据包含开盘价、收盘价、当前价、最高、最低、成交量

2. 批量抓取多个股票

stocks = ["sh600000", "sz000001", "sh600519"]
url = "http://hq.sinajs.cn/list=" + ",".join(stocks)
res = requests.get(url)
res.encoding = "gbk"
lines = res.text.strip().split("\n")

data_list = []
for line in lines:
    parts = line.split(",")
    info = {
        "symbol": parts[0].split('"')[1],
        "name": parts[0].split('"')[0][-6:],
        "price": float(parts[3]),
        "volume": int(parts[8])
    }
    data_list.append(info)

df = pd.DataFrame(data_list)
print(df)

三、实时抓取与异步处理

当股票数量很多时，可以使用 aiohttp + asyncio 提高抓取效率：

import asyncio
import aiohttp
import pandas as pd

stocks = ["sh600000", "sz000001", "sh600519"]
url_template = "http://hq.sinajs.cn/list={}"

async def fetch(session, url):
    async with session.get(url) as resp:
        text = await resp.text(encoding="gbk")
        return text

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url_template.format(stock)) for stock in stocks]
        results = await asyncio.gather(*tasks)

        data_list = []
        for line in results:
            parts = line.split(",")
            info = {
                "symbol": parts[0].split('"')[1],
                "name": parts[0].split('"')[0][-6:],
                "price": float(parts[3]),
                "volume": int(parts[8])
            }
            data_list.append(info)
        df = pd.DataFrame(data_list)
        print(df)

asyncio.run(main())

亮点：

多股票同时抓取，提高抓取速度
异步 IO 减少阻塞

四、实时数据分析

1. 计算涨跌幅

df["change_pct"] = (df["price"] - df["prev_close"]) / df["prev_close"] * 100

2. 计算移动均线

# 假设 df_stock 为历史数据
df_stock["ma5"] = df_stock["price"].rolling(5).mean()
df_stock["ma10"] = df_stock["price"].rolling(10).mean()

3. 波动率计算

df_stock["returns"] = df_stock["price"].pct_change()
volatility = df_stock["returns"].rolling(10).std()

五、可视化实时数据

1. Matplotlib 简单折线图

import matplotlib.pyplot as plt

plt.plot(df_stock["price"], label="Price")
plt.plot(df_stock["ma5"], label="MA5")
plt.plot(df_stock["ma10"], label="MA10")
plt.legend()
plt.show()

2. Plotly 实时交互图

import plotly.graph_objects as go

fig = go.Figure()
fig.add_trace(go.Scatter(y=df_stock["price"], name="Price"))
fig.add_trace(go.Scatter(y=df_stock["ma5"], name="MA5"))
fig.add_trace(go.Scatter(y=df_stock["ma10"], name="MA10"))
fig.show()

3. Streamlit 实时仪表盘

import streamlit as st
import pandas as pd
import time

st.title("实时股票行情")

while True:
    df = get_stock_data(stocks)  # 封装前面抓取逻辑
    st.dataframe(df)
    time.sleep(5)  # 每5秒更新一次