实战：构建每日热门个股行情自动采集系统

实战：构建全国城市空气质量自动采集系统

小象实战讲义 · 量化交易Python入门：从咖啡馆到华尔街

本节我们将使用 Scrapy 框架进行网络数据采集，以全国城市空气质量实时数据为例，编写爬虫从中国环境监测总站发布平台抓取 PM2.5、PM10、SO₂ 等污染物浓度，为后续量化分析输入第一手另类数据。

💡 核心导读

本节将带领读者使用Scrapy框架采集全国城市空气质量实时数据。通过完整项目演练，掌握动态网页数据抓取、XPath选择器使用、数据持久化存储等核心技能，最终构建可自动获取PM2.5、SO₂等6项空气质量指标的数据采集系统。重点学习反爬应对策略和结构化数据存储方案。

环境准备与目标分析

工具栈选择

import pandas as pd
import numpy as np
print("导入 pandas 和 numpy 成功。")

数据源分析

目标网站：中国环境监测总站全国城市空气质量实时发布平台

导入 pandas 和 numpy 成功。

Scrapy项目实战

1. 创建爬虫项目

scrapy startproject aqi_monitor
cd aqi_monitor
scrapy genspider aqi "https://air.cnemc.cn:18007"

2. 定义数据模型

class AqiMonitorItem:
    def __init__(self):
        self.city = None
        self.aqi = None
        self.pm25 = None
        self.pm10 = None
        self.so2 = None
        self.no2 = None
        self.co = None
        self.time = None

item = AqiMonitorItem()
item.city = "北京"
item.aqi = 45
item.time = "2024-01-01 12:00:00"
print(f"模拟创建数据项: 城市={item.city}, AQI={item.aqi}, 时间={item.time}")

模拟创建数据项: 城市=北京, AQI=45, 时间=2024-01-01 12:00:00

3. 核心爬虫逻辑

class AqiSpider:
    name = 'aqi'
    custom_settings = {
        'DOWNLOAD_DELAY': 2,
        'USER_AGENT': 'Mozilla/5.0 (Windows NT 10.0)'
    }

    def start_requests(self):
        urls = ['https://air.cnemc.cn:18007/']
        print(f"模拟爬虫开始请求，目标URL: {urls[0]}")
        return urls

    def parse(self, response):
        print("模拟解析响应，提取数据行。")
        rows_data = [
            {'city': '上海', 'aqi': 38, 'time': '2024-01-01 12:00:00'},
            {'city': '广州', 'aqi': 52, 'time': '2024-01-01 12:00:00'}
        ]
        for data in rows_data:
            print(f"提取到数据: {data}")
        return rows_data

spider = AqiSpider()
spider.start_requests()
parsed_data = spider.parse("模拟响应对象")
print(f"共解析 {len(parsed_data)} 条记录。")

模拟爬虫开始请求，目标URL: https://air.cnemc.cn:18007/
模拟解析响应，提取数据行。
提取到数据: {'city': '上海', 'aqi': 38, 'time': '2024-01-01 12:00:00'}
提取到数据: {'city': '广州', 'aqi': 52, 'time': '2024-01-01 12:00:00'}
共解析 2 条记录。

数据存储与导出

JSON存储配置

import datetime
now = datetime.datetime(2024, 1, 1, 12, 0, 0)
FEEDS = {
    f'aqi_data_{now.strftime("%Y%m%d_%H%M%S")}.json': {'format': 'json', 'encoding': 'utf8'}
}
print("模拟 Scrapy FEEDS 设置:")
for feed_path, settings in FEEDS.items():
    print(f"  文件: {feed_path}, 格式: {settings['format']}")

模拟 Scrapy FEEDS 设置:
  文件: aqi_data_20240101_120000.json, 格式: json

数据质量检查

import pandas as pd
data = [
    {'city': '北京', 'aqi': 45, 'time': '2024-01-01 12:00:00'},
    {'city': '上海', 'aqi': 38, 'time': '2024-01-01 12:00:00'},
    {'city': '广州', 'aqi': 52, 'time': '2024-01-01 12:00:00'},
    {'city': '北京', 'aqi': 48, 'time': '2024-01-01 13:00:00'}
]
df = pd.DataFrame(data)
print(f"获取城市数量: {df['city'].nunique()}")
print(f"数据时间范围: {df['time'].min()} 至 {df['time'].max()}")
print("数据预览:")
print(df)

获取城市数量: 3
数据时间范围: 2024-01-01 12:00:00 至 2024-01-01 13:00:00
数据预览:
  city  aqi                 time
0   北京   45  2024-01-01 12:00:00
1   上海   38  2024-01-01 12:00:00
2   广州   52  2024-01-01 12:00:00
3   北京   48  2024-01-01 13:00:00

反爬策略应对方案

1. 请求头伪装

headers = {
    'Accept-Language': 'zh-CN,zh;q=0.9',
    'Referer': 'https://air.cnemc.cn/',
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0)'
}
print("模拟 HTTP 请求头:")
for key, value in headers.items():
    print(f"  {key}: {value}")

模拟 HTTP 请求头:
  Accept-Language: zh-CN,zh;q=0.9
  Referer: https://air.cnemc.cn/
  User-Agent: Mozilla/5.0 (Windows NT 10.0)

2. IP轮换策略

class ProxyMiddleware:
    def process_request(self, request, spider):
        request_meta = {'proxy': 'http://proxy_pool:8000'}
        print(f"模拟代理中间件为请求添加代理: {request_meta['proxy']}")
        return request_meta

middleware = ProxyMiddleware()
result_meta = middleware.process_request("模拟请求对象", "模拟爬虫对象")
print(f"处理后的请求元数据: {result_meta}")

模拟代理中间件为请求添加代理: http://proxy_pool:8000
处理后的请求元数据: {'proxy': 'http://proxy_pool:8000'}

3. 验证码处理

def handle_captcha(response):
    print("模拟处理验证码逻辑。")
    captcha_url = "https://example.com/captcha.jpg"
    if captcha_url:
        solved_text = "ABCD"
        print(f"从URL {captcha_url} 识别验证码为: {solved_text}")
        return solved_text
    return None

result = handle_captcha("模拟响应对象")
print(f"验证码处理结果: {result}")

模拟处理验证码逻辑。
从URL https://example.com/captcha.jpg 识别验证码为: ABCD
验证码处理结果: ABCD

📝 动手练一练

第 1 题 · 概念应用

在构建量化策略的数据采集模块时，你发现目标网站（如某金融数据平台）返回的 JSON 数据中，中文股票名称和行业分类信息被存储为 \u4e0a\u6d77\u7535\u6c14 这样的 Unicode 转义序列。为了在后续的 Pandas 分析中直接使用可读的中文，你需要在 Scrapy 项目的 settings.py 中进行一项关键配置。

以下关于这项配置的说法，哪一项是正确的？

• A. 在 settings.py 中设置 FEED_EXPORT_ENCODING = 'gbk'，因为这是 Windows 系统的默认编码，可以避免乱码。
• B. 在 settings.py 中设置 FEED_EXPORT_ENCODING = 'utf-8'，并确保爬虫解析时使用 response.json() 而非 response.text 来获取数据。
• C. 在 settings.py 中设置 FEED_EXPORT_ENCODING = 'utf-8'，这能确保导出的 JSON 文件中的中文字符按原样（如“上海电气”）保存，而不是 \uXXXX 编码。
• D. 这个问题无法通过 Scrapy 配置解决，必须在数据存入数据库后，使用 Python 的 json.loads() 配合 ensure_ascii=False 参数重新处理。

第 2 题 · 代码填空

在完成空气质量数据爬取后，你需要将数据导出为 CSV 格式，以便用 Pandas 进行后续的量化分析（例如计算不同城市的 AQI 指数与股市波动的相关性）。请补全以下 Scrapy 命令行指令，使其能将名为 aqi_spider 的爬虫抓取的数据导出到 aqi_data.csv 文件中。

scrapy crawl aqi_spider -o aqi_data.______

scrapy crawl aqi_spider -o aqi_data.csv

本章小结

通过本实战项目，我们构建了完整的城市空气质量监测数据采集系统，关键收获包括：

1. Scrapy框架的Item Pipeline机制
2. 动态网页数据的XPath提取技巧
3. 结构化数据存储的最佳实践
4. 生产级爬虫的反爬应对方案

📋 行动清单

• 运行示例爬虫，修改XPath提取其他污染物指标
• 实现MySQL存储管道，将数据保存到数据库
• 添加异常处理逻辑，当 AQI>300（严重污染）时发送邮件告警

—— 小象教研组