写这篇的动力，源自一次被数据源坑惨的经历：夜深人静调完策略回测，收益率曲线漂亮得不像话，第二天实盘一跑，全变样了。最后定位到问题——数据源里的复权因子算错了一天，把回测结果直接带偏。从那以后，我对"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_code='000001.SZ', start_date='20240101', end_date='20241231' ) # 按日期升序排列 df = df.sort_values('trade_date').reset_index(drop=True) 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_code='000001.SZ') # 合并到行情数据 df = df.merge(adj_factor[['trade_date', 'adj_factor']], on='trade_date', how='left') # 计算前复权价格 = 原始价格 * 当前最新因子 / 当日因子 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_status='L', # L上市中，D已退市，P暂停上市 fields='ts_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_status='D'就能拿到。
3.4 分钟线数据：一个参数引发的血泪教训
分钟线数据的坑，和日线完全不同。Tushare的分钟接口：
# 获取1分钟K线数据（需较高积分） df_min = pro.stk_mins( ts_code='000001.SZ', freq='1min', start_date='2024-12-26 09:30:00', end_date='2024-12-26 15:00:00' )真正的坑是：分钟线数据不带复权因子。如果股票在样本期内有除权除息，分钟K线会在除权日出现价格跳变。处理办法是把复权因子按日下钻到分钟级别——除权日当天开盘前，把当日之前所有分钟价格乘以因子变化比例。这块做不好，高频策略的回测会得出一堆虚假信号。
3.5 组装一个通用获取函数
真正的工程化代码应该长这个样子，把逻辑封装起来：
class StockDataFetcher: def __init__(self, token, data_source='tushare'): 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(f"Unsupported data source: {self.data_source}") def get_daily(self, ts_code, start_date, end_date, adjust='qfq'): 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_code=ts_code, start_date=start_date, end_date=end_date) return df def _apply_adjust_factor(self, df, ts_code, adjust): adj = self.client.adj_factor(ts_code=ts_code) df = df.merge(adj[['trade_date', 'adj_factor']], on='trade_date', how='left') 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'], keep='last') df = df.sort_values('trade_date').reset_index(drop=True) return df这套结构中，_apply_adjust_factor和_clean独立成方法，方便后续替换数据源或者调整清洗逻辑。数据获取（fetch）、加工（adjust+clean）、使用（外部调用）各层不互相污染，是工程上最舒服的状态。
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(exchange='SSE', start_date='20240101', end_date='20241231') 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_code='000001.SZ', fields='ts_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( symbol='000001', period='daily', start_date='20240101', end_date='20241231', adjust='qfq' ) 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']], on='trade_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_code=code, start_date=start_date, end_date=end_date, adjust='qfq') if df.empty: print(f"No data for {code} from {start_date} to {end_date}") return df.to_sql('daily_kline', conn, if_exists='append', index=False)注意：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, interval=3): while True: for symbol in symbols: url = f"https://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_message=on_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, retries=3, backoff=2): for i in range(retries): try: return func() except Exception as e: wait = backoff ** i print(f"Request 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值得。真正该节省的是排查脏数据的时间。
永远保留原始数据。加工后的数据（比如前复权价格）可以覆盖，但原始未复权价格、原始财务字段，必须保留一份不可修改的副本。原因很简单：复权因子会随分红事件增加而变化，一年前的"最新前复权价格"和今天算出来的不一样。如果每次回测都用最新因子去算历史价格，那历史收益率的可比性就丧失了。
数据源越多不是越好，但至少需要两个。一个主源负责生产，一个备用源负责校验和容灾。两个源数据对不上时，别急着判谁对，先查除权除息、停牌、新股上市这些特殊事件。
把数据获取当成软件工程，而不是脚本。数据版本管理、日志记录、异常告警、幂等重跑，这些问题在数据量小的时候无所谓，数据量上来后一个都不能少。
以后有人再问我"股票数据到底怎么拿"，我大概还是会先反问一句：你拿到了数据之后要做什么？想清楚了，再来选工具、写代码。这套流程走一遍，数据质量心里有底，策略的每一步才有依据。
