4、RESTful API接入实战:以新浪财经为例,获取实时股票行情
好,咱们今天来点真家伙。
前面讲了那么多理论,什么RESTful风格、HTTP协议、JSON格式……说白了,都是为了这一刻——真正把数据从接口里拽出来。我个人习惯是,学技术一定要有个具体的抓手。新浪财经的免费行情接口,就是最好的练手对象。
4.1 为什么选新浪财经?
你可能会问,市面上那么多数据源,为什么偏偏拿新浪开刀?
原因有三:
- 免费——不用申请Token,不用签协议,拿来就用
- 稳定——我在项目中跑过三年,基本没断过
- 简单——接口设计直白,适合教学
嗯,这里要注意:免费的东西往往有坑。我曾经在生产环境里吃过一次亏,后面会专门讲。
4.2 接口地址与参数解析
先看核心接口:
GET https://hq.sinajs.cn/list=sh600519,sz000001,sz300750
参数说明:
| 参数 | 说明 | 示例 |
|---|---|---|
| list | 股票代码列表,逗号分隔 | sh600519,sz000001 |
| 前缀 | sh=上海,sz=深圳,hk=港股 | sh600519 |
你想想看,这个设计其实挺巧妙的。一个参数搞定多只股票,批量查询效率很高。我在做实时行情监控系统时,一次拉50只股票完全没问题。
4.3 实战:用Python拉取数据
直接上代码。这是我常用的写法:
import requests
def fetch_realtime_quote(codes):
"""
获取新浪财经实时行情
:param codes: list of str, 如 ['sh600519', 'sz000001']
:return: list of dict
"""
url = "https://hq.sinajs.cn/list=" + ",".join(codes)
headers = {
"Referer": "https://finance.sina.com.cn",
"User-Agent": "Mozilla/5.0"
}
try:
resp = requests.get(url, headers=headers, timeout=5)
resp.encoding = "gbk" # 注意编码!
return resp.text
except Exception as e:
print(f"请求失败: {e}")
return None
# 调用示例
data = fetch_realtime_quote(["sh600519", "sz000001"])
print(data[:200]) # 先看前200个字符
这里有个坑,我当年踩过——编码问题。新浪返回的是GBK编码,如果你直接用UTF-8解析,中文会变成乱码。我曾经因为这个排查了半小时,最后发现是编码没指定。
4.4 解析返回数据
返回的数据长这样:
var hq_str_sh600519="贵州茅台,1520.00,1518.50,1535.00,1540.00,1510.00,...";
说白了,就是个逗号分隔的字符串。我们需要把它拆开:
def parse_sina_quote(raw_text):
"""
解析新浪财经返回的行情数据
"""
lines = raw_text.strip().split("\n")
result = []
for line in lines:
if not line.startswith("var hq_str_"):
continue
# 提取股票代码
code = line.split("_")[2].split("=")[0]
# 提取数据部分
data_part = line.split("\"")[1]
fields = data_part.split(",")
quote = {
"code": code,
"name": fields[0],
"open": float(fields[1]),
"prev_close": float(fields[2]),
"current": float(fields[3]),
"high": float(fields[4]),
"low": float(fields[5]),
"volume": int(fields[8]),
"amount": float(fields[9])
}
result.append(quote)
return result
嗯,这里要注意:字段索引千万别搞错。我刚开始写的时候,把开盘价和昨收价搞反了,导致策略回测数据全错。后来我养成了一个习惯——每次解析完,先打印一条数据人工核对。
4.5 避坑指南
我曾经踩过的三个坑:
- 请求频率限制——新浪虽然没有明说,但如果你每秒请求超过10次,IP会被临时封禁。我建议加个0.2秒的延迟。
- 节假日数据为空——非交易日返回的是空字符串或旧数据。一定要加个数据新鲜度校验。
- 港股前缀不同——港股代码前缀是"hk",比如"hk00700"是腾讯。别用sh/sz去查港股。
4.6 核心逻辑流程图
我把整个流程画成了SVG图,方便你理解:
4.7 完整示例:批量获取并保存
最后,给你一个可以直接跑的生产级代码片段:
import time
import json
def batch_fetch_and_save(codes, output_file="quotes.json"):
"""
批量获取行情并保存到文件
"""
all_quotes = []
for i in range(0, len(codes), 10): # 每批10只
batch = codes[i:i+10]
raw = fetch_realtime_quote(batch)
if raw:
parsed = parse_sina_quote(raw)
all_quotes.extend(parsed)
time.sleep(0.3) # 避免触发限流
with open(output_file, "w", encoding="utf-8") as f:
json.dump(all_quotes, f, ensure_ascii=False, indent=2)
print(f"成功保存 {len(all_quotes)} 条行情数据")
return all_quotes
# 实战调用
codes = ["sh600519", "sz000001", "sz300750", "hk00700"]
data = batch_fetch_and_save(codes)
我的个人建议: 生产环境中,一定要加异常重试机制。网络抖动是常态,不是异常。我曾经因为没加重试,凌晨3点数据采集任务挂了,第二天早上才发现——那叫一个酸爽。
好了,RESTful API接入实战就讲到这里。你拿着这段代码,换个数据源(比如腾讯财经、东方财富)也能用,核心逻辑是一样的。记住:先跑通,再优化,最后上生产。
公众号:蓝海数据掘金营,微信deep3321