这次我们来看一条典型的股市收评：26 年 8 月 12 日，指数和个股的位置比较关键，只要这个位置不破，反弹还有延续空间，操作上不追高、不满仓，而是用移动跟踪防守的方式拿着。一句话看起来是盘面观点，但背后其实藏着一套可以程序化的复盘逻辑：什么位置算关键位置，怎么判断“破了”还是“没破”，移动防守线怎么跟着价格动，以及每天收盘后如何快速生成一份带结论的收评草稿。
本文不搞行情预测，也不构成任何投资建议。我要做的是把这条收评里的“位置不破 + 移动跟踪防守”翻译成可执行的技术规则，并给出一套用 Python 实现的“收盘自动复盘”工作流。熟悉数据清洗、指标计算和定时任务的读者，可以直接把代码改成自己的策略模板；不做交易的朋友，也可以把它当成一个数据处理和报告生成的练手项目。
先给结论，这套工作流适合谁：已经有一套看盘方法、但不想每天手工翻几十只股票的人；想把自己口头上的“支撑位、止损位、反弹延续”变成确定性规则的人；以及想学 Python 数据分析，拿 A 股日线数据做复盘项目的人。不适合谁：想靠代码自动赚钱、期望输出直接变成买卖信号的人，这类需求风险极高，脚本做不了风控。下面的内容先讲规则怎么落地，再讲代码怎么写，最后讲怎么批量和排错。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 项目类型 | 股市收评复盘辅助工具，把口头观点规则化 |
| 输入数据 | 指数/个股日线数据，包含日期、代码、开高低收、成交量 |
| 核心功能 | 关键位置判定、支撑压力计算、移动跟踪防守位、收评草稿生成、批量个股扫描 |
| 运行方式 | Python 脚本 + 定时任务 |
| 是否支持批量任务 | 支持，按股票列表批量扫描并输出 Markdown 日报 |
| 是否支持接口 API | 不固定，可以将报告生成模块封装成 HTTP 服务 |
| 硬件要求 | 无 GPU 需求，普通 CPU 即可 |
| 输出格式 | 控制台日志、Markdown 收评草稿、CSV 跟踪清单 |
| 适合场景 | 每日收盘复盘、个股止损跟踪、收评草稿素材整理 |
再看这里的策略参数，建议一开始用一组保守的默认值：
| 参数 | 默认值 | 作用 |
|---|---|---|
| 枢轴点周期 | 当日 | 计算 P、R1、S1 三个基础位置 |
| 关键位置确认 | 收盘价 vs S1 | 收盘不破 S1 视为反弹延续条件仍在 |
| 跌破确认 | 连续两日收盘低于 S1 | 降低单日假突破的干扰 |
| 移动跟踪方式 | ATR 倍数跟踪 | 防守位随价格上移，不回退 |
| 批量扫描范围 | 自定义股票列表 | 只处理关注清单，不扫全市场 |
这套参数不是唯一答案。不同行情、不同波动率下，S1 和 ATR 倍数都可能要调整。先把流程跑通，再根据自己观察的经验改默认值。
2. 适用场景与使用边界
先说清楚，这不是自动交易系统，更不是荐股工具。它解决的核心问题是：每天收盘后，你不需要再几十只股票来回翻，只需要运行一个脚本，脚本会把每只股票的关键位置、是否站稳、移动防守位全部算好，并生成一段可以改写的收评草稿。
适合用这套流程的场景包括：
- 盘后复盘：查看关注列表里哪些股票还站在关键防守位上方，哪些已经跌破。
- 止损跟踪：给每一只持仓标的维护一个移动防守位，收盘后检查是否触发。
- 收评素材整理：把指数和核心个股的位置判断汇总成 Markdown 草稿，再人工补充盘面细节。
不适合的场景也要讲清楚：
- 盘中实时交易信号：本文用的是日线收盘数据，最早只能在收盘后确认，不适合做盘中抢反弹。
- 替代基本面研究：技术位置只是复盘的一部分，公司基本面、行业变化、市场风险都不在脚本范围内。
- 无风控的自动执行：即便脚本算出“跌破防守位”，最终是否卖出仍要人去判断，不能直接接券商接口乱下单。
合规方面，所有行情数据来源要确认授权，不要爬取来路不明的付费数据。文章里所有代码只是技术示例，输出结果不能直接作为买卖依据。尤其涉及个股时，不要因为脚本说“位置未破”就认为一定会上涨。
3. 环境准备与前置条件
开始之前，先把工程目录和 Python 环境准备好。操作系统不限，Windows、Linux、macOS 都可以跑；唯一要求是能够正常安装 pandas、numpy、requests 这类常用库。
建议建一个独立目录，避免脚本文件散落：
stock_review/ ├── data/ # 行情数据缓存 ├── output/ # 收评草稿与跟踪表 ├── config.yaml # 参数配置 ├── load_data.py # 数据读取与清洗 ├── indicators.py # 关键位置和移动防守计算 ├── generate_report.py # 收评草稿生成与批量扫描 ├── run_daily.py # 每日执行入口 └── requirements.txt # Python 依赖Python 环境建议使用虚拟环境：
python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate然后安装基础依赖，版本使用当前最新稳定版即可：
pip install pandas numpy requests openpyxl如果你的行情数据源需要额外 SDK，按数据源官方文档安装。本文后续代码尽量只依赖 pandas 和 numpy，方便你替换成自己的数据接口。
4. 收盘数据获取与本地缓存
真实行情数据的获取方式很多，常见的是从数据接口订阅日线数据，也可以手动导出 CSV。为了让流程可测试，我先定义一个统一的数据读取函数。标准字段如下：
| 字段 | 类型 | 说明 |
|---|---|---|
| date | datetime | 交易日期 |
| code | str | 股票或指数代码 |
| name | str | 名称 |
| open | float | 开盘价 |
| high | float | 最高价 |
| low | float | 最低价 |
| close | float | 收盘价 |
| volume | float | 成交量 |
下面是 load_data.py，支持传入内存中的 DataFrame，也支持读取本地缓存文件：
import pandas as pd from pathlib import Path DATA_DIR = Path("data") CACHE_FILE = DATA_DIR / "daily_close.csv" def load_close_data(source_df=None) -> pd.DataFrame: """ 把行情数据源返回的数据统一成标准 DataFrame。 如果 source_df 为空，则读取本地缓存 CSV， 方便在没有真实数据源时用模拟数据测试流程。 真实使用时，把数据源返回结果传给 source_df 即可。 标准列：date, code, name, open, high, low, close, volume """ if source_df is not None: df = source_df.copy() else: df = pd.read_csv(CACHE_FILE, parse_dates=["date"]) # 统一列名，缺失值直接报错，避免后续计算静默出错 required_cols = ["date", "code", "name", "open", "high", "low", "close", "volume"] for col in required_cols: if col not in df.columns: raise ValueError(f"数据缺少必要字段: {col}") df = df.sort_values(["code", "date"]).reset_index(drop=True) df = df.drop_duplicates(subset=["code", "date"], keep="last") return df if __name__ == "__main__": df = load_close_data() print(df.tail())这段代码会在拿到数据后按“代码 + 日期”排序，并去掉重复行。真实数据接入时，优先在前端处理好复权问题。计算技术指标通常建议用前复权数据，否则除权当天会出现价格跳空，导致防守位失真。
5. 关键位置与移动跟踪防守计算
这是整篇的核心。收评里说的“这个位置不破”，需要先定义“位置”是什么。
5.1 用枢轴点计算当日关键位置
一种常见方法是用枢轴点（Pivot Point）体系。以当日的最高价、最低价、收盘价为基础，计算出一个中枢位 P，以及上方阻力 R1、下方支撑 S1：
def add_pivot_points(df: pd.DataFrame) -> pd.DataFrame: """ 基于当日最高、最低、收盘价计算枢轴点位置。 """ df = df.copy() pivot = (df["high"] + df["low"] + df["close"]) / 3 df["pivot"] = pivot df["r1"] = 2 * pivot - df["low"] df["s1"] = 2 * pivot - df["high"] return df得到 S1 之后，就能把“位置不破”转成一个可判断条件：当日收盘价是否仍然大于等于 S1。
5.2 判断“位置是否被有效跌破”
单日收盘低于 S1 不一定代表有效跌破，尤其在市场波动较大的时候容易假突破。比较保守的做法是加一个“连续两日收盘确认”：
def judge_support_break(df: pd.DataFrame) -> pd.DataFrame: """ 判断收盘价是否有效跌破 S1。 规则：连续两日收盘价 < S1，才认定为有效跌破。 """ df = df.copy() df["below_s1"] = df["close"] < df["s1"] df["confirmed_break"] = ( df["below_s1"].rolling(2).sum() >= 2 ) return df这个规则的好处是过滤掉单日插针的情况。坏处是信号会滞后一天。你可以按自己的风格选择单日判断还是两日判断。
5.3 移动跟踪防守位
“移动跟踪防守”是另一条关键规则。最常见的实现是 ATR 跟踪止损：用过去一段时间的平均真实波幅计算波动范围，然后让防守线跟随价格上移，并且只上移、不回退。
def add_atr_trailing_stop( df: pd.DataFrame, atr_period: int = 14, atr_multiplier: float = 2.0 ) -> pd.DataFrame: """ 计算 ATR 移动跟踪防守位。 默认逻辑： 1. 计算真实波幅 TR，再取 ATR。 2. 防守位 = 最高收盘价 - ATR * 倍数。 3. 使用 cummax 保证防守位只上移，不回退。 """ df = df.copy() prev_close = df["close"].shift(1) tr = pd.concat( [ df["high"] - df["low"], (df["high"] - prev_close).abs(), (df["low"] - prev_close).abs(), ], axis=1, ).max(axis=1) atr = tr.ewm(alpha=1 / atr_period, min_periods=atr_period).mean() df["atr"] = atr # 当前防守位 df["trail_stop_raw"] = df["close"] - atr_multiplier * atr # 只上移：取历史最大防守位 df["trail_stop"] = df["trail_stop_raw"].cummax() return df这个计算里有个细节很重要：防守位一旦抬高，在后续行情中即使价格涨得不够高，防守位也不会下降。这就是“移动跟踪”的含义，和固定止损有明显区别。
5.4 一句话判断示例
把以上函数组合起来，可以这样判断一只股票的状态：
from stock_review.indicators import add_pivot_points, add_atr_trailing_stop, judge_support_break def summarize_last_state(df: pd.DataFrame) -> dict: df = add_pivot_points(df) df = judge_support_break(df) df = add_atr_trailing_stop(df) last = df.iloc[-1] if last["confirmed_break"]: position_text = "已有效跌破关键防守位，反弹延续条件减弱。" elif last["close"] >= last["s1"]: position_text = "仍站在关键防守位上方，位置不破则反弹结构未被破坏。" else: position_text = "单日低于关键防守位，但尚未连续确认，需要继续观察。" return { "current_close": float(last["close"]), "s1": float(last["s1"]), "trail_stop": float(last["trail_stop"]), "position_text": position_text, }这是整篇文章最核心的“规则化”过程。收评里的“位置不破”，到这里就变成了close >= s1；收评里的“移动跟踪防守”，到这里就变成了trail_stop.cummax()。
6. 批量任务与收评草稿生成
单只股票的判断做好之后，批量扫描就很简单了。按code分组，对每一只股票执行同样的计算，最后把结果汇总起来。
6.1 批量扫描
def scan_stocks(df: pd.DataFrame) -> dict: """ 对多只股票批量计算关键位置和移动防守位。 返回一个 dict：{code: DataFrame计算结果} """ results = {} for code, group in df.groupby("code"): group = group.sort_values("date") group = add_pivot_points(group) group = judge_support_break(group) group = add_atr_trailing_stop(group) results[code] = group return results这里没有用复杂的并发，因为日线数据量不大，普通笔记本顺序跑也很快。如果你的股票数量非常大，比如几百只连续几年数据，再考虑用multiprocessing或concurrent.futures加速。
6.2 生成收评草稿
批量计算完成后，最实用的输出是一份 Markdown 收评草稿。我用一个很简单的模板示例：
def build_review_text(name: str, last_state: dict) -> str: close = last_state["current_close"] s1 = last_state["s1"] trail_stop = last_state["trail_stop"] position_text = last_state["position_text"] text = ( f"### {name}\n\n" f"收盘价：{close:.2f}\n\n" f"关键防守位：{s1:.2f}\n\n" f"移动跟踪防守位：{trail_stop:.2f}\n\n" f"位置判断：{position_text}\n\n" ) return text def generate_daily_report(results: dict, output_path: str = "output/review.md") -> None: lines = ["# 每日收评草稿", ""] for code, df in results.items(): last_state = summarize_last_state(df) name = str(df["name"].iloc[-1]) lines.append(build_review_text(name, last_state)) report = "\n".join(lines) with open(output_path, "w", encoding="utf-8") as f: f.write(report)这样输出文件就是一个可以直接复制的 Markdown 草稿。你不一定直接发布，但至少不需要从零开始写。
6.3 可选：接口 API 封装
如果想把它接到自己的内部系统，可以把generate_daily_report再包一层 HTTP 接口。一个最小 Flask 示例：
from flask import Flask, request, jsonify from generate_report import scan_stocks, generate_daily_report app = Flask(__name__) @app.route("/review", methods=["POST"]) def create_review(): """ 请求体格式：{"path": "data/daily_close.csv"} 真实项目中，可以让请求体直接传 DataFrame 的 JSON 数据。 """ payload = request.get_json(force=True) data_path = payload.get("path") if not data_path: return jsonify({"error": "missing path"}), 400 # 实际使用请替换为 load_close_data 的读取逻辑 # df = load_close_data(...) # results = scan_stocks(df) # generate_daily_report(results) return jsonify({"status": "ok", "message": "review generated"}) if __name__ == "__main__": app.run(host="127.0.0.1", port=8087)注意，这个接口示例只是说明可以把批量复盘能力暴露给外部系统，端口号和请求体字段都要按你自己项目改。接口服务不要直接暴露到公网，放在内网或本地访问更安全。
7. 定时任务与每日自动运行
收评的时间点通常是收盘后。A 股收盘时间是 15：00，日线数据一般在收盘后陆续更新。定时任务建议设置在数据更新完成后，比如 17：30。Windows 和 Linux 的设置方式不同。
7.1 Windows 使用计划任务
打开 cmd 或 PowerShell，创建每日任务：
schtasks /create /tn "StockDailyReview" /tr "python D:\stock_review\run_daily.py" /sc daily /st 17:30如果 Python 不在系统 PATH 中，/tr里要写完整路径，例如：
D:\venv\Scripts\python.exe D:\stock_review\run_daily.py7.2 Linux 使用 cron
30 17 * * 1-5 cd /path/to/stock_review && /usr/bin/python3 run_daily.py >> output/cron.log 2>&1这里1-5表示周一到周五，周六周日不执行。日期、时区、数据源更新时间不同，需要按实际情况调整。
7.3 入口脚本
run_daily.py的职责很单一：读数据、算指标、生成报告。日志要打印清楚每一步的状态：
import logging from datetime import datetime from pathlib import Path from load_data import load_close_data from generate_report import scan_stocks, generate_daily_report logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s", ) OUTPUT_DIR = Path("output") OUTPUT_DIR.mkdir(exist_ok=True) def main(): logging.info("开始读取行情数据") df = load_close_data() logging.info("开始批量计算指标，共 %d 条记录", len(df)) results = scan_stocks(df) output_path = OUTPUT_DIR / f"review_{datetime.now().strftime('%Y%m%d')}.md" generate_daily_report(results, output_path) logging.info("收评草稿已生成: %s", output_path) if __name__ == "__main__": main()这套定时任务的作用是保证每天固定时间产出一份报告，不需要人盯着。真正需要人做的，是第二天开盘前看报告，确认哪些位置发生了变化。
8. 功能测试与效果验证
拿到脚本之后，不要直接接真实数据，先用模拟数据测试。这里提供一个可复现的测试片段，验证三类核心逻辑：
- 连续上涨后，移动防守位是否只上移不回退。
- 收盘价跌破 S1 后，位置判断是否能识别。
- 连续两日跌破 S1 后，是否输出“有效跌破”。
import pandas as pd import numpy as np # 构造 30 个交易日的模拟数据 rng = np.random.default_rng(42) dates = pd.bdate_range("2026-07-01", periods=30) close = np.linspace(3000, 3180, 30) + rng.normal(0, 6, 30) df_test = pd.DataFrame( { "date": dates, "code": "TEST", "name": "测试指数", "open": close - 2, "high": close + 12, "low": close - 12, "close": close, "volume": 100000 + rng.integers(0, 5000, 30), } ) # 手动把最后两天收盘价调低，模拟跌破 S1 df_test.loc[28, "close"] = df_test.loc[28, "s1"] - 5 if "s1" in df_test.columns else df_test.loc[28, "close"] * 0.98 df_test.loc[29, "close"] = df_test.loc[28, "s1"] - 8 if "s1" in df_test.columns else df_test.loc[29, "close"] * 0.97 df_result = add_pivot_points(df_test) df_result = judge_support_break(df_result) df_result = add_atr_trailing_stop(df_result) print(df_result[["date", "close", "s1", "below_s1", "confirmed_break", "trail_stop"]].tail())测试通过的标准是：前段行情中trail_stop只增不减；最后两日below_s1和confirmed_break变成 True。如果发现防守位出现明显下降，说明cummax没有生效，或者数据里有缺失值。
9. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 脚本找不到数据文件 | 路径不一致 | 检查 data 目录和文件名 | 把代码里的路径改为绝对路径 |
| 数据接口返回空 | 行情源字段变动 | 打印原始返回样例 | 修改列名映射 |
| 防守位一直不变 | 上涨后回踩不足 | 检查 trail_stop_raw 和 cummax | 确认使用了“只上移不回退”逻辑 |
| 有效跌破判断不出来 | 未使用 rolling 确认 | 检查 confirmed_break 列 | 两日确认需要至少两天数据 |
| 报告生成乱码 | 文件编码问题 | 查看文件打开方式 | 写入时指定 encoding="utf-8" |
| 定时任务不执行 | 路径或权限问题 | 查看 cron/schtasks 日志 | 脚本路径和 Python 路径都用绝对路径 |
| 除权后数据不连续 | 未做复权处理 | 对比除权日前后价格 | 换取前复权数据源 |
| 停牌日数据缺失 | 行情源不返回该日期 | 检查缺失日期 | 用 ffill 或直接跳过空值 |
重点讲一下停牌和缺失值。有些股票停牌几天，复牌后交易数据会跳过一段日期，直接用rolling计算时不会把停牌日算进去，这是正常的。但如果你把停牌日当成普通空值填充，反而会干扰 ATR 计算。建议保留原始交易日期序列，不额外插入无交易日期。
另一个常见坑是不同数据源的字段精度不一致。有的数据源返回字符串，有的返回浮点数。统一在load_close_data里做一次转换：
for col in ["open", "high", "low", "close", "volume"]: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.dropna(subset=["high", "low", "close"])这样能避免字符串格式导致后续比较报错。
10. 最佳实践与合规建议
跑通这套流程之后，你每天收盘后要做的只有一件事：打开 output 目录下的收评草稿，看关键位置是否被收盘价击穿，再看移动防守位有没有上移。但有几个工程和合规上的习惯，建议从一开始就建立。
第一，先小样本测试。第一次运行只放 3 到 5 只股票，用模拟数据验证计算逻辑，再接入真实数据。不要第一天就全市场批量跑，否则出问题时很难定位。
第二，保留最小可运行配置。把config.yaml固定下来，包含股票列表、ATR 周期、ATR 倍数、S1 确认天数。每次调参前先记录当时的参数，方便回测对比。
第三，输入、输出、脚本分目录管理。原始行情数据、计算结果、最终报告不要混在一起，否则定时任务跑几个月后目录会非常乱。
第四，批量任务要加日志和失败重试。本文示例只写了日志，真实场景建议在数据读取环节加 try/except，失败时把错误写入 error.log，继续处理下一只股票，而不是整个脚本崩掉。
第五，接口服务要限制访问范围。如果你把复盘功能封装成了 HTTP 接口，只允许本机或内网访问，不要直接暴露到公网。
第六，涉及个股、指数位置、投资结论的内容，发布前要复核。脚本只能给出“位置未破”“已跌破”这类技术状态，不能保证未来涨跌。任何公开文章都要加上风险提示，不能把脚本输出包装成确定性的收益预测。
第七，数据源授权和隐私问题。A 股行情数据要确认数据服务商的授权范围，不能使用不合法的抓取渠道。如果你把脚本分享给其他人，不要把带个人持仓的数据打包进去。
总结与下一步
这条收评里的“位置不破”和“移动跟踪防守”，本质上都可以转换成非常明确的量化规则：关键位置可以用枢轴点 S1 定义，破位可以用连续两日收盘跌破确认，防守位可以用 ATR 跟踪止损实现。把这套规则写成 Python 脚本，再挂上定时任务，每天收盘后就能自动得到一份复盘草稿。
建议你拿到代码后先做三件事：跑通模拟数据测试，确认trail_stop只上移不回退；接入自己的行情数据源，验证字段是否正确；用 3 只股票做一周的对比观察，确认输出结果和你的手工复盘是否一致。
最容易踩的坑是复权和缺失值。除权不做前复权，防守位会失真；停牌日强行补数据，ATR 会失真。这两点直接影响“位置不破”判断的可靠性，一定要优先处理。
下一步可以扩展的方向很多：把收评草稿转换成带 K 线图的网页报告；加入指数和个股的涨跌家数统计；把移动跟踪防守位做成可视化曲线；或者把这份脚本接入到内部数据平台，形成更完整的复盘工作流。核心逻辑已经放在上面，剩下的按自己习惯调整即可。建议先收藏备用，周末用模拟数据完整跑一遍，再决定要不要接入真实行情。
