简介这是一套面向Python初学者与金融数据分析入门者的完整实践项目聚焦股票数据的自动化采集、量化分析与多维可视化可直接用于高校《Python程序设计》《金融数据分析》等课程设计或期末大作业。资源包含19个文件以10个核心Python脚本如stockInfo.py数据爬取、backTest.py回测、visualization.py绘图、strategies.py策略实现为主体辅以6张关键分析图表含双均线资金曲线、指标对比图等、README.md说明文档及基础工程配置文件整体包体仅850KB轻量易部署。已有845人学习下载项目经导师指导获评97分高分代码结构清晰、模块职责明确、注释充分所有脚本经实测可一键运行无需额外修改即可复现完整分析流程涵盖从原始数据获取、清洗预处理、技术指标计算、交易策略模拟到结果可视化全链路。1. 用 Python 搭建股票数据爬虫分析可视化框架不是写几个脚本而是建立可复用、可扩展、可验证的数据工作流很多刚接触金融数据的同学以为“爬股票数据”就是 requests BeautifulSoup 抓个网页再 pandas 读一读 CSV 就完事——但真实场景中你很快会撞上接口限频、字段缺失、复权错乱、时间对齐失败、多源数据冲突、图表无法复现等问题。这个标题里的“框架”二字恰恰是区分脚本与工程的关键它意味着统一的数据入口支持 Yahoo Finance、AkShare、Tushare 等多源适配、标准化的清洗管道处理停牌、除权、前复权/后复权自动识别、模块化的分析单元技术指标计算、波动率滚动窗口、动量因子分组、以及声明式可视化配置同一份分析结果一键切换折线图/热力图/箱线图。它面向的是需要持续跑回测、生成周报、对接下游风控或投研系统的从业者而不是一次性跑通就扔的练习代码。如果你正在维护一个手动更新 Excel 的股票跟踪表或者每次改个均线参数就要重写三段代码那这个框架的结构设计和参数组织方式比具体某行代码更值得你花时间细读。2. 构建可插拔数据源层用抽象基类统一管理 Yahoo Finance、AkShare 与 Tushare 接口2.1 为什么不能直接写死 requests 调用——接口稳定性与字段契约问题Yahoo Finance 的 HTML 结构每月都可能微调Tushare 的 token 过期机制和权限分级基础版仅开放日线专业版才开放分钟级AkShare 的函数命名风格stock_zh_a_histvsfund_etf_fund_daily_em各不相同。若每个数据获取函数各自为政一旦某源失效就得逐个 grep 修改更严重的是不同源返回的字段名不一致close/Close/收盘/tradedate/datetime/日期volume/vol/成交量。框架的第一道防线就是用 Python 的abc.ABC定义统一契约强制所有数据源实现.fetch(symbol, start_date, end_date)方法并返回严格符合pd.DataFrame结构的标准化结果。from abc import ABC, abstractmethod import pandas as pd from typing import Optional, Tuple class StockDataSource(ABC): abstractmethod def fetch(self, symbol: str, start_date: str, end_date: str, adjust: str qfq) - pd.DataFrame: 标准化数据获取接口 :param symbol: 股票代码如 600519.SH 或 AAPL :param start_date: YYYY-MM-DD 格式起始日期 :param end_date: YYYY-MM-DD 格式结束日期 :param adjust: 复权方式qfq(前复权) / hfq(后复权) / none :return: 包含列 [date, open, high, low, close, volume] 的 DataFrame date 列为 datetime64[ns] 类型索引为 date升序 pass提示adjust参数不是所有源都原生支持比如 Yahoo Finance 默认返回前复权价而 Tushare 需显式传adjqfq。框架层需在子类中做适配转换而非让下游调用者感知差异。2.2 实现 AkShare 数据源规避反爬、处理中文字段映射AkShare 是国内最稳定的免费源但其返回字段为中文且无内置复权逻辑。我们需在fetch()中完成三件事1捕获网络异常并重试2将中文列名映射为英文标准字段3调用akshare内置复权工具或自行实现简单前复权。关键点在于symbol的标准化处理——AkShare 要求 A 股代码带.SH/.SZ后缀而 Yahoo Finance 用AAPL即可。import akshare as ak import time import pandas as pd class AkShareSource(StockDataSource): def fetch(self, symbol: str, start_date: str, end_date: str, adjust: str qfq) - pd.DataFrame: # 步骤1标准化代码格式AkShare 要求 .SH/.SZYahoo 不要 if symbol.endswith((.SH, .SZ)): code symbol else: # 假设输入为数字代码补全交易所后缀简化逻辑实际应查映射表 if symbol.startswith(6) or symbol.startswith(9): code f{symbol}.SH else: code f{symbol}.SZ # 步骤2调用 akshare捕获超时与空数据异常 for _ in range(3): try: df ak.stock_zh_a_hist(symbolcode, perioddaily, start_datestart_date, end_dateend_date, adjustadjust) break except Exception as e: time.sleep(1) continue else: raise ConnectionError(fAkShare 获取 {symbol} 数据失败已重试3次) # 步骤3字段映射关键确保下游分析模块不依赖源字段名 column_map { 日期: date, 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume, 成交额: amount, # 可选保留用于换手率计算 } df df.rename(columnscolumn_map)[[date, open, high, low, close, volume]] # 步骤4强制类型转换与排序 df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df.set_index(date)2.2.1 字段映射表必须覆盖所有可能字段变体实际项目中我们维护一个field_mapping.yaml文件包含yahoo,tushare,akshare三套映射规则由配置驱动而非硬编码。例如 Tushare 的trade_date→datevol→volumechange→pct_change后者用于后续分析模块的涨跌幅计算。这样新增数据源时只需扩充 YAML无需修改核心类。2.3 Tushare 数据源接入Token 管理与权限降级策略Tushare 需要 token且免费版不支持分钟线。框架采用tushare官方 SDK并在fetch()中加入权限检查当请求分钟线但 token 权限不足时自动降级为日线并记录 warning。同时token 存储在环境变量TUSHARE_TOKEN中避免硬编码泄露。import tushare as ts import os class TushareSource(StockDataSource): def __init__(self): self.token os.getenv(TUSHARE_TOKEN) if not self.token: raise ValueError(TUSHARE_TOKEN 未设置请在环境变量中配置) ts.set_token(self.token) self.pro ts.pro_api() def fetch(self, symbol: str, start_date: str, end_date: str, adjust: str qfq) - pd.DataFrame: # Tushare 代码格式600519无后缀 code symbol.split(.)[0] # 剥离 .SH/.SZ # 免费版仅支持日线强制使用 daily 接口 df self.pro.daily(ts_codecode, trade_date, start_datestart_date.replace(-, ), end_dateend_date.replace(-, )) # 字段映射与类型转换 df df.rename(columns{ trade_date: date, open: open, high: high, low: low, close: close, vol: volume })[[date, open, high, low, close, volume]] df[date] pd.to_datetime(df[date]) df df.sort_values(date).set_index(date) return df参数取值说明是否必填框架默认值symbol支持600519.SH,AAPL,000001.SZ等格式自动识别交易所是—start_dateYYYY-MM-DD早于end_date是—end_dateYYYY-MM-DD晚于start_date是—adjustqfq前复权,hfq后复权,none不复权否qfq3. 设计可组合分析管道用 Pandas Pipe 链式调用实现技术指标与因子计算3.1 分析模块的核心约束输入输出必须是标准 DataFrame且不修改原始索引所有分析函数如计算 MACD、布林带、市盈率分位数必须满足1接收pd.DataFrame列含open,high,low,close,volume2返回同结构 DataFrame新增列以_开头如ma_20,macd_line3不改变原始date索引顺序。这保证了任意两个分析函数可自由组合且结果可被下游可视化模块直接消费。def add_ma(df: pd.DataFrame, window: int 20) - pd.DataFrame: 添加 N 日移动平均线 df df.copy() df[fma_{window}] df[close].rolling(windowwindow).mean() return df def add_macd(df: pd.DataFrame, fast: int 12, slow: int 26, signal: int 9) - pd.DataFrame: 添加 MACD 指标DIF, DEA, MACD柱 close df[close] ema_fast close.ewm(spanfast, adjustFalse).mean() ema_slow close.ewm(spanslow, adjustFalse).mean() dif ema_fast - ema_slow dea dif.ewm(spansignal, adjustFalse).mean() macd (dif - dea) * 2 df df.copy() df[macd_dif] dif df[macd_dea] dea df[macd_hist] macd return df # 链式调用示例一行代码叠加多个指标 result_df ( raw_data .pipe(add_ma, window5) .pipe(add_ma, window20) .pipe(add_ma, window60) .pipe(add_macd, fast12, slow26, signal9) .pipe(lambda x: x.dropna()) # 清洗 NaN 行 )注意pipe()的优势在于可读性与调试性。每个函数独立测试出错时可定位到具体环节而传统嵌套写法add_macd(add_ma(raw_data))难以插入断点检查中间态。3.2 实现动量因子分组基于过去 N 月收益率的行业轮动信号动量策略常需按申万一级行业分组计算过去 3 个月收益率并取 Top3 行业。框架提供group_momentum_rank()函数接受行业映射字典symbol_to_industry和回溯周期months3返回每只股票所属行业的动量排名。def group_momentum_rank( df: pd.DataFrame, symbol_to_industry: dict, months: int 3 ) - pd.DataFrame: 计算行业动量排名基于过去 N 月收益率均值 :param df: 原始行情 DataFrameindex 为 datecolumns 含 close :param symbol_to_industry: {symbol: industry_name} 映射字典 :param months: 回溯月数 :return: 添加 industry, industry_momentum_rank 列的 DataFrame # 步骤1计算个股月度收益率假设 df 已按日频需先 resample monthly_ret df[close].resample(ME).last().pct_change().dropna() # 步骤2按行业分组计算各行业月度收益均值 industry_rets {} for symbol, industry in symbol_to_industry.items(): if symbol in monthly_ret.columns: industry_rets.setdefault(industry, []).append(monthly_ret[symbol]) # 步骤3聚合行业收益排序 industry_avg {ind: pd.concat(rets).mean() for ind, rets in industry_rets.items()} ranked_industries sorted(industry_avg.items(), keylambda x: x[1], reverseTrue) # 步骤4为每只股票打标签 rank_map {ind: i1 for i, (ind, _) in enumerate(ranked_industries)} df[industry] df.index.map(lambda x: symbol_to_industry.get(x, 未知)) df[industry_momentum_rank] df[industry].map(rank_map) return df3.2.1 关键参数表动量计算中的时间窗口选择依据参数可选值推荐值业务含义对结果影响months1, 3, 6, 123回溯计算收益率的月份数短周期1月敏感但噪音大长周期12月稳定但滞后resample_ruleME月末,MS月初ME收益率计算的时间粒度ME更符合公募基金持仓披露习惯pct_change_methodsimple简单收益率,log对数收益率simple收益率计算方式简单收益率便于解释对数收益率适合多期连乘4. 声明式可视化引擎用 Plotly Express 配置文件生成可复用图表模板4.1 为什么不用 Matplotlib 手写绘图——维护成本与一致性问题Matplotlib 每次画图都要重复设置字体、网格、图例位置、颜色映射且难以保证同一份数据在不同报告中图表风格统一。框架采用 Plotly Expresspx作为底层因其支持链式语法、内置主题、响应式交互更重要的是——它允许用 YAML 配置文件定义图表行为实现“数据不变图表即变”。# chart_config.yaml charts: - name: price_and_ma type: line x: date y: [close, ma_5, ma_20, ma_60] title: 股价与多周期均线 labels: date: 交易日期 close: 收盘价 ma_5: 5日均线 ma_20: 20日均线 ma_60: 60日均线 color_discrete_sequence: [#1f77b4, #ff7f0e, #2ca02c, #d62728] template: plotly_white - name: macd_analysis type: scatter x: date y: [macd_dif, macd_dea] error_y: macd_hist title: MACD 指标分析 labels: date: 日期 macd_dif: DIF 线 macd_dea: DEA 线 macd_hist: MACD 柱 template: seaborn4.2 加载配置并渲染图表一行命令导出 HTML/PNG框架提供render_chart()函数读取 YAML 配置自动匹配 DataFrame 列并调用对应 Plotly 函数。关键逻辑在于1根据type选择px.line/px.scatter/px.bar2动态构建**kwargs3支持output_format: html或png。import plotly.express as px import yaml import plotly.io as pio def render_chart(df: pd.DataFrame, config_path: str, chart_name: str, output_path: str): with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) chart_def next((c for c in config[charts] if c[name] chart_name), None) if not chart_def: raise ValueError(f未找到图表配置: {chart_name}) # 构建 kwargs过滤掉 DataFrame 中不存在的列 y_cols [col for col in chart_def[y] if col in df.columns] if not y_cols: raise ValueError(f配置中指定的 Y 轴列 {chart_def[y]} 在数据中不存在) kwargs { x: chart_def[x], y: y_cols, title: chart_def[title], labels: chart_def.get(labels, {}), template: chart_def.get(template, plotly), } # 根据图表类型调用 if chart_def[type] line: fig px.line(df, **kwargs) elif chart_def[type] scatter: error_y chart_def.get(error_y) if error_y and error_y in df.columns: kwargs[error_y] error_y fig px.scatter(df, **kwargs) else: raise NotImplementedError(f不支持的图表类型: {chart_def[type]}) # 输出 if output_path.endswith(.html): fig.write_html(output_path) elif output_path.endswith(.png): pio.write_image(fig, output_path, width1200, height600) else: raise ValueError(仅支持 .html 和 .png 输出格式) # 使用示例 render_chart( dfresult_df, config_pathchart_config.yaml, chart_nameprice_and_ma, output_pathoutput/price_ma.html )4.2.1 配置文件支持的高级特性特性配置语法作用示例多 Y 轴y: [close, volume]同一图表叠加价格与成交量自动适配双 Y 轴条件着色color: industry_momentum_rank按因子值自动分配颜色Top3 行业高亮显示子图布局facet_col: industry按行业分面展示每个行业一个子图交互提示hover_data: [pct_change]鼠标悬停显示额外字段查看当日涨跌幅5. 框架落地技巧用 conda 环境隔离 Makefile 一键执行全流程5.1 环境管理为什么推荐 conda 而非 pip股票分析依赖numpy,pandas,akshare,tushare,plotly等包其中akshare依赖requests的特定版本plotly与orjson有编译兼容性问题。conda 的environment.yml可精确锁定二进制依赖避免pip install时因系统 OpenSSL 版本导致的 SSL 错误。# environment.yml name: stock-framework channels: - conda-forge - defaults dependencies: - python3.9 - pandas1.5.3 - numpy1.24.3 - akshare1.10.80 - tushare1.3.5 - plotly5.18.0 - pyyaml6.0.1 - pip - pip: - redis4.6.0 # 若后续接入缓存执行conda env create -f environment.yml即可创建完全隔离的环境conda activate stock-framework后所有依赖版本确定团队协作零差异。5.2 Makefile 自动化把“爬取→清洗→分析→绘图”串成一条命令手工执行python crawl.py,python analyze.py,python visualize.py极易遗漏步骤或顺序错误。Makefile 将流程固化为依赖关系make all自动判断哪些文件过期并重跑。# Makefile .PHONY: all clean DATA_DIR : data OUTPUT_DIR : output CONFIG_FILE : config.yaml all: $(OUTPUT_DIR)/price_ma.html $(OUTPUT_DIR)/macd_analysis.png $(OUTPUT_DIR)/price_ma.html: $(DATA_DIR)/600519.csv python visualize.py --config chart_config.yaml --chart price_and_ma --input $ --output $ $(OUTPUT_DIR)/macd_analysis.png: $(DATA_DIR)/600519.csv python visualize.py --config chart_config.yaml --chart macd_analysis --input $ --output $ $(DATA_DIR)/600519.csv: crawl.py $(CONFIG_FILE) python crawl.py --symbol 600519.SH --start 2023-01-01 --end 2023-12-31 --output $ clean: rm -rf $(DATA_DIR) $(OUTPUT_DIR) # 快捷命令一键更新全部 update: clean all执行make update即完成清空旧数据 → 重新爬取 → 生成图表。开发者只需维护config.yaml中的股票列表和日期范围无需记忆命令参数。5.3 验证框架是否正常工作的最小闭环测试在tests/test_framework.py中编写端到端测试不依赖网络用pytestunittest.mock模拟数据源返回import pytest from unittest.mock import patch import pandas as pd from src.data_source import AkShareSource def test_akshare_fetch_returns_standard_df(): # Mock akshare.stock_zh_a_hist to return fixed data mock_data pd.DataFrame({ 日期: [2023-01-01, 2023-01-02], 开盘: [100, 101], 最高: [102, 103], 最低: [99, 100], 收盘: [101, 102], 成交量: [1000, 1100], }) with patch(akshare.stock_zh_a_hist, return_valuemock_data): source AkShareSource() result source.fetch(600519.SH, 2023-01-01, 2023-01-02) # 断言字段名、索引类型、数据完整性 assert list(result.columns) [open, high, low, close, volume] assert result.index.dtype datetime64[ns] assert len(result) 2 assert result.loc[2023-01-01, close] 101运行pytest tests/ -v即可验证数据层契约是否被破坏。这是框架可持续演进的基石——任何修改都不能让此测试失败。本文还有配套的精品资源点击获取 Python股票数据工程框架：多源适配、管道分析与声明式可视化
简介这是一套面向Python初学者与金融数据分析入门者的完整实践项目聚焦股票数据的自动化采集、量化分析与多维可视化可直接用于高校《Python程序设计》《金融数据分析》等课程设计或期末大作业。资源包含19个文件以10个核心Python脚本如stockInfo.py数据爬取、backTest.py回测、visualization.py绘图、strategies.py策略实现为主体辅以6张关键分析图表含双均线资金曲线、指标对比图等、README.md说明文档及基础工程配置文件整体包体仅850KB轻量易部署。已有845人学习下载项目经导师指导获评97分高分代码结构清晰、模块职责明确、注释充分所有脚本经实测可一键运行无需额外修改即可复现完整分析流程涵盖从原始数据获取、清洗预处理、技术指标计算、交易策略模拟到结果可视化全链路。1. 用 Python 搭建股票数据爬虫分析可视化框架不是写几个脚本而是建立可复用、可扩展、可验证的数据工作流很多刚接触金融数据的同学以为“爬股票数据”就是 requests BeautifulSoup 抓个网页再 pandas 读一读 CSV 就完事——但真实场景中你很快会撞上接口限频、字段缺失、复权错乱、时间对齐失败、多源数据冲突、图表无法复现等问题。这个标题里的“框架”二字恰恰是区分脚本与工程的关键它意味着统一的数据入口支持 Yahoo Finance、AkShare、Tushare 等多源适配、标准化的清洗管道处理停牌、除权、前复权/后复权自动识别、模块化的分析单元技术指标计算、波动率滚动窗口、动量因子分组、以及声明式可视化配置同一份分析结果一键切换折线图/热力图/箱线图。它面向的是需要持续跑回测、生成周报、对接下游风控或投研系统的从业者而不是一次性跑通就扔的练习代码。如果你正在维护一个手动更新 Excel 的股票跟踪表或者每次改个均线参数就要重写三段代码那这个框架的结构设计和参数组织方式比具体某行代码更值得你花时间细读。2. 构建可插拔数据源层用抽象基类统一管理 Yahoo Finance、AkShare 与 Tushare 接口2.1 为什么不能直接写死 requests 调用——接口稳定性与字段契约问题Yahoo Finance 的 HTML 结构每月都可能微调Tushare 的 token 过期机制和权限分级基础版仅开放日线专业版才开放分钟级AkShare 的函数命名风格stock_zh_a_histvsfund_etf_fund_daily_em各不相同。若每个数据获取函数各自为政一旦某源失效就得逐个 grep 修改更严重的是不同源返回的字段名不一致close/Close/收盘/tradedate/datetime/日期volume/vol/成交量。框架的第一道防线就是用 Python 的abc.ABC定义统一契约强制所有数据源实现.fetch(symbol, start_date, end_date)方法并返回严格符合pd.DataFrame结构的标准化结果。from abc import ABC, abstractmethod import pandas as pd from typing import Optional, Tuple class StockDataSource(ABC): abstractmethod def fetch(self, symbol: str, start_date: str, end_date: str, adjust: str qfq) - pd.DataFrame: 标准化数据获取接口 :param symbol: 股票代码如 600519.SH 或 AAPL :param start_date: YYYY-MM-DD 格式起始日期 :param end_date: YYYY-MM-DD 格式结束日期 :param adjust: 复权方式qfq(前复权) / hfq(后复权) / none :return: 包含列 [date, open, high, low, close, volume] 的 DataFrame date 列为 datetime64[ns] 类型索引为 date升序 pass提示adjust参数不是所有源都原生支持比如 Yahoo Finance 默认返回前复权价而 Tushare 需显式传adjqfq。框架层需在子类中做适配转换而非让下游调用者感知差异。2.2 实现 AkShare 数据源规避反爬、处理中文字段映射AkShare 是国内最稳定的免费源但其返回字段为中文且无内置复权逻辑。我们需在fetch()中完成三件事1捕获网络异常并重试2将中文列名映射为英文标准字段3调用akshare内置复权工具或自行实现简单前复权。关键点在于symbol的标准化处理——AkShare 要求 A 股代码带.SH/.SZ后缀而 Yahoo Finance 用AAPL即可。import akshare as ak import time import pandas as pd class AkShareSource(StockDataSource): def fetch(self, symbol: str, start_date: str, end_date: str, adjust: str qfq) - pd.DataFrame: # 步骤1标准化代码格式AkShare 要求 .SH/.SZYahoo 不要 if symbol.endswith((.SH, .SZ)): code symbol else: # 假设输入为数字代码补全交易所后缀简化逻辑实际应查映射表 if symbol.startswith(6) or symbol.startswith(9): code f{symbol}.SH else: code f{symbol}.SZ # 步骤2调用 akshare捕获超时与空数据异常 for _ in range(3): try: df ak.stock_zh_a_hist(symbolcode, perioddaily, start_datestart_date, end_dateend_date, adjustadjust) break except Exception as e: time.sleep(1) continue else: raise ConnectionError(fAkShare 获取 {symbol} 数据失败已重试3次) # 步骤3字段映射关键确保下游分析模块不依赖源字段名 column_map { 日期: date, 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume, 成交额: amount, # 可选保留用于换手率计算 } df df.rename(columnscolumn_map)[[date, open, high, low, close, volume]] # 步骤4强制类型转换与排序 df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df.set_index(date)2.2.1 字段映射表必须覆盖所有可能字段变体实际项目中我们维护一个field_mapping.yaml文件包含yahoo,tushare,akshare三套映射规则由配置驱动而非硬编码。例如 Tushare 的trade_date→datevol→volumechange→pct_change后者用于后续分析模块的涨跌幅计算。这样新增数据源时只需扩充 YAML无需修改核心类。2.3 Tushare 数据源接入Token 管理与权限降级策略Tushare 需要 token且免费版不支持分钟线。框架采用tushare官方 SDK并在fetch()中加入权限检查当请求分钟线但 token 权限不足时自动降级为日线并记录 warning。同时token 存储在环境变量TUSHARE_TOKEN中避免硬编码泄露。import tushare as ts import os class TushareSource(StockDataSource): def __init__(self): self.token os.getenv(TUSHARE_TOKEN) if not self.token: raise ValueError(TUSHARE_TOKEN 未设置请在环境变量中配置) ts.set_token(self.token) self.pro ts.pro_api() def fetch(self, symbol: str, start_date: str, end_date: str, adjust: str qfq) - pd.DataFrame: # Tushare 代码格式600519无后缀 code symbol.split(.)[0] # 剥离 .SH/.SZ # 免费版仅支持日线强制使用 daily 接口 df self.pro.daily(ts_codecode, trade_date, start_datestart_date.replace(-, ), end_dateend_date.replace(-, )) # 字段映射与类型转换 df df.rename(columns{ trade_date: date, open: open, high: high, low: low, close: close, vol: volume })[[date, open, high, low, close, volume]] df[date] pd.to_datetime(df[date]) df df.sort_values(date).set_index(date) return df参数取值说明是否必填框架默认值symbol支持600519.SH,AAPL,000001.SZ等格式自动识别交易所是—start_dateYYYY-MM-DD早于end_date是—end_dateYYYY-MM-DD晚于start_date是—adjustqfq前复权,hfq后复权,none不复权否qfq3. 设计可组合分析管道用 Pandas Pipe 链式调用实现技术指标与因子计算3.1 分析模块的核心约束输入输出必须是标准 DataFrame且不修改原始索引所有分析函数如计算 MACD、布林带、市盈率分位数必须满足1接收pd.DataFrame列含open,high,low,close,volume2返回同结构 DataFrame新增列以_开头如ma_20,macd_line3不改变原始date索引顺序。这保证了任意两个分析函数可自由组合且结果可被下游可视化模块直接消费。def add_ma(df: pd.DataFrame, window: int 20) - pd.DataFrame: 添加 N 日移动平均线 df df.copy() df[fma_{window}] df[close].rolling(windowwindow).mean() return df def add_macd(df: pd.DataFrame, fast: int 12, slow: int 26, signal: int 9) - pd.DataFrame: 添加 MACD 指标DIF, DEA, MACD柱 close df[close] ema_fast close.ewm(spanfast, adjustFalse).mean() ema_slow close.ewm(spanslow, adjustFalse).mean() dif ema_fast - ema_slow dea dif.ewm(spansignal, adjustFalse).mean() macd (dif - dea) * 2 df df.copy() df[macd_dif] dif df[macd_dea] dea df[macd_hist] macd return df # 链式调用示例一行代码叠加多个指标 result_df ( raw_data .pipe(add_ma, window5) .pipe(add_ma, window20) .pipe(add_ma, window60) .pipe(add_macd, fast12, slow26, signal9) .pipe(lambda x: x.dropna()) # 清洗 NaN 行 )注意pipe()的优势在于可读性与调试性。每个函数独立测试出错时可定位到具体环节而传统嵌套写法add_macd(add_ma(raw_data))难以插入断点检查中间态。3.2 实现动量因子分组基于过去 N 月收益率的行业轮动信号动量策略常需按申万一级行业分组计算过去 3 个月收益率并取 Top3 行业。框架提供group_momentum_rank()函数接受行业映射字典symbol_to_industry和回溯周期months3返回每只股票所属行业的动量排名。def group_momentum_rank( df: pd.DataFrame, symbol_to_industry: dict, months: int 3 ) - pd.DataFrame: 计算行业动量排名基于过去 N 月收益率均值 :param df: 原始行情 DataFrameindex 为 datecolumns 含 close :param symbol_to_industry: {symbol: industry_name} 映射字典 :param months: 回溯月数 :return: 添加 industry, industry_momentum_rank 列的 DataFrame # 步骤1计算个股月度收益率假设 df 已按日频需先 resample monthly_ret df[close].resample(ME).last().pct_change().dropna() # 步骤2按行业分组计算各行业月度收益均值 industry_rets {} for symbol, industry in symbol_to_industry.items(): if symbol in monthly_ret.columns: industry_rets.setdefault(industry, []).append(monthly_ret[symbol]) # 步骤3聚合行业收益排序 industry_avg {ind: pd.concat(rets).mean() for ind, rets in industry_rets.items()} ranked_industries sorted(industry_avg.items(), keylambda x: x[1], reverseTrue) # 步骤4为每只股票打标签 rank_map {ind: i1 for i, (ind, _) in enumerate(ranked_industries)} df[industry] df.index.map(lambda x: symbol_to_industry.get(x, 未知)) df[industry_momentum_rank] df[industry].map(rank_map) return df3.2.1 关键参数表动量计算中的时间窗口选择依据参数可选值推荐值业务含义对结果影响months1, 3, 6, 123回溯计算收益率的月份数短周期1月敏感但噪音大长周期12月稳定但滞后resample_ruleME月末,MS月初ME收益率计算的时间粒度ME更符合公募基金持仓披露习惯pct_change_methodsimple简单收益率,log对数收益率simple收益率计算方式简单收益率便于解释对数收益率适合多期连乘4. 声明式可视化引擎用 Plotly Express 配置文件生成可复用图表模板4.1 为什么不用 Matplotlib 手写绘图——维护成本与一致性问题Matplotlib 每次画图都要重复设置字体、网格、图例位置、颜色映射且难以保证同一份数据在不同报告中图表风格统一。框架采用 Plotly Expresspx作为底层因其支持链式语法、内置主题、响应式交互更重要的是——它允许用 YAML 配置文件定义图表行为实现“数据不变图表即变”。# chart_config.yaml charts: - name: price_and_ma type: line x: date y: [close, ma_5, ma_20, ma_60] title: 股价与多周期均线 labels: date: 交易日期 close: 收盘价 ma_5: 5日均线 ma_20: 20日均线 ma_60: 60日均线 color_discrete_sequence: [#1f77b4, #ff7f0e, #2ca02c, #d62728] template: plotly_white - name: macd_analysis type: scatter x: date y: [macd_dif, macd_dea] error_y: macd_hist title: MACD 指标分析 labels: date: 日期 macd_dif: DIF 线 macd_dea: DEA 线 macd_hist: MACD 柱 template: seaborn4.2 加载配置并渲染图表一行命令导出 HTML/PNG框架提供render_chart()函数读取 YAML 配置自动匹配 DataFrame 列并调用对应 Plotly 函数。关键逻辑在于1根据type选择px.line/px.scatter/px.bar2动态构建**kwargs3支持output_format: html或png。import plotly.express as px import yaml import plotly.io as pio def render_chart(df: pd.DataFrame, config_path: str, chart_name: str, output_path: str): with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) chart_def next((c for c in config[charts] if c[name] chart_name), None) if not chart_def: raise ValueError(f未找到图表配置: {chart_name}) # 构建 kwargs过滤掉 DataFrame 中不存在的列 y_cols [col for col in chart_def[y] if col in df.columns] if not y_cols: raise ValueError(f配置中指定的 Y 轴列 {chart_def[y]} 在数据中不存在) kwargs { x: chart_def[x], y: y_cols, title: chart_def[title], labels: chart_def.get(labels, {}), template: chart_def.get(template, plotly), } # 根据图表类型调用 if chart_def[type] line: fig px.line(df, **kwargs) elif chart_def[type] scatter: error_y chart_def.get(error_y) if error_y and error_y in df.columns: kwargs[error_y] error_y fig px.scatter(df, **kwargs) else: raise NotImplementedError(f不支持的图表类型: {chart_def[type]}) # 输出 if output_path.endswith(.html): fig.write_html(output_path) elif output_path.endswith(.png): pio.write_image(fig, output_path, width1200, height600) else: raise ValueError(仅支持 .html 和 .png 输出格式) # 使用示例 render_chart( dfresult_df, config_pathchart_config.yaml, chart_nameprice_and_ma, output_pathoutput/price_ma.html )4.2.1 配置文件支持的高级特性特性配置语法作用示例多 Y 轴y: [close, volume]同一图表叠加价格与成交量自动适配双 Y 轴条件着色color: industry_momentum_rank按因子值自动分配颜色Top3 行业高亮显示子图布局facet_col: industry按行业分面展示每个行业一个子图交互提示hover_data: [pct_change]鼠标悬停显示额外字段查看当日涨跌幅5. 框架落地技巧用 conda 环境隔离 Makefile 一键执行全流程5.1 环境管理为什么推荐 conda 而非 pip股票分析依赖numpy,pandas,akshare,tushare,plotly等包其中akshare依赖requests的特定版本plotly与orjson有编译兼容性问题。conda 的environment.yml可精确锁定二进制依赖避免pip install时因系统 OpenSSL 版本导致的 SSL 错误。# environment.yml name: stock-framework channels: - conda-forge - defaults dependencies: - python3.9 - pandas1.5.3 - numpy1.24.3 - akshare1.10.80 - tushare1.3.5 - plotly5.18.0 - pyyaml6.0.1 - pip - pip: - redis4.6.0 # 若后续接入缓存执行conda env create -f environment.yml即可创建完全隔离的环境conda activate stock-framework后所有依赖版本确定团队协作零差异。5.2 Makefile 自动化把“爬取→清洗→分析→绘图”串成一条命令手工执行python crawl.py,python analyze.py,python visualize.py极易遗漏步骤或顺序错误。Makefile 将流程固化为依赖关系make all自动判断哪些文件过期并重跑。# Makefile .PHONY: all clean DATA_DIR : data OUTPUT_DIR : output CONFIG_FILE : config.yaml all: $(OUTPUT_DIR)/price_ma.html $(OUTPUT_DIR)/macd_analysis.png $(OUTPUT_DIR)/price_ma.html: $(DATA_DIR)/600519.csv python visualize.py --config chart_config.yaml --chart price_and_ma --input $ --output $ $(OUTPUT_DIR)/macd_analysis.png: $(DATA_DIR)/600519.csv python visualize.py --config chart_config.yaml --chart macd_analysis --input $ --output $ $(DATA_DIR)/600519.csv: crawl.py $(CONFIG_FILE) python crawl.py --symbol 600519.SH --start 2023-01-01 --end 2023-12-31 --output $ clean: rm -rf $(DATA_DIR) $(OUTPUT_DIR) # 快捷命令一键更新全部 update: clean all执行make update即完成清空旧数据 → 重新爬取 → 生成图表。开发者只需维护config.yaml中的股票列表和日期范围无需记忆命令参数。5.3 验证框架是否正常工作的最小闭环测试在tests/test_framework.py中编写端到端测试不依赖网络用pytestunittest.mock模拟数据源返回import pytest from unittest.mock import patch import pandas as pd from src.data_source import AkShareSource def test_akshare_fetch_returns_standard_df(): # Mock akshare.stock_zh_a_hist to return fixed data mock_data pd.DataFrame({ 日期: [2023-01-01, 2023-01-02], 开盘: [100, 101], 最高: [102, 103], 最低: [99, 100], 收盘: [101, 102], 成交量: [1000, 1100], }) with patch(akshare.stock_zh_a_hist, return_valuemock_data): source AkShareSource() result source.fetch(600519.SH, 2023-01-01, 2023-01-02) # 断言字段名、索引类型、数据完整性 assert list(result.columns) [open, high, low, close, volume] assert result.index.dtype datetime64[ns] assert len(result) 2 assert result.loc[2023-01-01, close] 101运行pytest tests/ -v即可验证数据层契约是否被破坏。这是框架可持续演进的基石——任何修改都不能让此测试失败。本文还有配套的精品资源点击获取