企业官网建设流程全解析

简介：围绕股票大数据分析与Python爬虫实战整理的学习资料包，面向希望把爬虫、数据处理与量化分析串成完整流程的金融数据初学者和中级开发者。内容覆盖用BeautifulSoup、Scrapy抓取股票行情与公告，借助Pandas清洗缺失值和异常值，使用NumPy、SciPy完成统计计算，并用Matplotlib、Seaborn绘制K线、成交量和波动率图表；同时延伸到Scikit-learn的回归、决策树、随机森林等预测模型，以及Tushare、Wind实时数据接口和Flask、Django结果展示。资源以7z压缩包分发，体积约4.47MB，文件总数显示为0、类型明细暂无数据，需下载后查看实际目录。已有3659人学习下载，适合按“采集—预处理—分析可视化—建模—部署”的路径查漏补缺，也可作为课程设计或小型量化项目的参考素材。

1. 从一份跑不通的股票分析脚本说起：这个项目到底能干什么

很多人第一次碰股票数据，都是从一段“看起来能跑”的 Python 爬虫脚本开始的：循环几十个代码，一页页翻接口，存成 CSV，再画两根均线。等到数据量涨到几年日线、上百只股票，脚本要么被限流封掉，要么内存直接爆掉，最后留下一堆半截文件。这个项目要解决的正是这个尴尬阶段——把“写个爬虫抓点数据”升级成一条能持续跑、能增量更新、能算指标、能出图的完整链路。它适合两类人：刚学完 pandas 想找个真实数据集练手的开发者，以及想把行情采集到分析这一套流程拆开改成自己工具的老手。整体技术栈围绕 Python 展开，采集侧重请求调度与异常重试，分析侧依赖 pandas 和 numpy 做指标计算，可视化用 matplotlib 收尾，数据落地常见做法是 SQLite 或本地 Parquet，项目结构按“采集—清洗—计算—展示”分层，不是一坨脚本堆在根目录。

2. 采集层拆解：一个行情请求从发起到落库的完整链路

2.1 采集模块的目录结构与职责切分

拿到这个项目，先别急着跑入口文件，先把目录看明白。采集相关的东西通常集中在collector/下，我见过的合格写法一般会拆成四个文件：fetcher.py负责单次请求与重试，parser.py负责把响应转成结构化字段，store.py管落库和去重，scheduler.py管批量循环和限速。这种拆法的好处是，接口换了只动 parser，存储换了只动 store，不会牵一发动全身。

stock-analysis/ ├── collector/ │ ├── fetcher.py # 请求封装：超时、重试、UA 轮换 │ ├── parser.py # 响应解析：字段对齐、类型转换 │ ├── store.py # 落库：增量写入、主键去重 │ └── scheduler.py # 批量调度：限速、断点续跑 ├── analysis/ │ ├── indicators.py # 均线、MACD、RSI 等指标计算 │ └── report.py # 汇总统计与导出 ├── config/ │ └── settings.py # 接口地址、并发数、存储路径 └── main.py # 串联入口

这个结构里最容易被忽略的是config/settings.py。很多人把接口地址和并发数硬编码在 fetcher 里，换一台机器跑就得改代码，血泪经验是第一次重构就应该把这些参数抽出来，用环境变量或配置文件覆盖默认值，后面换存储路径、调并发都不用碰业务代码。

2.2 请求封装：超时、重试与请求头轮换

采集层第一个要啃的硬骨头是稳定性。行情接口对高频请求的容忍度很低，裸写requests.get连续打几十次，大概率中途开始返回空数据或者直接拒绝。下面这段是我在这个项目里会采用的 fetcher 写法，核心是三件事：连接和读取分别设超时、对可重试状态做指数退避、每次请求带上一组像样的请求头。

import time import random import requests HEADERS_POOL = [ {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}, {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15"}, ] def fetch_json(url, params=None, retries=3, base_delay=1.0): """带超时与指数退避的 GET 请求，返回解析后的 JSON。""" last_err = None for attempt in range(retries): try: resp = requests.get( url, params=params, headers=random.choice(HEADERS_POOL), timeout=(3.05, 10), # (连接超时, 读取超时) ) if resp.status_code == 200: return resp.json() # 429/5xx 才值得重试，4xx 其余情况直接抛 if resp.status_code == 429 or resp.status_code >= 500: raise RuntimeError(f"retryable status: {resp.status_code}") resp.raise_for_status() except Exception as exc: last_err = exc delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5) time.sleep(delay) raise RuntimeError(f"fetch failed after {retries} tries: {last_err}")

timeout=(3.05, 10)这个元组是 requests 的连接/读取双超时，连接给短一点能快速跳过不可达的地址，读取给长一点是因为行情接口偶尔在整点附近响应变慢。退避公式base_delay * 2**attempt配上 0～0.5 秒的随机抖动，能避免重试请求在同一秒扎堆再次触发限流。这里只对 429 和 5xx 重试，是因为参数错误这类 4xx 重试多少次都没用，早抛早排查。

2.3 解析与增量落库：别让重复数据撑爆表

请求通了只是第一步，解析环节真正麻烦的是字段对齐。同一个接口，不同板块返回的字段可能少几个，直接pd.DataFrame(rows)会得到一堆 NaN 列。稳妥做法是在 parser 里定义一份字段白名单，缺的补 None，多的丢掉，保证每次写入的列结构一致。

import pandas as pd COLUMNS = ["code", "trade_date", "open", "high", "low", "close", "volume", "amount"] def parse_rows(raw_list): """把接口返回的原始列表规整成统一列结构的 DataFrame。""" records = [] for item in raw_list: records.append({ "code": item.get("symbol"), "trade_date": item.get("date"), "open": float(item.get("open") or 0), "high": float(item.get("high") or 0), "low": float(item.get("low") or 0), "close": float(item.get("close") or 0), "volume": float(item.get("volume") or 0), "amount": float(item.get("amount") or 0), }) return pd.DataFrame(records, columns=COLUMNS) def upsert_daily(conn, df): """以 (code, trade_date) 为主键做增量写入，重复数据直接忽略。""" df = df.drop_duplicates(subset=["code", "trade_date"]) df.to_sql("daily_bar", conn, if_exists="append", index=False, method=lambda table, c, keys, data: _insert_ignore(table, c, keys, data))

drop_duplicates只能消除本次批次内的重复，跨批次重复要靠数据库主键约束。SQLite 里用INSERT OR IGNORE，MySQL 用ON DUPLICATE KEY UPDATE，后面二次开发换存储时这层逻辑要跟着改。增量更新的判断依据一般是max(trade_date)，只拉比它更新的日期，避免每次全量拉几年数据。

2.4 批量调度：限速与断点续跑是刚需

批量跑几百只股票时，最怕的是跑到一半崩了，重新从第一只开始。调度器必须在每只股票完成后把进度写到一个日志文件或状态表里，下次启动先读进度，跳过已完成的代码。

import json, os, time def run_batch(codes, state_path="progress.json", sleep_range=(0.5, 1.5)): done = set() if os.path.exists(state_path): done = set(json.load(open(state_path, encoding="utf-8"))) for code in codes: if code in done: continue try: rows = fetch_json(API_URL, params={"symbol": code}) upsert_daily(conn, parse_rows(rows)) done.add(code) json.dump(sorted(done), open(state_path, "w", encoding="utf-8")) time.sleep(random.uniform(*sleep_range)) # 随机间隔，别用固定值 except Exception as exc: print(f"[skip] {code}: {exc}")

sleep_range用随机区间而不是固定 1 秒，是因为固定间隔太有规律，容易被识别成脚本行为。进度文件每完成一只就立即写盘，而不是整批跑完再写，这样即使进程被杀，重启时也能从断点继续。

3. 分析层拆解：从原始行情到指标与图表的核心实现

3.1 数据读取与清洗：复权、缺失与异常值处理

原始行情直接算指标基本都会偏。项目里分析层第一步通常是读库、按股票分组、按日期排序，然后处理两类问题：停牌造成的日期空洞，以及除权除息造成的价格跳变。停牌一般直接丢弃该行，指标计算按实际交易日对齐；复权则要看数据源是否提供复权因子，没有的话至少要用前收盘价做一次简单校验。

def load_and_clean(conn, code): df = pd.read_sql( "SELECT * FROM daily_bar WHERE code = ? ORDER BY trade_date", conn, params=(code,) ) df["trade_date"] = pd.to_datetime(df["trade_date"]) df = df.dropna(subset=["close"]).reset_index(drop=True) # 标记异常：单日涨跌幅超过阈值多半是复权没处理 df["pct_chg"] = df["close"].pct_change() df["suspect"] = df["pct_chg"].abs() > 0.21 return df

阈值 0.21 是按常见涨跌幅限制留的余量，超过这个数的记录不直接删，而是标记出来人工看一眼，因为有些确实是真实的大波动。这种“标记不删除”的思路在金融数据清洗里很实用，删掉容易丢信息，留着能做二次判断。

3.2 均线与 MACD 的向量化实现

指标计算不要写 for 循环逐行算，pandas 的 rolling 和 ewm 足够覆盖大多数常见指标。下面这段把均线、MACD 和 RSI 放在一起算，是这个项目分析层最核心的计算块。

def add_indicators(df, ma_windows=(5, 20, 60)): """批量计算均线、MACD、RSI，全部向量化，不写循环。""" for w in ma_windows: df[f"ma{w}"] = df["close"].rolling(window=w, min_periods=w).mean() # MACD: 快线12日、慢线26日、信号9日 ema_fast = df["close"].ewm(span=12, adjust=False).mean() ema_slow = df["close"].ewm(span=26, adjust=False).mean() df["dif"] = ema_fast - ema_slow df["dea"] = df["dif"].ewm(span=9, adjust=False).mean() df["macd"] = 2 * (df["dif"] - df["dea"]) # RSI: 14日，用 Wilder 平滑近似 delta = df["close"].diff() gain = delta.clip(lower=0).ewm(alpha=1/14, adjust=False).mean() loss = (-delta.clip(upper=0)).ewm(alpha=1/14, adjust=False).mean() rs = gain / loss.replace(0, 1e-9) df["rsi"] = 100 - 100 / (1 + rs) return df

min_periods=w保证前几行不会有半截窗口算出的假均线。MACD 里adjust=False表示不用复权后的加权方式，和我们常见的行情软件一致；df["macd"]乘 2 是国内软件柱状图的习惯画法。RSI 里把分母 0 替换成极小值，是为了避免连续下跌时出现除零，这个小改动能省掉一整类报错。

3.3 可视化输出：一张图里放下价格、均线和成交量

可视化不是把线画出来就完事，这个项目里比较实用的布局是上下两栏：上栏画收盘价和均线、MACD，下栏画成交量柱，共用 x 轴日期。用matplotlib.gridspec控制高度比例，比默认的等分布局顺眼很多。

import matplotlib.pyplot as plt from matplotlib import gridspec def plot_stock(df, code, save_path="chart.png"): fig = plt.figure(figsize=(14, 8)) gs = gridspec.GridSpec(3, 1, height_ratios=[3, 1, 1], hspace=0.05) ax1 = fig.add_subplot(gs[0]) ax1.plot(df["trade_date"], df["close"], label="close", linewidth=1) for w in (5, 20, 60): ax1.plot(df["trade_date"], df[f"ma{w}"], label=f"MA{w}", linewidth=0.8) ax1.legend(loc="upper left") ax1.set_title(code) ax2 = fig.add_subplot(gs[1], sharex=ax1) ax2.bar(df["trade_date"], df["volume"], width=1.0) ax2.set_ylabel("volume") ax3 = fig.add_subplot(gs[2], sharex=ax1) ax3.plot(df["trade_date"], df["macd"], linewidth=0.8) ax3.axhline(0, color="gray", linewidth=0.5) plt.setp(ax1.get_xticklabels(), visible=False) plt.savefig(save_path, dpi=140, bbox_inches="tight") plt.close(fig)

sharex=ax1让三栏横轴联动，setp(... visible=False)把上两栏的日期标签藏掉，避免重叠。width=1.0的成交量柱在日线数据上刚好铺满间隔，太宽会互相盖住。保存时用bbox_inches="tight"能去掉多余白边，后面批量出图时文件体积也小一些。

4. 避坑与排查：这套流程里最容易翻车的几处

4.1 现象：跑到几十只股票后开始连续返回空数据

原因多半是请求频率触发了服务端限流，而代码里的重试又在同一时间段内集中发起，把限流窗口越拖越长。解决办法是把退避时间改成指数增长，并在每只股票之间加随机间隔，同时把返回空的响应记进日志，第二天错峰补跑。

4.2 现象：指标图和行情软件对不上，均线整体偏移

检查日期排序和复权处理。常见原因是读库时没ORDER BY trade_date，SQLite 返回顺序不保证按写入顺序，导致 rolling 窗口错位。另一个原因是数据源本身返回的是不复权价格，遇到除权日附近指标就会跳。修复方式是在 SQL 里强制排序，并在清洗阶段标记出可疑跳变。

4.3 现象：增量更新后表里出现重复行，越跑越大

检查落库逻辑是否依赖了if_exists="replace"。这个参数会把整张表干掉重写，看起来“不重复”，但一旦中途失败，旧数据也没了。正确做法是建好唯一索引(code, trade_date)，用INSERT OR IGNORE写入，重复的直接跳过，增量更新只拉max(trade_date)之后的数据。

4.4 现象：批量出图时内存持续上涨直到进程被杀

plt.figure()创建后没有plt.close(fig)，图形对象一直挂在全局状态里。批量循环里每张图都savefig之后立刻close，并且不要在循环内反复调用plt.show()。另外pd.read_sql每只股票都重新建连接也会拖慢，连接对象应该在外面建好传进去复用。

4.5 现象：同一段代码换台机器就跑不通

十有八九是配置文件里的绝对路径和代理设置没跟着走。项目里所有路径都应该基于pathlib.Path(__file__).parent推算，存储目录、进度文件、图片输出目录统一从 settings 读。换机器时只需要改一个配置文件的根目录，其他都不动。

5. 二次开发：把这套流程改成自己顺手的工具

这套项目骨架跑通之后，改造成本其实很低，关键是找对扩展点。第一个值得动的是采集层的接口适配，如果换成另一个数据源，只需要在parser.py里改字段映射，fetcher 和调度逻辑不动。第二个是分析层，指标计算在indicators.py里是纯函数，输入 DataFrame 输出 DataFrame，加一个新指标就是加一个函数，然后在 report 里调用。第三个是输出层，现在出的是静态 PNG，想改成网页看板可以把 plot 函数换成 plotly，DataFrame 直接喂进去，改动集中在可视化模块。

验证改造是否成功，我一般用一个固定的老代码跑三遍：第一遍全量拉取，第二遍增量更新，第三遍只跑分析不拉数据，对比三次输出的指标值和行数是否一致。全量拉取验证采集链路，增量更新验证去重逻辑，纯分析跑验证计算可复现。三遍行数对不上就去查存储，指标值对不上就去查排序或复权。

提醒一点，改动采集频率之前先估算一下目标数据源的单日请求量，别一上来就把间隔调到 0.1 秒，跑几分钟就被关小黑屋。我第一次做这类项目就是图快，把随机间隔去掉直接全速跑，结果第二天换了个环境重跑才对上节奏。从那以后我每次调采集参数都先小批量试 20 只，观察返回状态码分布，再决定放不放开。希望帮到你。

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