ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

A02_Python 取 A 股数据:不装 SDK,只用 requests 的 3 行写

A02_Python 取 A 股数据:不装 SDK,只用 requests 的 3 行写 Python 取 A 股数据不装 SDK只用 requests 的 3 行写法含工程化客户端上一篇在浏览器地址栏里读到了接口契约。现在把同样的请求翻译成 Python。本文不引入任何专属 SDK——只用标准requests并逐步把它升级成一个可复用的客户端会话复用、响应校验、指数退避、DataFrame 落地以及一个真实的「日收益率」小分析。主机地址只出现在代码BASE常量里。1. 从 3 行代码开始importrequests BASEhttps://api.zhituapi.comTOKEN你的token# 访问凭证见第 5 节urlf{BASE}/hs/list/all?token{TOKEN}datarequests.get(url,timeout15).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 BASEhttps://api.zhituapi.comTOKENos.environ.get(QUOTE_API_TOKEN,你的token)classQuoteClient:def__init__(self,baseBASE,tokenTOKEN,timeout15):self.basebase self.tokentoken self.timeouttimeout self._sessionrequests.Session()# 复用连接defget(self,path,paramsNone,max_retries4):发 GET 请求返回解析后的 Python 对象。urlf{self.base}{path}q{token:self.token}ifparams:q.update(params)delay1.0forattemptinrange(max_retries):try:rself._session.get(url,paramsq,timeoutself.timeout)exceptrequests.exceptions.RequestExceptionase:time.sleep(delay)delay*2# 指数退避continuetextr.text.strip()# 关键错误是纯文本 code:msg不是 JSONiftext[:1]notin{[:ifr.status_code401:# 限频time.sleep(delay)delay*2continueraiseRuntimeError(fAPI错误{r.status_code}:{text[:120]})returnr.json()raiseRuntimeError(f重试{max_retries}次仍失败:{url})用法qcQuoteClient()# 平安银行历史日 K前复权klineqc.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 dfpd.DataFrame(kline)field_map{t:日期,o:开盘,c:收盘,h:最高,l:最低,v:成交量(手),a:成交额(元),pc:前收盘,sf:停牌标志,}df.rename(columnsfield_map,inplaceTrue)df[日期]pd.to_datetime(df[日期])print(fDataFrame shape:{df.shape})print(df[[日期,开盘,收盘,最高,最低,成交量(手)]].tail(3).to_string(indexFalse))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-DDo/c开盘 / 收盘元复权后h/l最高 / 最低元v成交量手1 手100 股a成交额元pc前收盘价元sf停牌标志0正常 / 1停牌df.to_csv(平安银行_日K_前复权.csv,indexFalse,encodingutf-8-sig)# Excel 不乱码4. 可复现分析日收益率序列把数据落 DataFrame 后立刻能做点有用的东西——比如日收益率dfdf.sort_values(日期).reset_index(dropTrue)df[收益率]df[收盘].pct_change()# 当日/前收 - 1df[对数收益](df[收盘]/df[收盘].shift(1)).apply(lambdax:x.ln()ifx0else0)df[年化波动率]df[收益率].rolling(252).std()*(252**0.5)print(df[[日期,收盘,收益率,年化波动率]].dropna().tail(3).to_string(indexFalse))这已经是一个最小可复现的量化研究骨架拿数据 → 清洗 → 计算衍生指标。后续接因子、接回测骨架不变。5. 批量拉取与限频实战场景拉 100 只股票的最新行情。免费档约 300 次/分钟需退避importtime codes[000001,000002,000006,000007,000008]rows[]forcodeincodes:barqc.get(f/hs/latest/{code}/d/f)# 最新日 Kbarbar[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(f5 只耗时含间隔约 1.3s推算 100 只约 25s)档位频率上限每次间隔100 只耗时免费档300 次/分0.2s~25s付费档1000~6000 次/分0.01~0.06s39s批量拉全市场 5000 只免费档约 17 分钟——对个人研究足够。6. 错误码体系代码里只需处理三类情况HTTP 状态错误码含义处理400104缺少 token 参数URL 补?token401—请求频率超限指数退避重试403102证书不存在/过期检查 token 是否正确客户端已在get()内统一处理文本错误抛RuntimeError限频自动退避。业务代码无需关心。7. 小结3 行最小代码requests.get(url).json()。工程化客户端Session复用 响应首字符校验防纯文本错误 指数退避。pandas 落地JSON→DataFrame→字段映射→CSV。可复现分析日收益率 / 年化波动率量化骨架即成型。限频退避免费档 300 次/分循环加sleep(0.2)。不装 SDK 反而少了一层黑箱——返回什么字段、错在哪一行看 JSON 就知道。下一篇把同一个BASE塞进 Excel 单元格。
返回列表