简介这是一套面向高校计算机与金融工程专业学生的Python股票数据分析全流程实践项目适用于课程大作业、毕业设计或量化入门学习。资源完整实现从网页爬取实时/历史股票数据、清洗分析到多维度可视化展示的端到端功能代码经实测可直接运行含详细注释与模块化结构便于理解与二次开发。压缩包共362个文件主体为38个核心Python脚本涵盖requests爬虫、pandas数据处理、matplotlib/seaborn/plotly可视化、59个HTML交互式报告页、39张PNG图表截图及148个GIF动图演示关键操作流程辅以JS/CSS前端渲染支持整体包体仅9.16MB轻量易部署。目前已有1805人学习下载项目结构清晰包含配置文件ini/json、SQL建表语句、Shell部署脚本及LayUI等前端框架样式资源兼顾后端逻辑与前端展示是少有的集爬虫、分析、可视化于一体的高分95教学级实战源码。1. 这不是又一个“爬股票然后画图”的玩具项目它是一套可直接嵌入实盘监控流程的轻量级数据管道如果你在期末大作业里交了一份用akshare爬几只股票、pandas算个涨跌幅、matplotlib画三条线的脚本老师给85分——那这份95分以上的源码包大概率会让你重新理解“框架”二字的分量。它不依赖 Jupyter Notebook 的交互式幻觉也不靠手动改代码来切换股票池它把数据获取、清洗校验、指标计算、存储归档、图表渲染全部封装成可配置、可调度、可复用的模块。核心逻辑跑在纯 Python 环境下无 Web 服务依赖命令行一键触发全链路输出结果自动存入 SQLite 并生成带时间戳的 HTML 报告。适合金融工程课设、量化兴趣小组快速验证策略逻辑、甚至作为实习中搭建本地监控看板的起点。对新手它暴露了真实金融数据流中的脏数据陷阱如停牌日缺失、复权因子跳变、交易所休市标识混乱对有经验者它的IndicatorEngine类设计和DataRouter配置机制提供了比backtrader更透明、比zipline更轻量的二次开发入口。2. 用 requests akshare 构建健壮的股票数据爬虫层绕过反爬、处理断点、统一字段语义2.1 为什么不用 tushare 或 baostock选型依据与协议层控制权tushare免费版接口调用频次受限且部分字段需付费解锁baostock依赖独立客户端进程部署隔离性差而akshare基于 HTTP 协议直连交易所/财经门户原始页面源码完全开源字段命名贴近原始披露口径如trade_date对应上交所日志字段且支持retry_times3、timeout15等底层参数透传。更重要的是该框架将akshare封装为StockDataFetcher类所有请求均通过自定义Session实例发出并注入User-Agent和Referer头模拟浏览器行为# data/fetcher.py import akshare as ak import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class StockDataFetcher: def __init__(self): self.session requests.Session() retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) self.session.mount(http://, adapter) self.session.mount(https://, adapter) self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://www.sse.com.cn/ }) def fetch_daily(self, symbol: str, start_date: str, end_date: str) - pd.DataFrame: try: df ak.stock_zh_a_hist(symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq) # 前复权确保价格连续 except Exception as e: logger.error(fFetch failed for {symbol}: {e}) return pd.DataFrame() return self._normalize_columns(df) def _normalize_columns(self, df: pd.DataFrame) - pd.DataFrame: # 统一字段名akshare 输出列名不一致如 日期/交易日期/date rename_map { 日期: trade_date, 交易日期: trade_date, date: trade_date, 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume, 成交额: amount } df df.rename(columnsrename_map) df[trade_date] pd.to_datetime(df[trade_date]).dt.strftime(%Y-%m-%d) return df.sort_values(trade_date).reset_index(dropTrue)提示ak.stock_zh_a_hist的adjustqfq参数必须显式指定否则默认返回未复权数据导致技术指标如 MA、MACD在除权日出现断崖式跳空。框架在fetch_daily方法末尾强制调用_normalize_columns解决不同股票代码如 000001.SZ 与 600000.SH返回列名不一致问题——这是学生项目中最常被忽略的“数据语义漂移”。2.2 断点续爬与增量更新避免每次全量拉取浪费带宽与时间框架引入SQLite表stock_metadata记录每只股票最新已入库交易日。DataPipeline.run()执行时先查表获取max(trade_date)再向fetch_daily传入start_date max_date pd.Timedelta(days1)。若某次网络中断导致000001.SZ只拉到 2024-03-15下次运行不会重拉 2024-01-01 至 2024-03-15而是从 2024-03-16 开始# pipeline.py def _get_latest_date(self, symbol: str) - Optional[str]: with sqlite3.connect(self.db_path) as conn: cursor conn.cursor() cursor.execute( SELECT MAX(trade_date) FROM stock_data WHERE symbol ?, (symbol,) ) result cursor.fetchone()[0] return result if result else 2020-01-01 def run(self, symbols: List[str]): for symbol in symbols: latest_date self._get_latest_date(symbol) today datetime.now().strftime(%Y-%m-%d) if latest_date today: logger.info(f{symbol} already up to date) continue df self.fetcher.fetch_daily(symbol, latest_date, today) if not df.empty: self._save_to_db(df, symbol)注意_get_latest_date返回2020-01-01作为兜底值而非None避免pd.date_range在start_dateNone时抛出异常。该设计使首次运行自动全量抓取后续运行严格增量——这是生产级数据管道的底线要求远超课程作业常见的“每次删库重跑”。2.3 处理三大典型脏数据停牌、一字板、交易所休市日原始数据中akshare对停牌日通常返回空 DataFrame 或填充NaN但volume0与closeopenhighlow的一字板需区分对待。框架在DataCleaner中定义三类校验规则触发条件处理动作业务含义volume 0且close open high low保留该行标记status limit_up或limit_down一字涨停/跌停属有效交易状态volume 0且close ! open删除该行记录警告WARN: price_change_without_volume数据异常可能为临时停牌或接口错误trade_date不在akshare.get_trade_days()返回列表中删除该行记录WARN: non_trading_day_in_data交易所休市日误入数据# data/cleaner.py def clean_daily_data(self, df: pd.DataFrame) - pd.DataFrame: trade_days ak.get_trade_days(start_date20200101, end_date20300101) trade_days [d.strftime(%Y-%m-%d) for d in trade_days] df df[df[trade_date].isin(trade_days)].copy() df[status] normal # 标记一字板 is_limit (df[volume] 0) (df[close] df[open]) (df[close] df[high]) (df[close] df[low]) df.loc[is_limit (df[close] df[close].shift(1)), status] limit_up df.loc[is_limit (df[close] df[close].shift(1)), status] limit_down # 删除无效零成交量 invalid_zero_vol (df[volume] 0) (df[close] ! df[open]) df df[~invalid_zero_vol].reset_index(dropTrue) return df这套规则让清洗后的数据可直接用于talib计算 MACD避免因停牌日NaN导致整列指标失效——这是95分项目与85分项目的分水岭。3. 基于 pandas-ta 的指标计算引擎支持自定义公式、批量回测、结果缓存3.1 为什么弃用 talibpandas-ta 的可调试性与国产适配优势ta-lib编译依赖复杂在 Windows 上常因Microsoft Visual C 14.0缺失报错其 C 扩展函数无法单步调试当MACD计算结果与通达信不一致时学生难以定位是参数fastperiod12设置错误还是signalperiod9与源平台默认值不同。pandas-ta完全基于 NumPy/Pandas 实现所有指标函数均为 Python 源码可直接import pandas_ta as ta; print(ta.macd.__code__)查看逻辑。框架将其封装为IndicatorEngine支持两种调用模式声明式配置在config/indicators.yaml中定义指标族编程式扩展继承BaseIndicator类实现自定义指标如“量能饱和度圆圈”# config/indicators.yaml macd: module: pandas_ta function: macd params: fast: 12 slow: 26 signal: 9 rsi: module: pandas_ta function: rsi params: length: 14 custom_saturation: module: indicators.saturation function: saturation_circle params: window: 203.2 批量计算与缓存机制避免重复解析同一段行情IndicatorEngine.compute_batch()接收symbol_list和date_range先检查cache/indicators/{symbol}_{start}_{end}.parquet是否存在。若存在且mtime last_fetch_time则直接加载否则调用pandas_ta计算并保存。Parquet 格式比 CSV 节省 60% 存储空间且支持按列读取如只需macd_macd列时不加载rsi列# indicators/engine.py def compute_batch(self, symbols: List[str], start_date: str, end_date: str): cache_key f{_.join(symbols)}_{start_date}_{end_date} cache_path Path(cache/indicators) / f{cache_key}.parquet if cache_path.exists(): mtime datetime.fromtimestamp(cache_path.stat().st_mtime) if mtime self.last_fetch_time: logger.info(fLoad indicators from cache: {cache_path}) return pd.read_parquet(cache_path) # 合并多只股票数据统一计算提升 pandas-ta 向量化效率 all_df pd.concat([ self._load_symbol_data(sym, start_date, end_date) for sym in symbols ], ignore_indexTrue) for indicator_name, config in self.config.items(): module importlib.import_module(config[module]) func getattr(module, config[function]) params config[params] # pandas-ta 支持直接传入 DataFrame 并添加新列 all_df func(all_df, **params) cache_path.parent.mkdir(exist_okTrue) all_df.to_parquet(cache_path, indexFalse) return all_df提示pandas-ta的macd()函数默认返回MACD_12_26_9,MACDh_12_26_9,MACDs_12_26_9三列框架在compute_batch后自动重命名为macd_line,macd_signal,macd_hist与国内主流软件命名对齐——这种细节正是高分作业的体现。3.3 自定义指标开发模板“量能饱和度圆圈1.00”的实现逻辑“量能饱和度圆圈”是 A 股常用情绪指标定义为(当日成交量 / 近 N 日平均成交量) * 100当值 150 时显示红色圆圈。框架提供indicators/saturation.py作为扩展入口# indicators/saturation.py import pandas as pd import numpy as np def saturation_circle(df: pd.DataFrame, window: int 20) - pd.DataFrame: 计算量能饱和度圆圈指标 :param df: 包含 volume 列的 DataFrame :param window: 均值窗口期默认20日 :return: 添加 saturation_circle 列的 DataFrame # 计算滚动均值使用 min_periods1 处理起始不足 window 日的情况 avg_vol df[volume].rolling(windowwindow, min_periods1).mean() # 饱和度 当日量 / 近 window 日均量 * 100 df[saturation_circle] (df[volume] / avg_vol * 100).round(2) # 标记状态150 为高饱和红圈80 为低饱和绿圈 df[saturation_status] normal df.loc[df[saturation_circle] 150, saturation_status] high df.loc[df[saturation_circle] 80, saturation_status] low return df该函数可直接被IndicatorEngine加载无需编译或安装额外包。学生修改window参数或阈值后重新运行pipeline.py即可生成新指标——这比修改talibC 源码现实得多。4. 使用 Plotly Jinja2 构建交互式可视化报告脱离 notebook 的独立 HTML 输出4.1 为什么不用 matplotlibPlotly 的动态交互与导出能力matplotlib生成静态 PNG无法缩放查看 K 线细节seaborn侧重统计分布不擅长金融时序叠加图。Plotly支持鼠标悬停显示精确价格与日期滚轮缩放任意时间段右键拖拽平移导出为 SVG 矢量图论文插图需求嵌入 HTML 时自动适配移动端框架将 K 线图、指标子图、成交量柱状图封装为ChartBuilder类每个图表均启用mode_bar_additions添加“下载为 PNG”按钮# viz/chart_builder.py import plotly.graph_objects as go from plotly.subplots import make_subplots class ChartBuilder: def build_kline_chart(self, df: pd.DataFrame, symbol: str) - go.Figure: fig make_subplots( rows3, cols1, shared_xaxesTrue, vertical_spacing0.05, subplot_titles(K线图, MACD, 成交量), specs[[{type: candlestick}], [{type: scatter}], [{type: bar}]] ) # K线主图 fig.add_trace( go.Candlestick( xdf[trade_date], opendf[open], highdf[high], lowdf[low], closedf[close], nameK线 ), row1, col1 ) # MACD 子图 fig.add_trace( go.Scatter(xdf[trade_date], ydf[macd_line], modelines, nameMACD线), row2, col1 ) fig.add_trace( go.Scatter(xdf[trade_date], ydf[macd_signal], modelines, name信号线), row2, col1 ) fig.add_trace( go.Bar(xdf[trade_date], ydf[macd_hist], nameMACD柱), row2, col1 ) # 成交量柱 fig.add_trace( go.Bar(xdf[trade_date], ydf[volume], name成交量), row3, col1 ) # 配置布局 fig.update_layout( titlef{symbol} 股票分析报告{df[trade_date].min()} 至 {df[trade_date].max()}, height800, showlegendTrue, mode_bar_additions[v1image, zoomIn2d, zoomOut2d, autoScale2d, downloadImage], xaxis_rangeslider_visibleFalse # 关闭底部缩略图用滚轮替代 ) return fig4.2 Jinja2 模板驱动报告生成插入指标统计摘要与风险提示HTML 报告非简单图表堆砌而是结构化文档。框架使用templates/report.html作为 Jinja2 模板动态注入summary_stats:df.describe()提取的close列均值、标准差、最大回撤risk_alerts: 标记出saturation_status high且rsi 70的日期超买预警indicator_table: 指标参数配置表来自indicators.yaml!-- templates/report.html -- h2 指标参数配置/h2 table border1 classdataframe theadtrth指标/thth参数/th/tr/thead tbody {% for name, config in indicators.items() %} tr td{{ name }}/td td{{ config.params|tojson }}/td /tr {% endfor %} /tbody /table h2⚠️ 风险提示/h2 ul {% for alert in risk_alerts %} li{{ alert.date }}{{ alert.message }}RSI{{ alert.rsi }}饱和度{{ alert.saturation }}/li {% endfor %} /ul生成命令python report_generator.py --symbol 000001.SZ --start 20240101 --end 20240630输出reports/000001.SZ_20240101_20240630.html双击即可在浏览器查看完整交互报告——这比提交.ipynb文件更符合企业交付规范。4.3 响应式布局与离线资源确保无网络环境可查看Plotly 默认从 CDN 加载 JavaScript导致离线打不开。框架在report_generator.py中启用include_plotlyjscdn→require并配合plotly.offline.plot()导出内联 JS# report_generator.py def generate_html_report(fig: go.Figure, output_path: str): # 导出为包含完整 JS 的 HTML约 3MB离线可用 fig.write_html( output_path, include_plotlyjsrequire, # 内联 plotly.js full_htmlTrue, config{displayModeBar: True} ) logger.info(fReport saved to {output_path})同时templates/base.html中head部分预置字体加载失败降级方案head link relstylesheet hrefhttps://fonts.googleapis.com/css2?familyNotoSansSC:wght300;400;500;700displayswap onerrorthis.hrefstatic/fonts.css /head当网络不可用时自动加载本地static/fonts.css含font-face定义保证中文不显示方块——这是金融类报告的基本体验底线。5. 实战技巧三步完成一只股票的全流程分析含参数速查表5.1 一条命令启动全链路从数据获取到 HTML 报告生成假设你要分析贵州茅台600519.SH2024年第二季度行情只需执行# 第一步安装依赖仅首次 pip install -r requirements.txt # 第二步配置股票列表编辑 config/symbols.txt echo 600519.SH config/symbols.txt # 第三步运行端到端流水线自动完成爬取→清洗→计算→绘图→生成HTML python main.py --start-date 20240401 --end-date 20240630 --symbols-file config/symbols.txtmain.py内部调用顺序为DataPipeline.run()→IndicatorEngine.compute_batch()→ChartBuilder.build_kline_chart()→report_generator.generate_html_report()。全程无交互输出日志清晰标注各阶段耗时INFO:root:Fetching data for 600519.SH (2024-04-01 to 2024-06-30)... INFO:root:Loaded 62 rows from cache/stock/600519.SH_20240401_20240630.parquet INFO:root:Computing indicators... (macd, rsi, saturation_circle) INFO:root:Generated chart for 600519.SH (height800px) INFO:root:Report saved to reports/600519.SH_20240401_20240630.html提示--symbols-file参数支持多只股票批量分析config/symbols.txt每行一个代码框架会自动并行处理concurrent.futures.ThreadPoolExecutor10只股票耗时仅比单只多 1.2 倍而非 10 倍。5.2 关键参数速查表修改哪里能改变什么结果配置文件参数路径默认值修改影响适用场景config/indicators.yamlmacd.params.fast12MACD 快线周期值越小越敏感短线交易者调至8config/indicators.yamlrsi.params.length14RSI 计算周期值越大越平滑长线投资者调至21config/pipeline.yamlcleaner.remove_invalid_zero_volumetrue是否删除无效零成交量行研究 ST 股票时设为falseconfig/viz.yamlchart.height800HTML 报告总高度像素屏幕分辨率低时设为600config/db.yamlsqlite.pathdata/stock.dbSQLite 数据库存储路径需长期保存时指向 SSD 盘所有参数均通过omegaconf加载支持 YAML 注释#开头行修改后无需重启 Python 进程下次运行自动生效。5.3 验证分析结果准确性的三个必做动作高分项目必须经得起推敲。交付前请执行以下验证比对权威源打开东方财富网https://quote.eastmoney.com/sh600519.html截图“K线图”与报告中600519.SH_20240401_20240630.html的 2024-05-10 日 K 线确认open/high/low/close/volume数值完全一致检查指标逻辑在报告 HTML 中右键 → “查看页面源代码”搜索macd_line找到某日数值如 2024-05-10 的macd_line12.34用 Excel 手动计算EMA(close,12)与EMA(close,26)验证差值是否匹配测试边界情况将config/symbols.txt改为000001.SZ平安银行运行python main.py --start-date 20200101 --end-date 20200101确认程序能正确处理单日数据无KeyError或IndexError。这三个动作覆盖了数据源一致性、算法正确性、鲁棒性三重验证是答辩时老师最可能追问的环节。本文还有配套的精品资源点击获取