4. 数据获取模块（上）：使用requests库获取实时行情、历史K线数据、处理API返回的JSON数据

做量化交易，第一步就是搞定数据。

没有数据，什么策略都是空谈。我个人习惯把数据获取比作「做饭前的买菜」——菜买不好，厨艺再高也白搭。

这一章，咱们就聊聊怎么用 Python 的 requests 库，从港股市场把数据「抓」回来。

4.1 为什么选 requests？

Python 里能发 HTTP 请求的库不少，但 requests 是我用得最顺手的。原因很简单：

• 语法简洁——两三行代码就能拿到数据
• 文档完善——遇到问题基本都能搜到答案
• 社区活跃——你踩过的坑，别人大概率也踩过

我在项目中遇到过用 urllib 写请求的代码，那叫一个啰嗦。后来全换成 requests，代码量直接砍半。

4.2 获取实时行情

实时行情，说白了就是「现在这一刻」的价格数据。

港股实时行情一般通过 REST API 获取。我以某券商提供的模拟接口为例（实际使用时替换成你自己的 API Key）：

import requests
import json

# 港股实时行情接口（示例）
url = "https://api.xxx.com/hk/quote"
params = {
    "symbol": "00700",  # 腾讯控股
    "api_key": "your_api_key_here"
}

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

if response.status_code == 200:
    data = response.json()
    print(f"当前价格: {data['price']}")
    print(f"涨跌幅: {data['change_pct']}%")
else:
    print(f"请求失败，状态码: {response.status_code}")

嗯，这里要注意：永远不要硬编码 API Key。我习惯用环境变量或者配置文件来管理。

4.3 获取历史K线数据

历史K线数据是回测的基础。没有历史数据，你没法验证策略到底行不行。

港股历史K线通常按「日线」、「周线」、「月线」来组织。接口一般长这样：

# 获取历史K线
url = "https://api.xxx.com/hk/kline"
params = {
    "symbol": "00700",
    "period": "day",      # day / week / month
    "start": "2024-01-01",
    "end": "2024-12-31",
    "api_key": "your_api_key_here"
}

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

# 打印前5条记录
for record in kline_data[:5]:
    print(f"日期: {record['date']}, 开盘: {record['open']}, "
          f"收盘: {record['close']}, 最高: {record['high']}, "
          f"最低: {record['low']}")

返回的数据结构通常是列表，每个元素是一个字典。你想想看，如果数据量很大，一次性拉取可能会超时。我一般会分批次拉取，比如一次拉3个月的数据。

4.4 处理API返回的JSON数据

API 返回的 JSON 数据，说白了就是嵌套的字典和列表。处理起来其实不复杂，但有几个坑要注意。

4.4.1 解析JSON

import json

# 方式一：直接解析 response
data = response.json()

# 方式二：先拿到文本，再解析
text = response.text
data = json.loads(text)

两种方式效果一样。我个人习惯用第一种，少写一行代码。

4.4.2 处理嵌套结构

港股 API 返回的数据经常是嵌套的。比如这样：

{
    "code": 0,
    "message": "success",
    "data": {
        "symbol": "00700",
        "name": "腾讯控股",
        "quotes": [
            {"time": "09:30", "price": 380.0, "volume": 1000},
            {"time": "09:31", "price": 380.5, "volume": 1500}
        ]
    }
}

要拿到里面的行情数据，你得一层层取：

quotes = data['data']['quotes']
for q in quotes:
    print(f"{q['time']} 价格: {q['price']} 成交量: {q['volume']}")

这里有个小技巧：用 .get() 方法代替直接索引，可以避免 KeyError。

# 安全取值
quotes = data.get('data', {}).get('quotes', [])

4.4.3 异常处理

网络请求嘛，总会遇到各种意外。超时、连接失败、数据格式不对……我建议你养成加异常处理的习惯：

try:
    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()  # 检查HTTP状态码
    data = response.json()
except requests.exceptions.Timeout:
    print("请求超时，请检查网络")
except requests.exceptions.ConnectionError:
    print("连接失败，请检查API地址")
except json.JSONDecodeError:
    print("返回数据不是有效的JSON格式")
except Exception as e:
    print(f"未知错误: {e}")

我曾经因为没加超时参数，程序在某个接口上卡了整整5分钟。后来所有请求都加上了 timeout，再也没遇到过类似问题。

4.5 知识体系总览

下面这张图，帮你把这一章的核心逻辑串起来：

4.6 实战小贴士

最后分享几个我在实战中总结的经验：

• 频率控制——别太频繁地请求API。港股交易所通常有频率限制，我一般每秒最多发3次请求。
• 数据缓存——历史数据拉下来后，存到本地文件或数据库里。下次直接用缓存，别重复请求。
• 日志记录——每次请求都打一条日志，方便排查问题。我习惯用 logging 模块，比 print 好用得多。

好了，这一章的内容就到这里。数据获取是量化交易的「地基」，地基打牢了，后面盖楼才稳当。