做金融数据方向的技术开发最常遇到的一个诉求是能不能每天早上自动拉一遍 A 股行情把自己关注的核心指标算好再生成一份干净的报告。手动打开行情软件一个个看效率低而且历史数据不好汇总用 Excel 去网页复制又容易受格式干扰。本文就围绕daily_stock_analysis这个典型项目从数据获取、指标计算、报告输出到定时自动化完整拆解一套可落地的每日股票分析方案适合有 Python 基础、想入门金融量化分析的开发者也适合后端工程师快速搭建自己的行情监控脚本。1. 背景与核心概念1.1 什么是每日股票分析每日股票分析简单来说就是通过程序定时获取股票行情数据再按照设定的规则计算技术指标或筛选条件最后输出分析结果。它解决的问题很明确把“每天需要人工打开多个页面查看数据”变成“脚本自动拉数、自动算、自动写报告”。从实现层面看一个完整的每日股票分析项目通常包含四个环节数据获取从行情接口拉取股票列表、实时快照、历史 K 线。指标计算根据原始行情计算涨跌幅、均线、换手率、量比等衍生指标。筛选或排序按自己的关注维度筛选出符合条件的股票。报告输出把结果保存为 Excel、CSV 或 Markdown 文件供后续查看。这类项目的技术难度并不高核心是熟悉 Python 的数据处理流程以及选对行情数据源。相比从零开始爬取财经网站现在更推荐直接使用成熟的 Python 行情库。1.2 常见应用场景每日自选股速览只统计自己关注的股票输出当日涨跌、成交量、均线位置。全市场热点扫描遍历 A 股全市场按涨幅、成交额、换手率等条件筛选异动股。技术指标监控持续记录沪深 300 成分股的 MA5、MA10、MA20 排列情况。定期报告生成每天早上生成一份行情 Excel 报告发送给同事或归档。投研数据整合把每日行情落库作为后续量化策略研究的基础数据。1.3 为什么不用手动方式手动方式的痛点主要体现在三个方面效率低每只股票都要单独打开页面查看操作重复。难汇总不同页面的数据格式不一致整理到 Excel 很痛苦。无法留痕缺少历史归档事后复盘时找不到当日数据。通过 daily_stock_analysis 这类自动化脚本可以实现数据获取、计算、归档一条龙这也是它适合作为 Python 数据分析练手项目的原因。2. 环境准备与版本说明2.1 运行环境本文示例以常见环境为例重点演示配置思路具体版本需要根据你的项目实际情况调整。操作系统Windows 10/11 或 CentOS 7 / Ubuntu 20.04。Python3.8 及以上版本推荐 3.10 或 3.11。包管理工具pip 或 conda。IDE推荐 VS Code 或 PyCharm方便调试和查看 DataFrame 数据。2.2 核心依赖库库名用途安装命令pandas数据处理与计算pip install pandasakshareA 股行情数据接口pip install akshareopenpyxl写入 Excel 文件pip install openpyxltabulateMarkdown 表格渲染pip install tabulatepython-dotenv读取本地环境变量pip install python-dotenv其中akshare是目前国内开发者使用较多的开源行情库数据来自东方财富等公开渠道免费且无需注册 token。需要注意它的接口和字段会随上游页面调整而变化使用过程中要关注版本更新。如果你想使用 tushare 或其他数据源也可以按照同一套代码结构替换数据获取函数后面会单独讲解替换思路。安装完成后可以执行一个简单验证import akshare as ak df ak.stock_zh_a_spot_em() print(df.head())如果输出了包含代码、名称、最新价、涨跌幅等字段的 DataFrame说明环境正常。由于网络环境和库版本差异若报错请先升级 akshare 到最新版。3. 核心设计思路3.1 项目整体流程整个 daily_stock_analysis 项目可以抽象成下面的执行链定时触发 - 拉取实时行情快照 - 清洗数据 - 计算指标 - 条件筛选 - 生成报告 - 归档文件为了便于理解和维护建议把代码拆成几个独立函数数据获取函数只负责和行情接口交互。指标计算函数只负责 DataFrame 的字段计算。筛选函数只负责按条件过滤。报告输出函数只负责写文件。主函数把以上步骤串起来。这样做的好处是当天行情接口变了只改数据获取函数想调筛选条件只改筛选函数。职责清晰排错也方便。3.2 数据字段说明A 股实时行情快照常见的字段如下字段含义代码股票代码例如 600519名称股票名称最新价当前最新成交价涨跌幅当日涨跌幅百分比涨跌额当日涨跌绝对值成交量当日成交量手成交额当日成交金额元换手率当日换手率百分比量比当日量比指标市盈率-动态动态市盈率总市值总市值元流通市值流通市值元这些字段基本能满足大多数日频筛选需求。历史 K 线数据则长这样字段含义日期交易日期开盘开盘价收盘收盘价最高当日最高价最低当日最低价成交量当日成交量手成交额当日成交金额元振幅当日振幅百分比涨跌幅当日涨跌幅百分比涨跌额当日涨跌额换手率当日换手率百分比3.3 筛选规则怎么设计筛选规则是整个项目最有价值的部分。建议先把规则写成常量或配置而不是硬编码在判断逻辑里。下面是一个偏保守的筛选示例剔除 ST、退市整理股票名称中不包含ST、*ST。涨跌幅要求当日涨幅在 3% 到 9% 之间避开涨停板上的无量股票。成交额要求大于 5 亿元保证流动性。换手率要求在 3% 到 20% 之间过滤掉过于冷门或过度投机的标的。市值要求流通市值在 50 亿到 500 亿之间。这些参数可以根据你的偏好调整例如把涨跌幅下限改成 5%或把成交额下限改成 2 亿。关键是不要把规则写死在大量重复代码里最好通过一个字典或配置文件管理。4. 完整实战案例下面进入实战环节。我们会实现一个最小可运行的daily_stock_analysis脚本功能包括拉取 A 股实时行情。清洗数据并剔除 ST 股票。计算均线、涨跌幅、量比等指标。按规则筛选股票。生成 Excel 报告和 Markdown 摘要。4.1 创建项目结构先在本地创建一个项目目录mkdir daily_stock_analysis cd daily_stock_analysis项目内文件结构如下daily_stock_analysis/ ├── config.py # 筛选参数配置 ├── data_fetcher.py # 行情数据获取 ├── analyzer.py # 指标计算与筛选 ├── reporter.py # 报告输出 ├── main.py # 主入口 ├── reports/ # 输出目录 └── requirements.txt # 依赖清单4.2 编写依赖清单创建requirements.txtpandas2.0.0 akshare1.15.0 openpyxl3.1.0 tabulate0.9.0 python-dotenv1.0.0安装依赖pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple使用国内镜像安装速度会快很多。不同 Python 环境需要根据实际情况调整如果部分库版本冲突建议新建一个虚拟环境再安装。4.3 编写配置文件创建config.py把筛选参数集中管理# 文件路径config.py # 说明集中管理筛选参数和数据源配置 # 筛选规则 FILTER_RULES { min_pct_change: 3.0, # 最低涨幅% max_pct_change: 9.0, # 最高涨幅%避免涨停无量 min_amount: 5e8, # 最低成交额元5亿 min_turnover: 3.0, # 最低换手率% max_turnover: 20.0, # 最高换手率% min_float_mv: 50e8, # 最低流通市值元50亿 max_float_mv: 500e8, # 最高流通市值元500亿 } # 是否过滤 ST 股票 FILTER_ST True # 报告输出目录 REPORT_DIR reports把参数集中到配置文件中后续调整筛选条件时不需要修改主逻辑只改config.py即可这也是工程化开发里比较推荐的写法。4.4 编写数据获取模块创建data_fetcher.py负责与 akshare 接口交互# 文件路径data_fetcher.py # 说明数据获取模块集中封装 akshare 的行情接口 import akshare as ak import pandas as pd def fetch_realtime_spot() - pd.DataFrame: 获取 A 股实时行情快照。 返回值包含代码、名称、最新价、涨跌幅、成交额、换手率等常用字段。 df ak.stock_zh_a_spot_em() if df is None or df.empty: raise ValueError(获取实时行情失败返回数据为空) return df def fetch_stock_history(symbol: str, days: int 60) - pd.DataFrame: 获取单只股票的历史日 K 数据用于计算均线等指标。 参数: symbol: 股票代码例如 600519 days: 需要的交易日数量默认 60 天 end_date pd.Timestamp.now().strftime(%Y%m%d) # 按自然日估算留出节假日余量 start_date (pd.Timestamp.now() - pd.Timedelta(daysint(days * 1.6))).strftime(%Y%m%d) df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, ) if df is None or df.empty: raise ValueError(f获取 {symbol} 历史行情失败) return df这里需要注意两点stock_zh_a_spot_em()返回的是全市场快照数据量比较大建议在非交易时段使用历史快照在交易时段能看到实时变化。历史 K 线接口返回的字段名是中文需要后续统一处理。4.5 编写指标计算与筛选模块创建analyzer.py# 文件路径analyzer.py # 说明指标计算和股票筛选模块 import pandas as pd from config import FILTER_RULES, FILTER_ST def clean_spot_data(df: pd.DataFrame) - pd.DataFrame: 清洗实时行情数据。 - 统一字段名 - 剔除 ST / *ST - 剔除停牌或缺失数据的股票 df df.rename(columns{ 代码: code, 名称: name, 最新价: price, 涨跌幅: pct_change, 涨跌额: change, 成交量: volume, 成交额: amount, 换手率: turnover, 量比: volume_ratio, 市盈率-动态: pe_ttm, 总市值: total_mv, 流通市值: float_mv, }) # 剔除缺失关键字段的数据 required_cols [code, name, price, pct_change, amount, turnover, float_mv] df df.dropna(subsetrequired_cols) # 过滤 ST 股票 if FILTER_ST: df df[~df[name].str.upper().str.contains(ST, naFalse)] # 剔除停牌或无成交额的数据 df df[df[amount] 0] return df.reset_index(dropTrue) def calculate_ma(df: pd.DataFrame, ma_window: int 20) - pd.DataFrame: 为 DataFrame 增加收盘价的 N 日均线列。 这里主要适用于单只股票的历史行情。 result df.copy() result[fMA{ma_window}] result[close].rolling(windowma_window).mean() return result def filter_stocks(df: pd.DataFrame) - pd.DataFrame: 按照配置的规则筛选股票。 筛选逻辑 1. 涨幅在 [min_pct_change, max_pct_change] 区间 2. 成交额大于 min_amount 3. 换手率在 [min_turnover, max_turnover] 区间 4. 流通市值在 [min_float_mv, max_float_mv] 区间 rules FILTER_RULES mask ( (df[pct_change] rules[min_pct_change]) (df[pct_change] rules[max_pct_change]) (df[amount] rules[min_amount]) (df[turnover] rules[min_turnover]) (df[turnover] rules[max_turnover]) (df[float_mv] rules[min_float_mv]) (df[float_mv] rules[max_float_mv]) ) result df[mask].copy() # 按涨跌幅降序排列 result result.sort_values(pct_change, ascendingFalse) return result.reset_index(dropTrue) def compute_top_stocks(df: pd.DataFrame, top_n: int 10) - pd.DataFrame: 在筛选结果中再选出涨幅最靠前的 N 只股票。 if df.empty: return df return df.head(top_n).reset_index(dropTrue)这段代码把“数据清洗”“指标计算”“筛选”拆成了独立函数方便单独测试。实际运行时你可以先执行clean_spot_data再执行filter_stocks最后用compute_top_stocks输出前 N 名。4.6 编写报告输出模块创建reporter.py负责把分析结果写入 Excel 和 Markdown# 文件路径reporter.py # 说明报告输出模块 import os from datetime import datetime import pandas as pd def save_to_excel(df: pd.DataFrame, report_dir: str reports) - str: 把 DataFrame 保存为 Excel 文件文件名带日期。 返回生成文件的完整路径。 os.makedirs(report_dir, exist_okTrue) today datetime.now().strftime(%Y%m%d) file_path os.path.join(report_dir, fstock_analysis_{today}.xlsx) # 写入 Excel使用 openpyxl 引擎 with pd.ExcelWriter(file_path, engineopenpyxl) as writer: df.to_excel(writer, indexFalse, sheet_name今日筛选) return file_path def save_to_markdown(df: pd.DataFrame, report_dir: str reports) - str: 把 DataFrame 保存为 Markdown 表格方便在文档中直接引用。 os.makedirs(report_dir, exist_okTrue) today datetime.now().strftime(%Y%m%d) file_path os.path.join(report_dir, fstock_analysis_{today}.md) with open(file_path, w, encodingutf-8) as f: f.write(f# 每日股票分析报告{today}\n\n) if df.empty: f.write(今日没有符合条件的股票。\n) else: f.write(df.to_markdown(indexFalse)) return file_path def print_summary(df: pd.DataFrame) - None: 在控制台打印简要分析摘要。 if df.empty: print(今日没有筛选出符合条件的股票。) return print(f今日共筛选出 {len(df)} 只股票涨幅前 5) print(df[[code, name, price, pct_change, amount]].head(5).to_string(indexFalse))4.7 编写主入口创建main.py把整个流程串起来# 文件路径main.py # 说明程序主入口串联数据获取、指标计算、筛选和报告输出 from data_fetcher import fetch_realtime_spot from analyzer import clean_spot_data, filter_stocks, compute_top_stocks from reporter import save_to_excel, save_to_markdown, print_summary def main(): print(正在拉取 A 股实时行情...) raw_df fetch_realtime_spot() print(f原始数据共 {len(raw_df)} 条) print(正在清洗数据...) clean_df clean_spot_data(raw_df) print(f清洗后剩余 {len(clean_df)} 条) print(正在按规则筛选...) filtered_df filter_stocks(clean_df) print(f筛选出 {len(filtered_df)} 条) top_df compute_top_stocks(filtered_df, top_n20) # 输出摘要 print_summary(top_df) # 保存报告 excel_path save_to_excel(filtered_df) md_path save_to_markdown(top_df) print(fExcel 报告已保存{excel_path}) print(fMarkdown 摘要已保存{md_path}) if __name__ __main__: main()4.8 运行与验证在项目根目录执行python main.py预期输出类似这样实际数值取决于当天行情正在拉取 A 股实时行情... 原始数据共 5394 条 正在清洗数据... 清洗后剩余 5318 条 正在按规则筛选... 筛选出 86 条 今日共筛选出 86 只股票涨幅前 5 code name price pct_change amount 300XXX 某某股份 12.35 8.12 7.360000e08 600XXX 某某科技 23.10 7.65 1.230000e09 ... Excel 报告已保存reports/stock_analysis_20250601.xlsx Markdown 摘要已保存reports/stock_analysis_20250601.md运行完成后reports目录下会生成对应的 Excel 和 Markdown 文件。Excel 可以用 WPS 或 Excel 直接打开Markdown 可以用于内部文档汇总。5. 定时自动化运行如果只是手动执行每天还要记得运行一次命令。更合理的做法是交给系统定时任务去跑。5.1 Windows 任务计划程序在 Windows 上可以使用任务计划程序在每天收盘后自动执行一次脚本。步骤如下打开“任务计划程序”点击“创建基本任务”。名称填写daily_stock_analysis。触发器选择“每天”设置时间例如 15:30A 股收盘后。操作选择“启动程序”。程序或脚本填写 Python 解释器路径例如C:\Python311\python.exe。添加参数填写main.py的绝对路径。起始于填写项目根目录例如D:\daily_stock_analysis。这样每天收盘后系统会自动运行脚本并生成报告。5.2 Linux crontab在 Linux 服务器上使用 crontab 实现定时运行crontab -e添加一行30 15 * * 1-5 cd /home/user/daily_stock_analysis /usr/bin/python3 main.py logs/run.log 21上面的配置表示周一到周五的 15:30 进入项目目录并运行脚本日志输出到logs/run.log。1-5代表周一到周五避开周末没有交易的情况。需要注意的是定时任务中的 Python 解释器路径可能和终端里的不同建议先用which python3查看完整路径。5.3 日志与错误捕获定时任务跑久了会遇到数据源偶尔超时、接口字段变化等问题。建议在入口文件里加上统一异常处理和日志记录。可以在main.py中增加日志输出import logging from datetime import datetime logging.basicConfig( filenameflogs/app_{datetime.now().strftime(%Y%m%d)}.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, )然后在main()里用 try-except 包裹核心逻辑避免中间一步失败导致后续全部中断def main(): try: run_analysis() except Exception as e: logging.exception(每日股票分析任务执行失败) raise这样即使某次运行失败也可以通过日志快速定位是数据源问题、字段问题还是本地环境问题。6. 常见问题与排查思路6.1 akshare 接口报错或返回空数据问题现象常见原因解决思路ak.stock_zh_a_spot_em()抛出异常akshare 版本过旧接口字段更新升级 aksharepip install --upgrade akshare返回 DataFrame 为空非交易时段或上游接口临时调整确认当前时间是否为交易时段稍后重试接口返回字段名发生变化数据源网页结构变更打印df.columns查看实际字段名更新清洗逻辑行情数据量过大导致内存占用高全市场快照数据超过数千行分市场获取或只拉取沪深 300 成分股akshare 属于社区维护的开源库接口变化相对频繁。如果某天脚本突然报错优先怀疑 akshare 版本问题而不是自己的代码逻辑。6.2 pandas 安装或导入失败常见原因是本地存在多个 Python 环境pip 安装到了错误环境。排查方式python -m pip install pandas python -c import pandas; print(pandas.__version__)如果安装很慢可以临时使用清华镜像pip install pandas -i https://pypi.tuna.tsinghua.edu.cn/simple6.3 Excel 文件打开乱码或格式异常如果使用df.to_csv()生成 CSV在 Excel 中打开可能出现中文乱码可以指定编码df.to_csv(output.csv, indexFalse, encodingutf-8-sig)如果使用pd.ExcelWriter报错缺少引擎说明没有安装 openpyxl执行pip install openpyxl6.4 明明有符合条件的股票却筛不出来这通常是字段单位问题。例如成交额有的是“元”有的是“万元”换手率有的是百分比数值有的是小数。建议在清洗步骤里统一单位并打印各字段的描述统计值print(df[[amount, turnover, float_mv]].describe())通过describe()可以快速看出数据量级是否合理避免因为单位问题导致筛选条件永远无法命中。6.5 定时任务没有输出文件定时任务执行失败时错误信息不会直接显示。排查步骤如下检查是否将日志输出到了文件。确认定时任务中的 Python 路径是否正确。确认项目路径是否正确脚本中是否存在相对路径问题。手动执行一次完整命令确认可以正常生成文件。7. 最佳实践与工程建议7.1 数据源选型目前 Python 生态常用的行情数据源有 akshare、tushare、yfinance 等选型时可以参考以下维度akshare免费、无需 token、接口丰富但接口稳定性一般需要关注版本更新。tushare需要注册并获取 token部分接口有积分限制数据质量稳定。yfinance覆盖全球市场但访问受网络环境影响A 股数据延迟可能较高。建议在代码中做一层数据源抽象让业务逻辑和数据源解耦。例如定义统一的get_realtime_data()接口内部根据配置决定调用 akshare 还是 tushare后续切换数据源时不需要改动筛选和报告逻辑。7.2 配置与敏感信息管理不要把 tushare token 等敏感信息写在代码里。推荐使用环境变量或.env文件import os from dotenv import load_dotenv load_dotenv() TUSHARE_TOKEN os.getenv(TUSHARE_TOKEN)同时在.gitignore中忽略.env文件避免提交到代码仓库。7.3 异常处理与告警交易数据获取类任务稳定性很重要。建议对网络超时设置重试机制最多重试 3 次。对关键节点增加日志输出。如果当天数据为空发送邮件或企业微信通知避免静默失败。输出文件落盘时使用日期命名保留历史归档。7.4 数据准确性校验在自动化任务中数据准确性比数据量更重要。建议在清洗后补充基础校验检查是否包含重复代码。检查最新价是否出现负数或 0。检查成交额是否超过合理范围。检查股票总数是否和前一天相差过大。如果校验失败宁可中断任务并告警也不要把错误数据写入报告。7.5 风险与合规提示这里要特别提醒股票分析脚本输出的所有指标、筛选结果仅用于技术研究和数据验证不构成任何投资建议。股市有风险实盘操作需要自行判断。对于生产环境的数据采集要注意合法合规个人学习场景下尽量控制请求频率避免对上游服务造成压力。不要用高并发方式反复请求免费接口。8. 总结与学习路线通过 daily_stock_analysis 这个项目你已经掌握了几个关键能力使用 Python 行情库获取 A 股实时数据。用 pandas 完成字段清洗、单位统一、缺失值删除。把筛选条件沉淀成配置文件保持逻辑与参数分离。输出 Excel 和 Markdown 两类报告。通过系统定时任务实现每日自动运行。对常见异常进行分类排查。下一步可以继续拓展的方向包括把每日行情数据写入 MySQL 或 SQLite积累历史数据。加入均线多头排列、MACD 金叉等更复杂的技术指标。接入企业微信机器人或飞书机器人每天推送报告摘要。用 Streamlit 做一个可视化行情看板。将项目容器化部署统一管理依赖。日常使用中优先关注数据源稳定性和筛选参数合理性这两点决定了自动化任务能否长期稳定运行。建议先跑通最小闭环再逐步叠加指标和推送能力。动手把项目跑起来才是真正掌握的开始。你可以先复制文中的config.py、data_fetcher.py、analyzer.py、reporter.py、main.py五个文件按照自己的需求调整筛选条件跑通之后再考虑定时任务和报告推送。