A02_Python 取 A 股数据:不装 SDK,只用 requests 的 3 行写
Python 取 A 股数据:不装 SDK,只用 requests 的 3 行写法(含工程化客户端)
上一篇在浏览器地址栏里读到了接口契约。现在把同样的请求翻译成 Python。本文不引入任何专属 SDK——只用标准
requests,并逐步把它升级成一个可复用的客户端:会话复用、响应校验、指数退避、DataFrame 落地,以及一个真实的「日收益率」小分析。主机地址只出现在代码BASE常量里。
1. 从 3 行代码开始
importrequests BASE="https://api.zhituapi.com"TOKEN="你的token"# 访问凭证,见第 5 节url=f"{BASE}/hs/list/all?token={TOKEN}"data=requests.get(url,timeout=15).json()print(f"获取到{len(data)}只股票")获取到 5209 只股票不装 SDK,不配专属环境,不需要pip install任何行情包。requests是 Python 生态最通用的 HTTP 库,几乎所有环境都有。
对比:用 Tushare 取同样的数据,需要
pip install tushare、注册、攒积分、调pro_api()。多出来的每一步都是潜在卡点。而纯 HTTP 接口的返回字段一目了然,调试时直接看 JSON。
2. 工程化客户端:会话复用 + 响应校验 + 退避
3 行能跑,但生产环境不能这么写。下面这个客户端解决了三件事:
- 会话复用:用
requests.Session()复用 TCP 连接,批量拉取更快。 - 响应校验:接口错误以纯文本(
104:.../102:...)返回而非 JSON,必须先判首字符再json(),否则会抛异常。 - 指数退避:限频(401)时按 2 的幂次等待,避免雪崩。
importosimporttimeimportrequests BASE="https://api.zhituapi.com"TOKEN=os.environ.get("QUOTE_API_TOKEN","你的token")classQuoteClient:def__init__(self,base=BASE,token=TOKEN,timeout=15):self.base=base self.token=token self.timeout=timeout self._session=requests.Session()# 复用连接defget(self,path,params=None,max_retries=4):"""发 GET 请求,返回解析后的 Python 对象。"""url=f"{self.base}{path}"q={"token":self.token}ifparams:q.update(params)delay=1.0forattemptinrange(max_retries):try:r=self._session.get(url,params=q,timeout=self.timeout)exceptrequests.exceptions.RequestExceptionase:time.sleep(delay)delay*=2# 指数退避continuetext=r.text.strip()# 关键:错误是纯文本 "code:msg",不是 JSONiftext[:1]notin"{[":ifr.status_code==401:# 限频time.sleep(delay)delay*=2continueraiseRuntimeError(f"API错误{r.status_code}:{text[:120]}")returnr.json()raiseRuntimeError(f"重试{max_retries}次仍失败:{url}")用法:
qc=QuoteClient()# 平安银行历史日 K(前复权)kline=qc.get("/hs/history/000001/d/f")print(f"获取到{len(kline)}根日K线")print(kline[-3:])获取到 8689 根日K线 [{'t':'2026-08-07','o':11.23,'c':11.19,'h':11.26,'l':11.10,'v':882977,'a':9.86e8,'pc':11.27}, {'t':'2026-08-10','o':11.18,'c':11.29,'h':11.38,'l':11.16,'v':889060,'a':1.01e9,'pc':11.19}, {'t':'2026-08-11','o':11.31,'c':11.28,'h':11.40,'l':11.24,'v':401133,'a':4.54e8,'pc':11.29}]平安银行自 1991 年上市至今 8689 个交易日的完整日 K,一个 GET 就拿到了。
路径规律:
/hs/history/{代码}/{周期}/{复权}。周期d/w/m/y/5/15/30/60;复权n/f/b。
3. 落 DataFrame:数据分析第一步
拿到 JSON 数组,第一步通常是转pandas.DataFrame并做字段映射:
importpandasaspd df=pd.DataFrame(kline)field_map={'t':'日期','o':'开盘','c':'收盘','h':'最高','l':'最低','v':'成交量(手)','a':'成交额(元)','pc':'前收盘','sf':'停牌标志',}df.rename(columns=field_map,inplace=True)df['日期']=pd.to_datetime(df['日期'])print(f"DataFrame shape:{df.shape}")print(df[['日期','开盘','收盘','最高','最低','成交量(手)']].tail(3).to_string(index=False))DataFrame shape: (8689, 9) 日期 开盘 收盘 最高 最低 成交量(手) 2026-08-07 11.23 11.19 11.26 11.10 882977 2026-08-10 11.18 11.29 11.38 11.16 889060 2026-08-11 11.31 11.28 11.40 11.24 401133| 字段 | 含义 | 单位 |
|---|---|---|
t | 交易日期 | YYYY-MM-DD |
o/c | 开盘 / 收盘 | 元(复权后) |
h/l | 最高 / 最低 | 元 |
v | 成交量 | 手(1 手=100 股) |
a | 成交额 | 元 |
pc | 前收盘价 | 元 |
sf | 停牌标志 | 0=正常 / 1=停牌 |
df.to_csv("平安银行_日K_前复权.csv",index=False,encoding='utf-8-sig')# Excel 不乱码4. 可复现分析:日收益率序列
把数据落 DataFrame 后,立刻能做点有用的东西——比如日收益率:
df=df.sort_values('日期').reset_index(drop=True)df['收益率']=df['收盘'].pct_change()# 当日/前收 - 1df['对数收益']=(df['收盘']/df['收盘'].shift(1)).apply(lambdax:x.ln()ifx>0else0)df['年化波动率']=df['收益率'].rolling(252).std()*(252**0.5)print(df[['日期','收盘','收益率','年化波动率']].dropna().tail(3).to_string(index=False))这已经是一个最小可复现的量化研究骨架:拿数据 → 清洗 → 计算衍生指标。后续接因子、接回测,骨架不变。
5. 批量拉取与限频
实战场景:拉 100 只股票的最新行情。免费档约 300 次/分钟,需退避:
importtime codes=['000001','000002','000006','000007','000008']rows=[]forcodeincodes:bar=qc.get(f"/hs/latest/{code}/d/f")# 最新日 Kbar=bar[0]rows.append({'代码':code,'日期':bar['t'],'收盘':bar['c'],'开盘':bar['o'],'最高':bar['h'],'最低':bar['l']})print(f"{code}: 收盘={bar['c']}日期={bar['t']}")time.sleep(0.2)# 免费档限频,每次间隔 ≥0.2sprint(f"5 只耗时含间隔约 1.3s;推算 100 只约 25s")| 档位 | 频率上限 | 每次间隔 | 100 只耗时 |
|---|---|---|---|
| 免费档 | 300 次/分 | 0.2s | ~25s |
| 付费档 | 1000~6000 次/分 | 0.01~0.06s | 39s |
批量拉全市场 5000 只,免费档约 17 分钟——对个人研究足够。
6. 错误码体系
代码里只需处理三类情况:
| HTTP 状态 | 错误码 | 含义 | 处理 |
|---|---|---|---|
| 400 | 104 | 缺少 token 参数 | URL 补?token= |
| 401 | — | 请求频率超限 | 指数退避重试 |
| 403 | 102 | 证书不存在/过期 | 检查 token 是否正确 |
客户端已在get()内统一处理:文本错误抛RuntimeError,限频自动退避。业务代码无需关心。
7. 小结
- 3 行最小代码:
requests.get(url).json()。 - 工程化客户端:
Session复用 + 响应首字符校验(防纯文本错误)+ 指数退避。 - pandas 落地:JSON→DataFrame→字段映射→CSV。
- 可复现分析:日收益率 / 年化波动率,量化骨架即成型。
- 限频退避:免费档 300 次/分,循环加
sleep(0.2)。
不装 SDK 反而少了一层黑箱——返回什么字段、错在哪一行,看 JSON 就知道。下一篇把同一个BASE塞进 Excel 单元格。