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 技术栈选型：每一层都选最不折腾的

这个项目的定位是「单机、可复现、可扩展的股票数据研究工程」，不是高并发的量化交易系统。所以选型原则很直接：哪个方案维护成本低、出了问题我能两小时定位，就选哪个。

为什么不选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.py，core 目录下的代码一行不用改。我见过不少项目把魔术数字散落在函数里，改一次费率要全文搜替换，这个项目从一开始就避开了那个坑。

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, period="daily", start_date="20150101", end_date="20241231", retries=3): # 示例地址，实际接入时替换成可用的公开行情接口 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, params=params, headers=HEADERS, timeout=10) 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(drop=True) 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 次。timeout=10 必须显式设置，否则 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_dir="data/raw", period="daily"): os.makedirs(raw_dir, exist_ok=True) 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, period=period, start_date=last_date or "20150101") if df.empty: continue if last_date is not None: df = df[df["date"] > last_date] df.to_csv(raw_file, mode="a", header=False, index=False) else: df.to_csv(raw_file, index=False) # 随机延时，避免固定间隔被识别 time.sleep(random.uniform(0.5, 1.5))

这里的核心细节是追加模式 mode="a" 加 header=False：只把增量行写到 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, method="forward"): # 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_period=14): df = df.sort_values("trade_date").reset_index(drop=True) # 移动平均线：rolling 窗口内 adj_close 的均值 for win in ma_windows: df[f"ma{win}"] = df["adj_close"].rolling( win, min_periods=win).mean() # RSI：用简单平均法算涨跌幅均值比 delta = df["adj_close"].diff() gain = delta.clip(lower=0).rolling(rsi_period).mean() loss = (-delta.clip(upper=0)).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_periods=win 表示窗口内数据不足时不输出值，避免前期数据不足时出现虚假均线。delta = diff() 是相邻交易日收盘价之差，clip(lower=0) 把负涨幅截断为 0，loss 那一行先用 clip(upper=0) 把正值截断为 0 再取负，这样 gain 和 loss 都是正数序列。RSI 分母出现 0（全程上涨）时会得到 inf，实际项目中我会在返回前加一层 fillna(50)，把极值温和地拉回中性区。布林带用 20 日和 2 倍标准差属于行业默认值，改这两个参数会直接改变交易信号灵敏度。

4.3 回测逻辑：信号次日生效，手续费一分不能少

回测模块是整套代码里最容易写出自欺欺人结果的地方。最常见的问题是未来函数——当天收盘产生信号，却假设当天收盘就能成交，实盘根本做不到。这个项目的处理方式是信号用 shift(1) 延后一天：

def run_backtest(df, fee_rate=0.0003, slippage=0.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(drop=True)。把这两行写在所有指标计算之前，后续任何涉及时间窗口的操作都以它为前提。排查时先在清洗输出里打印前五行和后五行，日期升序且连续，再继续往下走。

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, fast=5, slow=20): results = [] for code in stock_codes: raw = load_raw(f"data/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 里导出特征表。从那以后我每次改回测逻辑，都强制先跑一遍参数扫描、确认手续费和滑点参数没被注释掉，再谈曲线好不好看。希望帮到你。

本文还有配套的精品资源，点击获取