第三章 数据采集基础：API接口调用、网络爬虫基础、数据格式解析

做量化投资的朋友都知道，数据是咱们的命根子。尤其是另类数据，很多时候没有现成的数据库给你用，得自己去想办法搞。这一章，我就把数据采集这块的看家本领拿出来，跟大伙儿聊聊。

说白了，数据采集就三条路：调API、写爬虫、解析格式。这三样玩转了，市面上百分之八十的数据源你都能搞定。我这些年踩过的坑不少，今天一并分享出来。

3.1 API接口调用——正规军的打法

API调用，说白了就是跟数据服务商说"嘿，给我来点数据"，然后人家按规矩给你。这是最规范的方式，也是我最推荐的方式。

我个人习惯，拿到一个API的第一件事不是写代码，而是先看文档。你想想看，文档都不看就开干，那不是给自己挖坑吗？

3.1.1 RESTful API 基础

现在市面上百分之九十的API都是RESTful风格的。核心就四个动作：GET（查）、POST（增）、PUT（改）、DELETE（删）。做数据采集，我们主要用GET和POST。

举个例子，调个股票行情API：

import requests
import json

# 这是我在某数据平台用的API
url = "https://api.xxx.com/v1/stock/daily"
params = {
    "symbol": "000001.SZ",
    "start_date": "2024-01-01",
    "end_date": "2024-01-31",
    "api_key": "your_api_key_here"
}

response = requests.get(url, params=params)

# 检查状态码，这个习惯一定要养成
if response.status_code == 200:
    data = response.json()
    print(f"成功获取 {len(data['data'])} 条记录")
else:
    print(f"请求失败，状态码：{response.status_code}")
    print(f"错误信息：{response.text}")

3.1.2 处理API限流

做量化的人都知道，数据服务商不是慈善家。他们通常会限制你的请求频率。比如每分钟最多60次，或者每天最多10000次。

我曾经遇到过一个问题：写了个循环去拉历史数据，结果跑了没两分钟就被封IP了。后来学乖了，加了个限流机制：

import time
from datetime import datetime

class APIRateLimiter:
    """API请求限流器，我自己的项目里一直在用"""
    def __init__(self, max_calls_per_minute=60):
        self.max_calls = max_calls_per_minute
        self.call_times = []
    
    def wait_if_needed(self):
        now = time.time()
        # 清理一分钟之前的记录
        self.call_times = [t for t in self.call_times if now - t < 60]
        
        if len(self.call_times) >= self.max_calls:
            # 等最旧的那个请求过期
            wait_time = 60 - (now - self.call_times[0])
            print(f"达到限流上限，等待 {wait_time:.2f} 秒")
            time.sleep(wait_time)
        
        self.call_times.append(time.time())

# 使用示例
limiter = APIRateLimiter(max_calls_per_minute=30)
for i in range(100):
    limiter.wait_if_needed()
    # 这里放你的API请求代码
    print(f"第 {i+1} 次请求")

3.2 网络爬虫基础——游击队的战术

API虽然好，但不是所有数据都有API。这时候就得靠爬虫了。说白了，爬虫就是模拟浏览器去访问网页，然后把页面里的数据提取出来。

嗯，这里要注意：爬虫不是万能的，而且有法律风险。我个人的原则是——只爬公开数据，不爬需要登录的、不爬有明确robots.txt禁止的、不爬频率太高给人家服务器造成压力。

3.2.1 静态页面爬取

最简单的爬虫，就是直接拿HTML页面，然后用解析库提取数据。我常用的是requests + BeautifulSoup的组合：

import requests
from bs4 import BeautifulSoup
import pandas as pd

# 爬取某财经网站的板块数据
url = "https://example.com/sectors"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

response = requests.get(url, headers=headers)
soup = BeautifulSoup(response.text, 'html.parser')

# 找到表格数据
table = soup.find('table', class_='sector-table')
rows = table.find_all('tr')

data = []
for row in rows[1:]:  # 跳过表头
    cols = row.find_all('td')
    data.append({
        'sector': cols[0].text.strip(),
        'change_pct': cols[1].text.strip(),
        'volume': cols[2].text.strip()
    })

df = pd.DataFrame(data)
print(df.head())

3.2.2 动态页面处理

现在的网页越来越"聪明"了，很多数据是通过JavaScript动态加载的。你用requests直接拿，拿到的只是一个空壳子。

遇到这种情况，我一般用Selenium或者Playwright。说白了，就是让浏览器真的去渲染页面，然后你再从渲染后的页面里拿数据：

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# 启动浏览器
driver = webdriver.Chrome()
driver.get("https://example.com/dynamic-data")

# 等待数据加载完成
try:
    element = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.ID, "data-table"))
    )
    # 数据加载完成，开始提取
    rows = driver.find_elements(By.CSS_SELECTOR, "#data-table tr")
    for row in rows:
        print(row.text)
finally:
    driver.quit()

3.3 数据格式解析——把数据变成能用的样子

数据拿回来了，但格式五花八门。JSON、XML、CSV是最常见的三种。解析不好，后面全白搭。

3.3.1 JSON解析

JSON是现在最流行的数据交换格式。Python自带的json库就够用，但处理大量数据时，我推荐用orjson，速度能快好几倍：

import json
# 或者用 orjson
# import orjson as json

# 解析API返回的JSON
json_str = '{"code": 0, "data": {"stocks": [{"symbol": "000001", "price": 10.5}]}}'
parsed = json.loads(json_str)

# 提取数据
if parsed['code'] == 0:
    stocks = parsed['data']['stocks']
    for stock in stocks:
        print(f"{stock['symbol']}: {stock['price']}")

# 处理嵌套JSON，用递归更优雅
def extract_values(obj, key):
    """递归提取JSON中所有指定key的值"""
    results = []
    if isinstance(obj, dict):
        for k, v in obj.items():
            if k == key:
                results.append(v)
            elif isinstance(v, (dict, list)):
                results.extend(extract_values(v, key))
    elif isinstance(obj, list):
        for item in obj:
            results.extend(extract_values(item, key))
    return results

3.3.2 XML解析

XML虽然老了，但在金融数据领域还是有不少应用。比如一些交易所的历史数据还是用XML格式提供的。

我解析XML有两个选择：数据量小用xml.etree.ElementTree，数据量大用lxml：

import xml.etree.ElementTree as ET

# 解析XML格式的股票数据
xml_data = """
<market>
    <stock symbol="000001">
        <name>平安银行</name>
        <price>10.50</price>
        <volume>1234567</volume>
    </stock>
    <stock symbol="000002">
        <name>万科A</name>
        <price>15.80</price>
        <volume>2345678</volume>
    </stock>
</market>
"""

root = ET.fromstring(xml_data)
stocks = []
for stock in root.findall('stock'):
    stocks.append({
        'symbol': stock.get('symbol'),
        'name': stock.find('name').text,
        'price': float(stock.find('price').text),
        'volume': int(stock.find('volume').text)
    })

print(stocks)

3.3.3 CSV解析

CSV是最简单的格式，但坑也最多。编码问题、分隔符问题、引号问题，我全都遇到过。

我的建议是：永远用pandas.read_csv()，别自己手写解析逻辑。pandas把各种边界情况都处理好了：

import pandas as pd

# 标准CSV读取
df = pd.read_csv('stock_data.csv')

# 处理常见问题
df = pd.read_csv(
    'stock_data.csv',
    encoding='utf-8',           # 编码问题
    sep=',',                    # 分隔符
    quotechar='"',              # 引号字符
    na_values=['', 'NA', 'N/A'], # 缺失值处理
    parse_dates=['date'],       # 日期解析
    dtype={'symbol': str}       # 指定数据类型，防止股票代码被当成数字
)

# 大文件分块读取
chunk_size = 10000
for chunk in pd.read_csv('big_file.csv', chunksize=chunk_size):
    process_chunk(chunk)  # 你的处理逻辑

3.4 知识体系总览

说了这么多，我画张图把整个数据采集的脉络理一理。这样你心里有个谱，知道每一步该干什么：

这张图把数据采集的流程讲得很清楚了。从上到下，先确定数据源，再选采集方式，最后解析成你能用的格式。每一步都有对应的工具和技巧，别跳步，也别混着来。