收盘后30分钟比盯盘4小时更重要：用Python实现量化盘后自动化（多因子选股+信号复盘+CSV导出实战）
写在前面
很多刚接触量化的朋友，把 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 给加分）、资金流因子（当日主力净流入给加分）。每个因子归一化后加权求和。
import pandas as pd
import numpy as np
def score_stock(row, weights=None):
"""对单只股票做多因子打分，返回 0~100 的综合分"""
w = weights or {"pe": 0.25, "roe": 0.25, "momentum": 0.3, "flow": 0.2}
score = 0.0
# 估值因子：PE 越低分越高（限制在 0~1）
pe_score = max(0.0, min(1.0, (40 - row["pe"]) / 40)) if row["pe"] > 0 else 0.0
score += 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.3
score += w["momentum"] * mom_score * 100
# 资金流因子：主力净流入为正给分
flow_score = max(0.0, min(1.0, row["main_net_in"] / 1e8))
score += w["flow"] * flow_score * 100
return 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, axis=1)
top_n = data.sort_values("score", ascending=False).head(3)
print("盘后选股 Top3：")
print(top_n[["code", "score"]].to_string(index=False))
运行后你会得到一份按分数排序的清单。真实系统里，这部分接的是三级缓存（内存 → SQLite → API），盘后批量计算不占用交易时间，也不会触发行情接口限流。
📌 数据来源：船长自研量化系统选股引擎（10 种模式）实测逻辑（2026 年）；因子权重可在配置文件按需调整，无需改代码；三层信号过滤逻辑见 screener 源码。
二、信号复盘：用 pandas 按策略统计胜率
选完股只是第一步。更关键的是：你得知道哪些策略在说谎。我的系统每天把每个策略产生的真实信号记下来，盘后统一复盘。下面用 pandas 演示"按策略统计胜率"的标准做法。
胜率定义：某策略发出的信号中，后续 N 日该股票涨幅超过阈值的比例。这里用groupby聚合，一行就能算出来。
import pandas as pd
# 模拟某日各策略产生的信号流水
# hit=1 表示信号发出后达标（视为命中），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", ascending=False)
print("信号复盘（按策略胜率）：")
print(review.to_string(index=False))
输出会告诉你：macd胜率 100%、ma_cross约 67%、rsi0%。这时候你就该警惕了——一个策略如果长期 0 命中，要么参数坏了，要么它适用的市场状态已经变了。这正是"信号复盘"比单纯看回测更有价值的地方：它用真实信号反推回测假设。
📌 数据来源：船长自研量化系统信号复盘模块实测统计（2026 年）；胜率统计支持按策略 / 按时间 / 按日期三种维度。
三、模拟盘 CSV 导出：把每天的交易留痕
盘后流程的关键一步，是把模拟盘当天的交易导出成 CSV。每一笔成交的进场价、出场价、止损触发点、持仓时长都在里面，第二天早上喝咖啡时复盘用。
下面演示用paper_broker的成交记录构造 DataFrame 并导出。真实系统里，仓位、盈亏、手续费全部按真实规则算（佣金万三 + 单边印花税），唯一区别是钱是假的——但每一笔决策都按真实世界的规则来算。
import pandas as pd
from datetime import datetime, timedelta
def export_paper_trades(trades, path="paper_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() / 3600
rows.append({
"code": t["code"],
"entry": t["entry"],
"exit": t["exit"],
"pnl_pct": round(pnl_pct, 4),
"trigger": t["trigger"], # stop_loss / take_profit / trailing
"hold_hours": round(hold, 1),
})
df = pd.DataFrame(rows)
df.to_csv(path, index=False, encoding="utf-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(index=False))
导出后用utf-8-sig编码，Excel 打开中文不会乱码。复盘时重点看两列：trigger（是止损还是止盈触发，检验风控有没有生效）和hold_hours（持仓时长，太短说明追涨杀跌，太长说明该止盈没止）。连续亏损 3 次，系统会自动暂停交易——拦截的是"忍不住补仓"的冲动，不是机会。
四、数据层：三级缓存抗限流
盘后要计算几千只股票，最怕两件事：外部接口限流、盘中抢资源。我的数据层用三级缓存把这俩都摁住了，逻辑非常直白：
def get_price(code):
"""三级缓存：内存 -> 磁盘(SQLite) -> API"""
# 1. 内存缓存（热点数据，零延迟）
if code in mem_cache:
return mem_cache[code]
# 2. 磁盘缓存（SQLite，2 小时有效期）
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 触发器，和"手动点一下"说再见：
from apscheduler.schedulers.blocking import BlockingScheduler
def post_market():
"""盘后自动化主入口：15:30 触发"""
# 1. 盘后选股，取 Top N 推送到控制台
picks = run_screener(top_n=10) # 见第一节 score_stock
push_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", minute=30, hour=15, day_of_week="0-4")
sched.start() # 进程常驻，收盘后自动干活
把这几步自动化之后，你的角色就变了：盘中是系统的，盘后是你的。你不再被分时图绑架，而是每天用 30 分钟，看着系统交出来的清单做判断。更重要的是，系统不会"今天累了不想看"就漏掉机会，也不会"情绪上头"就乱买——纪律，是自动化最被低估的优势。
船长的话：散户最贵的成本不是本金，是注意力。把盯盘的 4 小时交给系统，把复盘的那 30 分钟留给自己，才是量化真正的意义。工具干工具的活，人做人的判断——别反过来。你省下的不是时间，是少做蠢决定的机会。
完整工程我放在了量化系统项目里（选股引擎 + 信号复盘 + 模拟盘模块 + Web 控制台，纯本地离线运行）。如果你也在搭自己的盘后流水线，上面几段代码就是最小可用骨架，先跑通再逐步加因子、加模式。
觉得有用，点个赞 + 收藏，下次盘后不知道干嘛的时候翻出来照着改。