Python股票数据分析实战：爬虫清洗与回测全流程-北京尧图网络

简介围绕股票大数据分析的Python实战资源适合金融数据分析初学者及有Python基础、希望系统掌握爬虫采集与量化分析流程的读者。资源为7z压缩包大小4.47MB站内暂未列出文件明细结合主题内容可预期包含Python爬虫脚本、Pandas/NumPy数据处理代码、Matplotlib/Seaborn可视化与Scikit-learn机器学习建模示例等。已有3654人学习下载。通过阅读和实践可以掌握从编写高效爬虫、处理反爬策略到清洗缺失值与异常值、统一数据格式从绘制K线图、分析成交量、用箱型图观察波动性到尝试线性回归、决策树、随机森林或神经网络模型预测价格走势最终搭建实时数据获取与分析展示工具。资源将分散知识串联为完整实战链路适合配套博文逐步上手也为后续量化策略研究打下基础。1. 股票大数据分析实战别让K线图骗了你最值钱的是洗数据那部分如果只靠行情软件看K线很多人做股票数据分析的第一步就输在数据上想拉五年日线手动导出Excel得点几十次格式还不统一复权口径五花八门。这套基于股票大数据分析的Python实战项目把整条链路串了起来——用Python爬虫自动抓历史行情pandas清洗入库再把均线、RSI、布林带和回测一次性算完。它不承诺预测涨跌只解决一个实际问题让研究者在十分钟内拿到一份可复现、可追溯的股票数据分析报告。适合两类人想练Python爬虫加数据处理的新手以及不想被商用软件绑定的进阶研究者。别被K线图迷惑——这个项目里最花时间的其实是处理复权因子和缺失数据的脏活这两件事做不好后面画的线都是自欺欺人。2. 技术选型与工程结构requests pandas SQLite为什么不杀鸡用牛刀2.1 技术栈选型每一层都选最不折腾的这个项目的定位是「单机、可复现、可扩展的股票数据研究工程」不是高并发的量化交易系统。所以选型原则很直接哪个方案维护成本低、出了问题我能两小时定位就选哪个。环节本项目选择常见替代方案为什么这样选数据采集requests 重试机制scrapy / 框架爬虫单机按天调度请求量不大轻量脚本更容易调试数据处理pandas numpy纯 Python 手算滚动窗口、向量化计算是刚需纯手写循环太慢数据存储SQLiteMySQL / 直接存CSV单机零配置SQL查询比pandas内存操作更可靠可视化matplotlibpyecharts / plotly生成静态PNG便于存档和写入报告任务调度手动执行 / 系统定时任务Airflow / APScheduler项目阶段没必要上调度框架cron一条命令解决为什么不选scrapy这个场景是「周期性批量抓取」不是「全网爬取」。scrapy的中间件、管道、爬虫框架在这里反而成了负担requests配合重试和限速足够直观报错时定位栈也短。为什么不选MySQL数据量撑死在几百万行以内SQLite单文件随项目走换台机器拷过去就能继续分析省掉一套数据库运维。为什么不直接用现成的财经数据包那些包的数据口径和复权逻辑是个黑匣子自己抓一遍才能完全掌控每个字段的含义出了问题也知道去哪看。2.2 目录结构一个请求从进入到报告的完整路径拿到源码先看目录项目模块划分得很清楚每个目录职责单一stock_analysis/ ├── config/ │ └── config.py # 股票代码、周期、时间范围、费率等参数 ├── core/ │ ├── fetcher.py # 行情抓取请求封装、重试、断点续爬 │ ├── cleaner.py # 清洗与复权排序、去重、因子调整 │ ├── indicators.py # 技术指标计算MA / RSI / BOLL │ └── backtest.py # 回测与绩效统计信号、持仓、手续费 ├── data/ │ ├── raw/ # 抓下来的原始CSV按股票代码命名 │ └── stock.db # 清洗后写入的SQLite数据库 ├── output/ │ ├── charts/ # 生成的K线和净值曲线PNG │ └── report.csv # 多股票回测结果汇总 └── main.py # 编排入口抓取 → 清洗 → 分析 → 报告数据流是一条直线main.py 读 config → fetcher 抓原始 CSV → cleaner 清洗并写入 SQLite → indicators 计算指标 → backtest 输出绩效和图表。这种线性结构最大的好处是每一层都可以单独跑、单独验证抓下来的数据对不对看 raw 目录里的 CSV清洗对不对查 SQLite 里的数策略结果对不对看 output 里的净值曲线。这样出了问题不用从头排查直接定位到某一层。2.3 数据库设计一张表装下日线序列SQLite 里只建了一张核心表字段设计得很克制没有冗余CREATE TABLE IF NOT EXISTS daily_kline ( stock_code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL, volume INTEGER NOT NULL, adj_factor REAL NOT NULL DEFAULT 1.0, PRIMARY KEY (stock_code, trade_date) );联合主键 (stock_code, trade_date) 天然防重复同一只股票同一天的记录插入第二次会直接报错清洗层不需要额外写去重逻辑。价格用 REAL 是因为浮点精度足够支撑回测计算成交量用 INTEGER 是因为它是整数计数存成浮点会在聚合时引入无意义的精度误差。adj_factor 字段是留给复权用的抓取时先存原始价清洗阶段再根据除权除息数据回填因子这样原始数据和派生数据分离想换复权口径不用重新抓数据。配套的 config.py 把所有可调参数集中在最上层STOCK_CODES [600001, 600002] # 示例代码替换成你关注的标的 PERIOD daily START_DATE 20150101 END_DATE 20241231 FEES { fee_rate: 0.0003, # 佣金加印花税近似按万三算 slippage: 0.001 # 滑点按千一算 }这里的设计意图是「改参数不动逻辑」。回测改费率、换股票、调整时间范围都只动 config.pycore 目录下的代码一行不用改。我见过不少项目把魔术数字散落在函数里改一次费率要全文搜替换这个项目从一开始就避开了那个坑。3. 数据采集模块实战从行情接口到CSV落地的完整链路3.1 请求封装把网络请求变成可重试的函数爬虫最容易翻车的不是解析逻辑而是网络层——超时、被拒、返回半截数据任何一样都能让整个任务中断。fetcher.py 第一层先把 requests 包成一个带重试的取值函数import requests import pandas as pd import time HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Accept: application/json, } def fetch_kline(stock_code, perioddaily, start_date20150101, end_date20241231, retries3): # 示例地址实际接入时替换成可用的公开行情接口 url https://api.example.com/kline params { code: stock_code, period: period, start: start_date, end: end_date, adjust: none, # 先抓原始价复权在清洗阶段做 } for attempt in range(1, retries 1): try: resp requests.get(url, paramsparams, headersHEADERS, timeout10) resp.raise_for_status() # 4xx/5xx 直接抛异常 data resp.json() df pd.DataFrame(data[data][items]) df.columns [date, open, high, low, close, volume] df[date] pd.to_datetime(df[date]) return df.sort_values(date).reset_index(dropTrue) except Exception as exc: # 指数退避第1次等2秒第2次等4秒第3次等8秒 print(f[{stock_code}] 第 {attempt} 次请求失败: {exc}) time.sleep(2 ** attempt) return pd.DataFrame() # 三次全失败返回空表由上层决定是否续跑参数说明period 支持 daily / weekly / monthly对应日线、周线、月线start_date 和 end_date 是字符串格式的闭区间retries 控制失败重试次数默认 3 次。timeout10 必须显式设置否则 requests 默认没有超时一个坏连接能把任务挂到天荒地老。resp.raise_for_status() 的作用是把 HTTP 400/500 状态码转成异常而不是让程序带着错误数据继续跑。返回前先 sort_values 保证 DataFrame 按时间升序后续所有计算都依赖这个顺序。注意实际接入行情源时先打印一次原始返回结构确认 items 字段的列顺序再写 df.columns 映射。不同数据源返回的字段名经常不一样这一步最容易踩坑。3.2 断点续爬失败从上次中断处接着跑抓几十只股票五年日线不是一次能跑完的中间断网、报错、电脑休眠都会中断。全量重抓既浪费时间又可能触发接口限流。fetcher.py 第二层用「本地文件断点」解决这个问题每次抓取前先读已有 CSV 的最后日期只补增量部分。import os import random def fetch_with_resume(stock_codes, raw_dirdata/raw, perioddaily): os.makedirs(raw_dir, exist_okTrue) for code in stock_codes: raw_file os.path.join(raw_dir, f{code}_{period}.csv) last_date None if os.path.exists(raw_file): existing pd.read_csv(raw_file, dtype{date: str}) if len(existing): last_date existing[date].max() # 从上次断点继续抓第一次跑则从默认起点开始 df fetch_kline(code, periodperiod, start_datelast_date or 20150101) if df.empty: continue if last_date is not None: df df[df[date] last_date] df.to_csv(raw_file, modea, headerFalse, indexFalse) else: df.to_csv(raw_file, indexFalse) # 随机延时避免固定间隔被识别 time.sleep(random.uniform(0.5, 1.5))这里的核心细节是追加模式 modea 加 headerFalse只把增量行写到 CSV 尾部不重复写表头。读取时用 dtype{date: str} 强制日期按字符串读避免 pandas 自动转类型导致比较出错。随机延时 0.5 到 1.5 秒目的是把请求节奏打散——固定间隔在服务端日志里非常显眼随机延时更接近人工操作的习惯。如果某次抓取中途崩溃重新跑脚本时会自动从上次最后日期续起已抓的数据不会浪费。3.3 采集纪律限速、异常捕获和日志可观测采集模块里有一个很容易被忽略但值得单独说的设计所有请求都打印带时间戳的日志而不是静默失败。print(f[{time.strftime(%H:%M:%S)}] 抓取 {stock_code} f增量数据本地已到 {last_date})跑批任务挂在后台时人不可能盯着屏幕。真正出问题的时候靠的是日志判断卡在哪只股票、停了多久、错误是什么类型。抓取纪律总结成三条第一请求频率控制在每秒一次以内公开接口普遍有频控超频的结果是临时封禁反而更慢第二重试上限设为 3 次且指数退避不要无限重试把服务器打成筛子第三HTTP 错误时先打印状态码和响应体再抛异常很多反爬提示藏在响应体里只打印状态码看不出原因。这套纪律在合规的前提下把「能跑」变成「能稳定跑完」。4. 数据清洗与信号分析复权、技术指标和回测的核心实现4.1 复权处理为什么直接拿原始价格算收益率会翻车股票分红送股后价格会向下跳空但账户资产其实没变。如果直接用原始价格计算收益率除权日那天会被误判成暴跌均线也会出现假死叉。所以清洗阶段必须做复权处理常见做法是拿复权因子调整所有价格字段def adjust_price(df, methodforward): # df 必须包含 adj_factor 字段取数时先按 trade_date 排序 df df.sort_values(trade_date) last_factor df[adj_factor].iloc[-1] first_factor df[adj_factor].iloc[0] if method forward: # 前复权以最新价格为基准历史价格向当前看齐 df[adj_close] df[close] * df[adj_factor] / last_factor else: # 后复权以上市首日为基准更利于计算真实收益率 df[adj_close] df[close] * df[adj_factor] / first_factor # 开高低价按复权比例同步调整保证K线形态不变 ratio df[adj_close] / df[close] df[adj_open] df[open] * ratio df[adj_high] df[high] * ratio df[adj_low] df[low] * ratio return df说明三个容易混淆的点。第一前复权和后复权的区别前复权让最新价格保持真实价适合看当下形态后复权让首日价格为真实价适合计算长期累计收益率。回测里我更推荐后复权因为前复权在长期大比例送转的股票上可能出现历史价格为负的怪象后复权不会。第二为什么 high / low / open 也要跟着调只调 close 会导致K线的影线失真信号计算也会偏差。第三ratio 是用 adj_close 除以原始 close 得到的再用这个比例去乘 open、high、low保证复权后的K线形态和原始K线完全一致。4.2 技术指标计算MA、RSI、布林带一次性生成indicators.py 的核心思路是「输入一个干净 DataFrame输出带全部指标列的扩展 DataFrame」纯函数式设计方便在回测里反复调用def add_indicators(df, ma_windows(5, 10, 20, 60), rsi_period14): df df.sort_values(trade_date).reset_index(dropTrue) # 移动平均线rolling 窗口内 adj_close 的均值 for win in ma_windows: df[fma{win}] df[adj_close].rolling( win, min_periodswin).mean() # RSI用简单平均法算涨跌幅均值比 delta df[adj_close].diff() gain delta.clip(lower0).rolling(rsi_period).mean() loss (-delta.clip(upper0)).rolling(rsi_period).mean() rs gain / loss df[rsi] 100 - 100 / (1 rs) # 布林带20日均线加减2倍标准差 df[boll_mid] df[adj_close].rolling(20).mean() df[boll_std] df[adj_close].rolling(20).std() df[boll_up] df[boll_mid] 2 * df[boll_std] df[boll_low] df[boll_mid] - 2 * df[boll_std] return df关键参数说明ma_windows 里的 min_periodswin 表示窗口内数据不足时不输出值避免前期数据不足时出现虚假均线。delta diff() 是相邻交易日收盘价之差clip(lower0) 把负涨幅截断为 0loss 那一行先用 clip(upper0) 把正值截断为 0 再取负这样 gain 和 loss 都是正数序列。RSI 分母出现 0全程上涨时会得到 inf实际项目中我会在返回前加一层 fillna(50)把极值温和地拉回中性区。布林带用 20 日和 2 倍标准差属于行业默认值改这两个参数会直接改变交易信号灵敏度。4.3 回测逻辑信号次日生效手续费一分不能少回测模块是整套代码里最容易写出自欺欺人结果的地方。最常见的问题是未来函数——当天收盘产生信号却假设当天收盘就能成交实盘根本做不到。这个项目的处理方式是信号用 shift(1) 延后一天def run_backtest(df, fee_rate0.0003, slippage0.001): df df.copy() # 信号ma5 上穿 ma20 持仓下穿空仓 df[signal] 0 df.loc[df[ma5] df[ma20], signal] 1 df.loc[df[ma5] df[ma20], signal] 0 # 关键position 取 signal 的昨日值当天信号次日才生效 df[position] df[signal].shift(1).fillna(0) # 策略收益持仓收益 - 手续费 - 滑点 df[ret] df[adj_close].pct_change().fillna(0) df[trade_ret] ( df[position] * df[ret] - df[position].diff().abs().fillna(0) * (fee_rate slippage) ) df[nav] (1 df[trade_ret]).cumprod() # 绩效指标 max_drawdown (df[nav] / df[nav].cummax() - 1).min() annual_return df[nav].iloc[-1] ** (252 / len(df)) - 1 sharpe df[trade_ret].mean() / df[trade_ret].std() * (252 ** 0.5) return df, { annual_return: annual_return, max_drawdown: max_drawdown, sharpe: sharpe, }这一段的工程细节值得逐行说。position 是 signal 的 shift(1)意味着今天看到的持仓状态来自昨天的信号从机制上杜绝了未来函数。手续费的计算方式是取 position 的 diff 绝对值再乘费率——position 从 0 变 1 或从 1 变 0 都算一次换仓每次换仓扣一次费用这比笼统地在每日收益上减固定值准确得多。滑点参数代表成交价与信号价的偏差用千分之一是保守估计。年化收益按 252 个交易日换算夏普比率是年化超额收益除以年化波动标准口径。跑完这个函数会得到一个包含净值曲线的 DataFrame 和一份绩效统计字典两部分分别用于画图和汇总报告。5. 踩坑排查数据、复权与回测里最常翻车的四件事5.1 日期乱序导致K线时间轴横跳现象画出的K线图时间轴前后颠倒或者均线在图中上下乱窜整体形态完全不对。原因行情接口返回的日期是字符串格式且顺序不固定有的源倒序返回有的源把日期格式混成两种pandas 读进来后没有统一转换排序就参与计算。解决所有数据进清洗层之后第一件事就是 df[trade_date] pd.to_datetime(df[trade_date])然后 sort_values(trade_date).reset_index(dropTrue)。把这两行写在所有指标计算之前后续任何涉及时间窗口的操作都以它为前提。排查时先在清洗输出里打印前五行和后五行日期升序且连续再继续往下走。5.2 前复权出现负价格指标变成一串 inf现象某只常年大比例送转的股票回测里计算出的历史价格出现负数收益率直接算成 NaN 或者 inf。原因前复权以最新价为基准历史除权次数太多后累计调整因子过大早年价格被压到负值。这不是代码逻辑错误而是口径问题。解决回测计算统一改用后复权价格后复权以首日价为基准永远不会出现负价格且计算真实收益率更准确。展示K线图时如果想贴近真实交易价格再用前复权两套口径分场景使用。我会在配置里加一个 adjust_method 参数默认 backward需要看图时切到 forward。5.3 停牌日缺失导致均线断崖现象均线不是平滑曲线而是突然出现跳变或者某几天的 MA 是 NaN策略信号在停牌后第一天异常触发。原因股票停牌日没有成交记录rolling 窗口按行数滑动而不是按自然日滑动停牌前后的窗口覆盖时间跨度不一致导致均线失真。解决先用交易日历补全缺失日期再对价格字段做前向填充最后计算指标# 用交易日历 reindex缺失日期补 NaN df df.set_index(trade_date).reindex(trade_calendar) df[adj_close] df[adj_close].ffill() # 停牌日沿用前收盘价 df df.reset_index()止损止盈相关的信号判断也必须放在补全之后否则停牌复牌那天的信号会基于错误窗口触发。ffill 前向填充是处理停牌的标准做法它隐含的假设是停牌期间价格不变这在回测里是合理的保守估计。5.4 回测收益虚高实盘完全对不上账现象策略回测年化收益翻倍模拟盘跑起来却稳定亏损。原因十有八九是没算手续费和滑点或者用了未来函数——当日收盘产生信号当日成交实盘里信号产生时当日已经收盘根本成交不了。解决先检查所有 signal 是否做了 shift(1)确认信号必然次日生效再确认 trade_ret 里扣了 (fee_rate slippage) 乘以换手次数。验证方法很简单把手续费率调成 0.01 跑一次如果收益曲线几乎不变说明手续费计算代码根本没执行到回去查 position.diff() 是不是全为 0。6. 二次开发把单股票分析扩展成多股票批量回测拿到这套代码后最顺手的改造方向是把单股票分析扩成多股票批量扫描。核心思路是保持 core 层纯函数不变在 main.py 里加一个循环把结果合并成一张汇总表。def batch_backtest(stock_codes, fast5, slow20): results [] for code in stock_codes: raw load_raw(fdata/raw/{code}_daily.csv) # 已有函数 df add_indicators(raw) # 生成 MA df, stats run_backtest(df) # 跑回测 stats[code] code results.append(stats) report pd.DataFrame(results).set_index(code) report.to_csv(output/report.csv) return report改造时先只跑三只风格不同的股票验证输出格式再扩大到全列表。我一般会在报告里加两列——收益排名和最大回撤排名用来快速筛选「收益尚可且回撤可控」的标的纯看收益排序容易被高风险标的带偏。另一条常用的改动路径是把 MA 窗口改成可扫描参数外面套两层循环找出不同标的各自表现最好的参数组合这就是最朴素的参数寻优。这套工程留出的扩展空间在于fetcher 只管拿数据cleaner 只管洗干净backtest 只管算账每一层都可以单独替换。想接更多数据源就重写 fetcher想加新策略就新增一个信号函数想上机器学习就先从 SQLite 里导出特征表。从那以后我每次改回测逻辑都强制先跑一遍参数扫描、确认手续费和滑点参数没被注释掉再谈曲线好不好看。希望帮到你。本文还有配套的精品资源点击获取

Python股票数据分析实战：爬虫清洗与回测全流程

相关推荐

Spark 3.2.0预编译包Hadoop版本兼容性深度解析

kmodes实战：用Python搞定分类数据聚类

SQLMap绕过WAF实战指南：从原理到参数调优

更多资讯

GET与POST发送普通参数全解析：编码、Content-Type与联调避坑

AI写学术专著靠谱吗？从工具横评到高效流程的完整攻略

Python65：核心语法-函数进阶-默认参数

Excel/WPS 里像写公式一样查 A 股行情：free-stockdb STOCK_DATA 宏快速上手指南

DeepSeek-V3-0324 本地加载实战：从 685B 权重到推理跑通

WebSocket群聊架构实战：握手鉴权、消息广播与断线重连

今日更新

智能硬件视频会议集成：从芯片选型到场景落地的全栈实践

FFmpeg音视频处理与5路RTMP推流实战：从原理到118克盒子部署

RK3528+1GB内存如何变身直播节点？硬编解码与六协议实战复盘

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？