Python爬虫构建股票大数据采集清洗与多因子分析

简介面向希望把Python用于股票数据采集、分析与预测实战的金融数据分析学习者内容围绕爬虫抓取、数据清洗、统计计算、可视化及机器学习建模展开帮助读者打通从原始行情到可解释分析结果的基本流程。压缩包采用7z格式分发大小约4.47MB文件总数与类型明细上游暂未提供。已有3659人学习或下载具备一定参考热度。内容涉及Scrapy等爬虫框架、HTTP与反爬处理Pandas缺失值和异常值清洗NumPy、SciPy统计计算Matplotlib、Seaborn绘制K线图、成交量与箱型图同时覆盖线性回归、决策树、随机森林、神经网络及Scikit-learn建模并延伸到Tushare、Wind实时数据获取和Flask、Django结果展示。适合按知识点查阅、对照练习和搭建课程项目。1. 需求先立住股票大数据分析为什么要自己攒一份数据带过几届毕业设计之后我发现股票大数据分析这类题目最容易出问题的地方不在算法而在数据行情软件导出的往往是不复权收盘价字段残缺、口径混乱一旦要跑批量回测就露馅。这套基于 Python 爬虫的方案解决的就是这件事——用爬虫把日线行情、复权因子、基础信息和分钟线抓下来落成一套可查询、可复算的本地数据资产再往上做指标计算、多因子打分和可视化看板。它适合两类人正在做金融数据分析方向毕业设计或课程设计、需要一套完整可跑通结构的学生以及要给团队交付一条行情数据管道、不想每次分析都手工导表的从业者。整套流程在一台普通笔记本上就能跑不依赖任何商业数据终端。2. 数据采集层设计爬虫任务怎么分层、怎么限频、怎么断点续传采集层是整个项目的地基。很多人在这一步图快写一个 for 循环把几千只股票一口气刷完结果跑到一半连接被掐断前面的数据全丢只能从头再来。我的做法是把采集拆成“维表全量、事实表增量、失败补采”三类任务每类任务各自有独立的进度记录和重试策略任何一类中断都不会污染另外两类。这一章把这条链路的骨架讲清楚代码可以直接改成自己的接口地址。2.1 采集对象分层四类数据各走各的任务需要抓的数据可以归成四类股票基础信息代码、名称、上市日期、所属行业、是否退市、日线行情开盘、最高、最低、收盘、成交量、成交额、复权因子分红送转带来的价格调整系数、分钟线可选用于日内分析。基础信息变化频率极低一周同步一次就够日线行情每天收盘后追加复权因子在除权除息日附近才会变动需要单独判断分钟线数据量大通常只抓自选池里的几十只股票。分层的意义在于失败隔离。基础信息抓失败不影响当天行情入库分钟线抓一半断了也不影响日线表的完整性。我在表设计里给每类任务留了一条进度记录任务名就是主键重跑时先读进度再决定从哪天开始。2.2 采集器骨架带超时、退避与抖动的请求函数下面这段是单次请求的核心函数看起来朴素但超时元组、指数退避和随机抖动这三样东西缺一不可。超时元组把连接时间和读取时间分开设避免某次请求卡在读取阶段把整个进程拖死退避让服务端压力降下来抖动能防止所有重试在同一毫秒打过去。# collector.py —— 增量采集骨架 import time import random import sqlite3 import requests BASE_URL https://market-api.example.invalid/v1/daily # 换成自己的行情接口地址 DB_PATH stock.db session requests.Session() session.headers.update({User-Agent: Mozilla/5.0 (compatible;>def load_checkpoint(conn: sqlite3.Connection, task: str, default: str 20200101) - str: row conn.execute( SELECT last_value FROM etl_checkpoint WHERE task ?, (task,) ).fetchone() return row[0] if row else default def save_checkpoint(conn: sqlite3.Connection, task: str, value: str) - None: conn.execute( INSERT INTO etl_checkpoint(task, last_value) VALUES(?, ?) ON CONFLICT(task) DO UPDATE SET last_value excluded.last_value, (task, value), ) conn.commit()进度值的粒度建议按“交易日”而不是“股票”。按股票记会让进度表膨胀到几千行恢复逻辑也要判断“这只股票跑到第几天”复杂度陡增。按交易日记最坏情况是重跑当天的全部股票而当天数据写库用的是INSERT OR REPLACE重跑天然幂等相当于给自己留了一颗后悔药。2.4 限频与并发把采集速度当成可调参数采集速度不是越快越好它应该是一个可以在配置里调的参数。我的经验值是单线程、请求间隔 0.3 到 0.5 秒、带随机抖动几千只股票的日线跑完大概十几分钟这个耗时完全可以接受而稳定性比并发高十倍。真要提速优先考虑“按交易日批量接口”而不是加大并发。def throttle(min_interval: float 0.35) - None: 按最小间隔节流抖动避免固定节奏 time.sleep(min_interval random.uniform(0, 0.15)) def collect_daily(codes, trade_date, conn): for idx, code in enumerate(codes, 1): row fetch_day(code, trade_date) if row: conn.execute( INSERT OR REPLACE INTO fact_daily VALUES (?,?,?,?,?,?,?,?), (code, trade_date, row[open], row[high], row[low], row[close], row[volume], row[amount]), ) throttle() if idx % 200 0: # 每 200 只提交一次兼顾性能与可恢复性 conn.commit() print(f{trade_date} 已完成 {idx}/{len(codes)}) conn.commit()批量提交的间隔也要想清楚。逐条 commit 会让磁盘 IO 成为瓶颈几千条数据能跑出几分钟全程不 commit 又会在异常退出时丢掉全部进度。两百条一提交是个折中配合幂等写入即使中途退出也只损失最后一批。2.5 调度收盘后跑失败第二天自动补调度用标准库之外的轻量方案就够一个 cron 表达式在每天收盘后一小时触发先跑补采再跑当日增量。补采任务的逻辑是查最近五个交易日里哪些日期在fact_daily里的股票数量明显少于维表总数把缺的日期重新入队。这个“数量对不齐就补”的判断比精确比对每一只股票便宜得多实际项目中能拦住绝大多数漏采。3. 清洗与存储三张表把原始数据变成可计算资产原始接口数据直接拿来算指标迟早会出问题。同一只股票在除权日前后价格断层不复权数据算出来的均线毫无意义停牌日如果补零会直接把收益率算成 -100%。这一章把存储结构和清洗规则定死后面所有分析都建立在这套口径上。3.1 表结构维表、事实表、因子表分开建三张核心表加一张进度表字段不多但每个都有明确用途。dim_stock存属性fact_daily存不复权行情fact_adj_factor存复权因子查询时再算复权价原始数据不覆盖。这样做的原因是复权口径会随分红方案变化把复权价直接写死进事实表过一段时间数据就对不上了。-- schema.sql CREATE TABLE IF NOT EXISTS dim_stock ( code TEXT PRIMARY KEY, -- 股票代码 name TEXT NOT NULL, list_date TEXT, -- 上市日期 YYYYMMDD industry TEXT, is_active INTEGER DEFAULT 1 -- 1 在市0 退市 ); CREATE TABLE IF NOT EXISTS fact_daily ( code TEXT NOT NULL, trade_date TEXT NOT NULL, -- YYYYMMDD open REAL, high REAL, low REAL, close REAL, volume REAL, -- 手 amount REAL, -- 元 PRIMARY KEY (code, trade_date) ); CREATE INDEX IF NOT EXISTS idx_daily_date ON fact_daily(trade_date); CREATE TABLE IF NOT EXISTS fact_adj_factor ( code TEXT NOT NULL, trade_date TEXT NOT NULL, adj_factor REAL NOT NULL, -- 累计复权因子 PRIMARY KEY (code, trade_date) );主键设成(code, trade_date)而不是自增 ID是为了让INSERT OR REPLACE直接具备幂等性重跑不会产生重复行。日期索引用在“按交易日取全市场截面”这种高频查询上选股打分和回测都依赖它。3.2 清洗规则缺失、停牌与异常值分开处理清洗不是把所有异常都删掉而是分类打标。成交量与成交额同时为 0 的记录判为停牌保留行但标记suspended计算收益率时跳过价格为空但成交量非空的记录判为接口异常进补采队列单日涨跌幅超过板块限制的先怀疑复权因子没对齐不要急着当脏数据删。涨跌幅限制因板块而异硬编码一个 10% 会误伤。我的做法是把阈值做成配置主板、创业板、科创板各一行超出阈值的记录写入etl_anomaly表人工看一眼而不是直接过滤。血泪经验是被自动过滤掉的那批“异常值”里往往藏着真正的除权日和停牌复牌日。3.3 复权计算原始价存库复权价现算后复权价等于原始价乘以累计复权因子前复权价则要用最新因子做归一化。两种口径的差别在回测里非常致命用前复权价算历史收益率等于用未来的分红信息修正过去的价格收益率会被系统性高估。import pandas as pd def to_adjusted(df: pd.DataFrame, mode: str hfq) - pd.DataFrame: df 需含 code, trade_date, close, adj_factor 四列 df df.sort_values([code, trade_date]).copy() if mode hfq: # 后复权历史价格不变最新价放大 df[close_adj] df[close] * df[adj_factor] else: # 前复权以最新因子为基准归一 latest df.groupby(code)[adj_factor].transform(last) df[close_adj] df[close] * df[adj_factor] / latest return df参数mode建议在配置里全局指定一次全项目只允许一种口径。因子计算、回测、看板三条链路读同一个配置才不会出现“看板显示前复权、回测用后复权”这种玄学差异。缺复权因子的日期用前值填充ffill但只允许向后填充绝不能向前填充否则又把未来信息漏进来了。3.4 入库前校验让错误在写库之前暴露校验函数放在写库之前检查行数、主键重复、价格逻辑关系和日期合法性。跑完一批数据立刻校验比第二天画图时发现曲线断成两截要好排查得多。def validate_batch(df: pd.DataFrame, trade_date: str) - list: errors [] if df.empty: return [f{trade_date} 批次为空] dup df.duplicated(subset[code, trade_date]).sum() if dup: errors.append(f主键重复 {dup} 行) bad_price df[(df[high] df[low]) | (df[close] df[high]) | (df[close] df[low])] if not bad_price.empty: errors.append(f价格逻辑异常 {len(bad_price)} 行: {bad_price[code].tolist()[:5]}) if df[trade_date].nunique() ! 1: errors.append(批次内混入多个交易日) return errors校验失败不抛异常中断整个任务而是把错误写日志并跳过该批次让后面的股票继续跑。等全量跑完再统一看日志一次性修数据比反复中断重跑效率高得多。4. 指标计算与看板从均线因子到能演示的分析界面数据攒齐之后真正的分析工作量其实不大难的是把“能算”变成“能讲清楚”。这一章从因子计算讲到多因子打分再补一个简易回测验证因子有没有信息量最后落到一个可交互的看板让答辩现场有东西可演示。4.1 因子计算向量化写法与参数口径用groupby加transform做分组滚动计算不要写 for 循环遍历股票。几千只股票乘以几年的日线向量化写法几秒钟跑完循环写法要几分钟而且中间结果容易出错。import pandas as pd def add_factors(df: pd.DataFrame) - pd.DataFrame: 输入含 code, trade_date, close_adj, amount 的日线表 df df.sort_values([code, trade_date]).copy() g df.groupby(code, group_keysFalse) df[ma5] g[close_adj].transform(lambda s: s.rolling(5).mean()) df[ma20] g[close_adj].transform(lambda s: s.rolling(20).mean()) df[ret1d] g[close_adj].pct_change() df[vol20] df[ret1d].transform(lambda s: s.rolling(20).std()) df[amt5] g[amount].transform(lambda s: s.rolling(5).mean()) return dftransform返回与原表等长的序列索引能直接对齐避免merge带来的排序问题。滚动窗口里如果遇到停牌日rolling会把停牌行算进窗口导致均线被拉平。处理办法是先把停牌行置为NaN再滚动min_periods设为窗口的一半这样均线只在有效交易日上计算。4.2 多因子打分把几个指标合成一张选股清单单看均线容易选出已经涨完的票加一点量能和波动率做平衡。打分用横截面百分位排名避免不同量纲的指标直接相加。def score_cross_section(df: pd.DataFrame, date: str, top_n: int 20) - pd.DataFrame: snap df[df[trade_date] date].dropna(subset[ma20, vol20]).copy() snap[r_ma] snap[close_adj] / snap[ma20] - 1 # 偏离度 snap[score] ( snap[r_ma].rank(pctTrue) * 0.4 # 趋势权重 snap[amt5].rank(pctTrue) * 0.3 # 流动性权重 (-snap[vol20]).rank(pctTrue) * 0.3 # 低波动权重 ) return snap.nlargest(top_n, score)[[code, close_adj, r_ma, score]]权重不是调出来的最优解而是一个可解释的起点。答辩时评委更关心“为什么是 0.4、0.3、0.3”而不是权重本身多精确。我的回答模板是趋势因子给最高权重是因为样本期内动量效应更稳定量能和波动率作为风险约束做平衡权重取值做了几组敏感性对比结论在合理区间内不敏感。4.3 回测雏形验证因子有没有信息量不需要完整的回测框架一个持有期收益统计就够说明问题。按打分日买入前 N 名持有 5 个交易日后统计平均收益和全市场等权收益对比。def forward_return(df: pd.DataFrame, hold: int 5) - pd.DataFrame: df df.sort_values([code, trade_date]).copy() # shift(-hold) 只用于评估绝不能作为特征输入模型 df[fwd_ret] df.groupby(code)[close_adj].shift(-hold) / df[close_adj] - 1 return df def evaluate(df, dates, hold5, top_n20): rows [] for d in dates: pick score_cross_section(df, d, top_n)[code].tolist() day df[(df[trade_date] d) (df[code].isin(pick))] rows.append({date: d, pick_ret: day[fwd_ret].mean(), mkt_ret: df[df[trade_date] d][fwd_ret].mean()}) res pd.DataFrame(rows) res[excess] res[pick_ret] - res[mkt_ret] return resshift(-hold)是典型的未来函数陷阱它只能出现在评估代码里一旦被误用到因子计算回测收益会漂亮得离谱。我在项目里会把评估函数单独放一个模块命名带上eval_only从物理上降低被误引用的概率。4.4 可视化看板三张图讲完一个分析结论看板不求功能多求能在三分钟内讲清楚。一张 K 线加均线图说明个股趋势一张全市场涨跌分布直方图说明当日情绪一张选股清单表格给出可执行的代码列表。后端用 Flask 提供两个接口前端用最朴素的 HTML 表格加 canvas 图表即可。from flask import Flask, request, jsonify import pandas as pd app Flask(__name__) DF pd.read_pickle(daily_with_factors.pkl) app.route(/api/rank) def api_rank(): date request.args.get(date) or DF[trade_date].max() items score_cross_section(DF, date).round(3).to_dict(records) return jsonify({date: date, count: len(items), items: items}) app.route(/api/kline/code) def api_kline(code): sub DF[DF[code] code].tail(120) cols [trade_date, open, high, low, close_adj, ma5, ma20] return jsonify(sub[cols].round(2).to_dict(records))数据用 pickle 缓存而不是每次查库是因为看板只读、数据量在几十万行级别读进内存后响应能压到几十毫秒。注意这个缓存要在每日采集任务跑完后重建一次否则看板会显示昨天的数据。5. 避坑与常见问题排查五类最容易翻车的点这一章是我在几个类似项目里踩出来的清单每一条都真实发生过按现象、原因、解决三段写。做这类项目的人大概率会撞上其中三条以上提前看一眼能省不少调试时间。5.1 数据口径类复权混用与停牌补零现象回测收益曲线漂亮得不像真的年化收益高到离谱或者某只股票在某个日期出现一根垂直插针。原因一是复权口径混用因子计算用了前复权、收益统计用了不复权二是停牌日被当成正常交易日补了零pct_change算出 -100% 的收益率把均值彻底带偏。解决全项目只保留一个ADJ_MODE配置项所有读取函数统一从配置取停牌行保留但价格置为NaN用ffill取前一日价格收益率自然变成 0。入库时加一条断言单日收益率绝对值超过 30% 的记录必须进异常表人工确认。5.2 工程稳定性类连接被掐断与请求频率失控现象采集跑到一半大量超时日志里全是ConnectionResetError第二天重跑还是同样的位置断。原因请求间隔太短短时间内的密集请求触发了服务端的频率保护。很多人第一反应是加大重试次数结果压力更大形成恶性循环。解决把并发降到单线程请求间隔拉到 0.3 秒以上并加随机抖动重试次数控制在三次以内且必须带退避。另外把“每天的全量任务”拆成“按交易日的小批次”单批失败只影响一天。这里最容易犯的错是把稳定性问题当成代码 bug 去调实际上是节奏问题。5.3 工程稳定性类逐条 commit 拖垮写库现象采集十分钟写库一小时stock.db文件锁频繁报database is locked。原因每条记录一次commitSQLite 每次提交都要 fsync几千条就是几千次磁盘同步同时如果有另一个进程在读库写锁和读锁互相等待。解决改成每 200 到 500 条提交一次写库进程和查询进程分离查询走只读连接。数据量再往上走就把 SQLite 换成列式存储但对毕业设计这个量级SQLite 加批量提交完全够用不要过早引入复杂度。5.4 数据口径类交易日历缺失导致日期错位现象按自然日遍历日期周末和节假日也去请求返回全是空分钟线数据跨日边界错乱同一天出现两个开盘。原因没有独立的交易日历表用自然日当天数用分钟线的时间戳带时区本地时间和接口时间混用。解决建一张dim_trade_date表从行情数据里反推交易日有成交记录的日期即为交易日所有遍历都以它为准。时间戳统一转成东八区的日期字符串再入库分钟线额外保留HH:MM字段跨日判断用日期字段而不是时间戳。5.5 分析逻辑类未来函数与幸存者偏差现象回测效果极好实盘或换一段时间就崩选股清单里经常出现已经退市的股票代码或者某只股票在退市前的数据特别“亮眼”。原因一是因子计算里混入了shift(-n)这类未来数据二是股票池只取了当前在市股票历史上退市的样本被系统性剔除收益率被高估。解决所有涉及未来数据的函数单独放一个模块并加命名前缀股票池从dim_stock全量取用is_active字段区分而不是过滤掉回测时按当日是否在市决定能否买入。这一条最隐蔽通常要等到换样本期验证才会暴露。6. 验收与答辩怎么证明这套方案真的跑通了验收这件事评委或甲方看的不是你写了多少行代码而是三件事数据是不是真的、流程是不是闭环、结论是不是可复现。我的建议是准备一张验收清单把每条都做成可现场执行的命令或页面操作而不是 PPT 上的文字描述。验收项检查方式合格标准数据完整性查fact_daily最新交易日股票数不少于维表在市数量的 95%数据正确性随机抽 5 只股票与公开行情比对收盘价误差为 0复权价换算一致采集可恢复手动中断任务后重启从断点继续无重复行因子可复现重跑同一日打分输出结果完全一致回测可解释展示超额收益与样本期能说明因子逻辑与失效场景看板可用打开页面切换日期3 秒内返回结果答辩演示建议按“一条数据的一生”来讲从接口请求一条原始记录开始展示它怎么进库、怎么清洗、怎么算因子、怎么进选股清单、怎么在看板上呈现。这条链路走一遍比讲十个算法名词有说服力。演示前一定要把数据缓存预热好现场跑全量采集是答辩大忌网络一抖就冷场。被问到“这个策略能赚钱吗”时别硬答能。稳妥的回答是把话题拉回工程这套方案交付的是一条可复现的数据管道和一套可验证的因子框架收益结论依赖样本期和参数选择当前样本内超额收益为正但未做严格样本外检验。评委通常更认可这种边界清晰的表述而不是拍胸脯保证。最后给一个具体技巧答辩前把项目里所有硬编码的路径、日期、参数集中到一个配置文件现场改一个参数重跑一遍展示结果随之变化。这个动作能同时证明“流程是活的”和“参数是可调的”比任何架构图都直观。从那以后我每次带人做这类项目都会强制对方先把数据口径写成一张表贴在文档第一页再谈模型和指标——口径不清的项目后面每一步都是返工。希望帮到你。本文还有配套的精品资源点击获取