一、环境准备

1.1 依赖库安装

pip install requests

本项目主要使用以下库：

• requests：发送HTTP请求
• csv：数据存储
• time、random：实现延时策略

其余模块均为Python标准库，无需额外安装。

1.2 项目结构

雪球网股票分析/
├── xueqiu_stocks.py    # 主程序
└── xueqiu_stocks.csv    # 输出数据文件

二、核心代码实现

2.1 完整代码

import time
import csv
import os
import random
import requests
from datetime import datetime


BASE_URL = "https://xueqiu.com/service/v5/stock/screener/quote/list"
HOME_URL = "https://xueqiu.com"

HEADERS_TEMPLATE = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/125.0.0.0 Safari/537.36"
    ),
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Referer": "https://xueqiu.com/hq/detail?market=CN&first_name=0&second_name=0&type=sh_sz",
    "Origin": "https://xueqiu.com",
}

# 请求参数（可按需修改）
PARAMS = {
    "page": 1,
    "size": 90,           # 每页条数，最大 90
    "order": "desc",       # 排序方向
    "order_by": "percent",  # 排序字段
    "market": "CN",
    "type": "sh_sz",       # sh_sz = 沪深
}

OUTPUT_FILE = "xueqiu_stocks.csv"
MIN_DELAY = 3      # 请求间隔下限（秒）
MAX_DELAY = 6      # 请求间隔上限（秒），随机 sleep [MIN, MAX) 之间
MAX_RETRIES = 3    # 单页最大重试次数


def get_session_with_cookies():
    """
    先访问雪球首页获取有效 Cookie
    返回带 Cookie 的 requests.Session。
    """
    session = requests.Session()
    session.headers.update(HEADERS_TEMPLATE)

    print(f"[{datetime.now():%H:%M:%S}] 正在获取 Cookie（访问首页）...")
    resp = session.get(HOME_URL, timeout=15)
    resp.raise_for_status()

    return session


def fetch_page(session, page):
    """采集单页数据，返回 JSON 或 None"""
    params = {**PARAMS, "page": page}

    for attempt in range(1, MAX_RETRIES + 1):
        try:
            resp = session.get(BASE_URL, params=params, timeout=15)
            if resp.status_code == 403:
                print(f"  [!] 第{page}页 403 Forbidden，Cookie 可能已过期")
                return None
            resp.raise_for_status()
            data = resp.json()

            if data.get("error_code") != 0:
                print(f"  [!] 第{page}页 API 返回错误: {data.get('error_description', '未知')}")
                return None

            return data

        except requests.RequestException as e:
            print(f"  [!] 第{page}页 请求失败 (尝试 {attempt}/{MAX_RETRIES}): {e}")
            if attempt < MAX_RETRIES:
                time.sleep(random.uniform(MIN_DELAY * 2, MAX_DELAY * 2))
            else:
                return None


def extract_stocks(data):
    """从 API 响应中提取股票列表"""
    stock_list = []
    if not data or "data" not in data or "list" not in data["data"]:
        return stock_list

    for item in data["data"]["list"]:
        stock_list.append({
            "代码": item.get("symbol", ""),
            "名称": item.get("name", ""),
            "当前价": item.get("current", ""),
            "涨跌幅(%)": item.get("percent", ""),
            "涨跌额": item.get("chg", ""),
            "成交量(手)": item.get("volume", ""),
            "成交额(亿)": (
                round(item["amount"] / 100_000_000, 2)
                if item.get("amount") is not None else ""
            ),
            "振幅(%)": item.get("amplitude", ""),
            "换手率(%)": item.get("turnover_rate", ""),
            "市盈率(动态)": item.get("pe_ttm", ""),
            "市净率": item.get("pb", ""),
            "总市值(亿)": (
                round(item["market_capital"] / 100_000_000, 2)
                if item.get("market_capital") is not None else ""
            ),
            "流通市值(亿)": (
                round(item["float_market_capital"] / 100_000_000, 2)
                if item.get("float_market_capital") is not None else ""
            ),
            "量比": item.get("volume_ratio", ""),
            "每股收益": item.get("eps", ""),
        })
    return stock_list


def save_to_csv(stocks, file_path, mode="w"):
    """保存股票数据到 CSV 文件"""
    if not stocks:
        print("  [-] 无数据可保存")
        return

    fieldnames = list(stocks[0].keys())
    is_new = not os.path.exists(file_path) or mode == "w"

    with open(file_path, mode=mode, newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if is_new:
            writer.writeheader()
        writer.writerows(stocks)

    print(f"  [+] 已保存 {len(stocks)} 条记录 -> {file_path}")


def crawl(total_pages=10):
    """
    主采集流程

    Args:
        total_pages: 采集页数（如需全部页可传入较大的数，API 会自动截断）
    """
    os.makedirs(os.path.dirname(OUTPUT_FILE) or ".", exist_ok=True)

    session = get_session_with_cookies()
    all_count = 0
    start_time = time.time()

    print(f"\n{'='*50}")
    print(f"启动采集 | 目标页数: {total_pages} | 每页: {PARAMS['size']} 条")
    print(f"{'='*50}\n")

    for page in range(1, total_pages + 1):
        print(f"[{datetime.now():%H:%M:%S}] 正在采集第 {page} 页...")

        data = fetch_page(session, page)
        if data is None:
            print(f"  [-] 第 {page} 页采集失败，终止翻页")
            break

        stocks = extract_stocks(data)
        if not stocks:
            print(f"  [-] 第 {page} 页无数据，采集结束")
            break

        # 总数据量（仅在首页获取）
        total_count = data["data"].get("count", 0)
        if page == 1:
            print(f"  [i] 共匹配 {total_count} 只标的")

        # 写入文件（首页覆盖，后续追加）
        mode = "w" if page == 1 else "a"
        save_to_csv(stocks, OUTPUT_FILE, mode=mode)
        all_count += len(stocks)

        # 随机间隔，避免触发反爬
        if page < total_pages:
            delay = random.uniform(MIN_DELAY, MAX_DELAY)
            print(f"等待 {delay:.1f} 秒后翻页...")
            time.sleep(delay)

    elapsed = time.time() - start_time
    print(f"\n{'='*50}")
    print(f"采集完成！共获取 {all_count} 条记录，耗时 {elapsed:.1f} 秒")
    print(f"数据文件: {os.path.abspath(OUTPUT_FILE)}")
    print(f"{'='*50}")


if __name__ == "__main__":
    # 采集所有页
    crawl(total_pages=57)

三、代码详解

3.1 请求头配置

HEADERS_TEMPLATE = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/125.0.0.0 Safari/537.36"
    ),
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Referer": "https://xueqiu.com/hq/detail?market=CN&first_name=0&second_name=0&type=sh_sz",
    "Origin": "https://xueqiu.com",
}

关键点说明：

字段	作用
User-Agent	模拟Chrome浏览器，避免被识别为爬虫
Referer	伪装请求来源为雪球行情详情页
Origin	跨域请求标识，与目标站点一致

3.2 Cookie获取机制

雪球网对未登录用户的API请求有严格的Cookie校验，必须先访问首页获取有效Cookie才能调用数据接口。

def get_session_with_cookies():
    session = requests.Session()
    session.headers.update(HEADERS_TEMPLATE)
    resp = session.get(HOME_URL, timeout=15)
    resp.raise_for_status()
    return session

Session的好处：

• 自动管理Cookie生命周期
• 保持连接复用，提高效率
• 统一管理请求头

3.3 反爬策略设计

# 1. 随机延时
delay = random.uniform(MIN_DELAY, MAX_DELAY)  # 3-6秒随机间隔

# 2. 失败重试（指数退避）
if attempt < MAX_RETRIES:
    time.sleep(random.uniform(MIN_DELAY * 2, MAX_DELAY * 2))

三重反爬策略：

策略	实现方式	效果
随机延时	random.uniform(3, 6)	模拟人类操作间隔
失败重试	最多3次，延时加倍	应对临时性限制
Cookie管理	Session自动维护	保持登录状态

3.4 数据提取与清洗

"成交额(亿)": (
    round(item["amount"] / 100_000_000, 2)
    if item.get("amount") is not None else ""
),

原始API返回的金额单位是元，通过除以1亿转换为更直观的"亿元"单位，并保留2位小数。

采集字段说明：

字段	含义	原始单位	处理后单位
代码	股票代码	-	-
名称	股票名称	-	-
当前价	最新价格	元	元
涨跌幅(%)	涨跌百分比	-	-
成交额(亿)	成交总金额	元	亿元
总市值(亿)	总市值	元	亿元
市盈率(动态)	PE-TTM	-	-
换手率(%)	换手率	-	-

3.5 参数配置说明

PARAMS = {
    "page": 1,
    "size": 90,           # 每页条数，最大 90
    "order": "desc",       # 排序方向
    "order_by": "percent",  # 排序字段
    "market": "CN",
    "type": "sh_sz",       # sh_sz = 沪深
}

可调整的参数：

• order_by：可改为"amount"（成交额）、"volume"（成交量）等
• order："asc"升序或"desc"降序
• type："sh_sz"沪深、"hk"港股、"us"美股

四、运行结果展示

五、常见问题与解决方案

5.1 403 Forbidden错误

原因： Cookie过期或请求频率过高

解决：

• 确认Cookie获取成功
• 增加延时（修改MIN_DELAY和MAX_DELAY）
• 检查User-Agent是否为最新版本

5.2 数据格式异常

原因： API接口返回结构变化

解决：

• 在fetch_page函数中添加调试输出
• 检查data["data"]["list"]结构是否变化
• 必要时更新extract_stocks中的字段映射

5.3 编码问题

with open(file_path, mode=mode, newline="", encoding="utf-8-sig") as f:

使用utf-8-sig编码，确保Excel打开CSV时中文正常显示。

六、注意事项

免责声明： 本文仅供技术学习交流，请勿用于商业用途或高频采集。使用爬虫时务必遵守网站robots.txt协议，合理控制请求频率，避免对目标服务器造成压力。

总结

本文详细介绍了一个完整的雪球网股票数据采集方案，涵盖了Cookie获取、分页采集、反爬策略、数据清洗等核心技术点。该方案具有以下优势：

✅ 稳定性高：多重异常处理和重试机制
✅ 反爬友好：随机延时模拟人类行为
✅ 扩展性强：可轻松扩展为港股、美股采集
✅ 数据完整：覆盖15个核心行情字段

完整代码可直接运行，只需根据需求修改采集页数即可。如果你对代码有任何疑问或有其他数据采集需求，欢迎在评论区留言交流！