这两年自己做点小策略回测、看看板块异动最烦的事不是写策略本身而是数据。各大免费软件导出数据动不动加个格式限制字段还不全手动复制粘贴又快把人搞疯。后来有一段时间我干脆用爬虫去抓页面结果页面结构一改代码就废一半维护成本远超收益。直到换了TUShare这套数据接口才把“把行情数据落成本地CSV”这件事彻底跑顺。这篇文章就围绕一个很具体的场景怎样使用TUShare把股票数据存入csv文件。我会从选型理由、环境准备、核心实现、增量更新、常见坑这几个角度展开把代码和踩坑记录都放出来。适合正在做量化入门、数据复盘、或者单纯想把自己关注的股票历史数据存成本地文件的人参考。内容不复杂但每一步我都会解释清楚为什么这么做方便你改造成自己的数据管道。1. 为什么我最终把行情数据接口锁定在TUShare上1.1 数据源选择的三个现实痛点先说数据源这件事。市面上能拿A股行情数据的方式不少但真正常用、能稳定跑批的方案其实有限。我自己的筛选标准就三条第一数据覆盖面得够股票列表、日线行情、交易日历、复权因子要能一起拿到避免多源拼接对不齐第二接口必须简单最好几行代码就能拿到标准的表格数据省去自己解析HTML和JS的过程第三本地存储要方便拿到手的数据能直接按我自己的规则落盘而不是被平台导出格式绑死。按这个标准筛下来TUShare给出的方案基本是最省事的。它提供得比较完整的是一套HTTP接口Tushare Pro通过Token鉴权返回的是结构化表格数据天然适合转成DataFrame再存CSV。对比那些还需要模拟登录、维护Cookie、解析动态网页的方案开发成本完全不在一个量级上。我当时在本地跑了一个小测试用十几行代码就把全市场5000多只股票的基础信息拉了下来字段包括代码、名称、行业、上市日期等存成CSV之后大概几百KB后续所有筛选都能直接基于这个本地文件操作。这个体感比之前用爬虫抓网页强太多了。1.2 和手动下载、爬虫方案的具体对比有不少读者会问那我直接用行情软件的“数据导出”功能不就行了确实行但问题出在自动化上。手动导出适合一次性取数一旦你需要每天都更新、或者要处理几千只股票手动操作就没法持续了。爬虫方案能自动化但维护成本高页面改版、反爬策略、字段缺失都是坑。TUShare这类专业数据接口则把“数据获取”本身做成了标准服务你只需要关注“拿哪些数据、怎么存”不需要关心“数据从哪来、怎么解析”。我用一个表格总结一下对比可能更直观方案自动化程度字段完整度长期维护成本上手难度行情软件手动导出差中等高手动费时低自写爬虫抓网页高低随页面结构变高页面一改就崩高TUShare接口高高字段稳定低接口文档完善低另外提一句并不是说爬虫一无是处如果你拿的是非标准公开页面数据那自然得爬。但单纯做A股行情数据落盘用成熟接口明显是更理性的选择。TUShare在你的数据体量还没大到需要自建数据中心之前作为个人或小团队的数据层完全够用。2. 第一步不是写代码而是把Token和权限边界搞清楚2.1 安装和Token申请很多人一上来就写代码调接口结果第一步就卡在鉴权上。TUShare的接口要求每个请求都带Token所以你得先在它的开放平台注册账号然后在个人主页创建一个Token注意这是一串比较长的字符串后面所有代码都要用到我建议放在环境变量里而不是硬编码在脚本中。安装库本身没什么特别的一行命令的事pip install tushare装完之后官网文档建议的是初始化接口对象import tushare as ts pro ts.pro_api(你的Token)如果你不想把Token写在代码里可以从环境变量读取import os import tushare as ts token os.environ.get(TUSHARE_TOKEN) pro ts.pro_api(token)2.2 权限边界和积分规则这是新手最容易忽略、也最影响体验的一部分。Tushare Pro并不是所有接口都无限制开放的很多高频接口尤其是分钟线、实时行情对账号的积分有门槛积分多少直接影响你能拉取的数据范围和调用频率。我刚用的时候没注意以为所有接口都一样能调结果一开始调分钟线接口就报权限错误排查半天才发现积分不够。常用的几类接口门槛大概是这些接口功能对应方法权限说明股票基础列表stock_basic低门槛新注册即可用日线行情daily低门槛适合大多数回测场景交易日历trade_cal低门槛建议先拉取保存复权因子adj_factor中等门槛做策略回测推荐使用分钟线pro_bar高门槛需积攒积分一个很实用的技巧是把低频且数据量小的基础表股票列表、交易日历先拉下来存成CSV或数据库之后所有计算都基于本地文件这样既能减少对接口的调用次数也能避开频率限制。积分不够的普通用户用daily接口做日线策略回测其实完全够用。3. 从拉取股票列表到批量写CSV的完整实现3.1 先拿市场基础信息股票列表和交易日历真正动手之前你需要理解TUShare的数据逻辑它不是一次性给你全部股票的所有数据而是按照“基础信息—行情数据”两级结构提供。先有股票代码表你才知道自己要拉哪些股票先有交易日历才知道哪些日期有交易避免把休市日当成无数据处理。第一步拿全市场股票基础信息import tushare as ts pro ts.pro_api(你的Token) # 获取当前所有正常上市的A股股票包括代码、名称、行业、上市日期等 stock_list pro.stock_basic( exchange, list_statusL, fieldsts_code,symbol,name,area,industry,market,list_date ) print(stock_list.head()) print(stock_list.shape) # 存为CSV后续可以直接读取 stock_list.to_csv(stock_basic.csv, indexFalse, encodingutf-8-sig)这里有几个参数值得注意。list_status有三个值L表示上市中D表示已退市P表示暂停上市。做回测的时候一般建议拉L就好避免样本里混入大量退市股影响结果。fields按需选择就行不用每次都把所有字段拉完减轻网络传输和存储压力。存CSV时我特别用encodingutf-8-sig是因为直接用电脑自带表格软件打开CSV时容易中文乱码加个BOM头能解决这个问题后面我会专门提。交易日历同样要优先拉取# 拉取2020年至今的交易日历 cal pro.trade_cal( exchangeSSE, start_date20200101, end_date20251231, is_open1 ) cal.to_csv(trade_cal.csv, indexFalse, encodingutf-8-sig) print(cal.head())这里is_open1表示只保留开市日期这个文件在后面的增量更新里会非常有用。你可以用它来判断某一天是不是交易日决定要不要触发数据更新。3.2 核心操作按股票代码拉取日线数据并落盘基础信息拿到之后就要进入这篇文章的正题了如何把某只股票的历史日线数据存入csv文件。日线接口的调用方式如下df pro.daily( ts_code000001.SZ, start_date20180101, end_date20241231 ) print(df.head())返回的字段主要有ts_code、trade_date、open、high、low、close、pre_close、change、pct_chg、vol和amount。这些字段的含义从名字就能猜个大概但做数据落盘时得注意几个点。首先是排序习惯。daily接口返回的数据默认按交易日倒序排列也就是最新日期在最前面。如果你要存成时间序列建议先按trade_date升序排列再写CSVdf df.sort_values(trade_date)其次是存储路径的组织。如果一次性要拉几百只股票就不能每只股票都堆在一个文件夹里我习惯按“股票代码.csv”来命名按行业或板块分子目录import os import pandas as pd def save_daily_to_csv(ts_code, start_date20180101, end_date20241231, dir_namedaily_data): # 确保目录存在 os.makedirs(dir_name, exist_okTrue) df pro.daily(ts_codets_code, start_datestart_date, end_dateend_date) if df is None or df.empty: print(f{ts_code} 无数据) return df df.sort_values(trade_date) file_path os.path.join(dir_name, f{ts_code}.csv) df.to_csv(file_path, indexFalse, encodingutf-8-sig) print(f已保存 {ts_code}共 {len(df)} 条记录 - {file_path}) save_daily_to_csv(000001.SZ) save_daily_to_csv(600519.SH)这段代码虽然简单但已经构成了一个可以循环执行的落盘函数。注意dir_name是个可配置参数这样你可以按板块、按日期范围、按策略分组存储不用每次都在代码里改路径。3.3 批量循环拉取时的节奏控制如果你要拉全市场几千只股票一口气循环调用接口大概率会被限流。原因很简单接口对单账号的每分钟调用次数有上限超了会返回错误或者干脆拒绝服务。实际跑批的时候我建议加入一个简单的循环加延时控制import time import os os.makedirs(daily_data, exist_okTrue) # 这里只以stock_list的前50只为示范 stock_codes stock_list[ts_code].head(50).tolist() success_count 0 fail_count 0 for code in stock_codes: try: df pro.daily(ts_codecode, start_date20240101, end_date20241231) if df is not None and not df.empty: df df.sort_values(trade_date) df.to_csv(fdaily_data/{code}.csv, indexFalse, encodingutf-8-sig) success_count 1 else: print(f{code} 没有数据) fail_count 1 except Exception as e: print(f{code} 拉取失败: {e}) fail_count 1 # 每拉10个停1秒避免触发频率限制 if success_count % 10 0: time.sleep(1) print(f完成成功 {success_count} 只失败 {fail_count} 只)关于延时设置我的经验是先不要无脑sleep先跑几只看日志如果发现接口报错提示频率限制再加延时重试。各账号的权限不同不能照抄网上别人的延时参数这一点得自己摸清楚。另外Tushare Pro的接口本身就是按“日”提供历史数据的日常更新频率也就一天一次完全没必要用毫秒级的并发请求去轰它。4. 增量更新与数据校验CSV落盘的进阶玩法4.1 每次都全量覆盖不是好习惯很多用户把数据存了一次之后隔天又手动跑一遍全量代码结果整个CSV被重新拉取、覆盖一遍。短期内没什么问题但拉的数据越来越多、接口调用频率受限之后这种全量覆盖的做法就很浪费。更理性的方式是做增量更新本地已有的数据不动只拉最后一次交易日期到今天的新数据。实现增量更新的核心是“读取本地旧CSV的最后日期”然后以这个日期为起点去请求新数据import pandas as pd import os def update_stock_daily(ts_code, data_dirdaily_data): file_path os.path.join(data_dir, f{ts_code}.csv) # 如果文件不存在就全量拉取 if not os.path.exists(file_path): df pro.daily(ts_codets_code, start_date20180101, end_date20251231) if df is not None and not df.empty: df df.sort_values(trade_date) df.to_csv(file_path, indexFalse, encodingutf-8-sig) print(f{ts_code}: 首次创建{len(df)} 条) return # 读取已有数据 old_df pd.read_csv(file_path, dtype{ts_code: str, trade_date: str}) if old_df.empty: return # 找到本地最后日期 last_date old_df[trade_date].max() print(f{ts_code}: 本地最新日期 {last_date}) # 增量拉取 new_df pro.daily(ts_codets_code, start_datelast_date, end_date20251231) if new_df is None or new_df.empty: print(f{ts_code}: 没有新增数据) return new_df new_df.sort_values(trade_date) # 去掉可能重复的日期接口可能把last_date当日再次返回 new_df new_df[new_df[trade_date] last_date] if new_df.empty: print(f{ts_code}: 没有新增数据) return # 合并并覆盖存储 merged_df pd.concat([old_df, new_df], ignore_indexTrue) # 按日期去重防止意外重复 merged_df merged_df.drop_duplicates(subset[trade_date], keeplast) merged_df merged_df.sort_values(trade_date) merged_df.to_csv(file_path, indexFalse, encodingutf-8-sig) print(f{ts_code}: 新增 {len(new_df)} 条总计 {len(merged_df)} 条) update_stock_daily(000001.SZ)有几个细节需要解释一下。为什么我在读CSV时要把ts_code和trade_date指定为字符串类型因为直接用Pandas默认读取时trade_date这种“20240102”格式会被当成整数丢掉前导零不说后续字符串比较还会出问题。把日期统一当字符串处理按字典序排序恰好就是时间序非常方便。另一个细节是去重逻辑接口的起始日期设为本地最后日期时它会包含当天数据所以要用new_df[trade_date] last_date过滤一遍避免把最后一天重复追加进去。4.2 用交易日历做完整性校验增量更新做到最后你会发现一个问题怎么知道本地数据是不是“完整”的比如某一天因为接口超时漏拉了一只股票它并不会自动告警你很难察觉。这里就要用到前面存的交易日历了。用交易日历校验的思路是对于一只股票理论上每个交易日都应该有一条日线数据停牌股除外所以可以把本地CSV里的trade_date集合和交易日历文件里的日期集合做差集找出缺失的交易日。当然实际使用时停牌、新股上市等都会造成合法缺失所以做校验的时候要限定合理范围比如只核对该股票上市日期之后、且最近一个完整季度之前的区间。简单实现import pandas as pd # 读取交易日历 cal pd.read_csv(trade_cal.csv, dtype{cal_date: str}) trade_dates set(cal[cal_date].tolist()) # 读取某只股票的本地日线 df pd.read_csv(daily_data/000001.SZ.csv, dtype{trade_date: str}) local_dates set(df[trade_date].tolist()) # 找出交易日历里有、但本地数据里没有的日期 missing sorted(trade_dates - local_dates) print(f交易日总数: {len(trade_dates)}本地记录数: {len(local_dates)}缺失日期数: {len(missing)}) print(missing[:20]) # 只看前20个缺失日期如果缺失数量巨大说明全量拉取就没有成功如果只有少数几天缺失大概率是停牌或复权数据调整造成的。两种情况要分开对待。这个思路很多人用TUSHARE拉到数据后就扔在一边等要用时才发现数据缺了一段到时从头补拉的代价就大了很多。5. 我在实际使用中踩过的坑和目前仍在用的方案5.1 中文编码和字段格式的坑CSV文件用记事本或Excel打开出现中文乱码几乎人人都会遇到一次。这个问题根源在于编码不一致Tushare返回的是UTF-8编码而Windows上老版本表格软件默认用GBK。解决办法就是前面提到的写入CSV时加上encodingutf-8-sig也就是带BOM的UTF-8这样两边都能正确识别。另一个容易忽略的是BigInt字段。amount、vol这些交易量字段的数值可能比较大有时候Pandas自动识别成int64在CSV里会被写成科学计数法。为了保证精度和可读性我在保存前会把关键数字字段的格式固定下来# 对数值字段控制在两位小数 df[change] df[change].round(2) df[pct_chg] df[pct_chg].round(2) df[amount] df[amount].round(2)5.2 接口返回None和限流错误用得久了你会发现就算参数正确偶尔也会出现返回空DataFrame的情况。常见原因有三种该股票确实停牌或暂无数据开始日期晚于上市日期查询区间内没有记录接口因为频率限制临时拒绝服务返回空或抛异常。区分办法很简单在异常处理里打印状态码和信息再结合交易日历判断是合理缺失还是接口问题。我在批处理脚本里一般这样写try: df pro.daily(ts_codecode, start_date20240101, end_date20241231) if df is None or len(df) 0: print(f[{code}] 返回空数据可能停牌或区间无记录) continue except Exception as err: print(f[{code}] 拉取异常{err}) # 异常往往是频率限制先等一会再继续 time.sleep(3) continue5.3 推荐的文件组织方式最后分享一个我目前仍在用的文件组织方式。项目里一般分三类目录basic放股票列表和交易日历daily_data放个股日线文件名直接是股票代码backtest放策略结果。数据更新用一个update.py统一管理先更新股票列表、再增量更新所有日线最后生成一份更新日志CSV。这样做的好处是回测代码只需要读CSV文件完全不依赖网络数据更新脚本单独跑哪里断了从哪里继续日志也能清楚看到每只股票的更新状态。目录结构大概长这样project/ ├── basic/ │ ├── stock_basic.csv │ └── trade_cal.csv ├── daily_data/ │ ├── 000001.SZ.csv │ ├── 600519.SH.csv │ └── ... ├── backtest/ │ └── results/ ├── update.py └── config.py这样做的好处不仅是整洁更关键的是把“数据获取”和“策略研究”两件事彻底解耦。数据层出问题就查数据脚本策略出问题就查策略代码不会互相干扰。在使用这套方案的过程中最大的感受就是数据管道这东西最开始看着简单真正跑起来之后细节非常多。但只要把Token管理、权限边界、CSV格式、增量更新这些点都理顺了后续维护成本确实非常低。现在每天更新一遍全市场的日线数据基本就是双击运行一个脚本的事再也不用打开行情软件手动导出了。