我最近把daily_stock_analysis这个开源项目完整跑了一遍又顺手在它的思路上做了不少改动。作为一个天天和A股行情打交道的人我太清楚每日复盘这件事有多折磨人了收盘后要拉全市场几千只股票的数据计算各种技术指标再对照自己的交易规则筛选异动标的最后整理成能看的报告。这套流程如果全靠手动每天至少耗掉一两个小时而且特别容易漏数据、算错复权。这个项目要解决的正是把每日股票数据分析这件高频、重复、规则相对固定的事情改造成一条自动化流水线。本文就围绕这个项目的架构、数据层设计、指标计算、筛选规则、定时任务和踩坑记录做一次尽可能完整的拆解。无论是想入门量化投研的Python开发者还是正在搭建个人复盘工具的交易者跟着这篇文章走一遍基本就能搭出自己版本的每日分析系统。1. 项目定位把每天收盘后的复盘工作自动化1.1 每日复盘到底在复盘什么别把每日复盘想得太玄乎它的本质就是回答几个固定问题今天全市场涨跌情况如何哪些股票放量了、突破了、走强了哪些板块在领涨我自己关注的股票池里有没有触发买入或卖出信号的票这些问题看起来简单但落到数据层面就有一堆细节。比如放量要怎么定义是量比大于2还是成交量超过5日均量的一倍突破是突破20日均线还是60日新高不同的人有不同的规则但绝大多数规则都建立在几个基础数据之上开高低收、成交量、成交额、换手率、涨跌幅。也就是说当日复盘的信息增量并不大真正的差异在于你把这几个字段加工成什么信号。daily_stock_analysis的设计思路就是先把这些基础数据收齐、算好指标再把你自己的规则写进筛选器里每天自动出一份结果。它不替你决策但把所有繁琐的数据工作接了过去。1.2 daily_stock_analysis的整体架构设计这个项目的整体架构并不复杂但分层很清晰我按自己的理解拆成了四层。第一层是数据采集层负责从公开行情源拉取全市场股票的快照数据与历史日线数据。第二层是数据存储层把拉到的原始数据清洗后落到本地数据库我用的是SQLite单机跑完全够用。第三层是计算分析层基于日线数据计算均线、MACD、RSI、量比、换手率等指标然后跑筛选规则输出符合条件的股票列表。第四层是输出层生成人可读的Markdown报告和机器可读的CSV文件同时配合定时任务实现无人值守。这样做的好处很明显每层之间通过数据格式解耦你换了数据源只需要改动采集团队的接口改了筛选规则不需要动数据层想换个报告模板更不用碰分析逻辑。我见过不少人的个人分析脚本把行情抓取、指标计算、结果打印全部糊在一个几百行的文件里跑起来没问题但每改一个需求都要提心吊胆。分层设计虽然前期多写一点代码后面省下的时间远超过投入。2. 数据层行情源选型、表结构与入库策略2.1 行情数据源怎么选数据源是整个系统最底层的依赖选不好后面全是坑。目前个人项目常用的免费方案有三类各有各的脾气。一类是akshare接口丰富、无需注册、开箱即用而且专门面向开源社区维护A股数据覆盖比较全。我目前的主力数据源就是它。缺点也有接口偶尔会因上游页面改版而失效你最好锁住版本别随便升级。另一类是tushare它的Pro版接口更稳定但需要注册并攒积分才能调用部分接口高频或大范围拉数据时积分门槛很现实。第三类是yfinance这类境外接口拿A股数据也能拿不过速度偏慢稳定性也一般更适合做多市场统一拉取。我的建议是主备结合主力用akshare同时把数据获取封装成独立函数万一它某个接口挂了可以快速切换到tushare的对应接口。这个切换成本在项目初期就该考虑进去而不是等服务崩了再改。封装的原则是对外只暴露fetch_daily(symbol, start_date, end_date) - DataFrame内部再按数据源分派。2.2 本地库表设计与增量更新全市场A股股票大概5000多只每天一条日线记录也就是每天新增5000多条数据。这个量级对SQLite来说几乎没有压力哪怕攒十年的日线数据也就几百万行的规模单机查询依然很快。所以本地存储不需要上MySQL、PostgreSQLSQLite一个文件搞定备份也方便。核心表结构我建议这样设计CREATE TABLE IF NOT EXISTS stock_daily ( trade_date TEXT NOT NULL, stock_code TEXT NOT NULL, stock_name TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL, volume REAL, amount REAL, amplitude REAL, pct_chg REAL, change REAL, turnover REAL, PRIMARY KEY (trade_date, stock_code) ); CREATE INDEX IF NOT EXISTS idx_stock_daily_code ON stock_daily(stock_code);用(trade_date, stock_code)做联合主键可以天然去重。很多时候接口重复拉取或者你手动重跑某一天的任务如果表里已有数据直接INSERT OR REPLACE就能幂等写入不会产生脏数据。更新策略上我分两步走。初次建库时从数据源拉过去三年左右的历史日线做一次全量灌入。之后每天收盘后只拉当天全市场快照以增量方式写入。这里要注意历史数据最好用前复权方式从行情接口一次性拉取而不是每天攒否则前期建库要跑很久而且复权基准不一致会导致历史价格对不上。2.3 数据获取层的统一封装数据获取层我单独放了一个data_fetcher.py核心就是几个函数。下面这段代码是我在实际项目里用的骨架精简了异常处理的细节import time import random import akshare as ak import pandas as pd def fetch_spot_all() - pd.DataFrame: 获取全市场A股实时快照返回标准化列名 raw ak.stock_zh_a_spot_em() raw.columns [ code, name, price, pct_chg, change, volume, amount, amplitude, high, low, open, pre_close, volume_ratio, turnover, pe, pb, total_mv, float_mv ] return raw def fetch_daily_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: 获取单只股票的历史日线前复权 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq ) df.columns [ trade_date, open, close, high, low, volume, amount, amplitude, pct_chg, change, turnover ] return df标准化列名这一步非常关键。akshare不同接口返回的列名并不统一有的是中文有的是英文有的带单位。如果你不在一开始做映射后面每个指标计算函数都要处理脏列名代码会越写越乱。我这个版本的列名统一用英文小写价格就是open/high/low/close成交量就是volume换手率是turnover后面所有计算模块都只认这套列名。3. 核心算法技术指标计算与异动筛选规则3.1 指标计算均线、MACD、RSI、量比数据入库之后下一步就是算指标。我的原则是指标层只做计算不做判断判断逻辑全部放到筛选层。这样指标可以被任何规则复用而不是和某条规则绑死。常用指标里均线、MACD、RSI、量比这四类是每日复盘最基础的。均线很简单就是滚动平均但要注意窗口的起点和对齐方式。MACD 涉及指数移动平均不要自己手写累加式的算法直接用pandas的ewm方法又快又不容易错。下面这段代码是我项目里的实际实现def calc_indicators(df: pd.DataFrame) - pd.DataFrame: 在日线数据基础上计算常用技术指标输入df必须按trade_date升序 df df.sort_values(trade_date).reset_index(dropTrue) # 均线 df[ma5] df[close].rolling(5).mean() df[ma10] df[close].rolling(10).mean() df[ma20] df[close].rolling(20).mean() df[ma60] df[close].rolling(60).mean() # MACD ema12 df[close].ewm(span12, adjustFalse).mean() ema26 df[close].ewm(span26, adjustFalse).mean() df[dif] ema12 - ema26 df[dea] df[dif].ewm(span9, adjustFalse).mean() df[macd] (df[dif] - df[dea]) * 2 # RSI(14) delta df[close].diff() gain delta.clip(lower0).rolling(14).mean() loss (-delta.clip(upper0)).rolling(14).mean() rs gain / loss df[rsi14] 100 - (100 / (1 rs)) # 量比当日成交量 / 过去5日平均成交量日线近似 df[volume_ma5] df[volume].shift(1).rolling(5).mean() df[volume_ratio] df[volume] / df[volume_ma5] return df这段代码里有几个细节要特别注意。第一MACD公式里的柱状值国内软件一般显示为(DIF - DEA) * 2这样才和同花顺、通达信的数字对得上否则你对比行情软件时会发现数值总差一倍。第二RSI里的gain和loss建议用rolling(14).mean()做简单平均而不是 Wilder 的平滑方式两者结果有差异但对筛选规则影响不大简单平均更容易理解。第三量比的标准定义是当日每分钟平均成交量比过去5日每分钟平均成交量日线层面没有分钟数据我用当日成交量比过去5日平均日成交量做近似逻辑上等价响应稍慢一点但足够用。3.2 筛选规则怎么落地成代码指标算完以后筛选层就轻松了。我建议把每一条规则都写成一个独立的函数输入是带指标的DataFrame输出是布尔Series。这样新增规则就是在文件里加一个函数改规则就是改这个函数的内部逻辑不会互相影响。我的项目里初始写了五条规则覆盖了最常见的几种打法规则名触发条件逻辑说明放量突破close ma20 且 volume_ratio 2 且 pct_chg 3量价配合突破中期均线均线多头ma5 ma10 ma20 ma60短中长均线多头排列零轴金叉dif 上穿 dea 且 dif 0MACD零轴上方金叉趋势转强创60日新高close 过去60日最高价突破前期平台连续放量连续3天 volume_ratio 1.5资金持续关注其中零轴金叉需要用到前一天的dif和dea所以不能只算当天的数据必须把历史指标也算出来再取最后一行的结果。这也是为什么指标计算要放在筛选之前、并且要用完整历史序列来算的根本原因。代码实现大致是这样的结构def rule_breakout_ma20(row: pd.Series) - bool: return row[close] row[ma20] and row[volume_ratio] 2 and row[pct_chg] 3 def rule_macd_gold_cross(hist: pd.DataFrame) - bool: last hist.iloc[-1] prev hist.iloc[-2] return prev[dif] prev[dea] and last[dif] last[dea] and last[dif] 0筛选时先把每只股票最近一段历史都跑一遍calc_indicators然后取最后一行判断条件。注意rule_macd_gold_cross这类规则依赖前一行数据所以历史窗口不能只取最近几天建议至少取60个交易日保证均线和MACD都进入稳定状态。3.3 为什么盘后筛选比盘中判定更稳我最早也想过盘中实时监控后来发现拿免费数据源做盘中判断痛点非常多。第一数据延迟严重免费接口的实时快照往往滞后一两分钟对短线信号影响很大。第二盘中价格波动剧烈一个放量突破信号可能在上午10点触发下午又跌回去了每天报警一堆实际能用的没几个。第三你没法在盘中验证信号质量因为当时的均线、量比都在动态变化。daily_stock_analysis的定位是盘后分析我认为这是非常正确的选择。盘后数据是确定的、完整的算法跑出来的结果可复现、可回测。你可以把某一天筛选出来的股票列表保存下来过几天再回头看哪些真的涨了信号胜率是多少。这种可验证性是做自动化分析的基础。如果真需要盘中提醒也应该基于这个框架在收盘前或午间休市时各跑一次而不是实时盯盘。4. 自动化落地报告生成与定时任务4.1 报告结构怎么设计才适合每天看报告是每天分析结果的出口设计得不好再准的信号也没人愿意看。我的报告分四块市场概况、涨停跌停统计、筛选结果、自选池跟踪。每天都看同样的结构信息密度高检索也方便。市场概况部分放涨跌家数、成交额、上涨占比、涨停跌停数量。涨停跌停的地域和板块分布能让你快速知道今天资金偏好。筛选结果部分是核心每个触发规则的股票都要给出代码、名称、收盘价、涨跌幅、换手率、量比以及触发了哪条规则。自选池部分单独拎出来把你自己关注的股票列成表跟踪它们是不是出现了新信号。输出格式我用了Markdown和CSV双轨。Markdown报告给人看直接渲染或放到笔记软件里都方便CSV文件给程序用方便后续统计和分析。报告文件按日期命名比如report_20250110.md每天一个文件不覆盖历史记录。下面是一个精简版的Markdown报告片段# 每日股票分析报告 2025-01-10 ## 市场概况 - 上涨家数: 3210 / 下跌家数: 1840 - 成交额: 9876亿 / 涨停: 45 / 跌停: 12 ## 触发信号个股 | 代码 | 名称 | 收盘价 | 涨跌幅 | 换手率 | 量比 | 触发规则 | | --- | --- | --- | --- | --- | --- | --- | | 600000 | 示例股份 | 10.25 | 5.02% | 3.21 | 2.45 | 放量突破 |报告里不写任何主观建议只列事实和信号因为主观判断应该留给使用者规则系统只负责把符合条件的东西挑出来。4.2 定时任务的正确打开方式Linux环境下每天下午收盘后自动跑一次分析最靠谱的方式是crontab。不是schedule库不好而是脚本一旦退出依赖进程内调度器的方案就全失效了。系统级crontab可以保证只要机器开着任务就会执行。我的crontab配置长这样30 15 * * 1-5 cd /home/user/daily_stock_analysis /usr/bin/python3 main.py logs/run.log 21每周一到周五的15点30分执行一次主程序。这里强调一个问题数据更新时间窗口。A股15点收盘但行情接口的当日数据并不是15点整就立刻完整可用的上游数据源还要做清算、校对。我个人实测下来15点30分跑成功率比较高但偶尔也会遇到个别股票数据缺失。为了解决这个问题我在主程序里加了一个补拉逻辑跑完第一次后如果检测到涨停股票数量异常偏少或某只重点股票当日数据为空就间隔15分钟自动重试一次最多重试3次。Windows用户也别担心用任务计划程序就能实现同样的效果。关键是触发时间要定在收盘后半小时以上别卡在15点整这是个用时间换稳定性的取舍。4.3 报告留档与数据积累很多个人复盘工具跑完当天就完事了报告也不留数据也不存回头想统计过去一个月的胜率发现无据可查。我在这个项目里养成了一个习惯报告和数据都做长周期留档。报告按日期存到reports/目录数据继续按主键(trade_date, stock_code)写入stock_daily表。每周末可以对本周信号做一次回看哪些股票触发信号后次日继续上涨哪些是假突破。这种统计不需要复杂工具直接用SQL按日期范围聚合就能出结果。数据积累的时间越长你对筛选规则的信赖程度就越有依据——你会发现有些规则在震荡市里完全失效在趋势市里胜率很高这种经验是复盘软件给不了你的。5. 实操避坑复权、停牌、限流与数据对齐5.1 复权方式选错指标全部失真这是我在复现过程中踩过最深的坑也是很多新手最容易忽略的问题。股票分红除息或送转除权后价格会突然向下跳空。如果不做复权处理收盘价就会出现一个毫无意义的断裂均线和MACD会被这个断裂严重扭曲。用前复权数据可以解决这个问题。前复权是保持最新价格不变把历史价格按照分红比例向下调整这样技术指标的形态就连续了。在akshare里拉历史日线时直接指定adjustqfq。但这里有一个需要留意的地方前复权的历史价格会随每次分红事件而改变基准你上个月拉的历史价格和这个月拉的历史价格可能不一样。所以历史数据入库后不要频繁做全量重建建议以增量更新为主每季度或每年做一次全量重算即可。5.2 停牌、ST、新股对统计结果的干扰全市场数据里混着大量无效样本。比如停牌股票当天没有交易成交量是0却依然会返回一条快照记录。这样的股票如果进入筛选逻辑volume_ratio会被算成0或无穷大各种统计也会被带偏。所以数据清洗的第一条铁律是过滤掉成交量小于等于0的股票。ST股票也要特殊处理。很多策略本身就是基于趋势跟踪的ST股的涨跌幅限制不一样而且基本面风险高若没有特殊需求建议直接排除。新股则是另一个大坑上市前五个交易日不设涨跌幅限制波动极大而且刚上市时没有足够的均线历史数据计算出来的指标没有任何参考意义。我一般在数据清洗阶段增加两个字段一个是is_st一个days_since_listing上市天数然后在筛选时统一排除。新增样本过滤逻辑def valid_universe(df: pd.DataFrame) - pd.DataFrame: 过滤停牌、ST、上市不足60日的股票 df df[df[volume] 0] df df[~df[name].str.contains(ST, caseFalse, naFalse)] df df[df[days_since_listing] 60] return df5.3 接口限流与失败重试免费数据接口都有频率限制。akshare的实时快照接口一次性返回全市场数据压力已经比较大了你在循环里逐只拉历史数据时一定要控制节奏。我的做法是每只股票之间随机暂停0.1到0.3秒避免固定频率触发上游风控。另一个容易忽略的问题是接口超时和偶发失败。网络抖动、上游临时维护都会导致某只股票的数据拉取失败。我的data_fetcher里封装了一个带重试的下载函数最多重试3次每次失败后 sleep 2 秒3次都失败就把这只股票记入失败日志跳过继续拉下一只。第二天启动时程序会先读取失败日志里的股票代码尝试补拉前一天缺失的数据这样就保证不会因为一只股票卡住整个流程。def fetch_with_retry(symbol: str, start: str, end: str, retries: int 3): for i in range(retries): try: return fetch_daily_history(symbol, start, end) except Exception as e: if i retries - 1: raise time.sleep(2)5.4 收盘后数据更新的时间窗口免费行情源的数据更新通常不是一次到位而是分批完成的。我的实测经验是15点收盘后大部分数据在15点10分左右已经可用但要等到所有股票的数据都齐尤其是中小市值和流动性差的股票往往要到15点30分以后。如果定时任务设在15点整你会在涨停统计里看到明显的缺失比如某只股票收盘明明涨停了快照里还是一个未收盘的价格。项目里我加了一个数据完整性检查模块。跑完主流程后统计当日写入的股票数量如果明显少于全市场股票总量就判定这次任务数据不完整记录状态等待重跑。重跑时程序会读取stock_daily表里当日已有的记录只补拉缺失的部分不会重复处理全部数据。这种设计虽然让代码复杂度上了一个台阶但保证了每天一定有一份完整报告这个底线。6. 后续演进从每日分析到更大框架6.1 规则参数化与多策略并存项目跑通之后你会发现每天最想改的是各种阈值 — 量比大于2还是大于2.5突破20日均线还是30日均线如果这些参数写死在代码里每调整一次都要改代码时间长了非常难受。我的做法是把规则和参数全部外置到config.yaml配置文件里。主程序启动时读取配置动态组装筛选规则。想试验新参数修改配置文件后重新运行一次即可不用动代码。更进一步你可以同时运行多套参数组合把结果分别存到不同目录周末统一比较哪套参数在本周表现更好。6.2 把信号接到通知渠道有人习惯每天早上打开电脑看报告也有人希望报告一出就自动推送到手机。在这个项目基础上扩展通知功能并不复杂。最简单的方案是接入企业微信机器人或钉钉机器人。只需要一个Webhook地址把当天筛选出的股票列表拼成文本用requests.post发过去即可。更轻量的方案是邮件Python标准库里的smtplib就能发但要注意邮件容易被当作垃圾邮件。我个人的习惯是重要信号用即时通知完整报告留在本地每天统一看。不要把每个信号都实时推送否则一天推几十条跟没推一样人会产生通知疲劳。6.3 向回测和组合管理延伸当stock_daily表积累了半年以上的数据后你其实已经拥有了一份可回测的数据资产。把筛选信号和未来N日收益关联起来就能评估每条规则的历史胜率。这一步不需要引入专门的回测框架用pandas就能做简单的事件研究对每个信号记录触发日收盘价然后计算未来3日、5日、10日的收益。按规则分组统计平均收益和胜率你就能快速淘汰掉那些看起来很美的规则。有了这一步daily_stock_analysis就从每日工具升级成了策略研究平台。不过要提醒一句历史胜率不代表未来表现尤其在市场环境发生大的变化时过去有效的规则完全可能失效这属于策略迭代的常态。6.4 开源协作时需要注意的事如果打算把这类项目开源有几点建议。第一不要在仓库里存放任何私人token或数据库文件配置项全部用环境变量或.env文件管理。第二akshare这类上游接口变动比较频繁项目里最好固定依赖版本并在README里注明测试时间。第三一定要把数据清洗规则写清楚因为不同使用者对ST、新股、复权方式的处理可能完全不同这些规则直接影响分析结果。我在实际复现这个项目的时候最大的感受是数据层的工程化程度决定了整个项目的上限。指标公式人人都会写筛选规则也不难难的是把几千只股票的数据每天稳定、干净、准点地准备好。复权、停牌、限流、重试这些看似琐碎的细节恰恰是项目能否长期跑下去的关键。如果你也想搭一套自己的每日股票分析系统我建议从今天开始先跑通一条最简单的流程连续记录两周再逐步加规则、加指标、加通知。数据积累得越久这套系统的价值就越大。