写这篇的动力源自一次被数据源坑惨的经历夜深人静调完策略回测收益率曲线漂亮得不像话第二天实盘一跑全变样了。最后定位到问题——数据源里的复权因子算错了一天把回测结果直接带偏。从那以后我对stock 信息获取这件事的态度彻底变了拿不到数据是小事拿到脏数据才是灾难。所以这篇不只是讲怎么把股票数据拉下来更想聊聊这几年代码和数据源打交道的完整思路从最基础的接口调用到数据清洗、复权处理、存储选型再到增量更新和实时行情最后落回如何验证数据对不对这个最容易被忽视的环节。1. 先用一张表看清股票信息到底分哪几类stock 信息获取听起来是一个词实际操作中是完全不同的几类数据搞混了会让后面的代码结构一团糟。我建议动手之前先在心里建一个分类框架。1.1 按业务用途划分的四类核心数据基础信息数据股票代码、名称、所属行业、上市日期、总股本、流通股本、注册资本、注册地址。这类数据变化频率极低通常一天拉一次甚至一周拉一次都够用。行情交易数据开高低收、成交量、成交额、换手率、振幅、涨跌幅。这是绝大多数策略的核心输入频率可以是日线、分钟线、甚至逐笔成交。数据量最大、最占存储也最容易出问题。财务基本面数据营收、净利润、ROE、毛利率、资产负债率、现金流、每股收益。季度更新一次但存在预告-快报-正式财报多个版本需要靠公告日期去对齐否则未来函数问题会很致命。资金面与情绪面数据北向资金流向、主力资金净流入、两融余额、龙虎榜、换手率分位数、涨跌停板数。这类数据对因子增强和择时判断很有价值但来源分散清洗难度高。1.2 按时间频率划分静止数据与动态数据另一个维度是按时间频率分低频静态数据股票列表、行业分类、股本结构。基本可以一次性全量拉取后续每天做增量同步。日频数据日K线、每日财务指标、每日资金流。盘中不用管收盘后拉一次就好。盘中高频数据分钟K线、Tick级行情、盘口五档委买委卖。策略越短周期对数据时间戳精度要求越高网络延迟影响也越大。先想清楚你要做什么再决定你需要哪几类数据。如果你的目标是周频调仓的选股策略一上来就搭Tick级数据管道纯属过度设计。2. 数据源选型免费的往往最贵付费的也要防坑这块是踩坑重灾区。每个数据源都有自己的脾气我一个个说清楚。2.1 免费数据源适合个人研究和策略原型验证Tushare Pro是老牌选手积分制积分门槛决定了你能调用的接口范围。日线基础接口门槛不高分钟数据、财务明细需要更高积分。它的优势是文档规范、字段命名统一社区活跃。缺点是服务器偶尔抽风以及积分限制可能让你在项目中期突然发现某个关键接口没权限。AkShare走的是爬虫聚合路线接口免费、开源、更新勤快数据源来自公开网页。好处是覆盖面极广——宏观数据、行业数据、期货、期权、基金都有。坏处也很明显上游网页改版AkShare接口就崩你必须不断升级版本。生产环境直接用AkShare大概率会出戏剧性的事故。Baostock提供免费的日线、分钟线和财务数据接口风格朴素对新手友好。缺点是数据类型相对基础缺少很多另类数据。2.2 商业数据源适合有一定预算的场景Wind和Choice属于机构标配数据质量、字段覆盖、服务响应都没得说缺点是贵个人开发者基本不考虑。聚宽、米筐、掘金这类量化平台提供的API本质是数据研究环境绑定。好处是数据质量可控本地化程度高回测框架都给你搭好了。缺点是换平台就有迁移成本数据也不是完全开放。2.3 我的选型建议使用场景推荐方案理由个人研究、策略验证Tushare Pro AkShare免费、覆盖广、社区成熟生产级自建数据仓库商业行情API / 券商柜台接口稳定性优先免费源出一次事故就得不偿失快速验证某个新想法AkShare接口全两三行代码就能拉数据落地高频策略研究专业行情服务商速度和精度是硬指标不能用免费源凑合有一点要提醒不要把所有鸡蛋放一个篮子里。就算以Tushare为主源也建议用AkShare或者Baostock做备用校验源。两个源同时出错的可能性远低于一个源出错。3. 核心实现代码这样写后面维护才不痛苦直接上干货。下面这套代码经过了多次重构设计原则就一条让数据获取、数据清洗、数据存储三个环节解耦。3.1 最简单的日线数据获取用Tushare Pro获取日线数据是最常见的起手式import tushare as ts import pandas as pd # 设置token在tushare.pro官网注册后获取 ts.set_token(your_token_here) pro ts.pro_api() # 获取平安银行最近一年的日线数据 df pro.daily( ts_code000001.SZ, start_date20240101, end_date20241231 ) # 按日期升序排列 df df.sort_values(trade_date).reset_index(dropTrue) print(df.head())输出长这样ts_code trade_date open high low close pre_close change pct_chg vol amount 0 000001.SZ 20240102 9.50 9.65 9.42 9.58 9.55 0.03 0.31 837276.04 799325.42 ...字段含义不复杂vol是成交量手amount是成交金额千元pct_chg是涨跌幅百分比。真正坑人的在后面。3.2 千万别忘复权处理日线接口返回的是未复权价格遇到分红送转价格会出现断崖式跳空直接算收益率会得到假信号。必须根据策略需求做前复权或后复权处理。Tushare提供复权因子接口# 获取复权因子 adj_factor pro.adj_factor(ts_code000001.SZ) # 合并到行情数据 df df.merge(adj_factor[[trade_date, adj_factor]], ontrade_date, howleft) # 计算前复权价格 原始价格 * 当前最新因子 / 当日因子 latest_factor df[adj_factor].iloc[-1] df[adj_open] df[open] * latest_factor / df[adj_factor] df[adj_close] df[close] * latest_factor / df[adj_factor]为什么手工算而不是直接用现成API因为掌握原理之后换数据源时不至于抓瞎。前复权的本质是让最新价格保持真实历史价格按因子等比缩放后复权则让首日价格保持真实适合长周期收益计算。3.3 股票列表获取与增量更新要监控全市场首先得有一份股票基础列表# 获取全部A股列表 stock_list pro.stock_basic( exchange, list_statusL, # L上市中D已退市P暂停上市 fieldsts_code,symbol,name,area,industry,market,list_date ) print(f当前A股上市公司数量: {len(stock_list)})每天盘后增量更新时不需要重新拉全量按list_date筛选当天新上市的即可today_new_stocks stock_list[stock_list[list_date] 20241231]退市股票必须要保留。做回测时如果用存活股票天然引入幸存者偏差策略虚高。正确的做法是获取历史上某一天的所有股票包括后来退市了的用list_statusD就能拿到。3.4 分钟线数据一个参数引发的血泪教训分钟线数据的坑和日线完全不同。Tushare的分钟接口# 获取1分钟K线数据需较高积分 df_min pro.stk_mins( ts_code000001.SZ, freq1min, start_date2024-12-26 09:30:00, end_date2024-12-26 15:00:00 )真正的坑是分钟线数据不带复权因子。如果股票在样本期内有除权除息分钟K线会在除权日出现价格跳变。处理办法是把复权因子按日下钻到分钟级别——除权日当天开盘前把当日之前所有分钟价格乘以因子变化比例。这块做不好高频策略的回测会得出一堆虚假信号。3.5 组装一个通用获取函数真正的工程化代码应该长这个样子把逻辑封装起来class StockDataFetcher: def __init__(self, token, data_sourcetushare): self.token token self.data_source data_source self._init_client() def _init_client(self): if self.data_source tushare: import tushare as ts ts.set_token(self.token) self.client ts.pro_api() else: raise ValueError(fUnsupported data source: {self.data_source}) def get_daily(self, ts_code, start_date, end_date, adjustqfq): raw_data self._fetch_daily(ts_code, start_date, end_date) if adjust and adjust ! none: raw_data self._apply_adjust_factor(raw_data, ts_code, adjust) raw_data self._clean(raw_data) return raw_data def _fetch_daily(self, ts_code, start_date, end_date): df self.client.daily(ts_codets_code, start_datestart_date, end_dateend_date) return df def _apply_adjust_factor(self, df, ts_code, adjust): adj self.client.adj_factor(ts_codets_code) df df.merge(adj[[trade_date, adj_factor]], ontrade_date, howleft) latest adj[adj_factor].iloc[-1] first adj[adj_factor].iloc[0] if adjust qfq: df[adj_factor_ratio] latest / df[adj_factor] elif adjust hfq: df[adj_factor_ratio] df[adj_factor] / first for col in [open, high, low, close]: df[f{col}_adj] df[col] * df[adj_factor_ratio] return df def _clean(self, df): df df.dropna(subset[trade_date]) df df.drop_duplicates(subset[ts_code, trade_date], keeplast) df df.sort_values(trade_date).reset_index(dropTrue) return df这套结构中_apply_adjust_factor和_clean独立成方法方便后续替换数据源或者调整清洗逻辑。数据获取fetch、加工adjustclean、使用外部调用各层不互相污染是工程上最舒服的状态。4. 数据清洗与质量校验决定策略生死的一环很多人的策略死在数据质量上却不自知。拿脏数据跑回测结果不但没用还会给出错误的自信。这一节专门讲怎么鉴别和清理脏数据。4.1 常见脏数据类型及判断方法缺失数据某天K线缺一根、财务字段整列为空。处理方式不是简单dropna而要区分是当天停牌还是接口漏数据。停牌时成交量应为0或者无记录漏数据则前后交易日记录正常。判断方法对比交易日历。重复数据同一代码同一交易日出现两行。多发生在增量更新没有做好幂等控制时。用drop_duplicates按ts_code trade_date去重即可。价格异常最高价小于最低价、收盘价超过当日最高价、涨跌幅超过11%A股主板涨跌幅限制10%ST股5%创业板/科创板20%。遇到这些必须逐条排查。单位不一致有的接口成交量单位是股有的是手金额有的是元有的是千元或万元。混用时因子计算会直接放大100倍。4.2 交易日历数据校验的基本参照系# 获取交易日历 cal pro.trade_cal(exchangeSSE, start_date20240101, end_date20241231) is_open cal[cal[is_open] 1][cal_date].tolist()有了交易日历就能做三件事校验行情数据是否缺交易日missing_days set(is_open) - set(df[trade_date])判断当天停牌交易日无K线记录大概率停牌需要单独标记而不是直接当缺失数据删掉对齐财务数据与价格数据保证财务公告日期是交易日避免未来函数4.3 财务数据的版本对齐与未来函数财务数据最隐蔽的坑是未来函数。季报披露日和报告期之间有数周到数月的时滞年报甚至可能拖到次年4月底如果直接用报告期数据做回测等于提前使用了未来才知道的信息。正解是使用公告日期ann_date而不是报告期end_date去对齐。Tushare的income接口里有这两个字段df_income pro.income_vip( ts_code000001.SZ, fieldsts_code,ann_date,end_date,revenue_ps,netprofit_yoy, ) # 回测使用时按ann_date对齐而不是end_date df_income df_income.sort_values(ann_date)财报快报/业绩预告同理如果策略需要提前反应就得引入预告数据但它的口径和正式财报不一样不能直接混用。4.4 一分钟数据的时间戳对齐问题分钟数据最容易忽视的是时区和交易时段。A股交易时段是9:30-11:30和13:00-15:00但不同数据源返回的时间戳可能包含集合竞价数据9:15-9:25也可能在中午休市时多返回一行11:30的K线。处理建议统一过滤非交易时段数据只保留时间在09:30-11:30和13:00-15:00之间的数据集合竞价数据要不要保留取决于策略。若开盘价需要真实开盘价保留9:30那一根即可4.5 两个数据源互相校验我的黄金法则我自己的做法是关键行情数据至少双源比对一次import akshare as ak # 用AkShare作为校验源 df_check ak.stock_zh_a_hist( symbol000001, perioddaily, start_date20240101, end_date20241231, adjustqfq ) df_check.columns [trade_date,open,close,high,low,volume,amount,amplitude,pct_chg,change,turnover] # 对比Tushare和AkShare的收盘价 merged df.merge(df_check[[trade_date, close]], ontrade_date, suffixes(_tushare, _akshare)) merged[diff] (merged[close_tushare] - merged[close_akshare]).abs() error_days merged[merged[diff] 0.01]如果某一日的两个源收盘价差异超过1分钱基本可以断定有一方数据出了问题。这个时候再去查除权除息、停牌、或者源站本身的数据错误。5. 从单次拉取到实时更新数据管道的设计思路拿到一次数据不算本事能日复一日自动把数据仓库维护好才是生产级的方案。5.1 存储选型SQLite够用PostgreSQL更好个人项目从SQLite起步完全够用零配置文件数据持久化简单单文件可备份。数据量超过几G或者需要多进程并发读写时迁到PostgreSQL更稳。import sqlite3 from datetime import datetime conn sqlite3.connect(stock_data.db) cur conn.cursor() # 创建日线表 cur.execute( CREATE TABLE IF NOT EXISTS daily_kline ( ts_code TEXT, trade_date TEXT, open REAL, high REAL, low REAL, close REAL, vol REAL, amount REAL, adj_close REAL, PRIMARY KEY (ts_code, trade_date) ) ) conn.commit()主键用(ts_code, trade_date)天然防重复重复插入同一交易日数据会直接抛异常或者用INSERT OR REPLACE覆盖。5.2 带幂等性的增量更新任务def update_daily_data(fetcher, conn, code, start_date, end_date): df fetcher.get_daily(ts_codecode, start_datestart_date, end_dateend_date, adjustqfq) if df.empty: print(fNo data for {code} from {start_date} to {end_date}) return df.to_sql(daily_kline, conn, if_existsappend, indexFalse)注意to_sql只有在表没有主键约束时才会简单追加。如果表里有主键重复插入会报错。所以生产环境我通常先查一遍数据库里已有的最大交易日max_date pd.read_sql( SELECT MAX(trade_date) as max_date FROM daily_kline WHERE ts_code ?, conn, params(code,) )[max_date][0]然后从max_date的下一个交易日开始拉取保证数据完整的同时也避免重复请求浪费API积分。5.3 实时行情推送轮询还是WebSocket分钟级别以下的数据适合用WebSocket推送。A股常见的免费方案是新浪/腾讯行情接口但稳定性和合规性都要自己评估。商业数据源通常提供官方WebSocket。轮询方案比较笨但简单适合分钟级import time import requests def poll_realtime_quote(symbols, interval3): while True: for symbol in symbols: url fhttps://your_realtime_api/?code{symbol} resp requests.get(url).json() # 处理报价数据 print(resp) time.sleep(interval)WebSocket方案更优雅适合秒级和Tick级import websocket def on_message(ws, message): # parse message pass ws websocket.WebSocketApp( wss://your_realtime_ws_endpoint, on_messageon_message ) ws.run_forever()图中的关键是WebSocket断线重连是必须处理的。盘中连接断了行情没推过来策略就停摆了。重连逻辑里要追包断线期间漏掉的数据从REST接口补一次。5.4 全套管道的时间表我自己的生产管道长这样时间点执行任务说明15:10拉取当日日线收盘后等行情稳定再拉15:30拉取当日复权因子更新因子表16:00拉取当日资金流向等Level2数据落地17:00拉取财务数据和公告处理公告日期对齐17:30数据校验生成日报双源比对异常告警20:00全量备份SQLite直接复制文件/PostgreSQL做dump数据管道不需要实时到秒级。A股收盘后各项数据逐步落地给你留够了时间窗口。6. 那些让我浪费过时间的坑踩坑记录与避坑清单最后总结一下这些年花真金白银买来的经验。这些坑每个都至少浪费过我一天时间。6.1 接口限流爬着爬着就断流免费数据源几乎都有限流。Tushare按积分和每分钟调用次数限制AkShare则完全取决于上游网站的心情。我踩过最狠的一次是循环拉全市场5000只股票的财务数据拉到3000只时触发了封禁整个IP被暂时限制。从那之后我学乖了所有批量任务必须加time.sleep控制频率并且设置重试机制import time def retry_call(func, retries3, backoff2): for i in range(retries): try: return func() except Exception as e: wait backoff ** i print(fRequest failed: {e}, retrying in {wait}s) time.sleep(wait) raise Exception(Max retries exceeded)6.2 字段名不统一同一字段在不同接口里叫法不同比如成交量有的叫vol有的叫volume还有的叫turnover_vol成交额更是五花八门amount、turnover、money。如果从多个源拉数据必须建立字段映射表统一规范否则后面每写一个策略都要猜一遍字段名。6.3 时区与时间戳毫秒之差订单天壤之别数据本身带时区是一回事自己处理时把时间当字符串存又是一回事。我建议所有时间字段统一用YYYYMMDD或YYYY-MM-DD HH:MM:SS格式存储入库时用pd.to_datetime统一解析绝不让2024/1/2和20240102并存。6.4 涨停/跌停数据看似没异常实则带陷阱一字涨停板的K线特征开盘价最高价最低价收盘价成交量极度萎缩。如果策略里有动量因子可能会把这些股票当成一字板买不进而错过也可能错误地当成无波动过滤掉。更麻烦的是有些数据源在涨停板当天返回的pct_chg可能因除权除息而不准确必须交叉验证。6.5 指数数据和个股数据不能混用同一套清洗逻辑指数没有股本、没有复权因子换手率和量比也无意义。如果一把梭把所有数据塞进同一个DataFrame做清洗指数数据是会出各种奇怪问题的。指数和个股要分开存储、分开清洗。6.6 异常告警让数据管道自己喊救命管道跑着跑着某个源挂了很正常。问题在于你如何第一时间知道。我的做法是加一个埋点每次更新完成后比对全市场股票数量、总成交额和历史均值偏差超过阈值就推送告警到手机。这样即使晚上睡觉时管道出问题第二天一早也能第一时间处理而不是等策略回测结果出来才发现数据从三天前就已经断更了。7. 最后再分享几条具体经验写stock 信息获取写了这么多年我最后的体会就几条先定策略再定数据。周频策略不需要分钟线日线就够日频策略连复权因子都要认真处理高频策略则要想清楚自己能否承受数据延迟和成本。数据粒度不是越细越好够用才是王道。宁可慢不可错。拉全市场日线慢一点没关系控制频率、加断点续传远比一把梭然后被封IP值得。真正该节省的是排查脏数据的时间。永远保留原始数据。加工后的数据比如前复权价格可以覆盖但原始未复权价格、原始财务字段必须保留一份不可修改的副本。原因很简单复权因子会随分红事件增加而变化一年前的最新前复权价格和今天算出来的不一样。如果每次回测都用最新因子去算历史价格那历史收益率的可比性就丧失了。数据源越多不是越好但至少需要两个。一个主源负责生产一个备用源负责校验和容灾。两个源数据对不上时别急着判谁对先查除权除息、停牌、新股上市这些特殊事件。把数据获取当成软件工程而不是脚本。数据版本管理、日志记录、异常告警、幂等重跑这些问题在数据量小的时候无所谓数据量上来后一个都不能少。以后有人再问我股票数据到底怎么拿我大概还是会先反问一句你拿到了数据之后要做什么想清楚了再来选工具、写代码。这套流程走一遍数据质量心里有底策略的每一步才有依据。