亲测最好用的AI编写量化策略工具，可以让 AI 直接写各个平台的策略代码，直接生成可运行的策略代码，代码质量远高于直接使用 DeepSeek、Trae 等平台。 大家可以直接用描述策略，然后一键生成可运行的完整策略代码，也可以把它当做一个API 查询工具。 最新消息，已经支持SuperMind等主流量化平台啦，并且实盘亲测过了，很适合小白用户，上线之后获得了非常多朋友的好评。 **🚀️ AI工具平台：https://iris.findtruman.io/ai/tool/ai-quantitative-trading/** 我们在做外汇量化策略回测时，经常遇到一个让人头疼的问题：夏令时切换当天，历史K线的时间归属会出现偏差。这个问题对分钟级和小时级策略影响尤其大，如果时间轴错位，策略信号可能完全失真。 最近我们又处理了一批历史数据，把踩过的坑和解决方案整理出来，分享给社区里同样做个人量化交易的朋友。 案例描述：夏令时切换如何影响K线归属 先说一下我们遇到的具体情况。在复盘一批外汇小时线数据时，发现某些日期的K线排列位置不太自然，价格走势本身没有异常，但K线看起来总是错开一点。进一步检查时间字段后，才发现是夏令时切换对行情时间产生了影响。 外汇市场的数据来源比较复杂，不同接口返回的时间格式可能不同。有的返回市场本地时间，有的返回UTC时间。如果历史数据没有记录夏令时状态，后续进行指标计算或者回测时，可能会出现时间轴不一致的问题。 以美国市场时间为例，冬令时期间纽约市场09:00对应UTC时间14:00，而进入夏令时后，同样的本地时间会对应UTC时间13:00。 时间状态 本地交易时间 UTC时间 冬令时 09:00 14:00 夏令时 09:00 13:00 如果系统按照固定时间规则生成K线，就可能导致当天部分数据归属错误。例如本应该属于某个小时周期的数据，被划分到了前一个周期或者后一个周期。这种影响在分钟K线和小时K线中更加明显。 历史数据增加时间标记：保留原始时间 处理历史K线时，我们更倾向于保留原始时间，同时增加额外字段记录时间状态，而不是直接修改时间。这样后续做策略调整或者重新回测时，可以回溯到最原始的数据。 常见做法是在数据表中增加： 字段 用途 utc_time 统一时间标准 local_time 市场本地时间 timezone 所属时区 dst_status 夏令时状态 例如一条行情数据可以保存为： { "symbol": "EURUSD", "local_time": "2026-03-08 09:00:00", "utc_time": "2026-03-08T13:00:00Z", "dst_status": "active" } 这样后续查看历史行情时，可以清楚知道这根K线对应的时间环境。我们在自己的量化数据库里强制要求这些字段，避免回测时出现时间错位。 K线生成不要依赖固定时间差 很多数据处理中会直接通过增加或者减少几个小时完成转换，这种方式处理普通日期没有明显问题，但面对夏令时切换日期时容易出现偏差。 更稳定的方法是使用时区规则进行转换，让程序根据具体日期自动判断时间变化。Python处理时可以这样实现： from datetime import datetime import pytz timezone = pytz.timezone("US/Eastern") time_str = "2026-03-08 09:00:00" local_time = datetime.strptime( time_str, "%Y-%m-%d %H:%M:%S" ) local_time = timezone.localize(local_time) utc_time = local_time.astimezone(pytz.utc) print(utc_time) 这种方式不需要人工维护夏令时规则，数据跨越不同年份时也能保持一致。我们在实际回测中对比过，动态时区转换能显著降低时间轴不一致带来的虚假信号。 实时行情和历史数据保持统一：AllTick API示例 在实际行情系统中，历史K线和实时tick往往需要连接在一起。如果两部分采用不同时间标准，新生成的数据可能无法和历史数据准确衔接。 我们在处理实时行情时，会先统一时间格式，再进入K线计算流程。以 AllTick API 为例，通过 WebSocket 获取实时行情后，会将收到的时间字段转换为统一格式，再参与分钟线和小时线生成。这个外汇api服务返回的时间戳是UTC格式，适合用来做统一处理。 import websocket import json from datetime import datetime import pytz def on_message(ws, message): data = json.loads(message) trade_time = data["tradeTime"] tz = pytz.timezone("US/Eastern") dt = datetime.strptime( trade_time, "%Y-%m-%d %H:%M:%S" ) dt = tz.localize(dt) utc_time = dt.astimezone(pytz.utc) print(data["symbol"], utc_time) ws = websocket.WebSocketApp( "wss://api.alltick.co/stock/websocket", on_message=on_message ) ws.run_forever() 实操建议：量化交易者的避坑清单 结合我们的经验，列出几条关键建议： 存储阶段就增加时间元数据，不要等到回测时再补。 内部统一使用UTC时间作为标准，本地时间只用于展示。 K线生成不要依赖固定时间差，一定用时区规则动态转换。 实时与历史数据采用同一套时间处理逻辑，避免回测和实盘之间出现时间轴不一致。 对于长期保存的行情数据，时间字段设计比单纯保存价格更加重要。价格可以重新计算，但时间归属一旦错误，后续的数据分析都会受到影响。外汇api提供的数据本身只是原始信息，真正稳定的行情系统还需要在存储阶段处理好时区和夏令时规则。把UTC作为内部标准，把本地时间作为展示内容，这种方式更适合不同市场之间的数据转换，也能减少后续分析中的偏差。 希望这些内容对大家的策略开发有帮助，欢迎在社区里一起讨论。 摘要：在量化研究与实盘策略运行中，实时Tick、K线数据的连续性直接影响回测复现、信号生成与策略执行。很多行情采集脚本仅实现WebSocket断线重连，忽略订阅状态恢复，出现连接正常但行情停滞的隐性故障。本文从实战角度梳理问题成因、处理流程，提供可调试的Python实现，同时说明实盘运行下的数据处理要点。 在量化策略的实盘运行环节，实时行情数据流的连续性是容易被低估的基础环节。多数研究者会把工作重心放在因子构建、回测逻辑、交易信号模型的打磨上，对于底层行情接入，常常默认WebSocket长连接建立之后，就可以稳定持续获取美股实时行情。 但程序进入长时间不间断运行之后，网络抖动、链路超时、服务端连接限制等客观因素，都会造成WebSocket链路静默断开。这类故障不会直接触发程序崩溃，进程依旧正常运行，只是行情推送停止。 对于量化工作而言，这种静默断连带来的数据缺口危害很大：Tick原始序列缺失会导致实盘和回测样本不一致，K线片段丢失会干扰技术指标计算，进一步造成策略信号失真。单纯把网络链路重新接通不足以解决问题，重连之后还需要复原原有订阅任务，才能保证行情数据流完整，这也是很多采集脚本存在的短板。 WebSocket断开会带来哪些量化层面的影响 WebSocket依靠长连接实现双向数据流，连接建立后服务端持续推送标的行情，本地客户端完成接收解析，供给后续模型、指标计算模块。 一旦连接关闭： 应用进程不会报错退出，外部很难直观感知故障； 行情数据流直接中断，数据采集模块停止产出； 若缺少状态检测逻辑，会持续产出残缺数据集，直接影响实盘信号，也会造成未来实盘和历史回测结果无法对齐。 工程实践中建议在采集程序内部维护两组状态：WebSocket连接健康状态、全部订阅元数据（标的代码、数据粒度类型等）。链路重建之后，读取保存的元数据重新发起订阅，无需人工干预重启，保障数据源持续输出，为策略模型提供完整输入。 核心误区：重连不等于订阅恢复 仅完成网络重连，没有重新下发订阅指令，现象就是连接状态显示正常，但服务端不会继续返回目标品种行情，采集程序持续拿到空数据流。 完整的故障自愈处理流程分为四步： 持续监测WebSocket连接的健康状态； 识别断开异常，重建WebSocket通信通道； 读取预先存储的订阅参数，重新提交订阅请求； 恢复行情报文接收，继续为指标、策略模块输出数据。 订阅参数必须做留存。当同时订阅多只美股标的时，需要保存完整订阅列表，否则故障恢复后，只能拿到部分标的行情数据。 Python实战示例：断线自动重连与订阅恢复 下面以Tick逐笔行情采集场景为例，给出基础实现代码。连接异常断开后，程序自动重建WebSocket连接，并恢复标的订阅，保障行情持续流入，可用于策略前置数据采集模块的原型调试。 import websocket import json import time def subscribe(ws): data = { "action": "subscribe", "symbol": "AAPL", "type": "tick", "source": "alltick" } ws.send(json.dumps(data)) def on_open(ws): print("连接成功") subscribe(ws) def on_message(ws, message): data = json.loads(message) print(data) def on_close(ws, code, msg): print("连接关闭") while True: try: ws = websocket.WebSocketApp( "wss://api.alltick.co/ws", on_open=on_open, on_message=on_message, on_close=on_close ) ws.run_forever() except Exception as e: print("异常:", e) time.sleep(5) 代码逻辑说明：当连接终止，程序等待数秒后重新建立WebSocket会话；新连接打开触发on_open回调，再次执行订阅函数，恢复行情推送。该版本适合原型验证，不能直接不经修改投入高强度实盘环境。 面向量化实盘的工程注意事项 以上示例为基础版本，在策略7×24小时运行场景，还需要补充多项逻辑，保障输入数据质量： 行情数据去重校验 重连之后存在重复推送相同Tick报文的情况。建议使用时间戳或者成交编号做去重判断，避免重复写入数据库，防止重复数据污染样本库，造成回测、统计指标偏差。 完整维护多标的订阅清单 多品种策略需要持久化全部订阅标的信息，防止重连后部分品种行情丢失，导致策略部分标的数据缺失。 合理控制重连重试频率 禁止无间隔循环重试连接。高频重试会增加接口侧压力，同时占用本地计算资源，影响策略主逻辑运行，设置梯度或固定休眠间隔提升稳定性。 补充异常告警机制（拓展建议） 可额外增加数据时间戳校验，当长时间没有新行情报文，触发日志告警，便于及时发现极端异常，避免策略基于过期数据运算。 小结 在量化研究与实盘运行当中，底层数据源的稳定性优先级很高。回测结果能否复现、实盘信号是否可靠，很大程度依赖行情数据的完整度。 WebSocket断线自动恢复属于底层基础能力，却直接决定采集数据集质量。在开发采集模块阶段，就将连接状态管理、订阅信息保存、报文校验去重纳入设计，才能为因子研究、回测验证、实盘策略提供可靠的数据底座。在原型调试阶段，可以使用AllTick API快速验证这套故障恢复逻辑，聚焦策略本身的研究工作。 交流探讨：在做行情采集的时候，大家还遇到过哪些影响量化数据质量的底层问题，欢迎一起讨论。 想用 AI 写量化策略？先给它一个好用的数据接口 2026 年，越来越多人开始用 ChatGPT、Claude、Cursor、Codex 来写量化代码。 思路很简单：把你的策略想法告诉 AI，让它直接生成可运行的 Python 代码。不用自己写循环，不用自己查 API 文档，不用自己调试数据格式。 但实际操作下来，很多人发现：AI 生成的代码跑不通。 不是 AI 不行，是你给它的数据接口太难用了。 AI 写代码的质量，取决于你给它什么工具 你让 AI 用 akshare 写一个选股脚本，它需要处理这些事情： 记住 stock_zh_a_hist 这个函数名（不是 get_kline，不是 stock_history） 知道列名是中文的（收盘、成交量，不是 close、volume） 知道要加 time.sleep 防封 IP 知道批量获取只能循环 知道不同函数的参数和返回格式不一样 AI 经常搞混这些细节。生成的代码一跑就报 KeyError: 'close'——因为 AI 习惯用英文列名，但 akshare 返回的是中文。 你让 AI 用 AlphaFeed 写同样的脚本，它只需要知道： af.klines.get(symbol, period, count, adjust, to_dataframe=True) af.klines.batch(symbols, ...) af.quotes.get(universes=..., to_dataframe=True) 列名统一是 open, high, low, close, volume, amount API 越简单、越一致，AI 写出来的代码越可能一次跑通。 实测：同一个需求，AI + 不同数据源的效果 我用完全相同的 prompt 让 AI 分别用 akshare 和 AlphaFeed 写代码，看结果差异。 需求 "帮我写一个 Python 脚本：获取全部 A 股的今日行情，筛选出涨幅大于 5% 且成交额大于 3 亿的股票，然后批量获取这些股票最近 20 天的日 K 线，计算 MA20，找出站上 MA20 的标的。" AI 用 akshare 生成的代码（典型输出） import akshare as ak import pandas as pd import time # 获取全市场行情 df = ak.stock_zh_a_spot_em() # 筛选 df["涨跌幅"] = pd.to_numeric(df["涨跌幅"], errors="coerce") df["成交额"] = pd.to_numeric(df["成交额"], errors="coerce") candidates = df[(df["涨跌幅"] > 5) & (df["成交额"] > 3e8)] # 获取 K 线并计算 MA20 results = [] for _, row in candidates.iterrows(): code = row["代码"] try: kdf = ak.stock_zh_a_hist(symbol=code, period="daily", start_date="20260801", adjust="qfq") if kdf is not None and len(kdf) >= 20: kdf["close"] = pd.to_numeric(kdf["收盘"]) # ← 这里经常出错 ma20 = kdf["close"].rolling(20).mean().iloc[-1] if kdf["close"].iloc[-1] > ma20: results.append({"code": code, "close": kdf["close"].iloc[-1], "ma20": ma20}) except Exception as e: print(f"Error: {code} - {e}") time.sleep(0.5) print(pd.DataFrame(results)) 常见问题： kdf["收盘"] 还是 kdf["close"]？AI 经常搞混，因为 akshare 不同版本列名可能不同 start_date 要手动算——AI 不知道"最近 20 天"从哪天开始（要排除周末节假日） 没有批量接口，AI 只能生成循环代码 AI 经常忘记加 time.sleep，你一跑就被封 IP 涨跌幅 的单位是百分比还是小数？AI 不确定，经常弄反 AI 用 AlphaFeed 生成的代码（典型输出） from alphafeed import AlphaFeed import pandas as pd af = AlphaFeed() # 获取全市场行情 df = af.quotes.get(universes="CN_Stock", to_dataframe=True) # 筛选 candidates = df[ (df["change_rate"].astype(float) > 0.05) & (df["amount"].astype(float) > 3e8) ] symbols = candidates["symbol"].tolist() print(f"筛选出 {len(symbols)} 只") # 批量获取 K 线 dfs = af.klines.batch(symbols, period="1d", count=25, adjust="forward", to_dataframe=True) # 计算 MA20 并筛选 results = [] for sym, kdf in dfs.items(): if len(kdf) < 20: continue ma20 = kdf["close"].rolling(20).mean().iloc[-1] if kdf["close"].iloc[-1] > ma20: results.append({"symbol": sym, "close": round(kdf["close"].iloc[-1], 2), "ma20": round(ma20, 2)}) print(pd.DataFrame(results)) 差异一目了然： 对比 akshare 版本 AlphaFeed 版本 能否一次跑通 ⚠️ 大概率需要手动修 ✅ 基本一次跑通 列名问题 经常中英文搞混 统一英文，AI 不会搞错 批量获取 只能生成循环 + sleep 直接用batch 日期计算 AI 需要手算 start_date count=25 直接指定条数 代码行数 ~25 行 ~15 行 为什么 API 设计影响 AI 代码质量 AI 生成代码的原理是：根据 API 的文档和常见用法模式来"推测"怎么写。 API 设计越规律、越一致，AI 推测得越准。 AlphaFeed 的 API 有几个对 AI 友好的特点： 1. 层级命名，可预测 af.klines.get() # 获取 K 线 af.klines.batch() # 批量获取 K 线 af.quotes.get() # 获取行情 af.depth.get() # 获取盘口 af.depth.batch() # 批量获取盘口 af.instruments.get() # 获取标的信息 AI 看到 klines.get 就能猜到有 klines.batch；看到 depth.get 就能猜到有 depth.batch。API 可预测 = AI 代码正确率高。 对比 akshare 的函数名： ak.stock_zh_a_hist() # A 股日 K 线 ak.stock_zh_a_spot_em() # A 股实时行情 ak.stock_us_daily() # 美股日线 ak.stock_hk_spot_em() # 港股行情 AI 无法从 stock_zh_a_hist 推测出实时行情的函数名是 stock_zh_a_spot_em。每个函数名都要查文档，AI 经常猜错。 2. 参数统一 # AlphaFeed 所有 K 线接口的参数风格一致： af.klines.get(symbol, period="1d", count=60, adjust="forward", to_dataframe=True) af.klines.batch(symbols, period="1d", count=60, adjust="forward", to_dataframe=True) af.klines.intraday(symbol, period="1m", to_dataframe=True) AI 学会一个函数的参数，就能正确使用所有函数。 3. 返回格式统一 不管查什么数据，DataFrame 的列名都是固定的：open, high, low, close, volume, amount。AI 不需要猜列名。 怎么用 AI + AlphaFeed 做量化 方式 1：直接在对话中告诉 AI 用 AlphaFeed Prompt 示例： 用 Python + AlphaFeed SDK 帮我写一个脚本： 1. 获取全部 A 股今日行情（af.quotes.get(universes="CN_Stock")） 2. 筛选出成交额 > 5 亿、涨幅 > 3% 的票 3. 批量拉这些票最近 30 天的日 K 线（af.klines.batch，前复权） 4. 找出 RSI14 < 30 且站上 MA20 的票 5. 输出到 CSV AlphaFeed 安装：pip install alphafeed 列名：open, high, low, close, volume, amount 告诉 AI 关键的 API 调用方式和列名，它生成的代码基本可以直接运行。 方式 2：在 Cursor / Codex 里写 如果你用 Cursor 或 Codex 这样的 AI 编程工具，可以在项目里放一个简短的说明文件，让 AI 知道怎么用 AlphaFeed： # 文件：CONTEXT.md 或者 Cursor Rules """ 本项目使用 AlphaFeed 作为数据源。 常用 API： - af.quotes.get(universes="CN_Stock", to_dataframe=True) — 全市场实时行情 - af.klines.get(symbol, period="1d", count=N, adjust="forward", to_dataframe=True) — K 线 - af.klines.batch(symbols_list, ...) — 批量 K 线 - af.depth.get(symbol) — 五档盘口 - af.klines.intraday(symbol, period="1m", to_dataframe=True) — 分钟线 DataFrame 列名：open, high, low, close, volume, amount 代码格式：600519.SH, AAPL.US, 0700.HK """ AI 读到这个文件后，后续写的所有量化代码都会自动用 AlphaFeed 的正确写法。 方式 3：让 AI 帮你迭代策略 AI 最强大的用法不是"帮你写第一版代码"，而是快速迭代。 第一轮：写一个双均线策略 → AI 5 秒生成 第二轮：加上 RSI 过滤 → 告诉 AI "加一个条件：RSI14 < 70 才入场" 第三轮：加止损 → "加一个 ATR 止损，2 倍 ATR" 第四轮：批量回测 → "在沪深 300 成分股上批量跑一遍" 第五轮：优化参数 → "把均线周期做成参数，测试 (5,20) (10,30) (20,60) 三组" 每一轮 AI 修改代码只需要几秒。如果数据接口够简洁（一个 batch 搞定批量），AI 修改起来不会出错。如果数据接口要循环 + sleep + try/except，每次修改 AI 都可能破坏原来的错误处理逻辑。 AI + 好数据接口 = 量化门槛降低 90% 传统做量化的门槛： 学 Python → 学数据获取 → 调试数据格式 → 学 pandas → 写策略逻辑 → 调试 → 回测 ↑ ↑ 2 周 1 周 AI + AlphaFeed 的做法： 告诉 AI 你的策略想法 → AI 生成代码 → 运行 → 看结果 → 让 AI 改进 ↑ ↑ 5 分钟 5 分钟 你不需要成为 Python 专家。你需要的是一个想法、一个 AI 工具、和一个 AI 能用好的数据接口。 AlphaFeed 的 API 设计——统一的函数命名、统一的参数风格、统一的返回格式、原生批量接口——让它成为目前最适合 AI 编程的量化数据源。 不信你试试：把上面的 prompt 复制到 ChatGPT 或 Claude 里，看它生成的代码能不能直接跑。 pip install alphafeed AlphaFeed 官网：https://alphafeed.org/ Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 外盘期货高频数据包（CME/NYMEX/COMEX）里面到底有什么？ 最近折腾外盘期货的高频数据，打开那个数据包的时候差点被文件夹名字绕晕——什么CME_ES_FUT_20250301_Tick.csv、NYMEX_CL_OPT_Level2_20250302.parquet，文件名长得像乱码，但拆开看其实挺有意思。干脆把里面到底塞了哪些字段、哪些数据包，按我扒拉出来的结果整理一下，免得每次都要点开预览。 先说好，这堆数据是真的大，一个Tick文件解压后动不动就几个G，电脑配置不够的别轻易全量下载，否则硬盘叫起来比空调外机还响。 数据包里到底有哪些交易所品种 我下的这个数据包覆盖了CME集团底下好几个交易所，不是只有CME，具体有这么几类： 交易所 常见品种 数据粒度 CME 标普500迷你期货（ES）、纳斯达克迷你（NQ）、欧元外汇（6E）等 Tick / Level 2 CBOT 美债期货（ZB、ZN）、大豆、玉米 Tick NYMEX 原油（CL）、天然气（NG）、精炼汽油（RB） Tick / Level 2 COMEX 黄金（GC）、白银（SI）、铜（HG） Tick / Level 2 有些数据包里还塞了期权链，比如NYMEX的原油期权，文件会单独标注OPT，容易看花眼，但下载的时候可以勾选过滤，不然全下下来解压到一半就想砸电脑。 Tick数据和Level 2数据到底差在哪 这两类数据包我刚开始也搞混，后来发现区别挺大，直接用表格对比一下： 对比维度 Tick数据 Level 2数据 记录频率 每一笔成交记录一次 通常以“快照”形式更新，频率可达几十毫秒甚至更高 核心字段 时间戳、最新价、成交量、持仓量、成交方向 多档买卖盘口（买1-买10、卖1-卖10），每档价格和挂单量 有没有盘口深度 没有 有，而且某些品种能到20档 适用场景 做成交分布、VWAP、波动率分析 做订单簿失衡、盘口剥头皮策略、高频做市研究 文件大小 一个品种一天大概几百MB到2GB 同品种Level 2可能是Tick的5-10倍，因为快照数据密度太高 Tick数据里有个坑：持仓量字段不是每笔成交都更新，有时候是累加到一定阶段才刷新，所以别直接用单笔成交的持仓变化去推PL，误差会很大。 数据包里具体有哪些字段（以Tick为例） 我拿CME的ES期货Tick文件拆开看过，字段跟国内期货的CTP结构不太一样，但该有的都有： TradeTime：交易所时间戳，东八区得自己转换，注意夏令时冬令时，这点很烦，我就在这上面栽过跟头，回测时差了1小时。 Price：最新成交价，小数点后位数根据品种不同，ES是0.25跳，原油是0.01。 Volume：该笔成交的成交量，注意是“该笔”不是累计，累计成交量得自己算。 OpenInterest：持仓量，这个字段不是实时变化的，有时候连着一堆成交，持仓量都是同一个值。 TradeCondition：成交条件，比如正常成交、价差成交、交易所内部撮合等，这个字段字母缩写很怪，需要查交易所的官方文档，我当时对照表找了好久。 Bid/Ask：有些Tick数据包会附带当时的买一卖一价，但不是所有交易所都有，我下的NYMEX原油Tick就没有，只有成交价，想要盘口得去Level 2文件里找。 Level 2就更复杂一点，除了上面的时间戳，还会有： BidPrice1~BidPrice10：买一到买十价格 BidSize1~BidSize10：对应挂单量 AskPrice1~AskPrice10：卖一到卖十价格 AskSize1~AskSize10：对应挂单量 有些深度能到20档，但文件名里会标Depth20，别下错了，否则拿到10档的还纳闷怎么没有更深。 怎么把数据搞到本地（别傻乎乎手动下载） 我一开始是打开网页一个个点下载，后来发现可以用Python直接调接口，省事很多。后来用了数据源：CMES金融数据库的接口，直接pip装一下，然后几行代码就能把数据拉到本地，不用再去网页上人肉翻页。 # 安装命令 pip install cmes from cmes import MarketData # CMES金融数据库的行情接口，注意入参正确，调用频率正常 client = MarketData(api_key='your_key_here') # 获取CME的ES期货2025年3月3日的Tick数据 df = client.get_tick( exchange='CME', symbol='ES', trade_date='2025-03-03', data_type='tick' # 可选 'tick' 或 'level2' ) print(df.head()) 入参里面trade_date格式要注意，有些数据是UTC日期，跟我们本地日期可能差一天，拉完数据记得检查一下边界。还有调用频率，测试的时候别拿循环猛刷，很容易被暂时限制，正常用完全够。 不同交易所数据包的一点小差别 这东西不是标准化到完全一致，不同交易所的数据包字段命名有时会抽风。比如CME的持仓量字段叫OpenInterest，CBOT的文件里可能叫OI，合并清洗的时候得自己统一一下，不然merge的时候直接报错。 还有，NYMEX的天然气合约代码是NG，但有时候数据包里会缩写成NGQ或者NGX，下载的时候最好先预览一下前几行，确认symbol命名的规律，免得下完一堆文件发现代码对不上自己策略里的品种。 期权数据就更乱了，期权链文件里会有StrikePrice、ExpirationDate、Right（认购/认沽），但是命名规则每个交易所不同，CME的期权代码是组合码，比如ESJ5 C4000，要自己解析合约月份和执行价，这个解析脚本我写了大半天，比想象中费劲。 反正数据本身是好数据，就是前期整理麻烦点，但搞完一次后面就顺了。 前阵子朋友让我帮他盯几只港美股，说是开盘时间他总在开会，错过了好几次加减仓的窗口。我嘴上答应，心里其实有点犯怵——总不能每五分钟刷一次行情软件吧，我也有自己的事。 于是花了一个周末，用 iTick 的免费档搭了个简单的行情预警机器人，WebSocket 长连接收实时推送，价格一碰到我设的阈值就推消息到手机上。这篇文章把整套接入过程写下来，代码都是能直接跑的，给想自己折腾实时行情接入的朋友当个参考。 为什么用 WebSocket 而不是轮询 一开始我也想过用 REST 接口定时去拉，写起来简单嘛。但算了一下账就放弃了：iTick 免费档 REST 接口是 5 次/分钟，如果盯 20 只股票，每只每 5 分钟拉一次就得 4 次/分钟，一个预警机器人就快把配额吃光了，更别说以后加标的。 WebSocket 是另一套逻辑，建一条长连接，服务端把行情主动推过来，你只需要订阅感兴趣的标的，不占 REST 的调用次数。免费档虽然 WebSocket 只给 1 条连接、3 个订阅，但对个人盯盘来说，3 个订阅刚好够我盯 3 只重点股票，REST 的 5 次/分钟配额还能留着做别的查询。 第一步：拉一把历史数据做参考 预警逻辑需要先有个基准价，比如昨收价或者最近 N 根 K 线的均价。我先用 REST 接口把腾讯控股（700）的日线拉出来，算一下过去 20 天的简单移动平均，拿这个当判断趋势的参考线。 iTick 的鉴权很简单，注册后在控制台生成一个 token，请求时放 header 里就行： import requests url = "https://api.itick.org/stock/klines?region=HK&codes=700&kType=8&limit=20" headers = { "accept": "application/json", "token": "你的token" } resp = requests.get(url, headers=headers).json() print(resp["code"], resp["msg"]) # 取出20根日线收盘价 closes = [k["c"] for k in resp["data"]["700"]] sma20 = sum(closes) / len(closes) print(f"700 最近20日收盘均价: {sma20:.2f}") 返回的每根 K 线是标准的 OHLCV 结构，o/h/l/c 是开高低收，v 是成交量，t 是毫秒时间戳。kType=8 是日线，想拉分钟线就改 1（1分钟）、2（5分钟）这些，映射关系文档里写得很清楚。 第二步：WebSocket 订阅实时行情 历史参考线有了，接下来接实时推送。iTick 官方有 Python SDK，装一下： pip install itick-sdk 然后建连接、注册回调、订阅标的： from itick.sdk import Client import time token = "你的token" client = Client(token) def on_message(message): # 收到的每一条实时行情都走这里 print(f"[{time.strftime('%H:%M:%S')}] {message}") def on_error(error): print(f"连接异常: {error}") client.set_message_handler(on_message) client.set_error_handler(on_error) # 连接股票 WebSocket client.connect_stock_websocket() # 订阅腾讯控股和阿里巴巴的报价 client.send_websocket_message( '{"action": "subscribe", "region": "HK", "codes": ["700", "9988"], "types": ["quote"]}' ) try: while True: time.sleep(1) except KeyboardInterrupt: client.close_websocket() SDK 内部把心跳和重连都处理好了：默认 30 秒发一次心跳，网络断开后按 5 秒间隔自动重连，最多重试 10 次，重连成功后订阅关系会自动恢复。这一点省了我很多事，自己手写 WebSocket 的话，心跳、断线重连、订阅恢复这三块至少要折腾大半天。 推送回来的 quote 消息里有最新价、涨跌幅、买一卖一这些字段，够做基础的盘中判断了。 第三步：加阈值预警逻辑 只打印消息没意义，得让它按我的规则提醒我。我在回调里加了简单的判断：最新价相对参考线（刚才算的 SMA20）的偏离超过 ±3% 就告警，并且用状态去重，同一只股票在告警解除前不重复提醒。 from itick.sdk import Client import time import json token = "你的token" client = Client(token) # 参考价：从REST拿的SMA20 REF_PRICE = {"700": 535.0, "9988": 140.0} THRESHOLD = 0.03 # 3% alerted = {} # 去重：记录每只股票是否在告警中 def on_message(message): try: data = json.loads(message) except Exception: return # 只处理实时报价 if data.get("type") != "quote": return code = data.get("code") last = data.get("last", data.get("ld")) if not code or last is None: return ref = REF_PRICE.get(code) if not ref: return pct = (last - ref) / ref if abs(pct) >= THRESHOLD and not alerted.get(code): alerted[code] = True direction = "突破" if pct > 0 else "跌破" print(f"⚠️ {code} {direction}参考线 {abs(pct)*100:.1f}%，最新价 {last}") # 这里可以接推送服务，比如 Server酱、企业微信机器人 if abs(pct) < THRESHOLD * 0.5: alerted[code] = False # 回到安全区间，解除告警 client.set_message_handler(on_message) client.connect_stock_websocket() client.send_websocket_message( '{"action": "subscribe", "region": "HK", "codes": ["700", "9988"], "types": ["quote"]}' ) try: while True: time.sleep(1) except KeyboardInterrupt: client.close_websocket() 告警触发后怎么通知自己，看个人习惯：想收到手机推送，可以在那个 print 的位置换成 Server酱或者企业微信群机器人的 webhook；想落库做后续分析，就写进 SQLite。逻辑都一样，改一行的事。 第四步：多市场怎么扩展 盯完港股，想顺手把美股也加上，不用另起一套代码。iTick 的接口对多市场是统一的，region 参数切换市场：US 是美股、HK 是港股、SH/SZ 是沪深。订阅消息里把 region 换掉就行： # 订阅美股 client.send_websocket_message( '{"action": "subscribe", "region": "US", "codes": ["AAPL", "TSLA"], "types": ["quote"]}' ) 免费档限制是 1 条连接、3 个订阅，所以同一时间只能盯一个市场的 3 个标的。想同时盯港股和美股，要么升级套餐加连接数，要么按时间段切换——比如 A 股收盘后再切到美股。个人用的话，我觉得按时间段切换其实够用。 几个踩过的坑 整个接入不算复杂，但有几个地方当时卡了我一阵，写出来免得大家重复踩： region 代码别记混：美股是 US，港股是 HK，一开始我写了个 HKG，返回空数据，排查了半天才发现是市场代码的问题。 订阅参数格式要对：不同接口版本对订阅消息的字段名要求不完全一样，我用的 SDK 版本里 codes 要传数组，传成字符串会静默失败，没有任何报错，容易误以为连上了。 告警要去重：实时行情一秒推好几条，不做状态去重的话，价格一过阈值手机能给你震到没电。 免费档 REST 有频率限制：5 次/分钟，写历史数据拉取的循环时记得加 time.sleep(12)，不然会被限流返回错误。 token 别写死在代码里：放环境变量，尤其是要把代码推到 GitHub 的时候，血的教训。 小结 这套预警机器人从零到跑通，大概花了我一个周末，其中大半时间花在研究订阅格式和调试告警去重上，纯接入的部分其实很快。WebSocket 收实时行情 + REST 拉历史参考线，两条路配合，个人盯盘、辅助决策完全够用。 如果你是刚开始接触实时行情接入，或者想给自家产品加个行情监控功能，建议先拿免费档把 WebSocket 这条链路跑通，重点看数据延迟和推送稳定性是否符合预期，再决定要不要上更高级的规格。详细的接口字段和订阅格式，可以翻 iTick 的官方文档，Python/Java/Go 的 SDK 都有现成的示例。 有问题欢迎在评论区交流，我踩过的坑，希望能帮你少走点弯路。 引言：交易者的“选择焦虑”与止损命门 在股市博弈中，最折磨人的往往不是行情的震荡，而是你内心摇摆不定的“主观预期”：抓到主升浪，涨一点就想跑，结果错失大牛股；身处下跌通道，总觉得“到底了”，结果抄底抄在半山腰。 为什么你总是拿不住牛股？为什么一买就跌？核心症结在于你缺乏一套定量化、无须主观判断的执行标准。别跟我说你已经记住了这些逻辑，当你真正身处亏损的阵痛中，人性会让你瞬间忘掉纪律。现在就收藏这份手册，它是你在亏钱、迷茫时唯一能把你拉回正轨的“交易救生圈”。今天，我们分享一种源自“顾比技术分析”的极简博弈策略：回溯3根K线，锁定生死线。 取舍的艺术：不再被波动惊扰的“持股逻辑” 在单边上升趋势中，投资者最容易犯的错误是“恐高”。为了过滤随机噪音，我们需要建立一道空间对冲的防御墙。如果你正在操作**缠论中的****“向上一笔”**行情，这套方法的有效性将更上一层楼。 核心操作步骤： 1.**确立基准点（0号线）： 在上涨波段中，找到那一根最高价未能创出新高**的K线，将其标记为“0”。 2.**定量回溯计数： 从“0”号K线开始，向历史方向（往回数）依次标记3根K线，分别为“1、2、3”。请注意，“3”号K****线是这组序列中最古老、最靠前的那一根**。 **3.**划定生死支撑： 以“3号”K线的最低价画一条水平线。 深度导师视角： 这种方法的魅力在于它的“反直觉性”。平庸的交易者盯着眼前的最高点惶恐不安，而实战派则通过回溯找到结构性的支撑底价。只要股价未跌破这根“3号线”，趋势逻辑就没有破坏，你根本不用跑。一旦跌破，必须执行止损指令，绝不拖泥带水。 “这根线只要没有跌破，根本就不用跑。一旦跌破就需要果断离场。” 反转的逻辑：在下跌趋势中锁定“进场哨音” 同样的定量分析，可以帮你彻底戒掉“盲目抄底”的毒瘾。在单边下跌趋势中，反转不是靠感觉，而是靠压力的有效突破。 核心操作步骤： 1.**确立基准点（0号线）： 在下跌波段中，找到那一根最低价未能创出新低**的K线，将其标记为“0”。 **2.**定量回溯计数： 同样向回溯源3根K线，标记为“1、2、3”。 **3.**锁定关键压力： 以“3号”K线的最高价画出一条水平线。 行动策略： ●**进场触发： 只有当行情强势突破3号线最高点**的当天，才是标准的进场点。这是定量的买入信号，不突破就绝不动手。 **●**持续观望： 只要股价还在该线下方运行，所有的反弹都是假动作，继续观望即可。 这种策略通过“空间对冲”的逻辑，将模糊的看盘感觉转化为精准的数学触发，让你变盲目猜测为有据可依。 交易的心法：与自己签一份“盈利契约” 作为导师，我必须严肃地告诉你：技术只是表象，对规则的敬畏才是你财富增值的本质。 在评论区或心中默念“一路长虹”，这不仅仅是一句美好的祝福，更是你与自己签下的一份执行契约。这一刻，你是在告诉自己的潜意识：从今往后，每一笔下单前都要对照这套K线逻辑。符合标准才买，跌破标准就走。 “你打‘一路长虹’的那一刻跟自己签了一份契约。以后每次下单前先对照知识点，符合标准才买，不符合标准就等。你管住了手，钱自然就来了。” 同时，这也是一种“环境算法”。当你不断关注这类硬货内容时，你的交易系统和社交算法都会自动过滤杂讯，为你推送更纯粹的实战乾货。 总结：知行合一的财富增值 技术分析的终点是“极简”。这套“3根K线回溯法”剥离了所有虚假指标，只留下价格最真实的刻度： ●**上升看支撑： 找不创新高的“0”，回溯3根看最低点**。 ●**下跌看压力： 找不创新低的“0”，回溯3根看最高点**。 盈利不是靠预测未来，而是靠对既定规则的严谨执行。当你不再依赖“我觉得会涨”，而是转而依赖“突破了3号线”时，你就已经跨过了职业交易者的门槛。下一次当你准备冲动下单时，你是否愿意多花一分钟，回溯那三根改变命运的K线？ A股五档tick行情与分钟线历史数据 最近在折腾量化回测，想着把A股的秒级盘口数据重新梳理一遍，结果发现之前存的一些免费数据要么缺字段，要么时间戳对不上，合并起来一堆脏数据，气得我直接删了重来。后来朋友随口提了句“CMES金融数据库”有现成的，我就试了下，没想到tick和分钟线都能直接拉，格式还干净，于是把这两个数据集的结构记下来，方便以后翻。 先说说五档tick行情。这东西其实就是交易所每3秒（或实时）推送的一笔快照，记录了当时盘口挂单的十层价格和量，再加上最后一笔成交的信息。做高频策略或者分析盘口深度的，基本都绕不开它。 我拿到的数据字段大概长这样，没完全列全，因为我只关心盘口变化，所以只捡了几个重点： 字段名 含义 我常用的理由 trade_time 时间戳（毫秒） 判断延迟，对齐行情 last_price 最新成交价 看价格跳动 volume 累计成交量 算日内VWAP必须用 amount 累计成交额 同上 bid_price1～5 买一到买五价格 看支撑位 bid_volume1～5 买一到买五挂单量 挂单撤单判断 ask_price1～5 卖一到卖五价格 压力位 ask_volume1～5 卖一到卖五挂单量 同上 没有列全，但上面这些做日内回转或者冰山订单检测已经够用了。我实际处理的时候发现，有些tick记录里买一和卖一价格倒挂，查了半天才发现是集合竞价期间的异常，这种数据如果不做过滤，策略回测会直接炸掉。所以拿到数据第一件事，我还得自己写个清洗脚本，把涨跌停时刻的盘口缺失也剔掉。 至于分钟级别历史数据，就是常见的K线了，只不过多了一些细节。我拉的是1分钟线，字段没什么花头： 字段 说明 time 分钟K线时间（如09:31） open/high/low/close 四价 volume 这一分钟的成交量 amount 这一分钟的成交额 pre_close 前一分钟收盘价（用于计算收益率） 分钟线我主要用来做因子挖掘，比如过去5分钟量比、买卖压力失衡什么的。数据量比tick小太多，跑起来也快，但缺陷是没了盘口细节，没法知道盘口深度变化。 这里岔开一句，我刚开始用的是自己爬的接口，限流严重，隔三差五被封Ip，后来在代码里直接换了CMES金融数据库的接口，本地用pip装完就能用，省了我不少反爬的精力。 # 安装：pip install cmes import cmes # CMES金融数据库的行情接口，注意入参正确，调用频率正常 # 获取A股五档tick数据，股票代码'000001.SZ'，日期2024-01-02 # 返回的DataFrame直接包含上面表格里的字段 tick_df = cmes.get_tick(symbol='000001.SZ', trade_date='2024-01-02', level=5) print(tick_df.head()) # 分钟线数据，同样的日期和股票 min_df = cmes.get_minute(symbol='000001.SZ', trade_date='2024-01-02', freq='1min') print(min_df.describe()) 接口里注释提醒调用频率正常，我就设了个0.5秒间隔，一天的单票tick数据大概也才几万行，不会触发限制。如果你要批量拉全市场，那得自己控制好节奏，不然内存和接口都会爆。 说实话，tick数据最大的坑是存储，2024年一天全市场五档tick压缩后大概十几个G，解压更夸张。我都是按月拉完就切分存成parquet，不然硬盘撑不住。分钟线就友好多了，一年的数据也没多少。所以如果你是刚入门，建议先从分钟线玩起，别一上来就啃tick，容易劝退。 另外，深交所和上交所的tick数据在盘口更新频率上有点差异，深交所盘口变化更频繁，体现在数据里就是bid_volume的变动次数更多。这个细节我也是被坑过，做跨交易所统计套利的时候，延迟不一致导致信号偏移，后来只能按本地时间戳对齐。 没什么特别要总结的，数据这东西，干净、齐全、好拿，比什么都重要 周末花 10 分钟，用 Python 自动复盘上周 A 股行情 周末是复盘的最佳时间。但大多数人的复盘方式是"打开行情软件，翻翻 K 线图，想想下周买什么"——没有结构、没有数据、没有积累。 这篇文章教你写一个周复盘脚本：周末跑一下，自动生成上周的市场概览、板块表现、自选股状态和关键信号，输出一份结构化的复盘报告。 复盘看什么 一份有用的周复盘应该回答这几个问题： 上周大盘怎么样？ — 主要指数涨跌、成交额变化 哪些板块最强/最弱？ — 板块轮动方向 我的自选股表现如何？ — 持仓或关注标的的周度表现 有没有值得注意的信号？ — 异动、突破、趋势变化 第一步：大盘周度概览 from alphafeed import AlphaFeed import pandas as pd import numpy as np from datetime import date, timedelta af = AlphaFeed() # 主要指数 ETF（用 ETF 代替指数，能拿到成交额） indices = { "沪深300": "510300.SH", "中证500": "510500.SH", "创业板": "159915.SZ", "科创50": "588000.SH", } print("=" * 60) print(f" A 股周复盘报告 | {date.today()}") print("=" * 60) print("\n📊 主要指数周度表现") print("-" * 50) for name, symbol in indices.items(): df = af.klines.get(symbol, period="1d", count=10, adjust="forward", to_dataframe=True) if len(df) < 5: continue # 取最近 5 个交易日（一周） week = df.tail(5) prev_week = df.iloc[-10:-5] if len(df) >= 10 else df.head(5) week_open = week["open"].iloc[0] week_close = week["close"].iloc[-1] week_high = week["high"].max() week_low = week["low"].min() week_ret = (week_close - week_open) / week_open week_amount = week["amount"].sum() prev_amount = prev_week["amount"].sum() if len(prev_week) == 5 else week_amount amount_change = (week_amount - prev_amount) / prev_amount if prev_amount > 0 else 0 print(f" {name:<6} 周涨跌: {week_ret*100:>+6.2f}% " f"最高: {week_high:.3f} 最低: {week_low:.3f} " f"成交额变化: {amount_change*100:>+5.1f}%") 第二步：全市场板块强弱 # 拉全市场行情，按板块分组看周表现 all_quotes = af.quotes.get(universes="CN_Stock", to_dataframe=True) all_quotes["change_rate"] = all_quotes["change_rate"].astype(float) all_quotes["amount"] = all_quotes["amount"].astype(float) all_quotes["last_price"] = all_quotes["last_price"].astype(float) # 用代码前缀做简易分板块 def get_board(symbol): code = symbol.split(".")[0] if code.startswith("688"): return "科创板" elif code.startswith("300") or code.startswith("301"): return "创业板" elif code.startswith("002"): return "中小板" elif code.startswith(("600", "601", "603", "605")): return "沪主板" elif code.startswith(("000", "001")): return "深主板" return "其他" all_quotes["board"] = all_quotes["symbol"].apply(get_board) # 从每个板块抽取前 20 只（按成交额），批量拉周 K 线看周涨幅 board_symbols = {} for board in ["沪主板", "深主板", "创业板", "科创板", "中小板"]: bdf = all_quotes[all_quotes["board"] == board] top = bdf.nlargest(20, "amount")["symbol"].tolist() board_symbols[board] = top all_syms = [] for syms in board_symbols.values(): all_syms.extend(syms) dfs = af.klines.batch(list(set(all_syms)), period="1d", count=5, adjust="forward", to_dataframe=True, show_progress=True) print("\n📈 板块周度表现（按龙头股均值）") print("-" * 50) board_results = {} for board, syms in board_symbols.items(): rets = [] for s in syms: if s in dfs and len(dfs[s]) >= 2: kdf = dfs[s] week_ret = (kdf["close"].iloc[-1] - kdf["open"].iloc[0]) / kdf["open"].iloc[0] rets.append(week_ret) avg_ret = np.mean(rets) if rets else 0 board_results[board] = avg_ret for board, ret in sorted(board_results.items(), key=lambda x: x[1], reverse=True): bar = "█" * int(max(0, ret * 200)) neg_bar = "░" * int(max(0, -ret * 200)) print(f" {board:<6} {ret*100:>+6.2f}% {bar}{neg_bar}") 第三步：自选股周报 # 你的自选股列表 my_watchlist = [ ("600519.SH", "贵州茅台"), ("000001.SZ", "平安银行"), ("300750.SZ", "宁德时代"), ("601318.SH", "中国平安"), ("002594.SZ", "比亚迪"), ("000858.SZ", "五粮液"), ("600036.SH", "招商银行"), ("601012.SH", "隆基绿能"), ] symbols = [s for s, _ in my_watchlist] name_map = {s: n for s, n in my_watchlist} # 批量拉最近 25 天数据（4 周 + 余量） dfs = af.klines.batch(symbols, period="1d", count=25, adjust="forward", to_dataframe=True) print("\n📋 自选股周度表现") print("-" * 60) watchlist_summary = [] for symbol in symbols: if symbol not in dfs or len(dfs[symbol]) < 5: continue kdf = dfs[symbol] week = kdf.tail(5) name = name_map.get(symbol, symbol) week_ret = (week["close"].iloc[-1] - week["open"].iloc[0]) / week["open"].iloc[0] week_high = week["high"].max() week_low = week["low"].min() week_vol_avg = week["volume"].mean() # 对比上周量能 prev_week = kdf.iloc[-10:-5] if len(kdf) >= 10 else None vol_change = "" if prev_week is not None and len(prev_week) == 5: prev_vol_avg = prev_week["volume"].mean() if prev_vol_avg > 0: vc = (week_vol_avg - prev_vol_avg) / prev_vol_avg vol_change = f"{vc*100:>+5.0f}%" # MA20 位置 if len(kdf) >= 20: ma20 = kdf["close"].rolling(20).mean().iloc[-1] above_ma20 = "✅" if week["close"].iloc[-1] > ma20 else "❌" else: above_ma20 = "—" watchlist_summary.append({ "symbol": symbol, "name": name, "week_ret": week_ret, "above_ma20": above_ma20, "vol_change": vol_change, }) print(f" {name:<6} {symbol} " f"周涨跌: {week_ret*100:>+6.2f}% " f"站上MA20: {above_ma20} " f"量能变化: {vol_change}") 第四步：关键信号检测 print("\n🔔 关键信号") print("-" * 60) signals = [] for symbol in symbols: if symbol not in dfs or len(dfs[symbol]) < 20: continue kdf = dfs[symbol] name = name_map.get(symbol, symbol) close = kdf["close"].values volume = kdf["volume"].values high = kdf["high"].values # 信号 1：周线突破 20 日新高 high_20 = high[-21:-1].max() if len(high) > 21 else high[:-1].max() if close[-1] > high_20: signals.append(f" 🔺 {name}({symbol}) 突破 20 日新高") # 信号 2：MA5 金叉 MA20（本周发生） ma5_arr = pd.Series(close).rolling(5).mean().values ma20_arr = pd.Series(close).rolling(20).mean().values for i in range(-5, 0): if (i-1 >= -len(ma5_arr) and ma5_arr[i-1] <= ma20_arr[i-1] and ma5_arr[i] > ma20_arr[i]): signals.append(f" ⬆️ {name}({symbol}) 本周 MA5 金叉 MA20") break if (i-1 >= -len(ma5_arr) and ma5_arr[i-1] >= ma20_arr[i-1] and ma5_arr[i] < ma20_arr[i]): signals.append(f" ⬇️ {name}({symbol}) 本周 MA5 死叉 MA20") break # 信号 3：周成交量异常放大 if len(volume) > 10: avg_vol_prev = volume[-15:-5].mean() avg_vol_week = volume[-5:].mean() if avg_vol_prev > 0 and avg_vol_week > avg_vol_prev * 2: signals.append(f" 📊 {name}({symbol}) 本周成交量是前两周均值的 " f"{avg_vol_week/avg_vol_prev:.1f} 倍") # 信号 4：连续下跌后止稳 if len(close) >= 8: prev_5 = [close[i] < close[i-1] for i in range(-6, -1)] if sum(prev_5) >= 4 and close[-1] > close[-2]: signals.append(f" 🔄 {name}({symbol}) 连续下跌后本周尾盘企稳") if signals: for s in signals: print(s) else: print(" 本周自选股无特殊信号") 第五步：输出完整报告 def generate_weekly_report(): """生成完整周复盘报告并保存为 Markdown""" lines = [] lines.append(f"# A 股周复盘 | {date.today()}\n") # 大盘部分 lines.append("## 主要指数\n") lines.append("| 指数 | 周涨跌 | 周最高 | 周最低 |") lines.append("| --- | --- | --- | --- |") for name, symbol in indices.items(): df = af.klines.get(symbol, period="1d", count=5, adjust="forward", to_dataframe=True) if len(df) < 2: continue ret = (df["close"].iloc[-1] - df["open"].iloc[0]) / df["open"].iloc[0] lines.append(f"| {name} | {ret*100:+.2f}% | {df['high'].max():.3f} " f"| {df['low'].min():.3f} |") # 自选股部分 lines.append("\n## 自选股\n") lines.append("| 名称 | 代码 | 周涨跌 | MA20 | 量能变化 |") lines.append("| --- | --- | --- | --- | --- |") for item in watchlist_summary: lines.append(f"| {item['name']} | {item['symbol']} | " f"{item['week_ret']*100:+.2f}% | " f"{item['above_ma20']} | {item['vol_change']} |") # 信号部分 if signals: lines.append("\n## 本周信号\n") for s in signals: lines.append(f"- {s.strip()}") report = "\n".join(lines) filename = f"weekly_review_{date.today():%Y%m%d}.md" with open(filename, "w", encoding="utf-8") as f: f.write(report) print(f"\n报告已保存到 {filename}") return report report = generate_weekly_report() 配合定时任务 # 每周六上午 10 点自动跑复盘脚本 0 10 * * 6 cd /your/path && uv run python weekly_review.py 加上推送，周六早上起来手机就有一份现成的复盘报告。 为什么手动复盘不如自动化 手动翻行情软件 Python 自动复盘 耗时 30-60 分钟 10 秒 覆盖面 凭记忆看几只票 系统扫描自选股 + 全市场 一致性 每周看的指标不一样 固定框架，纵向可比 积累 记在脑子里，下周就忘 Markdown 文件，永久可追溯 情绪干扰 亏了就不想看 不带情绪，客观输出 最后一条最重要。亏钱的那一周，恰恰是最需要复盘的一周。 自动化复盘不会因为你心情不好就跳过。 AlphaFeed 在这个场景中的作用 整个脚本的数据获取只有三处： af.klines.get() — 拉指数 ETF 的 K 线 af.quotes.get(universes=...) — 全市场行情做板块分析 af.klines.batch() — 批量拉自选股 K 线 三处总共花几秒钟。剩下的时间全部是分析逻辑。如果用循环 + sleep 的方式拉数据，光数据获取就要好几分钟——周末复盘变成了等数据。 AlphaFeed 官网：https://alphafeed.org/ Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 
