JSON数据流操作：解析股票行情API响应

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

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

💡 核心导读

JSON（JavaScript Object Notation）是现代数据交换的主流格式之一，本节将系统讲解Python环境下JSON文件的读写操作与实战技巧。通过本节学习，您将掌握：使用标准库json进行数据序列化与反序列化、处理嵌套JSON结构、Pandas的JSON转换方法，以及大规模JSON文件的优化处理策略。重点培养从本地文件系统高效提取结构化数据的能力。

JSON基础操作

数据结构转换原理

JSON与Python数据类型的对应关系：

+---------------+-------------------+
|   JSON类型    |  Python对应类型   |
+---------------+-------------------+
| object        | dict              |
| array         | list              |
| string        | str               |
| number        | int/float         |
| true/false    | True/False        |
| null          | None              |
+---------------+-------------------+

标准库基础操作

import json

employee = {
    "id": 101,
    "name": "小象咖啡",
    "skills": ["Python", "SQL", "Spark"],
    "contact": {
        "email": "zhang@example.com",
        "phone": "13800138000"
    }
}

json_str = json.dumps(employee, ensure_ascii=False, indent=2)
print("序列化结果：")
print(json_str)

loaded_data = json.loads(json_str)
print("\n反序列化结果类型：", type(loaded_data))

序列化结果：
{
  "id": 101,
  "name": "小象咖啡",
  "skills": [
    "Python",
    "SQL",
    "Spark"
  ],
  "contact": {
    "email": "zhang@example.com",
    "phone": "13800138000"
  }
}

反序列化结果类型： <class 'dict'>

进阶处理技巧

处理复杂嵌套结构

import json

company_data = {
    "company": "小象量化教研组",
    "departments": [
        {
            "name": "研发部",
            "members": [
                {"name": "王工", "level": "P7"},
                {"name": "李工", "level": "P6"}
            ]
        },
        {
            "name": "市场部",
            "members": [
                {"name": "赵经理", "level": "M3"}
            ]
        }
    ]
}

dev_members = [member["name"]
              for dept in company_data["departments"]
              if dept["name"] == "研发部"
              for member in dept["members"]]

print("研发部成员：", dev_members)

研发部成员： ['王工', '李工']

Pandas集成方案

import pandas as pd

nested_json = [
    {"id": 1, "metrics": {"cpu": 0.8, "memory": 0.6}},
    {"id": 2, "metrics": {"cpu": 0.4, "memory": 0.9}}
]

df = pd.json_normalize(nested_json)
print("标准化处理结果：")
print(df)

标准化处理结果：
   id  metrics.cpu  metrics.memory
0   1          0.8             0.6
1   2          0.4             0.9

性能优化策略

大文件处理方案

import json

large_data = [{"id": i, "data": "x"*5} for i in range(5)]
json_str = json.dumps({"item": large_data}, indent=2)
parsed_data = json.loads(json_str)

for record in parsed_data["item"]:
    print(f"处理记录ID: {record.get('id')}")

处理记录ID: 0
处理记录ID: 1
处理记录ID: 2
处理记录ID: 3
处理记录ID: 4

📝 动手练一练

第 1 题 · 概念应用

你正在编写一个数据采集模块，用于接收来自不同行情供应商的 API 响应。其中一家供应商返回的数据是一个包含股票列表的 JSON 格式字符串 market_data_str，其结构如下：

{"stocks": [{"symbol": "AAPL", "price": 175.25}, {"symbol": "MSFT", "price": 330.12}]}

另一家供应商则直接将同样的数据以 Python 字典对象 market_data_dict 的形式提供给你。

现在，你需要将这两份数据统一转换为 Pandas DataFrame 进行分析。以下哪种说法是正确的？

• A. 对于 market_data_str，应该使用 pd.json_normalize(market_data_str) 来解析。
• B. 对于 market_data_dict，可以直接使用 pd.DataFrame(market_data_dict) 来转换。
• C. 对于 market_data_str，必须先使用 json.loads() 将其转换为字典，才能用 pd.json_normalize() 处理。
• D. 对于 market_data_dict，必须先用 json.dumps() 将其序列化为字符串，才能用 pd.read_json() 读取。

import json
import pandas as pd

# 模拟供应商A的JSON字符串响应
market_data_str = '{"stocks": [{"symbol": "AAPL", "price": 175.25}, {"symbol": "MSFT", "price": 330.12}]}'
# 模拟供应商B的Python字典对象
market_data_dict = json.loads(market_data_str)  # 假设这就是直接给的对象

# 正确处理方法C
parsed_dict = json.loads(market_data_str)
df_from_str = pd.json_normalize(parsed_dict, 'stocks')
print("从字符串解析得到的DataFrame:")
print(df_from_str)
print()

# 对于字典对象的正确处理方法（非选项B）
df_from_dict = pd.DataFrame(market_data_dict['stocks'])
print("从字典对象转换得到的DataFrame:")
print(df_from_dict)

从字符串解析得到的DataFrame:
  symbol   price
0   AAPL  175.25
1   MSFT  330.12

从字典对象转换得到的DataFrame:
  symbol   price
0   AAPL  175.25
1   MSFT  330.12

第 2 题 · 动手题（代码填空）

你收到一份来自风控系统的持仓风险报告 risk_report_str：外层是报告日期，positions 是一个列表，列表中的每个字典就是一条持仓记录（账户、标的、权重、风险价值），只有一层嵌套。请补全下方代码中的 record_path，把每条持仓展平为 DataFrame 的一行，并让每一行都带上报告日期。

import json
import pandas as pd

risk_report_str = '''
{
  "report_date": "2023-10-27",
  "positions": [
    {"account_id": "ACC001", "symbol": "600519.SH", "weight": 0.15, "var": 0.02},
    {"account_id": "ACC001", "symbol": "000858.SZ", "weight": 0.10, "var": 0.015},
    {"account_id": "ACC002", "symbol": "AAPL", "weight": 0.25, "var": 0.03}
  ]
}
'''

# 将JSON字符串解析为Python字典
report_dict = json.loads(risk_report_str)

# record_path 指定"记录列表"所在的键；meta 中的外层字段会附加到每一行
df_flat = pd.json_normalize(report_dict,
                             record_path=______,
                             meta=['report_date'])

print("展平后的持仓风险数据：")
print(df_flat)

import json
import pandas as pd

risk_report_str = '''
{
  "report_date": "2023-10-27",
  "positions": [
    {"account_id": "ACC001", "symbol": "600519.SH", "weight": 0.15, "var": 0.02},
    {"account_id": "ACC001", "symbol": "000858.SZ", "weight": 0.10, "var": 0.015},
    {"account_id": "ACC002", "symbol": "AAPL", "weight": 0.25, "var": 0.03}
  ]
}
'''

# 将JSON字符串解析为Python字典
report_dict = json.loads(risk_report_str)

# record_path 指定"记录列表"所在的键；meta 中的外层字段会附加到每一行
df_flat = pd.json_normalize(report_dict,
                             record_path='positions',
                             meta=['report_date'])

print("展平后的持仓风险数据：")
print(df_flat)

展平后的持仓风险数据：
  account_id     symbol  weight    var report_date
0     ACC001  600519.SH    0.15  0.020  2023-10-27
1     ACC001  000858.SZ    0.10  0.015  2023-10-27
2     ACC002       AAPL    0.25  0.030  2023-10-27

本章小结

📋 行动清单

• 创建一个包含多层嵌套结构的JSON文件，使用json_normalize将其转换为平面表格结构
• 编写一个Python脚本，自动合并当前目录下所有.json文件中的”timestamp”字段，并输出时间范围
• 使用ijson库处理超过1GB的JSON样本文件，记录内存消耗与处理时间

—— 小象教研组