4、新闻数据抓取实战：使用requests库抓取新闻、处理JSON响应、数据清洗与结构化存储

好，咱们进入实战环节了。

前面几章讲了情绪因子的理论，讲了为什么新闻数据能赚钱。但说一千道一万，数据拿不到手，一切都是空谈。我见过太多人，策略想得天花乱坠，结果卡在数据获取这一步，项目直接烂尾。

今天我们就来亲手解决这个问题。用Python最经典的requests库，把新闻数据从网上拽下来，再把它收拾得干干净净，存进数据库里。

4.1 准备工作：环境与目标分析

动手之前，先看看我们要抓什么。

我个人习惯用东方财富的新闻接口做演示。为什么选它？因为它的JSON结构规整，字段齐全，而且不需要复杂的登录验证，非常适合教学。你想想看，如果一上来就搞反爬虫对抗，那这节课就变成逆向工程课了，偏离了我们的主题。

你需要安装的库：

pip install requests pandas sqlalchemy pymysql

嗯，这里要注意：sqlalchemy和pymysql是用于后续存数据库的，如果你只想先看看数据，只装requests和pandas就够了。

4.2 第一行代码：用requests抓取新闻

我们先写一个最简单的请求，看看能拿到什么。

import requests

url = "https://push2.eastmoney.com/api/qt/stock/get"
params = {
    "secid": "1.600519",  # 贵州茅台
    "fields": "f58,f57,f43,f170,f171"
}
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

resp = requests.get(url, params=params, headers=headers, timeout=10)
print(resp.status_code)
print(resp.text[:500])

跑一下，如果看到200，恭喜你，网络通了。但这不是新闻接口，这是行情接口。我们真正要的是新闻列表。

我在项目中遇到过一个问题：很多人拿到200状态码就以为成功了，结果解析JSON时报错。为什么？因为服务器返回的可能是一段HTML错误页面，而不是JSON。所以，养成好习惯——先看Content-Type响应头。

print(resp.headers.get("Content-Type"))
# 应该输出: application/json; charset=utf-8

4.3 新闻列表接口实战

好，我们换到真正的新闻接口。东方财富的新闻API长这样：

news_url = "https://push2.eastmoney.com/api/qt/stock/news"
params = {
    "secid": "1.600519",
    "count": 20,      # 取20条
    "offset": 0,      # 分页偏移
    "type": 1         # 1=全部新闻, 2=重要新闻
}
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Referer": "https://quote.eastmoney.com/"
}

resp = requests.get(news_url, params=params, headers=headers, timeout=10)
data = resp.json()
print(data.keys())
# 输出: dict_keys(['rc', 'rt', 'svr', 'lt', 'full', 'data'])

看到data字段了吗？新闻列表就藏在这里面。我们把它取出来看看结构：

news_list = data.get("data", [])
print(f"共获取 {len(news_list)} 条新闻")
print(news_list[0].keys())
# 输出: dict_keys(['art_code', 'title', 'content', 'date', 'time', ...])

核心字段说明：

  • art_code：文章唯一ID，去重用的
  • title：新闻标题
  • content：正文摘要（不是全文）
  • date / time：发布时间
  • source：来源，比如"证券时报"

4.4 处理JSON响应：别被坑了

JSON解析看起来简单，但有几个坑我必须提醒你。

坑一：响应不是JSON

我曾经在半夜跑数据抓取脚本，突然报json.decoder.JSONDecodeError。查了半天，原来是服务器临时返回了一段维护公告。所以，一定要加异常处理：

try:
    data = resp.json()
except ValueError as e:
    print(f"JSON解析失败: {e}")
    print(f"原始响应前200字符: {resp.text[:200]}")
    data = {}

坑二：字段可能缺失

不是每条新闻都有source字段。有些快讯类新闻，来源字段是空的。如果你直接item['source']，会报KeyError。我习惯用.get()方法：

title = item.get('title', '')
source = item.get('source', '未知来源')
content = item.get('content', '')

4.5 数据清洗：把脏数据洗干净

原始数据长什么样？我打印一条给你看：

{
  "title": "贵州茅台：2024年一季度净利润同比增长15.73%",
  "content": "    贵州茅台(600519)4月27日晚间披露...  ",
  "date": "2024-04-27",
  "time": "20:30:15",
  "source": " 证券时报 "
}

看到了吗？content前后有空格，source也有。这些不处理干净，后面做情绪分析时会影响分词效果。

我总结了一套清洗流程，直接上代码：

import pandas as pd
import re

def clean_news_item(item):
    """清洗单条新闻数据"""
    cleaned = {}
    
    # 1. 去除首尾空白
    cleaned['title'] = item.get('title', '').strip()
    cleaned['source'] = item.get('source', '').strip()
    
    # 2. 去除content中的多余空白和HTML标签
    content = item.get('content', '').strip()
    content = re.sub(r' ', ' ', content)  # 替换HTML空格
    content = re.sub(r'<[^>]+>', '', content)  # 去除HTML标签
    content = re.sub(r'\s+', ' ', content)     # 合并多个空格
    cleaned['content'] = content
    
    # 3. 合并日期和时间
    date_str = item.get('date', '')
    time_str = item.get('time', '')
    if date_str and time_str:
        cleaned['publish_time'] = f"{date_str} {time_str}"
    else:
        cleaned['publish_time'] = date_str
    
    # 4. 去重标识
    cleaned['art_code'] = item.get('art_code', '')
    
    return cleaned

# 批量清洗
cleaned_list = [clean_news_item(item) for item in news_list]
df = pd.DataFrame(cleaned_list)
print(df.head())

我的习惯：清洗后的数据，我会额外加一列crawl_time，记录当前抓取时间。这样以后排查数据延迟问题时，一眼就能看出来。

4.6 结构化存储：存进数据库

数据洗干净了，不能老放在内存里。我建议直接存进MySQL或SQLite。对于量化场景，我偏爱MySQL，因为支持并发读写，而且方便后续用SQL做分析。

先建表：

CREATE TABLE news_sentiment (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    art_code VARCHAR(64) UNIQUE NOT NULL,
    title VARCHAR(500),
    content TEXT,
    source VARCHAR(100),
    publish_time DATETIME,
    crawl_time DATETIME,
    stock_code VARCHAR(10),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_publish_time (publish_time),
    INDEX idx_stock_code (stock_code)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

注意这里用了UNIQUE约束在art_code上。为什么要这样做？因为新闻可能重复推送，你重复抓取时，INSERT会报错，但你可以用INSERT IGNORE或ON DUPLICATE KEY UPDATE来优雅处理。

Python写入代码：

from sqlalchemy import create_engine

# 连接数据库
engine = create_engine('mysql+pymysql://user:password@localhost:3306/quant_db')

# 批量写入
df.to_sql(
    name='news_sentiment',
    con=engine,
    if_exists='append',   # 追加模式
    index=False,
    method='multi'        # 批量插入，提升性能
)

print(f"成功写入 {len(df)} 条新闻")

注意：method='multi'在数据量很大时（比如一次插入几万条）可能会超过MySQL的max_allowed_packet限制。我建议单次插入控制在1000条以内，分批写入。

4.7 完整流程：从抓取到存储

我把整个流程串起来，形成一个可复用的函数。这也是我实际项目中用的模板：

def fetch_news_to_db(stock_code, count=50, engine=None):
    """
    抓取指定股票的新闻，清洗后存入数据库
    """
    # 1. 构建请求
    secid = f"1.{stock_code}" if stock_code.startswith('6') else f"0.{stock_code}"
    params = {
        "secid": secid,
        "count": count,
        "offset": 0,
        "type": 1
    }
    headers = {
        "User-Agent": "Mozilla/5.0",
        "Referer": "https://quote.eastmoney.com/"
    }
    
    # 2. 发送请求
    try:
        resp = requests.get(news_url, params=params, headers=headers, timeout=10)
        resp.raise_for_status()
        data = resp.json()
    except Exception as e:
        print(f"请求失败: {e}")
        return False
    
    # 3. 解析并清洗
    raw_list = data.get('data', [])
    if not raw_list:
        print("未获取到新闻数据")
        return False
    
    cleaned_list = [clean_news_item(item) for item in raw_list]
    df = pd.DataFrame(cleaned_list)
    
    # 4. 添加辅助字段
    df['stock_code'] = stock_code
    df['crawl_time'] = pd.Timestamp.now()
    
    # 5. 存入数据库
    if engine is not None:
        df.to_sql('news_sentiment', engine, if_exists='append', index=False, method='multi')
        print(f"股票 {stock_code}: 成功写入 {len(df)} 条")
    
    return df

# 使用示例
engine = create_engine('mysql+pymysql://user:pass@localhost/quant_db')
fetch_news_to_db('600519', count=30, engine=engine)
fetch_news_to_db('000858', count=30, engine=engine)  # 五粮液

4.8 本章知识体系

下面这张图，是我梳理的本章核心流程。你可以把它当作一个检查清单：

新闻数据抓取与存储流程 ① 请求构建 URL + 参数 + 请求头 ② 发送请求 requests.get() + 异常处理 ③ JSON解析 resp.json() + 判空 ④ 数据清洗 去除空白 / HTML标签 / 合并时间 字段补全 / 去重标识 ⑤ 结构化存储 MySQL / SQLite UNIQUE约束 / 索引优化 ⑥ 批量写入 to_sql() + method='multi' ⚠️ 常见坑点 • 非JSON响应 • 字段缺失 • 重复数据 • 编码问题 • 请求超时 • 数据库连接断开 • 批量插入过大 • 时间字段格式 核心原则：先验证 → 再清洗 → 后存储，每一步都要有异常处理

这张图把整个流程分成了6步。我个人建议你把它打印出来贴在工位上，每次写抓取脚本时对照着检查一遍，能省去很多调试时间。

4.9 避坑指南

最后，分享几个我踩过的坑：

  • 请求头缺失Referer：有些接口会校验来源，不加Referer直接返回403。我曾经排查了半小时才发现是这个问题。
  • 编码问题：新闻内容里偶尔会有特殊字符，比如\u200b（零宽空格）。清洗时最好用.encode('utf-8', errors='ignore').decode('utf-8')兜底。
  • 数据库连接池：如果你要批量抓取上千只股票，每次新建连接会非常慢。建议用sqlalchemy的连接池，或者用pymysql的connections复用。
  • 频率控制：别抓太快。我一般会在每次请求后time.sleep(0.5)，既保证速度，又不触发反爬。

好了，这一章的内容就到这里。代码你直接复制就能跑，但建议你亲手敲一遍。只有自己踩过坑，印象才深刻。

公众号:蓝海数据掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →