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 避坑指南

我曾经踩过的三个坑:

  1. 请求频率限制——新浪虽然没有明说,但如果你每秒请求超过10次,IP会被临时封禁。我建议加个0.2秒的延迟。
  2. 节假日数据为空——非交易日返回的是空字符串或旧数据。一定要加个数据新鲜度校验。
  3. 港股前缀不同——港股代码前缀是"hk",比如"hk00700"是腾讯。别用sh/sz去查港股。

4.6 核心逻辑流程图

我把整个流程画成了SVG图,方便你理解:

新浪财经行情接入流程 1. 构造请求URL 2. 发送HTTP GET请求 3. 接收GBK编码响应 4. 解析逗号分隔数据 5. 输出结构化JSON ⚠️ 关键注意点 • 必须加Referer头 • 编码用GBK • 加0.2秒请求间隔 • 校验数据新鲜度

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