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

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

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

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

小象量化极客小册 · 华尔街量化分析先锋系列 在量化交易中，历史回测数据往往以多种异构本地文件（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("序列化结果：\n", json_str)

with open('employee.json', 'w', encoding='utf-8') as f:
    json.dump(employee, f, ensure_ascii=False)

with open('employee.json', 'r', encoding='utf-8') as f:
    loaded_data = json.load(f)

print("\n反序列化结果类型：", type(loaded_data))

💡 概念小测：JSON方法区别

当我们向行情服务器发送 API 请求并获得了一个表示股票代码和最新价格的 JSON 格式字符串 json_str 时，应该使用哪种方法将其解析为 Python 字典？

• json.dump(json_str)
• json.load(json_str)
• json.loads(json_str)
• json.dumps(json_str)

📝 选择题

💡 概念小测：JSON方法区别 当我们向行情服务器发送 API 请求并获得了一个表示股票代码和最新价格的 JSON 格式字符串 json_str 时，应该使用哪种方法将其解析为 Python 字典？

• A. json.dump(json_str)
• B. json.load(json_str)
• C. json.loads(json_str)
• D. json.dumps(json_str)

答案：C（json.loads(json_str)）

进阶处理技巧

处理复杂嵌套结构

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

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("\n标准化处理结果：")
print(df)

df.to_json("output.jsonl", orient="records", lines=True)

性能优化策略

大文件处理方案

import ijson

def process_large_json(file_path):
    """流式处理超大JSON文件"""
    with open(file_path, "rb") as f:
        for record in ijson.items(f, "item"):
            # 逐条处理记录
            print(f"处理记录ID: {record.get('id')}")

large_data = [{"id": i, "data": "x"*100} for i in range(10000)]
with open('large.json', 'w') as f:
    json.dump({"item": large_data}, f)

process_large_json('large.json')

💡 代码填空：解析JSON文件

补全代码，从本地保存的行情文件 ticks.json 中读取并解析出股票字典对象：

import json

with open('ticks.json', 'r') as f:
    # 从文件对象中加载 JSON
    tick_dict = json.{blank}(f)
print(tick_dict['AAPL'])

• loads
• load
• dump

✏️ 填空题

💡 代码填空：解析JSON文件 补全代码，从本地保存的行情文件 ticks.json 中读取并解析出股票字典对象：

import json
with open('ticks.json', 'r') as f:
    # 从文件对象中加载 JSON
    tick_dict = json.___(f)
print(tick_dict['AAPL'])

答案：load

本章小结

行动清单

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

「小象量化教研组」