本地多源数据：量化历史回测的基石

本地多源数据：量化历史回测的基石

小象量化极客小册 · 华尔街量化分析先锋系列

在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

小象量化极客小册 · 华尔街量化分析先锋系列 在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

💡 核心导读

本章将带领读者掌握本地多源数据采集与操作的核心技能。通过学习，您将理解异构数据源的整合逻辑，掌握使用 Python 生态工具链进行本地文件（CSV/JSON/Excel）、数据库（SQLite/MySQL）和 API 接口数据的采集与清洗技术。重点包括：Pandas 的多格式 IO 操作、SQLAlchemy 的 ORM 交互、Requests 的 API 调用封装，以及多源数据的合并与标准化处理。

数据采集基础框架

环境配置与工具链

import pandas as pd
import sqlite3
from sqlalchemy import create_engine
import requests
import json

print("环境检查：")
print(f"Pandas {pd.__version__}, SQLAlchemy {sqlalchemy.__version__}")

多源数据采集架构

数据源类型      采集方式               输出格式
-----------   -------------------   ------------
本地文件       pd.read_*()          DataFrame
数据库        SQLAlchemy/原生驱动   DataFrame/ResultSet
API 接口      requests.get()        JSON/Dict

💡 概念小测：数据格式优缺点

在量化回测系统中，如果需要存储和快速读取高频 Stock Tick 报价数据（每秒数十条），通常不推荐使用以下哪种文件格式？

• HDF5 或 Parquet，因为它们支持二进制存储和列式读取，速度极快。
• Excel 格式（.xlsx），因为有行数限制且读写极度消耗内存和时间。
• 关系型数据库如 SQLite，因为支持 SQL 索引查询。
• CSV 压缩包（.csv.gz），以流式读取降低内存消耗。

📝 选择题

💡 概念小测：数据格式优缺点 在量化回测系统中，如果需要存储和快速读取高频 Stock Tick 报价数据（每秒数十条），通常不推荐使用以下哪种文件格式？

• A. HDF5 或 Parquet，因为它们支持二进制存储和列式读取，速度极快。
• B. Excel 格式（.xlsx），因为有行数限制且读写极度消耗内存和时间。
• C. 关系型数据库如 SQLite，因为支持 SQL 索引查询。
• D. CSV 压缩包（.csv.gz），以流式读取降低内存消耗。

答案：B（Excel 格式（.xlsx），因为有行数限制且读写极度消耗内存和时间。）

实战演练：三源数据整合

CSV 与 Excel 联合作业

sales_df = pd.read_csv(
    "sales_2023.csv",
    parse_dates=["order_date"],
    dtype={"customer_id": "string"}
)
products_df = pd.read_excel(
    "product_master.xlsx",
    sheet_name="active",
    usecols=["sku", "category", "price"]
)

merged_df = pd.merge(
    left=sales_df,
    right=products_df,
    how="left",
    on="sku"
)
merged_df["gmv"] = merged_df["qty"] * merged_df["price"]

print("合并后数据结构示例：")
print(merged_df.head(2).to_markdown())  # 输出 Markdown 格式表格

执行效果对比：

|    | order_date   | customer_id   |   qty | sku    | category   |   price |   gmv |
|---:|:-------------|:--------------|------:|:-------|:-----------|--------:|------:|
|  0 | 2023-01-05   | CUST-1001     |     2 | P-1092 | Electronics|   899.0 | 1798  |
|  1 | 2023-01-07   | CUST-1003     |     1 | P-2056 | Home       |   249.5 | 249.5 |

数据库交互实战

conn = sqlite3.connect(":memory:")
products_df.to_sql("products", conn, index=False)

engine = create_engine("sqlite:///:memory:")
query = """
SELECT 
    category,
    SUM(qty) as total_qty,
    AVG(price) as avg_price
FROM sales s
JOIN products p ON s.sku = p.sku
GROUP BY category
"""
report_df = pd.read_sql(query, engine)

print("\n品类销售报告：")
print(report_df.to_markdown(index=False))

API 数据采集技巧

带认证的请求处理

endpoint = "https://api.example.com/v1/orders"
headers = {
    "Authorization": "Bearer YOUR_ACCESS_TOKEN",
    "Content-Type": "application/json"
}
params = {
    "start_date": "2023-01-01",
    "end_date": "2023-01-31"
}

response = requests.get(
    url=endpoint,
    headers=headers,
    params=params,
    timeout=10
)
response.raise_for_status()  # 自动检测 HTTP 错误

api_data = response.json()
with open("orders_202301.json", "w") as f:
    json.dump(api_data, f, ensure_ascii=False)

print(f"采集到 {len(api_data['orders'])} 条订单记录")

💡 代码填空：文件基础读取

补全在 Python 中打开并读取包含小象咖啡门店交易日志 trade.log 的内置函数：

with {blank}('trade.log', 'r', encoding='utf-8') as f:
    logs = f.readlines()
print(f"共读取到 {len(logs)} 行交易日志")

• read
• open
• load

✏️ 填空题

💡 代码填空：文件基础读取 补全在 Python 中打开并读取包含小象咖啡门店交易日志 trade.log 的内置函数：

# 以只读模式打开本地文件
with ___('trade.log', 'r', encoding='utf-8') as f:
    logs = f.readlines()
print(f"共读取到 {len(logs)} 行交易日志")

答案：open

本章小结

行动清单

• 使用 Pandas 同时读取 CSV 和 Excel 文件，生成包含计算字段的合并报表
• 在 SQLite 中创建测试数据库，通过 SQLAlchemy 执行包含 JOIN 和 GROUP BY 的查询
• 调用任意开放 API（如 GitHub API），将返回的 JSON 数据转换为 DataFrame 并保存为 Parquet 格式

「小象量化教研组」数据工程团队编写