写在前面很多刚接触量化的朋友把 90% 的精力花在盘中盯盘上看着分时图心跳加速涨了欢天喜地跌了如坐针毡。但残酷的真相是——盘中的波动大多是噪音真正决定你账户长期走向的是收盘后那 30 分钟做了什么。我自己的 A 股量化系统跑了一段时间后把盘后流程彻底自动化了15:30 收盘后调度器自动做几件事——盘后选股、信号复盘、模拟盘 CSV 导出、夜间统计。本文把这几块的核心代码拆出来全部可运行你可以直接抄进自己的项目。环境Python 3.9依赖pandas / numpy。下面所有代码都用模拟数据复制即可跑。系统整体纯本地离线运行行情用三级缓存不依赖任何云账号。一、盘后选股10 种模式 三层信号 多因子打分盘后选股的本质是给全市场股票打分分数高的排前面。我的系统选股引擎内置10 种模式ma_cross质量股 MA5 上穿 MA30、动量突破、低估值修复、资金流入、均线多头、回调企稳、行业轮动、高股息、北向增持、破净修复——覆盖不同市况下的机会类型而不是一个模式打天下。以 ma_cross 为例它要过三层信号才入选① 价格结构MA5 上穿 MA30② 成交量确认突破日放量③ 质量门槛ROE / 市盈率达标。单指标触发不进避免假突破噪音。通过筛选的票再进入多因子打分排序。因子说明估值因子PE 越低越好、质量因子ROE 越高越好、动量因子MA5 上穿 MA30 给加分、资金流因子当日主力净流入给加分。每个因子归一化后加权求和。# -*- coding: utf-8 -*-import pandas as pdimport numpy as npdef score_stock(row, weightsNone):对单只股票做多因子打分返回 0~100 的综合分w weights or {pe: 0.25, roe: 0.25, momentum: 0.3, flow: 0.2}score 0.0# 估值因子PE 越低分越高限制在 0~1pe_score max(0.0, min(1.0, (40 - row[pe]) / 40)) if row[pe] 0 else 0.0score w[pe] * pe_score * 100# 质量因子ROE 越高分越高roe_score max(0.0, min(1.0, row[roe] / 30))score w[roe] * roe_score * 100# 动量因子MA5 MA30 给满分否则按比例mom_score 1.0 if row[ma5] row[ma30] else 0.3score w[momentum] * mom_score * 100# 资金流因子主力净流入为正给分flow_score max(0.0, min(1.0, row[main_net_in] / 1e8))score w[flow] * flow_score * 100return round(score, 2)# 模拟一篮子股票数据data pd.DataFrame([{code: 600519, pe: 28, roe: 25, ma5: 1720, ma30: 1680, main_net_in: 1.2e8},{code: 000858, pe: 18, roe: 22, ma5: 150, ma30: 155, main_net_in: -0.4e8},{code: 300750, pe: 35, roe: 18, ma5: 210, ma30: 205, main_net_in: 0.8e8},{code: 601318, pe: 8, roe: 12, ma5: 48, ma30: 46, main_net_in: 0.5e8},])data[score] data.apply(score_stock, axis1)top_n data.sort_values(score, ascendingFalse).head(3)print(盘后选股 Top3)print(top_n[[code, score]].to_string(indexFalse))运行后你会得到一份按分数排序的清单。真实系统里这部分接的是三级缓存内存 → SQLite → API盘后批量计算不占用交易时间也不会触发行情接口限流。 数据来源船长自研量化系统选股引擎10 种模式实测逻辑2026 年因子权重可在配置文件按需调整无需改代码三层信号过滤逻辑见 screener 源码。二、信号复盘用 pandas 按策略统计胜率选完股只是第一步。更关键的是你得知道哪些策略在说谎。我的系统每天把每个策略产生的真实信号记下来盘后统一复盘。下面用 pandas 演示按策略统计胜率的标准做法。胜率定义某策略发出的信号中后续 N 日该股票涨幅超过阈值的比例。这里用groupby聚合一行就能算出来。# -*- coding: utf-8 -*-import pandas as pd# 模拟某日各策略产生的信号流水# hit1 表示信号发出后达标视为命中0 表示未达标votes pd.DataFrame([{strategy: ma_cross, code: 600519, hit: 1},{strategy: ma_cross, code: 000858, hit: 0},{strategy: ma_cross, code: 601318, hit: 1},{strategy: rsi, code: 300750, hit: 0},{strategy: rsi, code: 600519, hit: 0},{strategy: macd, code: 000858, hit: 1},{strategy: macd, code: 300750, hit: 1},])# 按策略分组统计信号数与命中数算胜率review (votes.groupby(strategy).agg(signals(hit, count), hits(hit, sum)).reset_index())review[win_rate] (review[hits] / review[signals]).round(3)# 按胜率排序一眼看出谁在偷懒review review.sort_values(win_rate, ascendingFalse)print(信号复盘按策略胜率)print(review.to_string(indexFalse))输出会告诉你macd胜率 100%、ma_cross约 67%、rsi0%。这时候你就该警惕了——一个策略如果长期 0 命中要么参数坏了要么它适用的市场状态已经变了。这正是信号复盘比单纯看回测更有价值的地方它用真实信号反推回测假设。 数据来源船长自研量化系统信号复盘模块实测统计2026 年胜率统计支持按策略 / 按时间 / 按日期三种维度。三、模拟盘 CSV 导出把每天的交易留痕盘后流程的关键一步是把模拟盘当天的交易导出成 CSV。每一笔成交的进场价、出场价、止损触发点、持仓时长都在里面第二天早上喝咖啡时复盘用。下面演示用paper_broker的成交记录构造 DataFrame 并导出。真实系统里仓位、盈亏、手续费全部按真实规则算佣金万三 单边印花税唯一区别是钱是假的——但每一笔决策都按真实世界的规则来算。# -*- coding: utf-8 -*-import pandas as pdfrom datetime import datetime, timedeltadef export_paper_trades(trades, pathpaper_trades.csv):把模拟盘成交记录导出为 CSV 报告rows []for t in trades:pnl_pct (t[exit] - t[entry]) / t[entry]hold (t[exit_time] - t[entry_time]).total_seconds() / 3600rows.append({code: t[code],entry: t[entry],exit: t[exit],pnl_pct: round(pnl_pct, 4),trigger: t[trigger], # stop_loss / take_profit / trailinghold_hours: round(hold, 1),})df pd.DataFrame(rows)df.to_csv(path, indexFalse, encodingutf-8-sig)return df# 模拟当天 2 笔成交trades [{code: 600519, entry: 1680, exit: 1712, trigger: take_profit,entry_time: datetime(2026,7,10,9,35), exit_time: datetime(2026,7,10,14,20)},{code: 000858, entry: 155, exit: 148, trigger: stop_loss,entry_time: datetime(2026,7,10,10,2), exit_time: datetime(2026,7,10,13,50)},]df export_paper_trades(trades)print(模拟盘当日报告)print(df.to_string(indexFalse))导出后用utf-8-sig编码Excel 打开中文不会乱码。复盘时重点看两列trigger是止损还是止盈触发检验风控有没有生效和hold_hours持仓时长太短说明追涨杀跌太长说明该止盈没止。连续亏损 3 次系统会自动暂停交易——拦截的是忍不住补仓的冲动不是机会。四、数据层三级缓存抗限流盘后要计算几千只股票最怕两件事外部接口限流、盘中抢资源。我的数据层用三级缓存把这俩都摁住了逻辑非常直白# -*- coding: utf-8 -*-def get_price(code):三级缓存内存 - 磁盘(SQLite) - API# 1. 内存缓存热点数据零延迟if code in mem_cache:return mem_cache[code]# 2. 磁盘缓存SQLite2 小时有效期row db.query(SELECT close, ts FROM kline WHERE code?, code)if row and (now - row[ts]).seconds 7200:mem_cache[code] row[close]return row[close]# 3. API 兜底严格控频避免新浪限流data api.fetch(code) # 带 1.5s 翻页间隔db.execute(UPSERT kline ..., data)mem_cache[code] data[close]return data[close]内存没有的去 SQLite 取前两级都没有才请求外部接口。盘后批量计算走这套机制不占用交易时间也不怕新浪限流把任务卡死。这是离线优先设计的核心——系统 7×24 能跑靠的就是不依赖任何单一在线数据源。五、把一切串起来盘后自动化调度几块逻辑各自独立靠一个调度器在 15:30 串起来。核心就是一个post_market()入口按顺序调用选股、复盘、导出。调度本身用 APScheduler 的 cron 触发器和手动点一下说再见# -*- coding: utf-8 -*-from apscheduler.schedulers.blocking import BlockingSchedulerdef post_market():盘后自动化主入口15:30 触发# 1. 盘后选股取 Top N 推送到控制台picks run_screener(top_n10) # 见第一节 score_stockpush_to_console(picks)# 2. 信号复盘按策略统计胜率review review_by_strategy(load_today_votes()) # 见第二节log_review(review)# 3. 模拟盘当日 CSV 导出export_paper_trades(load_today_trades()) # 见第三节print(盘后自动化完成选股 %d 只复盘 %d 策略导出成交 %d 笔% (len(picks), len(review), len(load_today_trades())))# 每个交易日 15:30 自动触发cron分 时 日 月 周sched BlockingScheduler()sched.add_job(post_market, cron, minute30, hour15, day_of_week0-4)sched.start() # 进程常驻收盘后自动干活把这几步自动化之后你的角色就变了盘中是系统的盘后是你的。你不再被分时图绑架而是每天用 30 分钟看着系统交出来的清单做判断。更重要的是系统不会今天累了不想看就漏掉机会也不会情绪上头就乱买——纪律是自动化最被低估的优势。船长的话散户最贵的成本不是本金是注意力。把盯盘的 4 小时交给系统把复盘的那 30 分钟留给自己才是量化真正的意义。工具干工具的活人做人的判断——别反过来。你省下的不是时间是少做蠢决定的机会。完整工程我放在了量化系统项目里选股引擎 信号复盘 模拟盘模块 Web 控制台纯本地离线运行。如果你也在搭自己的盘后流水线上面几段代码就是最小可用骨架先跑通再逐步加因子、加模式。觉得有用点个赞 收藏下次盘后不知道干嘛的时候翻出来照着改。