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

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

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

💡 核心导读

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

数据采集基础框架

环境配置与工具链

import pandas as pd
import sqlite3
import json

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

环境检查：
Pandas 3.0.6

多源数据采集架构

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

实战演练：三源数据整合

CSV 与 Excel 联合作业

import pandas as pd

# 构造销售数据
sales_data = {
    'order_date': ['2023-01-01', '2023-01-01', '2023-01-02', '2023-01-02'],
    'customer_id': ['C001', 'C002', 'C003', 'C001'],
    'sku': ['P001', 'P002', 'P001', 'P003'],
    'qty': [2, 1, 3, 5]
}
sales_df = pd.DataFrame(sales_data)
sales_df['order_date'] = pd.to_datetime(sales_df['order_date'])
sales_df['customer_id'] = sales_df['customer_id'].astype('string')

# 构造产品数据
products_data = {
    'sku': ['P001', 'P002', 'P003'],
    'category': ['Electronics', 'Clothing', 'Electronics'],
    'price': [99.9, 49.5, 199.0]
}
products_df = pd.DataFrame(products_data)

# 数据合并与计算
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))

合并后数据结构示例：
  order_date customer_id   sku  qty     category  price    gmv
0 2023-01-01        C001  P001    2  Electronics   99.9  199.8
1 2023-01-01        C002  P002    1     Clothing   49.5   49.5

执行效果对比：

|    | 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 |

数据库交互实战

import pandas as pd
import sqlite3

# 构造销售数据
sales_data = {
    'order_date': ['2023-01-01', '2023-01-01', '2023-01-02', '2023-01-02'],
    'customer_id': ['C001', 'C002', 'C003', 'C001'],
    'sku': ['P001', 'P002', 'P001', 'P003'],
    'qty': [2, 1, 3, 5]
}
sales_df = pd.DataFrame(sales_data)

# 构造产品数据
products_data = {
    'sku': ['P001', 'P002', 'P003'],
    'category': ['Electronics', 'Clothing', 'Electronics'],
    'price': [99.9, 49.5, 199.0]
}
products_df = pd.DataFrame(products_data)

# 使用 SQLite 进行聚合查询
conn = sqlite3.connect(":memory:")
products_df.to_sql("products", conn, index=False, if_exists='replace')
sales_df.to_sql("sales", conn, index=False, if_exists='replace')
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, conn)

print("\n品类销售报告：")
print(report_df)

品类销售报告：
      category  total_qty   avg_price
0     Clothing          1   49.500000
1  Electronics         10  132.933333

API 数据采集技巧

带认证的请求处理

import json

# 模拟 API 响应数据
api_data = {
    "orders": [
        {"order_id": "O001", "date": "2023-01-01", "amount": 150.0},
        {"order_id": "O002", "date": "2023-01-02", "amount": 75.5},
        {"order_id": "O003", "date": "2023-01-03", "amount": 220.0}
    ]
}

# 将数据转换为 JSON 字符串并解析，模拟请求-响应过程
json_string = json.dumps(api_data, ensure_ascii=False)
parsed_data = json.loads(json_string)

print(f"采集到 {len(parsed_data['orders'])} 条订单记录")
print("第一条订单信息:", parsed_data['orders'][0])

采集到 3 条订单记录
第一条订单信息: {'order_id': 'O001', 'date': '2023-01-01', 'amount': 150.0}

📝 动手练一练

第 1 题 · 概念应用

小象咖啡的量化研究员正在构建一个股票因子回测系统。他需要从多个来源获取数据：每日收盘价（CSV格式）、公司基本面数据（JSON格式的API响应）以及一个存储了历史持仓记录的小型SQLite数据库。为了确保回测效率，他计划将所有数据最终整合到一个统一的DataFrame中进行向量化计算。

以下关于他数据整合方案的描述中，哪一项最可能导致性能瓶颈或实现困难？

• A. 使用pandas.read_csv读取收盘价文件，然后使用pandas.read_json解析API返回的JSON字符串，最后用pandas.read_sql查询数据库，并将三个DataFrame通过pd.concat按行合并。
• B. 使用pandas.read_csv和pandas.read_json分别获取价格和基本面数据后，先通过pd.merge将它们按股票代码和日期对齐，再将结果与从数据库读取的持仓记录通过pd.merge进行二次合并。
• C. 将所有数据源（CSV、JSON字符串、数据库连接）作为参数，编写一个统一的函数，内部依次调用对应的pandas.read_*方法，并立即将日期列转换为datetime类型、股票代码列转换为string类型，再进行合并。
• D. 为了追求极致的读取速度，他决定将所有CSV和JSON数据都先导入到同一个SQLite数据库中，然后编写一个复杂的多表JOIN的SQL语句，最后用一次pandas.read_sql查询得到最终结果。

第 2 题 · 代码填空

在回测中，经常需要将来自不同API（如行情API返回JSON，财报API返回CSV字符串）的数据进行整合。以下代码模拟了这一过程：它从两个模拟的“API响应”中分别获取了股票价格和基本面数据，并将它们合并为一个完整的分析数据集。请补全调用Pandas函数将CSV格式的字符串解析为DataFrame的代码。

import pandas as pd
import io

# 模拟来自行情API的JSON响应
price_api_response = {
    "data": [
        {"date": "2023-01-03", "symbol": "AAPL", "close": 175.2},
        {"date": "2023-01-03", "symbol": "MSFT", "close": 330.5},
        {"date": "2023-01-04", "symbol": "AAPL", "close": 176.1},
        {"date": "2023-01-04", "symbol": "MSFT", "close": 332.8}
    ]
}

# 模拟来自财报API的CSV格式响应（字符串）
fundamental_csv_response = """date,symbol,pe_ratio
2023-01-03,AAPL,28.5
2023-01-03,MSFT,32.1
2023-01-04,AAPL,28.7
2023-01-04,MSFT,32.3
"""

# 1. 处理JSON响应：直接由字典列表创建DataFrame
price_df = pd.DataFrame(price_api_response['data'])
price_df['date'] = pd.to_datetime(price_df['date'])

# 2. 处理CSV响应：将字符串转换为DataFrame
# 提示：pd.read_csv可以直接读取一个StringIO对象或字符串本身。
fundamental_df = pd.______(io.StringIO(fundamental_csv_response))
fundamental_df['date'] = pd.to_datetime(fundamental_df['date'])

# 3. 合并两个数据集
merged_df = pd.merge(price_df, fundamental_df, on=['date', 'symbol'], how='inner')

print("合并后的数据集：")
print(merged_df)

import pandas as pd
import io

# 模拟来自行情API的JSON响应
price_api_response = {
    "data": [
        {"date": "2023-01-03", "symbol": "AAPL", "close": 175.2},
        {"date": "2023-01-03", "symbol": "MSFT", "close": 330.5},
        {"date": "2023-01-04", "symbol": "AAPL", "close": 176.1},
        {"date": "2023-01-04", "symbol": "MSFT", "close": 332.8}
    ]
}

# 模拟来自财报API的CSV格式响应（字符串）
fundamental_csv_response = """date,symbol,pe_ratio
2023-01-03,AAPL,28.5
2023-01-03,MSFT,32.1
2023-01-04,AAPL,28.7
2023-01-04,MSFT,32.3
"""

# 1. 处理JSON响应：直接由字典列表创建DataFrame
price_df = pd.DataFrame(price_api_response['data'])
price_df['date'] = pd.to_datetime(price_df['date'])

# 2. 处理CSV响应：将字符串转换为DataFrame
# 提示：pd.read_csv可以直接读取一个StringIO对象或字符串本身。
fundamental_df = pd.read_csv(io.StringIO(fundamental_csv_response))
fundamental_df['date'] = pd.to_datetime(fundamental_df['date'])

# 3. 合并两个数据集
merged_df = pd.merge(price_df, fundamental_df, on=['date', 'symbol'], how='inner')

print("合并后的数据集：")
print(merged_df)

合并后的数据集：
        date symbol  close  pe_ratio
0 2023-01-03   AAPL  175.2      28.5
1 2023-01-03   MSFT  330.5      32.1
2 2023-01-04   AAPL  176.1      28.7
3 2023-01-04   MSFT  332.8      32.3

本章小结

📋 行动清单

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

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

—— 小象教研组