Python实战：5分钟搞定新浪股票API数据抓取与解析（附完整代码）
Python实战：5分钟搞定新浪股票API数据抓取与解析（附完整代码）
最近两年，量化投资在国内逐渐兴起，越来越多的个人开发者开始尝试通过程序化方式获取金融数据。作为国内最常用的免费数据源之一，新浪股票API因其接口稳定、数据全面而备受开发者青睐。今天我们就来手把手教你，如何用Python快速抓取并解析新浪股票数据。
1. 环境准备与接口分析
在开始编码前，我们需要先准备好开发环境。推荐使用Python 3.7及以上版本，主要依赖库包括：
pip install requests pandas新浪股票API主要提供两类数据接口：
- 实时行情接口：返回最新成交价、买卖盘等实时数据
- 历史K线接口：返回指定时间周期的历史行情数据
以实时行情接口为例，其基本URL格式为：
http://hq.sinajs.cn/list=sh601006其中sh601006为股票代码，上海股票以sh开头，深圳股票以sz开头。多个股票可以用逗号分隔。
2. 数据抓取实战
让我们从最简单的单只股票数据抓取开始。下面是一个完整的请求示例：
import requests def get_realtime_data(stock_code): url = f"http://hq.sinajs.cn/list={stock_code}" headers = { "Referer": "http://finance.sina.com.cn", "User-Agent": "Mozilla/5.0" } response = requests.get(url, headers=headers) return response.text # 获取大秦铁路(601006)实时数据 data = get_realtime_data("sh601006") print(data)这段代码需要注意几个关键点：
- 必须设置
Referer请求头，否则会被新浪服务器拒绝 - 返回的数据是JavaScript变量定义格式，需要进一步处理
3. 数据解析技巧
新浪返回的数据格式比较特殊，是一个JavaScript字符串，例如：
var hq_str_sh601006="大秦铁路,27.55,27.25,26.91,27.55,26.20,26.91,26.92,22114263,589824680,...";我们需要用Python将其解析为结构化数据。下面是完整的解析函数：
import re from typing import Dict def parse_sina_data(raw_data: str) -> Dict: # 使用正则提取数据部分 pattern = r'="(.*)"' match = re.search(pattern, raw_data) if not match: return None data_str = match.group(1) items = data_str.split(",") # 主要字段映射 fields = { "name": items[0], # 股票名称 "open": float(items[1]), # 今开 "pre_close": float(items[2]), # 昨收 "price": float(items[3]), # 当前价 "high": float(items[4]), # 最高 "low": float(items[5]), # 最低 "volume": int(items[8]), # 成交量(股) "amount": float(items[9]) # 成交额(元) } # 买卖盘数据(买1-买5，卖1-卖5) for i in range(5): fields[f"bid_{i+1}_price"] = float(items[11+i*2]) fields[f"bid_{i+1}_volume"] = int(items[10+i*2]) fields[f"ask_{i+1}_price"] = float(items[21+i*2]) fields[f"ask_{i+1}_volume"] = int(items[20+i*2]) fields["date"] = items[30] # 日期 fields["time"] = items[31] # 时间 return fields4. 数据存储与分析
将解析后的数据转为Pandas DataFrame会更便于分析。我们可以这样处理：
import pandas as pd def get_dataframe(stock_codes): all_data = [] for code in stock_codes: raw = get_realtime_data(code) parsed = parse_sina_data(raw) if parsed: parsed["code"] = code all_data.append(parsed) df = pd.DataFrame(all_data) # 计算换手率(假设流通股数为已知) df["turnover_rate"] = df["volume"] / 100000000 # 假设流通盘1亿股 return df # 获取多只股票数据 stocks = ["sh601006", "sz000001", "sh601318"] df = get_dataframe(stocks) print(df[["name", "price", "volume", "turnover_rate"]])对于历史K线数据，新浪提供了另一个接口：
def get_kline_data(code, scale=5, ma=5, datalen=1023): url = f"http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol={code}&scale={scale}&ma={ma}&datalen={datalen}" response = requests.get(url) return response.json() # 获取平安银行(sz000001)5分钟K线 kline = get_kline_data("sz000001") kline_df = pd.DataFrame(kline) kline_df["day"] = pd.to_datetime(kline_df["day"]) print(kline_df.head())5. 常见问题与优化建议
在实际使用中，你可能会遇到以下问题：
请求频率限制：新浪对高频请求会封禁IP
- 解决方案：添加随机延迟，使用代理IP池
数据缺失或异常：部分字段可能为空
- 解决方案：添加异常处理逻辑
def safe_float(x): try: return float(x) except: return None # 在parse_sina_data中使用safe_float替代直接float转换- 性能优化：批量请求时可以使用多线程
from concurrent.futures import ThreadPoolExecutor def batch_get_data(stock_codes, max_workers=5): with ThreadPoolExecutor(max_workers=max_workers) as executor: results = list(executor.map(get_realtime_data, stock_codes)) return [parse_sina_data(r) for r in results if r]- 数据存储建议：对于长期收集的数据，建议使用数据库存储
import sqlite3 def save_to_db(data, db_path="stocks.db"): conn = sqlite3.connect(db_path) df = pd.DataFrame(data) df.to_sql("realtime_data", conn, if_exists="append", index=False) conn.close()6. 完整代码示例
最后，我们把所有功能整合成一个完整的类：
import requests import re import pandas as pd from concurrent.futures import ThreadPoolExecutor from typing import List, Dict import time import random class SinaStockAPI: def __init__(self): self.headers = { "Referer": "http://finance.sina.com.cn", "User-Agent": "Mozilla/5.0" } def get_realtime(self, stock_codes: List[str]) -> List[Dict]: """获取实时行情数据""" def fetch_single(code): url = f"http://hq.sinajs.cn/list={code}" try: response = requests.get(url, headers=self.headers) return response.text except Exception as e: print(f"Error fetching {code}: {str(e)}") return None # 添加随机延迟避免被封 time.sleep(random.uniform(0.1, 0.5)) with ThreadPoolExecutor(max_workers=5) as executor: results = list(executor.map(fetch_single, stock_codes)) return [self.parse_data(r) for r in results if r] def parse_data(self, raw_data: str) -> Dict: """解析新浪返回的数据格式""" if not raw_data: return None pattern = r'="(.*)"' match = re.search(pattern, raw_data) if not match: return None items = match.group(1).split(",") if len(items) < 32: return None # 基础字段 data = { "name": items[0], "code": raw_data.split("=")[1].split("_")[-1], "open": self._safe_float(items[1]), "pre_close": self._safe_float(items[2]), "price": self._safe_float(items[3]), "high": self._safe_float(items[4]), "low": self._safe_float(items[5]), "volume": self._safe_int(items[8]), "amount": self._safe_float(items[9]), "date": items[30], "time": items[31] } # 买卖盘 for i in range(5): data[f"bid_{i+1}_price"] = self._safe_float(items[11+i*2]) data[f"bid_{i+1}_volume"] = self._safe_int(items[10+i*2]) data[f"ask_{i+1}_price"] = self._safe_float(items[21+i*2]) data[f"ask_{i+1}_volume"] = self._safe_int(items[20+i*2]) return data def get_kline(self, code: str, scale=5, ma=5, datalen=1023) -> pd.DataFrame: """获取K线数据""" url = f"http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol={code}&scale={scale}&ma={ma}&datalen={datalen}" try: response = requests.get(url, headers=self.headers) data = response.json() df = pd.DataFrame(data) df["day"] = pd.to_datetime(df["day"]) return df except Exception as e: print(f"Error fetching kline for {code}: {str(e)}") return pd.DataFrame() def _safe_float(self, x): try: return float(x) except: return None def _safe_int(self, x): try: return int(x) except: return None # 使用示例 if __name__ == "__main__": api = SinaStockAPI() # 获取多只股票实时数据 stocks = ["sh601006", "sz000001", "sh601318"] realtime_data = api.get_realtime(stocks) df = pd.DataFrame(realtime_data) print(df[["name", "code", "price", "volume"]]) # 获取K线数据 kline_df = api.get_kline("sz000001") print(kline_df.head())这个类封装了所有核心功能，包括：
- 多线程并发请求
- 数据解析与类型转换
- 异常处理
- K线数据获取
使用时只需要创建一个SinaStockAPI实例，然后调用相应方法即可。