这次我们来看一条典型的股市收评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。为了让流程可测试我先定义一个统一的数据读取函数。标准字段如下字段类型说明datedatetime交易日期codestr股票或指数代码namestr名称openfloat开盘价highfloat最高价lowfloat最低价closefloat收盘价volumefloat成交量下面是 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_dfNone) - 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(dropTrue) df df.drop_duplicates(subset[code, date], keeplast) return df if __name__ __main__: df load_close_data() print(df.tail())这段代码会在拿到数据后按“代码 日期”排序并去掉重复行。真实数据接入时优先在前端处理好复权问题。计算技术指标通常建议用前复权数据否则除权当天会出现价格跳空导致防守位失真。5. 关键位置与移动跟踪防守计算这是整篇的核心。收评里说的“这个位置不破”需要先定义“位置”是什么。5.1 用枢轴点计算当日关键位置一种常见方法是用枢轴点Pivot Point体系。以当日的最高价、最低价、收盘价为基础计算出一个中枢位 P以及上方阻力 R1、下方支撑 S1def 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(), ], axis1, ).max(axis1) atr tr.ewm(alpha1 / atr_period, min_periodsatr_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, encodingutf-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(forceTrue) 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(host127.0.0.1, port8087)注意这个接口示例只是说明可以把批量复盘能力暴露给外部系统端口号和请求体字段都要按你自己项目改。接口服务不要直接暴露到公网放在内网或本地访问更安全。7. 定时任务与每日自动运行收评的时间点通常是收盘后。A 股收盘时间是 1500日线数据一般在收盘后陆续更新。定时任务建议设置在数据更新完成后比如 1730。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 使用 cron30 17 * * 1-5 cd /path/to/stock_review /usr/bin/python3 run_daily.py output/cron.log 21这里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( levellogging.INFO, format%(asctime)s | %(levelname)s | %(message)s, ) OUTPUT_DIR Path(output) OUTPUT_DIR.mkdir(exist_okTrue) def main(): logging.info(开始读取行情数据) df load_close_data() logging.info(开始批量计算指标共 %d 条记录, len(df)) results scan_stocks(df) output_path OUTPUT_DIR / freview_{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, periods30) 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 列两日确认需要至少两天数据报告生成乱码文件编码问题查看文件打开方式写入时指定 encodingutf-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], errorscoerce) 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 线图的网页报告加入指数和个股的涨跌家数统计把移动跟踪防守位做成可视化曲线或者把这份脚本接入到内部数据平台形成更完整的复盘工作流。核心逻辑已经放在上面剩下的按自己习惯调整即可。建议先收藏备用周末用模拟数据完整跑一遍再决定要不要接入真实行情。