简介这是一份面向金融数据分析初学者与量化爱好者的Python股票数据处理源码包基于akshare库实现股票数据的自动化抓取、清洗与分析适合想用编程替代手工整理行情数据、搭建个人分析流程的开发者参考。压缩包共442个文件、约12.22MB其中352个py脚本承担数据抓取与处理的核心逻辑55个md与7个rst文档提供使用说明和接口文档另有少量js、yml、yaml、svg及Dockerfile、flake8、toml等配置与部署文件兼顾开发、文档与容器化运行。资源已有1170人学习下载说明其在同类工具中具备一定参考价值。读者可从中获得一套可直接运行的akshare数据接口调用示例、模块化的脚本组织方式、Jupyter交互环境与Docker部署配置以及项目文档、许可证与行为守则等协作规范便于快速理解金融数据处理的工程结构并迁移到自己的分析项目中。1. 用 akshare 拉 A 股数据为什么你的第一版脚本总在收盘后翻车很多人第一次用 Python 做股票金融数据处理都是从 akshare 开始的。装完库敲几行代码stock_zh_a_hist一跑数据哗啦啦出来感觉这事成了。然后第二天早上再跑发现昨天收盘后拉的数据缺了最后几根 K 线或者某只票直接报错返回空 DataFrame。这不是玄学是数据源在收盘后的结算窗口里还没把当日数据落库你抢跑了。这篇笔记要讲清楚一件事基于 Python 的 akshare 股票金融数据处理从拉取、清洗、复权对齐到本地存储一条能每天稳定跑的链路长什么样。适合两类人刚学完 pandas 想拿真实数据练手的新手以及已经在用 akshare 但被字段不一致、复权错位、接口限流折腾过的熟手。我不会只给你一段能跑的代码而是把每个环节为什么这么写、参数怎么调、哪里最容易翻车讲透。数据处理框架可以很复杂但这条链路的核心只有四个动作取、洗、对齐、存。2. akshare 取数接口怎么选日线、复权与字段差异2.1 三个最常用的 A 股行情接口对比akshare 里跟 A 股日线相关的接口不止一个新手最容易混。我一般只用下面三个按场景分工。接口返回内容复权适用场景stock_zh_a_hist东财日线含成交量成交额支持 qfq/hfq/不复权日常行情主数据stock_zh_a_daily新浪日线支持 qfq/hfq需要新浪口径做交叉验证stock_zh_a_spot_em东财实时快照不复权盘中监控、当日快照选stock_zh_a_hist当主力的原因很实际它返回的字段名是中文列顺序稳定日期/开盘/收盘/最高/最低/成交量/成交额/振幅/涨跌幅/涨跌额/换手率一次给全省掉大量重命名工作。新浪接口字段是英文且历史上改过列名维护成本高。2.2 复权参数到底怎么设adjust参数有三个值不复权、qfq前复权、hfq后复权。这里有个血泪经验做历史回测用前复权做长期收益统计用后复权做当日盯盘用不复权。原因是前复权以最新价为基准往前调整每次除权除息后历史价格会整体变化你今天存的前复权数据和明天存的对不上。后复权以上市首日为基准往后调整历史值一旦确定就不再变适合做需要长期稳定的因子计算。很多人不知道这个差异把前复权数据存进数据库当天量因子过了一个除权日发现整个序列跳变排查半天。import akshare as ak import pandas as pd def fetch_daily(symbol: str, start: str, end: str, adjust: str qfq) - pd.DataFrame: symbol: 6位代码如 600519 start/end: YYYYMMDD adjust: | qfq | hfq df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustadjust, ) if df is None or df.empty: return pd.DataFrame() # 统一列名方便后续入库 df df.rename(columns{ 日期: trade_date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 换手率: turnover, }) df[trade_date] pd.to_datetime(df[trade_date]) df[symbol] symbol return df.sort_values(trade_date).reset_index(dropTrue)逻辑说明先判空再重命名是因为接口在代码不存在或停牌期间会返回空表直接 rename 不会报错但后续操作会出问题。sort_values不能省东财返回的顺序大部分时候是升序但偶尔会有乱序尤其是跨年拉取时。参数说明period只填daily要周线月线自己在本地 resample别依赖接口的weekly那个口径和本地重采样经常对不上。start_date和end_date都是闭区间格式必须是 8 位数字字符串传datetime对象会报错。2.3 批量拉取时的限流与重试单只票拉没事一旦你要拉全市场 5000 多只问题就来了。东财接口对高频请求有软限制表现是返回空表或者连接超时不是明确的 429。我一般加两层保护请求间隔和失败重试。import time import random def fetch_batch(symbols, start, end, adjustqfq, sleep0.3, retry3): frames [] for i, sym in enumerate(symbols): for attempt in range(retry): try: df fetch_daily(sym, start, end, adjust) if not df.empty: frames.append(df) break except Exception as e: if attempt retry - 1: print(f{sym} 最终失败: {e}) else: time.sleep(sleep * (attempt 1) random.random()) # 每只票之间留间隔避免触发限流 time.sleep(sleep random.random() * 0.2) if (i 1) % 100 0: print(f已处理 {i1}/{len(symbols)}) return pd.concat(frames, ignore_indexTrue) if frames else pd.DataFrame()逻辑说明重试间隔用sleep * (attempt 1)做退避再加一个随机抖动避免所有请求在同一时刻重试形成脉冲。每 100 只打印进度方便中断后知道从哪继续。参数说明sleep0.3是实测比较稳的值全市场拉完大概 25 到 30 分钟。想快可以降到 0.1但失败率会明显上升反而更慢。retry3够用超过 3 次还失败的基本是代码本身有问题重试没意义。3. 数据清洗与复权对齐把脏数据挡在入库前3.1 停牌、涨跌停与异常值的识别拉回来的原始数据不是干净的。停牌日接口可能不返回也可能返回一条成交量为 0 的记录不同接口行为不一致。涨跌停日的价格是正常的但如果你做的是分钟级或 tick 级策略涨跌停日的成交额会异常小需要单独标记。我一般做三件事补交易日历、标记停牌、剔除明显异常。def clean_daily(df: pd.DataFrame, calendar: pd.DatetimeIndex) - pd.DataFrame: df df.set_index(trade_date) # 用交易日历 reindex缺失的交易日就是停牌 df df.reindex(calendar) df[is_suspended] df[volume].isna() | (df[volume] 0) # 价格字段前向填充但标记出来 price_cols [open, close, high, low] df[price_cols] df[price_cols].ffill() # 剔除价格全为 0 或负数的异常行 mask_bad (df[price_cols] 0).any(axis1) df df[~mask_bad] return df.reset_index().rename(columns{index: trade_date})逻辑说明reindex到完整交易日历是关键一步它把停牌日显式暴露出来而不是让它们悄悄消失导致后续计算错位。前向填充价格是为了让技术指标连续但必须用is_suspended标记否则你会把停牌日的价格当成真实成交价。参数说明交易日历用ak.tool_trade_date_hist_sina()获取返回的是历史所有交易日取你数据范围内的即可。注意这个日历不含未来日期做实时任务时要自己补当天。3.2 前复权与后复权的对齐陷阱前面说了前复权数据会随除权日变化。如果你每天增量拉取前复权数据然后追加到同一张表除权日之后你会发现新旧数据接不上价格序列出现断层。这是最常见的翻车点。正确做法有两种。第一种每次全量重拉前复权数据覆盖旧表。适合数据量不大、每天跑一次的场景。第二种存不复权数据加复权因子用的时候自己算。适合数据量大、需要灵活切换复权方式的场景。def align_qfq(full_df: pd.DataFrame, new_df: pd.DataFrame) - pd.DataFrame: 全量重拉模式用最新一次全量数据覆盖 full_df 和 new_df 都是 fetch_daily 的输出 combined pd.concat([full_df, new_df], ignore_indexTrue) combined combined.drop_duplicates(subset[symbol, trade_date], keeplast) return combined.sort_values([symbol, trade_date]).reset_index(dropTrue)逻辑说明keeplast保证新拉的数据覆盖旧的因为新数据是基于最新除权状态计算的前复权值。drop_duplicates的 subset 必须同时包含 symbol 和 trade_date只按日期去重会把不同股票的同日数据误删。参数说明这个函数假设你已经有了全量历史数据。如果是第一次跑full_df 传空 DataFrame 即可。增量场景下建议每天拉最近 60 个交易日的数据做覆盖而不是只拉当天因为除权会影响更早的历史值。3.3 字段类型与缺失值处理从接口拿到的 DataFrame数值列默认是 object 或 float64成交量可能是 int 也可能是 float取决于当天有没有成交。入库前统一类型能省掉后面无数麻烦。def normalize_dtypes(df: pd.DataFrame) - pd.DataFrame: float_cols [open, close, high, low, amount, turnover] int_cols [volume] for c in float_cols: if c in df.columns: df[c] pd.to_numeric(df[c], errorscoerce).astype(float64) for c in int_cols: if c in df.columns: df[c] pd.to_numeric(df[c], errorscoerce).fillna(0).astype(int64) df[symbol] df[symbol].astype(str).str.zfill(6) return df逻辑说明errorscoerce把无法转换的值变成 NaN而不是抛异常中断整个流程。成交量用fillna(0)是因为停牌日成交量就是 0不是缺失。zfill(6)保证股票代码始终是 6 位避免 1 和 000001 被当成两只票。参数说明turnover换手率在某些接口里是百分比字符串带%to_numeric处理不了需要先str.replace(%, )。我这里假设接口返回的是数值如果你的版本返回字符串加一步清洗。4. 本地存储选型CSV、Parquet 还是 SQLite4.1 三种存储方式的实测对比数据拉回来洗干净了存哪是个实际问题。我用过 CSV、Parquet 和 SQLite 三种各有适用场景。存储写入速度读取速度文件大小适用场景CSV慢慢大临时交换、人工查看Parquet快快小批量分析、长期归档SQLite中中中需要按条件查询、增量更新全市场 5000 只票 10 年日线CSV 大概 800MB 到 1GBParquet 压缩后 150MB 左右SQLite 约 400MB。读取全量做因子计算Parquet 比 CSV 快 5 到 10 倍。但如果你的需求是查某只票某段时间SQLite 加索引后比 Parquet 全表扫描快得多。4.2 用 Parquet 做批量归档Parquet 适合按股票分区存储每只票一个文件或者按年份分区。我一般按 symbol 分区因为查询模式大多是取某只票的全部历史。import os import pyarrow as pa import pyarrow.parquet as pq def save_parquet(df: pd.DataFrame, root: str): os.makedirs(root, exist_okTrue) for sym, group in df.groupby(symbol): path os.path.join(root, f{sym}.parquet) table pa.Table.from_pandas(group, preserve_indexFalse) pq.write_table(table, path, compressionsnappy) def load_parquet(root: str, symbol: str) - pd.DataFrame: path os.path.join(root, f{symbol}.parquet) if not os.path.exists(path): return pd.DataFrame() return pq.read_table(path).to_pandas()逻辑说明按 symbol 分组写入每个文件独立读取时只加载需要的票不用全表扫描。preserve_indexFalse避免把 pandas 索引写进去省空间也避免读取时多一列。参数说明compressionsnappy是速度和压缩比的平衡点追求更小体积可以用zstd但写入会慢一些。如果单只票数据量很大比如分钟级可以再按年份二级分区。4.3 用 SQLite 做增量更新与查询需要频繁按条件查询或者做增量更新时SQLite 更合适。建表时把(symbol, trade_date)设成联合主键插入时用INSERT OR REPLACE实现幂等。import sqlite3 def init_db(db_path: str): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS daily ( symbol TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, close REAL, high REAL, low REAL, volume INTEGER, amount REAL, turnover REAL, is_suspended INTEGER DEFAULT 0, PRIMARY KEY (symbol, trade_date) ) ) conn.execute(CREATE INDEX IF NOT EXISTS idx_date ON daily(trade_date)) conn.commit() return conn def upsert_daily(conn, df: pd.DataFrame): df df.copy() df[trade_date] df[trade_date].dt.strftime(%Y-%m-%d) cols [symbol, trade_date, open, close, high, low, volume, amount, turnover, is_suspended] df df.reindex(columnscols) placeholders ,.join([?] * len(cols)) sql fINSERT OR REPLACE INTO daily ({,.join(cols)}) VALUES ({placeholders}) conn.executemany(sql, df.itertuples(indexFalse, nameNone)) conn.commit()逻辑说明INSERT OR REPLACE依赖主键冲突检测所以主键必须设对。itertuples比iterrows快很多大批量插入时差距明显。日期存成字符串而不是日期类型是为了跨平台兼容SQLite 的日期类型本质也是字符串。参数说明idx_date索引是给查某天全市场这类查询用的如果你只按 symbol 查主键索引就够了可以不加。executemany一次提交不要每行 commit否则慢到怀疑人生。5. 避坑与排查akshare 数据处理最常见的 5 个翻车现场5.1 收盘后立刻拉数据最后一根 K 线缺失现象下午 3 点收盘3 点 05 分跑脚本返回的数据最后一根是昨天的今天的没有。原因东财数据落库有延迟收盘后通常要等 15 到 30 分钟才完整。不同接口延迟不一样stock_zh_a_hist一般比实时快照慢。解决定时任务设在收盘后 40 分钟再跑或者跑之前先判断当天数据是否存在不存在就等 10 分钟重试。别用实时快照接口拼日线口径对不上。5.2 前复权数据每天对不上历史值跳变现象昨天存的前复权收盘价今天重新拉同一时间段发现除权日之前的值全变了。原因前复权以最新价为基准除权除息后基准变了历史值整体平移。这是前复权的定义决定的不是 bug。解决要么每天全量覆盖要么存不复权加复权因子。做因子计算和回测建议用后复权历史值稳定。前复权只适合看盘和展示。5.3 批量拉取中途返回空表误判为停牌现象拉 5000 只票中间有几十只返回空 DataFrame检查发现这些票当天正常交易。原因接口限流或网络抖动不是停牌。停牌是返回一条成交量为 0 的记录不是空表。解决空表要重试重试 3 次还空才标记为异常。停牌判断用volume 0不要用df.empty。两者混在一起会导致你的停牌标记全是错的。5.4 股票代码前导零丢失000001 变成 1现象从 CSV 读回来或者从某些接口拿到代码000001变成1和1开头的其他票冲突。原因CSV 读取时 pandas 自动推断类型把字符串000001转成了整数1。接口返回的代码有时也是整数。解决读取时显式指定dtype{symbol: str}拿到后统一zfill(6)。入库前再检查一遍确保全是 6 位字符串。5.5 成交量单位不统一手和股混用现象不同接口拿到的成交量一个是以手为单位一个是以股为单位合并后数据量级差 100 倍。原因东财接口的成交量单位是手新浪接口是股。akshare 文档里写了但很容易忽略。解决统一转成股乘以 100。在fetch_daily里加一步df[volume] df[volume] * 100并在注释里写清楚。别指望后面发现量级错误会直接毁掉你的因子。6. 进阶技巧用交易日历和缓存把每日任务压到 3 分钟6.1 增量拉取的正确姿势全量拉 5000 只票 10 年数据第一次跑 30 分钟可以接受。但每天跑一次还全量拉就是浪费。增量拉取只取最近 N 天N 取 60 比较稳因为除权会影响历史值60 天覆盖了大部分除权调整窗口。from datetime import datetime, timedelta def incremental_update(symbols, db_path, lookback60): conn init_db(db_path) end datetime.now().strftime(%Y%m%d) start (datetime.now() - timedelta(dayslookback)).strftime(%Y%m%d) df fetch_batch(symbols, start, end, adjusthfq, sleep0.2) if df.empty: print(无新数据) return df normalize_dtypes(df) upsert_daily(conn, df) conn.close() print(f更新完成共 {len(df)} 条)逻辑说明用后复权做增量因为后复权历史值稳定增量覆盖不会产生断层。lookback60是自然日不是交易日60 个自然日大约覆盖 40 个交易日足够覆盖除权调整。参数说明sleep0.2比全量时的 0.3 小因为增量数据量小限流风险低。如果跑的时候发现失败率高调回 0.3。6.2 用本地缓存避免重复请求同一天内多次跑脚本没必要重复请求接口。加一层本地缓存按日期存原始返回第二次直接读缓存。import pickle from pathlib import Path CACHE_DIR Path(./cache) CACHE_DIR.mkdir(exist_okTrue) def fetch_with_cache(symbol, start, end, adjust): key f{symbol}_{start}_{end}_{adjust}.pkl cache_file CACHE_DIR / key if cache_file.exists(): with open(cache_file, rb) as f: return pickle.load(f) df fetch_daily(symbol, start, end, adjust) with open(cache_file, wb) as f: pickle.dump(df, f) return df逻辑说明缓存 key 包含所有影响结果的参数避免不同参数命中同一缓存。用 pickle 而不是 CSV因为 pickle 保留 dtype读回来不用重新转换。参数说明缓存文件会越积越多建议加一个清理逻辑删除 7 天前的缓存。或者按日期分目录每天一个文件夹定期整个删掉。6.3 验证数据完整性的三个检查每天任务跑完别直接关掉。加三个检查能提前发现大部分问题。第一检查当天交易日是否有数据。用ak.tool_trade_date_hist_sina()判断今天是不是交易日是的话检查最新日期是不是今天。第二检查股票数量。全市场正常在 5000 只左右如果某天突然变成 3000说明拉取出了问题。第三检查价格范围。A 股价格正常在 0.5 到 3000 之间出现负数或超过 10000 的标记出来人工看一眼。def validate(df: pd.DataFrame, trade_date: str): issues [] today_df df[df[trade_date] trade_date] if today_df.empty: issues.append(当日无数据) if len(today_df) 4000: issues.append(f股票数量异常: {len(today_df)}) bad_price today_df[(today_df[close] 0) | (today_df[close] 10000)] if not bad_price.empty: issues.append(f价格异常: {len(bad_price)} 条) return issues逻辑说明三个检查覆盖了最常见的三类问题数据缺失、数量异常、数值异常。返回问题列表而不是直接抛异常方便你决定是中断还是继续。参数说明4000这个阈值是经验值全市场 5000 多只正常不会低于 4500设 4000 留了缓冲。价格上限 10000 是因为 A 股历史最高价没超过这个数超过基本是数据错误。这套链路我跑了两年多从最开始每天手动跑脚本到现在定时任务自动完成中间踩的坑基本都在上面了。最大的教训是别相信接口返回的数据是干净的也别相信昨天的代码今天还能跑。数据源会变字段会变限流策略会变唯一不变的是你得每天看一眼验证结果。希望帮到你。本文还有配套的精品资源点击获取