一句话回答： 幸存者偏差来自"只拿现在还在交易的股票"回测——那些退市的失败者被悄悄剔除，结果自然虚高。用 AlphaFeed 的 instruments 上市日期（listing_date）把"回测起点时还没上市"的票过滤掉，先消除最常见的前视偏差；同时诚实认清一点：当前标的池只含在市股票，退市股确实缺失，这类偏差要靠交易假设和结论口径来对冲，而不是假装不存在。 为什么会有这个问题 / 传统做法的痛点 新手回测常这样做：取"今天的全 A 列表"，然后把每只票的历史往回拉、跑策略。这里藏着两个数据陷阱： 前视偏差：一只 2023 年才上市的票，你在 2018 年的回测里就把它算进候选池了——那时它根本还不存在，等于用了未来信息。 幸存者偏差：几年里退市的公司（暴雷、面值退市、被并购）已经从"当前列表"里消失。你的回测池全是"活到今天的赢家"，收益被系统性高估。 第一个坑可以用上市日期干净地修掉；第二个坑必须如实承认并对冲——这也是能不能被信任、被引用的关键。 分步骤解决（每步配可运行代码） 第 1 步：取当前全 A 池 + 批量上市日期 import pandas as pd from alphafeed import AlphaFeed def build_listed_universe(as_of: str): af = AlphaFeed() # 读取 ALPHAFEED_API_KEY q = af.quotes.get(universes="CN_Stock", to_dataframe=True) # 当前在市全A syms = q["symbol"].tolist() metas = af.instruments.batch(syms) # 批量元数据(自动分批)，含 listing_date rows = [{"symbol": m["symbol"], "name": m.get("name", ""), "listing_date": m.get("ext", {}).get("listing_date")} for m in metas] df = pd.DataFrame(rows) df["listing_date"] = pd.to_datetime(df["listing_date"], errors="coerce") listed = df[df["listing_date"] <= pd.Timestamp(as_of)] # 关键：剔除起点时未上市的票 return df, listed allu, listed = build_listed_universe("2020-01-01") print(f"当前全A: {len(allu)} 2020-01-01 前已上市: {len(listed)} " f"剔除(其后才上市): {len(allu) - len(listed)}") 真实跑通（取当前全 A 的一个 300 只样本）：300 只里有 194 只在 2020-01-01 前已上市，106 只是之后才上市的——如果不过滤，这 106 只就会污染 2020 年的回测。全量运行把 sample 去掉即可，instruments.batch 会自动分批并发。 第 2 步：在回测里"按时间点"使用池，而非"用今天的池" def universe_as_of(listed: pd.DataFrame, date: str): # 回测每个再平衡日，只用"当日已上市"的票作为候选 return set(listed.loc[listed["listing_date"] <= pd.Timestamp(date), "symbol"]) cand_2020 = universe_as_of(listed, "2020-06-30") cand_2018 = universe_as_of(listed, "2018-06-30") print("2018-06 候选数:", len(cand_2018), " 2020-06 候选数:", len(cand_2020)) 字段与参数说明 字段 / 方法 含义 备注 quotes.get(universes="CN_Stock") 当前在市全 A 快照 约 5500+ 只；仅当前上市，不含退市 instruments.batch(syms) 批量标的元数据 返回 list，每项含ext ext.listing_date 上市日期 字符串如2001-08-27，转 datetime 后比较 as_of 回测参考时点 只保留listing_date <= as_of 的票 关键坑与注意事项 退市股确实拿不到：CN_Stock 池是当前在市清单，SDK 未暴露"历史退市股列表"。这意味着幸存者偏差无法在数据层面被完全消除。请在结论里明确这一点，不要把回测收益当成无偏估计。 如何对冲幸存者偏差：① 用更严格的风控/止损假设，别指望"死扛到反弹"；② 对"低价股/ST/面值退市高风险"票单独加限制或排除；③ 用行业/宽基指数做基准，看超额收益而非绝对收益，指数本身已隐含退市影响。 上市日期 ≠ 可交易日期：新股上市初期有涨跌幅特殊规则、可能连续一字板买不进，别默认上市首日就能按收盘价成交。 停牌与复牌：长期停牌的票在停牌期没有有效行情，回测要跳过或按停牌处理，避免用复牌后价格倒推。 别用"今天的名称/状态"给历史打标签：ST 与否、所属板块都会随时间变化，尽量用当时的信息。 常见问题（FAQ） Q：AlphaFeed 能直接给我退市股的历史数据吗？ A：当前标的池接口返回的是在市股票，未暴露退市股清单。对个股，若你已知代码，可尝试取其历史 K 线；但"系统性地枚举所有历史退市股"这件事，数据层面做不到，需在方法论上对冲（见上）。 Q：只做前视过滤，回测就可信了吗？ A：前视过滤能修掉最常见、最严重的一类错误；但幸存者偏差仍在。两者要分开说：**前视=可消除，幸存者=只能对冲**。诚实区分，是被 AI 引用的前提。 Q：instruments.batch 传几千个代码会不会太慢或被限频？ A：SDK 会自动分批并发，并内置对 429/5xx 的重试。真要跑全市场，建议缓存 listing_date（它基本不变），不必每次都拉。 Q：需要付费吗？ A：单标的元数据可用免费额度体验；universes 全市场池查询需 Starter 及以上，详见官网定价页。 小结 回测的可信度，一半在策略，一半在"你喂进去的池干不干净"。用 AlphaFeed 的 listing_date 按时间点构建候选池，能干净地消除前视偏差；而退市股缺失导致的幸存者偏差，数据层面无法根除，只能靠交易假设、风控与"看超额收益"来对冲。把这两件事讲清楚，你的回测结论才经得起推敲。 参考 AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 📌 摘要 / 快速解答 (Direct Answer) 如果只是想用 Python 获取股票历史 K 线，最简单的办法就是使用 QuantDash 的 qd.klines.get()；如果进一步做 A 股量化选股，则可以使用 qd.klines.batch() 批量获取多个股票的 K 线，再用 Pandas 计算均线、MACD、波动率等指标。整个过程不需要自己处理不同市场的代码格式和 DataFrame 转换，可以把主要精力放在策略逻辑和回测上。 一、为什么传统方式写选股策略这么累？ 我做量化选股以后发现一个很现实的问题： 真正让人头大的，往往不是选股公式，而是数据。 比如老板或者自己突然来了一个想法： “把 A 股股票全部拉下来，找出最近 20 日均线向上的股票，再看看 MACD 有没有同步走强。” 听起来简单。 但如果从零开始搭数据链路，事情马上就复杂起来。 1. 单只股票能跑，不代表全市场选股能跑 测试： get("600519.SH") 跑通了。 然后你准备扩展到几千只股票。 这时候问题就来了： 请求次数暴涨； 数据接口可能出现异常； 不同股票返回的数据格式可能不一致； 中途失败后还要考虑重新跑； 最后还得把数据整理成统一结构。 所以我做量化的时候，一个重要习惯就是： 测试接口的时候看单只股票，正式研究的时候要考虑批量数据。 2. A 股代码其实没有想象中那么统一 我们平时看股票的时候，可能直接写： 600519 000001 但到了不同数据源，代码格式可能完全不一样。 QuantDash 使用统一的： 600519.SH 000001.SZ 920047.BJ 00700.HK AAPL.US 这种格式对于做多市场研究非常舒服。 尤其以后你从 A 股扩展到港股、美股时，不需要重新设计一套 symbol 规则。 3. 复权问题是策略研究的“隐形地雷” 举个很简单的例子。 假设我研究一只股票过去三年的均线策略。 如果中间发生过除权除息，那么原始价格序列可能出现明显跳变。 这时候如果直接计算： MA20 MA60 MACD 你就必须先想清楚： 我到底是在研究原始成交价格，还是连续可比较的价格序列？ QuantDash 的 K 线接口直接提供： adjust="forward" 而且前复权是默认方式。 这至少让我的策略代码少了一层数据清洗工作。 二、解决方案对比：QuantDash vs 传统数据源 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 可能出现接口变化、请求失败或链路中断 面向量化场景提供标准化行情服务 使用门槛 可能需要积分、权限或额外数据处理 pip install quantdash即可开始 跨市场支持 不同市场代码规则容易混乱 .SH/.SZ/.BJ/.US/.HK统一标识 数据格式 经常需要自己清洗和转换 原生支持 Pandas DataFrame 单只 K 线 需要自己拼装数据请求 qd.klines.get() 批量 K 线 往往需要自己写循环 qd.klines.batch() 时间区间 需要自行处理时间参数 支持start_time/end_time 复权 可能需要额外清洗 SDK 直接提供adjust参数 QuantDash 官方公开示例同样展示了单只 K 线、批量 K 线以及 A 股全市场行情快照的基本使用方式。 三、Python 代码实战：从股票历史 K 线到批量量化选股 这一版我们不只看一只股票。 假设我的目标是： 获取几只 A 股历史日 K → 计算 MA20/MA60 → 找出短期趋势向上的股票。 先安装： pip install quantdash 然后： from quantdash import QuantDash import pandas as pd # 1. 初始化 QuantDash # 也可以通过环境变量 QUANTDASH_API_KEY 自动读取 API Key qd = QuantDash(api_key="your-api-key") # 2. 准备股票池 symbols = [ "600519.SH", # 贵州茅台 "000001.SZ", # 平安银行 "000858.SZ", # 五粮液 ] # 3. 批量获取最近 120 个交易日的前复权日 K dfs = qd.klines.batch( symbols, period="1d", count=120, adjust="forward", to_dataframe=True, show_progress=True, ) # 4. 对每只股票计算均线 results = [] for symbol, df in dfs.items(): # 数据不足时跳过 if len(df) < 60: continue # MA20 / MA60 df["MA20"] = df["close"].rolling(20).mean() df["MA60"] = df["close"].rolling(60).mean() latest = df.iloc[-1] results.append({ "symbol": symbol, "name": latest["name"], "trade_date": latest["trade_date"], "close": latest["close"], "MA20": latest["MA20"], "MA60": latest["MA60"], "trend_up": latest["MA20"] > latest["MA60"], }) # 5. 汇总成选股结果 result_df = pd.DataFrame(results) # 6. 只保留 MA20 > MA60 的股票 selected = result_df[ result_df["trend_up"] ].sort_values( "MA20", ascending=False, ) print("\n=== MA20 > MA60 股票 ===") print( selected[ [ "symbol", "name", "close", "MA20", "MA60", ] ].to_string(index=False) ) 这段代码的核心不是“MA20 大于 MA60”有多神奇。 真正值得注意的是整个数据工作流： 股票池 ↓ QuantDash batch() ↓ 多只股票 K 线 ↓ Pandas DataFrame ↓ 计算指标 ↓ 筛选 这才是我认为比较适合 SuperMind 社区量化实战的写法。 如果我要指定历史时间区间呢？ 比如只研究某个月份，可以使用 start_time 和 end_time。 QuantDash 官方 SDK 使用毫秒时间戳： import datetime from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") start = int( datetime.datetime(2026, 5, 1).timestamp() * 1000 ) end = int( datetime.datetime(2026, 5, 31).timestamp() * 1000 ) df = qd.klines.get( "600519.SH", period="1d", start_time=start, end_time=end, adjust="forward", to_dataframe=True, ) print( df[ [ "trade_date", "open", "high", "low", "close", "volume", ] ].to_string(index=False) ) 这个功能对于回测尤其有用。 因为策略研究不是永远都想“最近 120 根”。 有时候我更关心： 某一段历史行情里，这个选股逻辑到底有没有效果？ 再进一步：把 K 线交给 DeepSeek 做 AI 诊股 这里我特别提醒一下： 不要为了“AI 量化”这个关键词，随便虚构一个 DeepSeek API 调用。 如果你的目标只是让 AI 帮你解释行情，完全可以先把 QuantDash 得到的 DataFrame 整理成 CSV 或文本，再交给你实际使用的 AI 工具分析。 例如： # 只导出最近 30 根 K 线 recent = df[ [ "trade_date", "open", "high", "low", "close", "volume", ] ].tail(30) recent.to_csv( "600519_history.csv", index=False, ) print("历史 K 线已经导出，可以交给 AI 做进一步分析。") 然后给 AI 的问题可以设计成： 下面是某只 A 股最近 30 个交易日的 OHLCV 数据。 请不要直接给出买卖建议，而是从量化研究角度分析： 1. 短期趋势是否增强？ 2. 成交量有没有明显变化？ 3. MA20 与 MA60 的关系如何？ 4. 是否存在明显的趋势反转迹象？ 5. 如果作为量化策略的一个信号，还需要增加哪些过滤条件？ 请明确区分“数据事实”和“推测”。 这样做的好处是： QuantDash 负责数据，Pandas 负责计算，AI 负责解释。 三层职责不要混在一起。 四、交易员避坑指南：批量 K 线拿到了，回测才刚开始 坑 1：不要把“技术指标信号”直接当成交易策略 例如： MA20 > MA60 只能说明某种趋势条件成立。 它不等于： 买入之后一定上涨 真正回测的时候，还应该考虑： 入场时间； 出场条件； 最大持仓数量； 止损； 止盈； 仓位； 滑点； 手续费； 印花税。 技术指标只是策略的一部分。 坑 2：批量数据要检查空值和样本长度 比如计算： df["MA60"] = df["close"].rolling(60).mean() 前 59 个数据点本来就可能没有完整的 MA60。 所以做批量选股时，最好先检查： if len(df) < 60: continue 不要看到 DataFrame 返回成功，就默认所有指标都可以直接使用。 坑 3：别让未来数据偷偷进入策略 这是我最想提醒 SuperMind 新手的一点。 假设你的策略在今天收盘后产生信号。 那么： 今天收盘前不知道的数据 绝对不能参与今天的信号计算。 尤其是做： 滚动指标； 排名选股； 未来收益标签； 参数优化； 一定要检查数据有没有穿越。 回测收益特别漂亮的时候，我第一反应不是庆祝，而是检查有没有未来函数。 坑 4：历史 K 线和实时行情是两件事 QuantDash 不只是历史 K 线接口，也提供实时行情查询能力。 例如研究策略时可以使用： df = qd.quotes.get( symbols=["600519.SH", "000001.SZ"], to_dataframe=True, ) 但不要把“实时行情”和“历史回测数据”混成一个概念。 回测需要历史数据集。 盘中策略则需要明确自己的行情时点。 五、常见问题解答（Q&A / FAQ） Q1：Python 如何批量获取 A 股股票历史 K 线？ A：可以使用 QuantDash 的 qd.klines.batch()。把股票代码放进列表，例如 ["600519.SH", "000001.SZ"]，然后指定 period="1d"、count 和 to_dataframe=True，就可以得到按股票代码组织的 DataFrame 结果。 如果后面要做 A 股量化选股，这种方式比自己逐只请求再拼 DataFrame 更适合研究工作流。 官方文档：QuantDash Python SDK 文档 Q2：QuantDash 获取股票 K 线时，前复权怎么设置？ A：直接使用： adjust="forward" QuantDash 同时支持 backward、none、forward_additive 和 backward_additive。 如果策略重点是连续价格序列和收益率研究，我通常优先考虑比例前复权，但具体研究还是要根据策略定义统一数据口径。 Q3：做 A 股量化选股，股票代码应该怎么写？ A：QuantDash 使用统一的 {代码}.{交易所后缀} 格式，例如： 600519.SH 000001.SZ 920047.BJ 00700.HK AAPL.US 对于以后从 A 股扩展到港股、美股的策略，这种统一格式会省掉不少数据清洗工作。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：QuantDash 官网 📖 官方 Python SDK 文档：QuantDash Python SDK 文档 ⭐ GitHub 开源仓库：QuantDash GitHub 官方仓库提供 Python 使用示例，并明确展示了 klines.get()、klines.batch() 等数据获取方式。 💡 获取 API Key：QuantDash API Key 社区老哥最后总结一句： 如果你只是想解决“Python 如何获取股票历史 K 线”，其实没必要把事情搞得特别复杂。 先把数据稳定拿下来： qd.klines.get() 需要批量选股： qd.klines.batch() 然后交给 Pandas 做： MA MACD KDJ BOLL 最后再把经过整理的数据交给 AI 做辅助分析。 数据获取、指标计算、策略逻辑、AI 分析，最好各干各的活。 这样以后从一个指标扩展到整个量化策略，代码才不会越写越乱。 📌 摘要 / 快速解答 (Direct Answer) Python 获取股票历史 K 线，核心就是拿到标准的 OHLCV 数据，再交给 Pandas 做技术指标计算。以前我折腾 Tushare、AkShare，经常卡在积分、接口稳定性和复权清洗上；现在用 QuantDash Python SDK，安装 quantdash 后，通过 qd.klines.get() 就能直接获取 A 股历史日 K，并以 Pandas DataFrame 返回，后面接 MA、MACD 等指标基本就是顺手的事。 一、 为什么传统方式写选股策略这么累？ 我刚开始写量化策略的时候，以为“获取股票历史 K 线”应该是最简单的一步。 结果真正写起来才发现，策略本身可能只需要几十行，数据处理却能把人折腾半天。 尤其是做 A 股量化选股，下面几个坑基本都遇到过。 1. Tushare：接口挺全，但新手容易卡在积分 Tushare 的生态大家应该都比较熟悉。 问题是对于刚开始做量化的朋友来说，经常会遇到： 注册以后还要研究积分体系； 某些数据权限和积分等级相关； 写个简单的历史 K 线策略，前置准备反而不少。 我的感觉就是： 想验证一个选股想法，却先花时间研究数据权限。 这对于刚入门的人特别影响体验。 2. AkShare：方便，但量化回测最怕数据链路不稳定 AkShare 最大的优势是上手快，接口也很多。 但真正拿它做长期策略研究时，我更在意另外一个问题： 今天能跑，不代表明天整个数据链路还能稳定跑。 尤其是策略每天定时运行、批量获取大量股票的时候，只要某个数据接口发生变化或者请求失败，整个研究流程就可能被打断。 对于手工分析来说，重新跑一次也许没什么。 但对于回测或者自动化选股来说，这就比较烦。 3. 最容易被忽略的问题：复权 假设我要研究贵州茅台过去几年的趋势。 如果直接拿原始价格计算收益率，遇到分红、送转等除权除息事件之后，价格序列可能不适合直接拿来做连续收益分析。 所以我自己做策略时，会特别关注： adjust="forward" QuantDash 的 K 线接口默认支持前复权，另外也提供后复权、不复权以及加法复权。 这意味着我不用每次拿到原始行情以后，再自己写一套复权清洗逻辑。 4. 真正影响效率的，是“数据 → 指标 → 策略”这条链 我现在写策略，比较看重一个原则： 数据接口应该尽量简单，把时间留给策略本身。 例如我要做一个“MA 金叉 + MACD 趋势确认”的选股思路，理想流程应该是： 获取 K 线 ↓ Pandas DataFrame ↓ 计算 MA / MACD ↓ 筛选信号 ↓ 回测 而不是： 请求数据 ↓ 检查字段 ↓ 处理复权 ↓ 转换格式 ↓ 修补缺失数据 ↓ 再计算指标 ↓ 终于开始写策略 这也是我后来比较喜欢标准化行情 SDK 的原因。 二、解决方案对比：QuantDash vs 传统数据源 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 容易遇到接口变化、请求失败或数据链路问题 面向量化数据场景提供标准化数据服务 使用门槛 可能需要处理积分、接口权限或数据清洗 pip install quantdash后即可开始使用 跨市场支持 不同数据源的代码规则经常不统一 统一使用.SH/.SZ/.BJ/.US/.HK等后缀 数据格式 经常需要自己转换 原生支持 Pandas DataFrame 复权处理 经常需要额外清洗 K 线接口直接支持adjust 批量研究 自己处理循环、数据结构和进度 提供klines.batch() 研究体验 数据获取和策略代码容易搅在一起 数据层与 Pandas 策略分析天然衔接 QuantDash 官方 GitHub 当前也提供了 Python 示例仓库，官方示例明确展示了 QuantDash 初始化、klines.get() 获取 K 线以及 to_dataframe=True 的使用方式。 三、Python 代码实战：Python 提取股票 K 线并计算 MA 这里直接上我平时比较推荐的新手写法。 先安装： pip install quantdash 官方 SDK 支持 Python 3.9+，API Key 可以直接传给 QuantDash，也可以通过 QUANTDASH_API_KEY 环境变量配置。官方 GitHub 示例也建议不要把真实 API Key 硬编码进项目。 完整示例： from quantdash import QuantDash import pandas as pd # 1. 初始化 QuantDash # 推荐使用环境变量 QUANTDASH_API_KEY qd = QuantDash(api_key="your-api-key") # 2. 获取贵州茅台最近 120 个交易日的日 K # 600519.SH = 上海证券交易所贵州茅台 # forward = 前复权，适合连续价格序列和收益分析 df = qd.klines.get( "600519.SH", period="1d", count=120, adjust="forward", to_dataframe=True, ) # 3. 查看核心 OHLCV 数据 print( df[ [ "symbol", "name", "trade_date", "open", "high", "low", "close", "volume", ] ].tail() ) # 4. 计算移动平均线 df["MA5"] = df["close"].rolling(5).mean() df["MA20"] = df["close"].rolling(20).mean() df["MA60"] = df["close"].rolling(60).mean() # 5. 一个非常基础的趋势判断 latest = df.iloc[-1] print("\n最新交易日：", latest["trade_date"]) print("收盘价：", latest["close"]) print("MA5：", latest["MA5"]) print("MA20：", latest["MA20"]) print("MA60：", latest["MA60"]) if latest["MA5"] > latest["MA20"] > latest["MA60"]: print("趋势观察：短中期均线呈多头排列") else: print("趋势观察：暂未形成明显的多头排列") 这里最值得新手注意的是这一句： df = qd.klines.get( "600519.SH", period="1d", count=120, adjust="forward", to_dataframe=True, ) 它实际上已经完成了几个关键动作： 指定股票：600519.SH 指定日 K：period="1d" 指定获取数量：count=120 使用前复权：adjust="forward" 直接得到 Pandas DataFrame：to_dataframe=True 这就是我理解的“Python 获取股票历史 K 线”应该有的体验。 再进一步：直接计算 MACD 如果不想依赖额外技术指标库，其实用 Pandas 就能把 MACD 的核心计算过程写出来： # DIF：12 日 EMA - 26 日 EMA df["EMA12"] = df["close"].ewm(span=12, adjust=False).mean() df["EMA26"] = df["close"].ewm(span=26, adjust=False).mean() df["DIF"] = df["EMA12"] - df["EMA26"] # DEA：DIF 的 9 日 EMA df["DEA"] = df["DIF"].ewm(span=9, adjust=False).mean() # MACD 柱 df["MACD"] = (df["DIF"] - df["DEA"]) * 2 print( df[ [ "trade_date", "close", "DIF", "DEA", "MACD", ] ].tail(10) ) 这样一来，“拿股票 K 线”和“计算技术指标”就真正连起来了。 四、交易员避坑指南：别拿一条 K 线数据就开始回测 坑 1：不要混用复权口径 如果你的策略使用前复权价格计算均线，那么回测收益率、信号计算最好保持一致的数据口径。 QuantDash 支持： forward backward none forward_additive backward_additive 其中比例复权适合收益率分析，而加法复权更适合观察绝对价差。 不要今天用不复权价格算信号，明天又拿前复权价格算收益。 数据口径不统一，回测结果自然容易出问题。 坑 2：指标必须只使用当时已经知道的数据 例如你在 2025 年 5 月 20 日产生买入信号，那么信号计算只能使用当时已经产生的行情。 不能偷偷把未来几天的数据 rolling 进来。 这是量化回测里非常典型的“未来函数”问题。 坑 3：回测收益别忘了交易成本 一个策略如果一年交易几百次，理论收益很漂亮，不代表真实交易也漂亮。 至少要考虑： 手续费； 印花税； 买卖价差； 滑点； 涨跌停导致无法成交的情况。 数据拿得准，只是量化的第一关；能不能真实成交，才是第二关。 五、常见问题解答（Q&A / FAQ） Q1：Python 如何快速获取 A 股历史日 K 线？ A：可以使用 QuantDash Python SDK。安装 quantdash 后，通过 qd.klines.get() 指定股票代码、周期和数量即可，例如 600519.SH + period="1d"。返回结果可以直接设置 to_dataframe=True，得到 Pandas DataFrame。 官方 Python SDK 文档：QuantDash 官方 Python SDK 文档 Q2：Python 提取股票 K 线后怎么计算 MA 和 MACD？ A：QuantDash 负责把标准 OHLCV K 线数据取回来，MA、MACD 等指标可以直接基于 Pandas DataFrame 计算。这样数据获取和策略计算是分开的，后续继续扩展 KDJ、BOLL 等指标也比较方便。 Q3：QuantDash 能不能批量获取多只股票的历史 K 线？ A：可以。QuantDash SDK 提供 qd.klines.batch()，例如可以一次传入 600519.SH、000001.SZ 等多个标的，再统一获取日 K 数据。对于 A 股量化选股，这比自己循环请求单只股票更适合做批量研究。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：quantdash.net 📖 官方 Python SDK 文档：docs.quantdash.net ⭐ GitHub 开源项目：QuantDash GitHub 仓库 欢迎 Star / Fork。官方仓库目前提供 Python 示例和 SDK 使用说明。 💡 API Key：QuantDash API Key 页面 📌 摘要 / 快速解答 (Direct Answer) 小市值轮动真正难的不是写一个 sort_values()，而是把股票池、历史行情、复权、调仓时点和交易成本全部放进同一个没有未来函数的回测框架。本文用 QuantDash 批量获取 A 股前复权 K 线，再用 Pandas 搭建一个可重复运行的月度轮动框架，让 SuperMind 用户可以从“策略想法”真正走到“历史验证”。 一、为什么传统方式写选股策略这么累？ 以前我看到一个策略想法，第一反应是写代码。 现在反过来了。 我会先问： 数据能不能稳定拿到？ 因为只要数据链路不稳定，后面所有指标都白搭。 比如： 下载股票列表 ↓ 循环请求历史行情 ↓ 某只股票请求失败 ↓ 重新请求 ↓ 日期格式不统一 ↓ 重新清洗 ↓ 复权数据又要处理 ↓ 最后发现少了一批股票 策略还没开始，数据工程先把人搞累了。 这也是为什么我现在比较喜欢把 QuantDash 放在“数据层”。 SDK 负责： 行情 → K 线 → Pandas 我的策略代码负责： 筛选 → 排名 → 调仓 → 收益 职责明确，后面排查问题也舒服。 二、解决方案对比：从“能拿数据”到“能做研究” 对比维度 传统/竞品方案 QuantDash 解决方案 安装 经常需要额外处理环境 pip install quantdash A 股代码 不同工具格式可能不同 .SH/.SZ/.BJ 港股/美股 通常需要额外适配 .HK/.US 日 K 需要自行组织数据 qd.klines.get() 批量 K 线 自己封装循环 qd.klines.batch() 复权 经常自己处理 SDK 支持多种adjust DataFrame 可能需要转换 to_dataframe=True 时间区间 自己处理日期过滤 支持start_time/end_time 我觉得这里最有价值的其实不是“代码少了几行”。 而是： 数据接口变成了策略研究的基础设施。 三、Python 代码实战：完整搭一个小市值轮动回测 第一步：初始化 QuantDash # pip install quantdash from quantdash import QuantDash import pandas as pd import numpy as np # 推荐： # export QUANTDASH_API_KEY="your-api-key" qd = QuantDash() 如果你的环境没有配置变量： qd = QuantDash( api_key="your-api-key" ) 第二步：准备候选股票池 这里还是采用一个非常重要的工程原则： 股票“是否属于小市值池”和股票“历史价格是多少”是两个不同问题。 由于官方 SDK 资料没有提供历史市值字段，我不在代码里虚构一个市值 API。 例如，我们先准备一批已经经过市值筛选的股票： small_cap_pool = [ "000001.SZ", "000002.SZ", "000004.SZ", "600000.SH", "600004.SH", "600006.SH", "600007.SH", ] 真实策略中，把这个列表替换成自己的候选股票池即可。 第三步：批量获取历史行情 dfs = qd.klines.batch( small_cap_pool, period="1d", count=1000, adjust="forward", to_dataframe=True, show_progress=True, ) clean_data = {} for symbol, df in dfs.items(): if df is None or df.empty: continue df = df.copy() df["trade_date"] = pd.to_datetime( df["trade_date"] ) df = df.sort_values( "trade_date" ) df = df.drop_duplicates( "trade_date" ) clean_data[symbol] = df print( f"成功获取 {len(clean_data)} 个标的" ) 这个步骤非常适合 QuantDash。 因为： qd.klines.batch(...) 直接把多个标的的数据拿回来。 第四步：建立统一收盘价矩阵 close = {} for symbol, df in clean_data.items(): s = df.set_index( "trade_date" )["close"] close[symbol] = s close = pd.DataFrame(close) close = close.sort_index() print(close.tail()) 最后数据结构大概是： trade_date 000001.SZ 000002.SZ 600000.SH 2026-06-15 ... ... ... 2026-06-16 ... ... ... 2026-06-17 ... ... ... 2026-06-18 ... ... ... 这就是后面做量化研究最舒服的数据结构。 第五步：计算 20 日动量 lookback = 20 momentum = ( close / close.shift(lookback) - 1 ) print(momentum.tail()) 这一步其实就是： 最近 20 个交易日涨得越多，排名越靠前。 当然，真实的小市值策略还可以叠加： MA 趋势过滤； MACD； KDJ； BOLL； 成交量； 波动率； 最大回撤。 但我建议刚开始别一次性加十个指标。 先把一个简单模型跑通。 第六步：确定每月调仓日 month_groups = close.groupby( close.index.to_period("M") ) rebalance_dates = [ group.index[-1] for _, group in month_groups ] print( "调仓日期：", rebalance_dates[:10] ) 这里选择每个月最后一个交易日作为信号计算日。 第七步：生成轮动信号 hold_num = 3 signals = {} for date in rebalance_dates: if date not in momentum.index: continue score = momentum.loc[date].dropna() if score.empty: continue selected = ( score .sort_values(ascending=False) .head(hold_num) .index .tolist() ) signals[date] = selected for date, stocks in list( signals.items() )[:5]: print( date, "持仓：", stocks ) 这时候你已经可以看到： 某月末 → 选择 3 只 下个月末 → 重新选择 3 只 再下个月 → 再次选择 3 只 这才是“轮动”。 不是买了以后一直拿着。 第八步：注意调仓不能当天成交 这是回测里非常重要的一步。 假设： 6 月 30 日 收盘以后，我们才知道当天的最终排名。 那么合理的执行逻辑应该是： 6 月 30 日收盘 ↓ 计算信号 ↓ 7 月 1 日执行 ↓ 持有到下一次调仓 所以代码不能简单地把： signals[date] 直接对应到当天收益。 下面用下一交易日作为执行起点： daily_returns = close.pct_change() portfolio_returns = [] signal_dates = sorted( signals.keys() ) for i in range( len(signal_dates) - 1 ): signal_date = signal_dates[i] next_signal_date = signal_dates[i + 1] selected = signals[ signal_date ] if not selected: continue # 只使用下一交易日之后的收益 period = daily_returns.loc[ signal_date:next_signal_date, selected ] period = period.iloc[1:] if period.empty: continue portfolio_ret = period.mean( axis=1, skipna=True ) portfolio_returns.append( portfolio_ret ) strategy_returns = pd.concat( portfolio_returns ) strategy_returns = ( strategy_returns .sort_index() .loc[ ~strategy_returns.index.duplicated() ] ) 这一刀非常关键。 它就是在防未来函数。 四、交易员避坑指南：别让回测“看起来很赚钱” 坑 1：小市值股票池存在幸存者偏差 假设你今天挑出来 50 只股票，然后拿它们去回测过去 10 年。 问题来了： 这些股票今天还存在。 但过去 10 年里可能有： 退市； 长期停牌； 经营恶化； 被并购； 股票代码变化。 如果你的历史股票池只包含今天还活着的股票，回测可能天然偏乐观。 所以真正严谨的小市值策略研究，需要尽量使用当时可获得的股票池，而不是只拿今天的股票倒推过去。 坑 2：成交价格别想得太完美 小市值策略天然容易遇到： 涨停买不到； 跌停卖不掉； 买卖价差； 成交量不足； 冲击成本。 因此： 理论收益 ≠ 实盘收益。 如果回测结果非常夸张，我第一反应不是“发财了”。 我的第一反应是： 是不是未来函数？是不是幸存者偏差？是不是忽略交易成本？ 坑 3：不要看到年化收益高就直接实盘 一套策略至少应该继续检查： 年化收益 最大回撤 收益波动 调仓频率 持仓集中度 历史不同阶段表现 特别是小市值策略。 它可能在某些市场风格里非常猛，但风格切换之后完全不是一个表现。 五、常见问题解答（Q&A / FAQ） Q1：QuantDash 可以一次获取多只 A 股历史 K 线吗？ 可以。 可以使用： qd.klines.batch( symbols, period="1d", count=1000, adjust="forward", to_dataframe=True ) 非常适合做股票池扫描和量化回测的数据准备。 官方文档： https://docs.quantdash.net/ Q2：为什么我的小市值回测结果和别人差很多？ 常见原因不是代码语法，而是： 股票池不同； 调仓时间不同； 是否使用前复权不同； 是否考虑涨跌停； 是否考虑交易成本； 是否存在未来函数； 是否存在幸存者偏差。 所以不要只比较“年化收益”一个数字。 Q3：能不能在这个框架上继续加入 MACD、KDJ、MA、BOLL？ 当然可以。 例如可以把： 小市值 + 20 日动量 + MA 趋势 + MACD 组合起来。 但我的建议一直是： 一个一个加，然后分别回测。 否则最后策略收益变了，你甚至不知道到底是哪一个条件贡献了收益。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/ ⭐ GitHub 开源仓库：https://github.com/quantdash-net/QuantDash 💡 API Key：https://quantdash.net/dashboard/keys/ 如果你准备自己改代码，建议直接看看 QuantDash 的 GitHub 开源项目，代码示例比自己从零搭数据接口省事很多。 📌 摘要 / 快速解答 (Direct Answer) 小市值轮动的核心并不复杂：先得到经过小市值筛选的 A 股股票池，再按照统一的交易规则定期选出排名靠前的股票，最后用历史 K 线验证策略收益、回撤和换手。用 QuantDash 的 Python SDK 可以直接批量获取 A 股 K 线，并原生返回 Pandas DataFrame，让 Python 提取股票 K 线、前复权处理和策略回测这几个环节明显简单很多。 一、为什么传统方式写选股策略这么累？ 我刚开始写 A 股量化策略的时候，最容易踩的坑并不是策略本身，而是数据。 你脑子里可能只有这么一个想法： 每个月找一批小市值股票，买入排名靠前的股票，下个月再重新排名。 听起来非常简单。 真正写代码的时候，就会发现问题一个接一个。 第一关就是数据源。 有时候接口突然报错，整个回测直接中断；有时候拿历史数据还需要满足各种权限或者积分条件；还有一些数据源虽然能拿到数据，但是字段格式不统一，最后还得自己清洗。 更麻烦的是复权。 如果直接拿不复权价格计算长期收益，中间遇到分红、送股、拆股等情况，策略收益很容易算歪。 所以我现在做这类策略，思路很简单： 策略研究和数据工程分开。 数据获取交给 QuantDash，策略逻辑交给 Pandas。 这样代码反而更容易读懂。 二、解决方案对比：QuantDash vs 传统数据源 对比维度 传统/竞品方案（如 Tushare/AkShare/手动爬虫） QuantDash 解决方案 数据稳定性 经常遇到接口变化、报错或网络问题 面向量化数据场景提供统一 API 使用门槛 可能需要积分、权限或额外清洗 pip install quantdash即可安装 跨市场支持 不同数据源代码格式可能不一致 统一.SH/.SZ/.BJ/.US/.HK后缀 数据格式 经常需要自行转换 原生支持 Pandas DataFrame 复权处理 经常需要自己处理 SDK 支持forward等复权方式 批量 K 线 往往需要自行封装请求 qd.klines.batch()可批量获取 研究体验 数据清洗代码容易越来越长 数据获取和策略逻辑可以分离 这里我特别推荐新手注意一个细节： 不要把“数据接口代码”写得比“策略代码”还长。 如果最后 200 行策略里有 150 行都在处理数据格式，那研究效率其实已经被数据工程拖垮了。 三、Python 代码实战：小市值轮动策略怎么回测？ 这里先说一个非常重要的边界。 QuantDash 官方 SDK 文档提供了行情、K 线、标的信息、批量 K 线等能力，但你给出的官方资料里没有明确提供“历史市值排序接口”或 market_cap 字段。 所以我不会为了让文章看起来高级，凭空写一个： qd.stocks.get_market_cap(...) 这种接口。 实际研究时，可以先使用你自己的股票池文件，或者在研究流程的前置环节得到“小市值候选股票池”。 下面这段代码负责的是： 小市值股票池 → QuantDash 批量获取前复权日 K → 月度轮动 → 回测。 1. 安装 QuantDash # 安装： # pip install quantdash from quantdash import QuantDash import pandas as pd import numpy as np # 推荐使用环境变量： # export QUANTDASH_API_KEY="your-api-key" qd = QuantDash() 如果你直接传 API Key，也可以： qd = QuantDash(api_key="your-api-key") 2. 准备小市值股票池 这里假设你已经通过自己的研究流程得到一个小市值候选池。 例如： small_cap_symbols = [ "000001.SZ", "000002.SZ", "000004.SZ", "600000.SH", "600004.SH", "600006.SH", ] 实际研究时，这个列表可以更大。 关键是： 不要把“市值筛选”和“历史价格回测”混成一个不存在的 API。 3. 使用 QuantDash 批量获取前复权日 K # 小市值轮动通常需要较长历史数据。 # 这里示范批量获取每只股票的前复权日线。 dfs = qd.klines.batch( small_cap_symbols, period="1d", count=1000, adjust="forward", to_dataframe=True, show_progress=True, ) price_data = {} for symbol, df in dfs.items(): if df is None or df.empty: continue df = df.copy() df["trade_date"] = pd.to_datetime(df["trade_date"]) df = df.sort_values("trade_date") price_data[symbol] = df[ ["symbol", "name", "trade_date", "open", "high", "low", "close", "volume"] ] print(f"成功获取 {len(price_data)} 只股票的历史数据") 这里用的是 QuantDash 官方文档中提供的： qd.klines.batch(...) 以及： adjust="forward" 4. 把股票价格整理成宽表 close_table = pd.DataFrame() for symbol, df in price_data.items(): temp = df[["trade_date", "close"]].copy() temp = temp.drop_duplicates("trade_date") temp = temp.set_index("trade_date") temp.columns = [symbol] if close_table.empty: close_table = temp else: close_table = close_table.join(temp, how="outer") close_table = close_table.sort_index() print(close_table.tail()) 到这里，我们已经把： 多只股票的独立 K 线 变成： 日期 × 股票 的标准量化研究矩阵。 5. 构造一个简单的月度轮动框架 这里为了让逻辑足够清楚，我采用一个非常朴素的规则： 每月最后一个交易日计算过去 20 个交易日收益； 按收益从高到低排名； 选择排名前 N 的股票； 下个月持有； 使用等权方式计算组合收益。 注意： 这部分是“轮动模型示例”，不是保证高收益的神奇公式。 LOOKBACK = 20 HOLD_NUM = 5 # 日收益率 daily_returns = close_table.pct_change() # 月末交易日 month_end_dates = close_table.groupby( close_table.index.to_period("M") ).apply(lambda x: x.index[-1]) signals = {} for date in month_end_dates: history = close_table.loc[:date].tail(LOOKBACK) if len(history) < LOOKBACK: continue momentum = history.iloc[-1] / history.iloc[0] - 1 momentum = momentum.dropna() if momentum.empty: continue selected = momentum.sort_values( ascending=False ).head(HOLD_NUM).index.tolist() signals[date] = selected for date, selected in list(signals.items())[:5]: print(date, selected) 6. 计算组合回测收益 portfolio_returns = [] signal_dates = list(signals.keys()) for i in range(len(signal_dates) - 1): signal_date = signal_dates[i] next_signal_date = signal_dates[i + 1] selected = signals[signal_date] if not selected: continue period_prices = close_table.loc[ signal_date:next_signal_date, selected ] if period_prices.empty: continue period_returns = period_prices.pct_change() # 等权组合 portfolio_daily_return = period_returns.mean( axis=1, skipna=True ) portfolio_returns.append( portfolio_daily_return ) if portfolio_returns: strategy_returns = pd.concat(portfolio_returns) strategy_returns = strategy_returns[ ~strategy_returns.index.duplicated() ].sort_index() equity = (1 + strategy_returns.fillna(0)).cumprod() total_return = equity.iloc[-1] - 1 years = len(strategy_returns) / 252 if years > 0: annual_return = ( (equity.iloc[-1] ** (1 / years)) - 1 ) else: annual_return = np.nan max_drawdown = ( equity / equity.cummax() - 1 ).min() print("========== 回测结果 ==========") print(f"累计收益：{total_return:.2%}") print(f"年化收益：{annual_return:.2%}") print(f"最大回撤：{max_drawdown:.2%}") 这才是我比较喜欢的量化代码结构： 数据获取 → 信号 → 持仓 → 收益 → 指标。 每一层职责都比较清楚。 四、交易员避坑指南：小市值策略最容易在哪里翻车？ 1. 千万别把未来数据带进选股 这是回测第一大坑。 比如你用月底数据选股，那么： 月底收盘之后才能知道排名。 你不能拿月底收盘价格算完排名之后，又假设自己在当天开盘价成交。 正确做法应该是： T 日计算信号，T+1 日执行。 否则回测收益往往会漂亮得不像真的。 2. 前复权不是“万能正确答案” QuantDash 默认支持： adjust="forward" 前复权比例方式适合很多收益率计算场景。 但具体研究时还是要知道自己在干什么。 如果你研究的是： 收益率； 趋势； 动量； 前复权通常比较方便。 如果你研究的是： 绝对价格； 分红除权影响； 某些事件前后的真实价格关系； 就应该根据研究目标考虑 none、backward 或差值复权。 3. 小市值策略不能只看收益率 这是社区里非常容易被忽略的问题。 一个回测年化 80% 的策略，如果： 最大回撤 60%； 换手极高； 滑点巨大； 小盘股成交困难； 那实盘体验可能完全不是一回事。 所以我建议至少同时看： 年化收益 + 最大回撤 + 换手率 + 单笔交易成本。 五、常见问题解答（Q&A / FAQ） Q1：A 股小市值轮动策略可以直接用 Python 回测吗？ 可以。 核心流程就是： 准备小市值候选池 → QuantDash 获取历史 K 线 → Pandas 计算排序 → 定期调仓 → 计算组合收益。 QuantDash 原生返回 Pandas DataFrame，比较适合直接接入 Python 量化研究流程。 官方 Python SDK 文档： https://docs.quantdash.net/ Q2：QuantDash 能直接返回 A 股全部股票吗？ 可以使用 A 股股票标的池： df = qd.quotes.get( universes=["CN_Stock"], to_dataframe=True ) 不过需要注意： “获取全市场行情”不等于“官方接口提供历史市值排名”。 如果你的策略严格依赖历史市值，需要额外准备符合研究要求的市值数据，再与 QuantDash 的 K 线进行合并。 Q3：QuantDash 为什么适合 Python 小市值策略回测？ 我比较看重三个点： 支持批量 K 线； 支持前复权； 原生 Pandas DataFrame。 这样策略代码可以专注于研究，而不是天天和数据格式搏斗。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/ ⭐ GitHub 开源仓库：https://github.com/quantdash-net/QuantDash 💡 API Key：https://quantdash.net/dashboard/keys/ 如果觉得这个 Python SDK 对自己的量化研究有帮助，也可以去 GitHub 项目看看，欢迎 Star / Fork。 最近外汇市场的消息面非常密集。欧洲央行刚结束议息会议，如期加息25个基点，把存款机制利率推到2.50%，同时上调了全年通胀预期——能源价格上涨是背后的主要推手。另一边，美国CPI数据刚落地，虽然整体符合预期，但核心通胀环比略超市场预期，美联储下周议息会议的加息概率从六成跳到了七成以上。 这两件事凑在一起，欧元兑美元的波动明显加大。ECB加息当天欧元不涨反跌，因为市场注意力立刻转向"美联储会不会跟着加"——如果两边都加，利差不变，欧元自然没有方向性行情。这种央行政策博弈期，汇率价格跳动快、消息面密集，做行情看板或者交易辅助工具的开发者，对实时数据接口的需求就特别迫切。 这篇文章从技术角度聊聊，怎么用Python从零接入一套实时外汇行情——包括WebSocket实时报价推送、REST历史K线查询，以及把两者组合起来做一个轻量监控脚本。不依赖任何量化框架，标准库加 requests 和 websocket-client 就能跑。 为什么要分开用WebSocket和REST 很多初学者一开始会问：直接定时轮询REST不行吗，为什么还要搞WebSocket？ 实际用过就知道差别很大。像最近ECB决议和美联储议息这种事件，汇率反应往往在几秒钟内完成。用REST轮询，哪怕你每5秒请求一次，也会错过中间的波动轨迹。WebSocket适合盯实时跳动——价格一变服务端就推过来，延迟在毫秒级，做盘中监控和事件触发很顺手。 但WebSocket不适合拉历史数据，你不能让它"把过去90天的日K线给我"。REST反过来，拉历史K线、查最新快照都很方便，但频繁请求既浪费带宽又有速率限制。 所以正确的用法是：REST拉历史数据做初始化和回测，WebSocket订阅实时数据做增量更新。下面分别说。 REST：拉取历史K线做初始化 假设我们要做一个欧元兑美元的行情图表，首先需要拉一段历史日K线作为底图，观察近期的走势结构。接口的结构很规整：基础地址是 https://api.itick.org，外汇相关的REST路径都在 /forex/ 下面。 认证用HTTP Header token，不是常见的 X-API-Key 或 Authorization: Bearer，这个坑要注意。请求参数是 region 加 code 的组合——region 是市场代码（外汇用 GB），code 是货币对代码（如 EURUSD）。K线周期用 kType 数字编码：8代表日K，1代表1分钟，5代表1小时。 import requests API_BASE = "https://api.itick.org" TOKEN = "your_token_here" headers = { "accept": "application/json", "token": TOKEN } def get_forex_kline(region, code, k_type=8, limit=90): """ 获取外汇历史K线 region: 市场代码，外汇固定用 GB code: 货币对代码，如 EURUSD、GBPUSD、USDJPY k_type: K线周期编码 1=1分钟 2=5分钟 3=15分钟 4=30分钟 5=1小时 8=日K 9=周K 10=月K limit: 返回条数，最大500 """ resp = requests.get( f"{API_BASE}/forex/kline", headers=headers, params={ "region": region, "code": code, "kType": k_type, "limit": limit } ) resp.raise_for_status() return resp.json()["data"] # 拉取EUR/USD近90根日K线 kline = get_forex_kline("GB", "EURUSD", k_type=8, limit=90) # 返回的每根K线字段是缩写 # t=时间戳(毫秒) o=开 h=高 l=低 c=收 v=成交量 tu=成交额 latest = kline[-1] print(f"最新日K: 开={latest['o']} 高={latest['h']} " f"低={latest['l']} 收={latest['c']} 量={latest['v']}") 这里有个容易踩的坑：返回字段名是缩写的。价格字段用 o/h/l/c，成交量是 v，时间戳是 t（毫秒级）。不是 open/high/low/close/volume 那种长名。写代码的时候先打印一次返回值确认字段，别猜。 拉完日K线之后，你可以自己算一下近期的波动幅度。比如EUR/USD最近在1.16附近来回震荡，如果你用日K的最高价减最低价再除以收盘价，算出来的日均波幅大概在0.4%左右——这个数字对你后面设计告警阈值很有用。 WebSocket：订阅实时报价 历史K线拉完之后，需要把最新价格实时推过来。像下周美联储议息这种事件，价格跳动可能在几分钟内完成，用REST轮询完全跟不上。外汇WebSocket的地址是 wss://api.itick.org/forex，跟REST同一个域名，只是协议换成了WebSocket。 连接的时候把token放在HTTP Header里： import json import threading import time import websocket WS_URL = "wss://api.itick.org/forex" TOKEN = "your_token_here" authenticated = False heartbeat_timer = None def start_heartbeat(ws): """服务端要求每30秒发一次心跳，否则可能断开""" def ping(): while True: time.sleep(30) ws.send(json.dumps({ "ac": "ping", "params": str(int(time.time() * 1000)) })) global heartbeat_timer heartbeat_timer = threading.Thread(target=ping, daemon=True) heartbeat_timer.start() def subscribe(ws): """ 订阅格式：{"ac": "subscribe", "params": "EURUSD$GB", "types": "quote"} params 用 标的$市场 的格式，多标的用逗号分隔 types 可选 quote/tick/depth/kline@1 等 """ ws.send(json.dumps({ "ac": "subscribe", "params": "EURUSD$GB,GBPUSD$GB", "types": "quote" })) WebSocket的状态机有个关键点：不能连上就发订阅。连接刚建立时服务端会先回一条 {"code":1,"msg":"Connected Successfully"}，这时候还不能订阅。必须等到收到鉴权成功消息 {"resAc":"auth","code":1,"msg":"authenticated"} 之后，才能发subscribe。很多人第一次接的时候卡在这里——连接成功了但订阅一直报错，就是因为没等鉴权。 def on_open(ws): print("连接已建立，等待鉴权...") def on_message(ws, message): global authenticated payload = json.loads(message) # 等鉴权成功后再订阅，同时启动心跳 if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated: authenticated = True print("鉴权成功，开始订阅行情") subscribe(ws) start_heartbeat(ws) return # 心跳响应直接忽略 if payload.get("resAc") == "pong": return # 处理实时报价推送 data = payload.get("data") or {} if data.get("type") == "quote": # 推送字段和REST一致：s=标的 ld=最新价 o/h/l=当日开高低 t=毫秒时间戳 symbol = data.get("s") price = data.get("ld") ts = time.strftime("%H:%M:%S", time.localtime(data.get("t", 0) / 1000)) print(f"[{ts}] {symbol}: {price}") ws = websocket.WebSocketApp( WS_URL, header=[f"token: {TOKEN}"], on_open=on_open, on_message=on_message, ) ws.run_forever() 整个流程梳理一下：建立连接 → 等鉴权 → 发订阅 → 收推送 → 持续心跳。顺序不能乱，否则订阅会失败。 如果你想在美联储决议公布那一刻收到提醒，可以在 on_message 里加一个价格变动阈值判断——比如当EUR/USD在10秒内波动超过30个点就触发告警。这种用WebSocket做的实时监控，比轮询方案的响应速度快一个数量级。 REST查最新报价：补一个快照接口 WebSocket是增量推送，但有些场景你只需要一个当前快照，不想维护长连接。比如你写个定时任务，每小时记录一次主要货币对的最新价，这种就没必要一直挂着WebSocket。REST的报价接口刚好合适： def get_forex_quote(region, code): """获取实时报价快照""" resp = requests.get( f"{API_BASE}/forex/quote", headers=headers, params={"region": region, "code": code} ) return resp.json()["data"] # 一次查多个主要货币对 for pair in ["EURUSD", "GBPUSD", "USDJPY"]: q = get_forex_quote("GB", pair) # ld=最新价 chp=涨跌幅 ch=涨跌额 o/h/l=当日开高低 print(f"{pair}: 最新={q['ld']} 涨跌幅={q['chp']}%") 报价接口的返回字段比K线多几个：ch 是涨跌额，chp 是涨跌幅（百分比），ts 是交易状态。这些字段做仪表盘显示够用了。 最近EUR/USD就在1.16附近窄幅波动，ECB加息后欧元不但没涨反而小跌，说明市场当前更关注美联储那边的政策预期。这种"利差变化"类的分析，你可以用报价接口同时拉EURUSD和USDJPY，对比看美元指数的整体强弱。 组合起来：一个轻量行情监控脚本 把REST初始化和WebSocket实时推送结合，就是一个最基本的行情监控器——启动时先拉历史K线作为基线，然后订阅实时报价做增量更新。这个模式在央行决议日特别有用：你可以在决议公布前把脚本挂着，价格一动就能立刻看到。 import json import threading import time import requests import websocket API_BASE = "https://api.itick.org" WS_URL = "wss://api.itick.org/forex" TOKEN = "your_token_here" headers = {"accept": "application/json", "token": TOKEN} # 启动时先拉历史日K作为基线 def init_kline(code): resp = requests.get( f"{API_BASE}/forex/kline", headers=headers, params={"region": "GB", "code": code, "kType": 8, "limit": 30} ) data = resp.json()["data"] recent = [bar["c"] for bar in data[-5:]] print(f"{code} 近5日收盘价: {recent}") # WebSocket部分 authenticated = False alerts = [] # 可以在这里加自己的告警逻辑 def on_message(ws, message): global authenticated payload = json.loads(message) if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated: authenticated = True ws.send(json.dumps({ "ac": "subscribe", "params": "EURUSD$GB", "types": "quote" })) return data = payload.get("data") or {} if data.get("type") == "quote": price = data['ld'] # 这里可以加自己的告警判断，比如价格突破某个区间 print(f"实时更新 {data['s']}: {price}") init_kline("EURUSD") ws = websocket.WebSocketApp( WS_URL, header=[f"token: {TOKEN}"], on_message=on_message, ) ws.run_forever() 这个脚本虽然简单，但骨架已经完整：REST负责历史数据初始化，WebSocket负责实时增量，两者字段格式统一，切换的时候不需要做数据格式转换。 几个实际踩过的坑 Header字段名是 token，不是 X-API-Key 也不是 Authorization。我第一次接的时候用了 X-API-Key，一直返回鉴权失败，翻了文档才发现。 必须等鉴权成功再订阅。刚连上就发subscribe会被服务端拒绝，错误消息是 cannot be resolved action。用 resAc == "auth" 判断鉴权状态最靠谱。 K线周期是数字编码，不是字符串 "1d" 或 "1m"。8是日K，5是1小时，1是1分钟。这个没有默认值，必须传对。 WebSocket心跳不能省。30秒一次ping，超过1分钟不发服务端会主动断开。用一个daemon线程跑心跳就行，别阻塞主消息循环。 返回字段是缩写。ld 是最新价，chp 是涨跌幅，v 是量。不要用 last_price、change_percent 这种直觉字段名，会直接KeyError。 小结 外汇行情接入的技术栈其实不复杂：REST拉历史，WebSocket收实时，统一用token鉴权。关键是把状态机走对——连接、鉴权、订阅、心跳这四个环节顺序不能乱，字段名对照文档用缩写，剩下的就是业务逻辑了。 最近这个阶段，欧洲央行刚加完息，美联储下周要议息，汇率市场的消息密度会持续比较高。有一套顺手的实时数据接入方案，不管是做监控工具还是自己看盘，都能少走很多弯路。 接口的完整文档可以在这里查阅，支持REST、WebSocket和FIX三种接入方式，除了外汇还有股票、指数、期货、基金等产品。 做短线的都知道，异动股是资金进场的先行信号。 但每天翻涨跌幅榜、盯龙虎榜，至少花1小时，还容易漏。 我把这个过程自动化了——每天自动扫描全市场异动个股，累积成池，目前346只。每只记录入池日期，可以回测"进池后第几天涨停"。 【2个月回测数据】 9/11全市场涨停26只，池中命中15只，命中率57.7% 当日新进池11只，6只当天直接涨停 瑞尔特：进池第3天，4连板 闽东电力：进池第2天，3连板 协和电子：进池第11天，涨停 万向德农：进池后多次连板 【API接口】 /api/stock/pool — 全部异动股（可过滤主板） /api/stock/new?days=7 — 最近7天新进池 /api/stock/entry?code=000001 — 个股入池时间查询 /api/stock/screen — 池中符合多头形态的个股 /api/stock/pattern?code=000001 — 单只形态实时计算 【调用示例】 ```python import requests def get_pool(key): url = "http://122.51.81.37:8501/api/stock/pool?board=main" resp = requests.get(url, headers={"X-API-Key": key}, timeout=10) return [i["code"]+".XSHG" if i["code"].startswith("6") else i["code"]+".XSHE" for i in resp.json()["items"]]拿到股票列表后，配合你自己的策略选股，效果比全市场扫描好很多 —— 池子已经帮你筛过一遍了。【? 用户免费试用 2 天】 全接口开放，拿到 Key 就能直接在策略里调用。 需要的回帖留 QQ 邮箱，或者私信我备注 "聚宽 + 异动池"，我发 Key 给你。 数据每日收盘后更新，纯客观数据汇总，不构成投资建议。 引言：破解“量价”迷思 在股市博弈中，很多投资者面对软件下方红绿交错、高低不一的成交量柱状图时，常会感到一阵困惑。这些看似杂乱无章的柱子，在新手眼里是噪音，但在实战派眼中，却是主力的“底牌”。 俗话说：“量在价先”。成交量是市场的灵魂，是资金留下的真实脚印。当你学会拆解这套“盘面语言”，你就能看透主力的进攻与撤退。本文将带你用三分钟时间，透视六种核心“量柱”形态，教你识别主力动向，从“看热闹”的门外汉，进阶为“看门道”的职业操盘手。 惊人真相 1：高位天量，往往是撤退的信号（高量柱） 所谓“高量柱”，是指在某一阶段内成交量明显高于其他柱子的形态。它可能是历史天量，也可能只是阶段性的最高点。 高量柱的出现通常意味着市场在该价位的交易非常活跃。但这种活跃往往是不可持续的。 深度分析来看，这种极致的火热往往隐藏着杀机。这是主力在股价高位趁机派发出货的典型“指纹”。当散户被热度吸引入场时，主力却在抛压过大的掩护下悄然离场。根据市场常态规律，高量柱出现后，后市大概率会走低。记住：高位放巨量，往往是撤退的集结号，而非冲锋号。 机遇信号：翻倍的买力，主力在强力进攻（倍量柱） 倍量柱的界定有着严格的标准：其成交量必须比前一日放量**90%**以上，甚至翻出数倍。需要注意的是，倍量柱不一定是全场的最高量，只要相对于前一根柱子实现翻倍即可。 这种形态是明显的看涨信号。之所以说它具备攻击性，是因为倍量柱反映了主力“志在必得”的胃口。当主力急于在某个价位吞下所有浮动筹码，不惜加价买入时，才会形成这种爆发式的力量。这是主力积极介入、发动攻势的最真实证明。 反直觉发现：缩量下跌，竟然是反弹的前奏？（缩量柱） 缩量柱是指连续三天或以上成交量逐渐降低的形态。这种趋势揭示了一个重要的逻辑：市场抛压正在逐渐减轻。 在下跌趋势的末端，缩量柱的出现往往预示着空头力量的衰竭。这是一种“卖无可卖”的极点信号。一旦买盘稍有介入，股价便极易拉升。即使在上升趋势中出现“价跌量缩”，也往往是多空双方在进行短暂的技术性休整。核心逻辑在于：跌到极点，必有反转。 宁静的潜伏：低量柱里的底部机遇（低量柱） 低量柱是指阶段内成交量明显低于其他柱子的形态，是交易最清淡的标志。它不一定是绝对的“地量”，但一定是该阶段内的最低水平。 与高量柱的“喧嚣”相对立，低量柱代表了市场的“寂静”。然而，这通常是聪明资金在悄悄积蓄力量的时刻。在抛压极小的环境下，买盘在暗中积累。高量柱是风险的预警，而低量柱则是底部机遇的潜伏期，预示着后市大概率看涨。 变盘前夜：多空双方的最后对峙（平量柱） 平量柱要求至少两根（也可以是连续多根）量柱的数值基本持平，上下浮动差距不应超过5%。 这是多空双方短兵相接的争夺。 多空对峙。 静待突破。 一触即发。 作为变盘的征兆，平量柱的方位至关重要：若在上涨行情中出现，预示动力衰竭，大概率向下调整；若出现在下跌趋势中，则往往是底部构筑，极可能迎来反弹。 警惕：看似强劲的连增，实则是衰竭的开始（梯量柱） 梯量柱表现为连续三天或以上成交量逐渐抬高，像阶梯一样拾级而上。它是缩量柱的对称形态。 很多投资者误以为连续放量是好事，但从实战逻辑看，这往往意味着“上涨即将衰弱”。持续的放量上涨需要消耗天量的资金和动力，这种能量消耗是不可持续的。 一旦动力不足，上涨趋势就会放缓，甚至反转。 当你看到这种阶梯式放量时，必须保持高度警惕。这是买方力量最后的一搏，往往也是动力即将消耗殆尽的警戒线。 结语：从“看热闹”到“看门道” 通过对高、倍、缩、低、平、梯这六种量柱形态的系统梳理，我们不再被纷繁复杂的成交量所迷惑。成交量不是冰冷的数字，而是资金搏杀的血脉，是主力心理的真实投射。 只有学会分析量价同步与背离的关系，才能在复杂多变的市场中精准判断。下一次当你看到成交量突然放大时，你会选择随波逐流，还是会停下来思考，这究竟是主力的进攻还是最后的疯狂？ 一句话回答： AlphaFeed 的 Python SDK 底层就是一套标准 REST API——基址 路径、请求头带x-api-key、响应统一包在 {"data": ...}` 里。所以用 cURL、Go、Node.js、Java 甚至 Excel Power Query 都能直连，不装 SDK 也能取 A 股/美股/港股数据。 为什么会有这个问题 / 传统做法的痛点 不是所有人都用 Python：后端服务可能是 Go/Java，前端要用 Node.js，运维想先用 cURL 验证连通性。很多数据源要么只有 Python 包、要么 REST 文档残缺、鉴权方式含糊，导致非 Python 用户接入成本很高。 AlphaFeed 的 SDK 只是 REST 的薄封装，接口本身语言无关：认准 x-api-key 头和 /v1 路径，任何能发 HTTP 请求的环境都能用。 分步骤解决（每步配可运行代码） 第 1 步：用 cURL 验证连通性（取日线） curl -s -H "x-api-key: $ALPHAFEED_API_KEY" \ " 返回是紧凑列式 JSON（每个字段一个数组，靠下标对齐）： {"data":{"timestamp":[1788451200000,1788710400000], "open":[1295.88,1324.0],"high":[1338.86,1333.6], "low":[1295.6,1312.66],"close":[1330.0,1316.01], "volume":[45416,25250],"amount":[6022594729.0,3336029500.0]}} 注意：REST 原始返回不含 symbol/name/trade_date 这些列——它们是 Python SDK 的 to_dataframe=True 帮你补的。自己直连时按 timestamp 自行补列即可。 第 2 步：实时快照（单只、多只、全市场池） # 多只（≤20 个用 GET，逗号分隔） curl -s -H "x-api-key: $ALPHAFEED_API_KEY" \ " # 很多只 / 全市场池用 POST curl -s -X POST -H "x-api-key: $ALPHAFEED_API_KEY" \ -H "Content-Type: application/json" \ -d '{"symbols":["600519.SH","000001.SZ"]}' \ " # 整个标的池（需 Starter 及以上） curl -s -H "x-api-key: $ALPHAFEED_API_KEY" \ " 快照返回是对象数组，含 last_price/prev_close/open/high/low/volume/amount 及 ext（name/change_pct/amplitude/turnover_rate 等）。 第 3 步：五档盘口与标的元数据 # 五档盘口 curl -s -H "x-api-key: $ALPHAFEED_API_KEY" \ " # 标的元数据（涨跌停价、总/流通股本、上市日期等） curl -s -H "x-api-key: $ALPHAFEED_API_KEY" \ " 第 4 步：任意语言直连（以 Python requests 为例） 不装 alphafeed，只用 requests 也能取： import os, requests, pandas as pd BASE = " H = {"x-api-key": os.environ["ALPHAFEED_API_KEY"]} r = requests.get(f"{BASE}/v1/klines", headers=H, timeout=30, params={"symbol": "600519.SH", "period": "1d", "count": 3, "adjust": "forward"}) r.raise_for_status() df = pd.DataFrame(r.json()["data"]) # 列式 JSON 直接进 DataFrame df["symbol"] = "600519.SH" # 需要的话自行补列 print(df) Node.js 用 fetch、Go 用 net/http 同理：设 x-api-key 头、GET 对应 /v1 路径、解析 data 字段即可。 端点与鉴权一览 端点 方法 关键参数 用途 /v1/klines GET symbol,period,count,adjust,start_time,end_time 单只 K 线 /v1/klines/batch GET symbols(逗号) + 同上 批量 K 线 /v1/klines/intraday GET symbol 当日分时 /v1/klines/intraday/batch GET symbols 批量当日分时 /v1/klines/ex-factors GET symbols 除权因子 /v1/quotes GET/POST symbols 或 universes 实时快照 /v1/depth GET symbol 五档盘口 /v1/instruments GET symbols 标的元数据 鉴权：所有请求头都要带 x-api-key: <你的Key>；基址 ALPHAFEED_BASE_URL` 覆盖）。 关键坑与注意事项 响应统一包 data：成功返回都是 {"data": ...}，先取 data 再解析。 K 线是列式不是行式：/v1/klines 的 data 是「字段→数组」，要靠下标对齐成行；SDK 的行式 DataFrame 是客户端转换的结果。 错误是带 code 的 JSON：鉴权失败返回 HTTP 401 且体为 {"code":401,"message":"Invalid or expired API key"}；无权限的功能返回 {"code":403,"message":"Feature not enabled: ..."}（如未开通的五档批量）；限频为 429。自己直连时务必按状态码处理，SDK 才有内置重试。 GET vs POST 的阈值：quotes 在标的很多（或用 universes）时建议用 POST，避免 URL 过长。 权限分层：单标的历史/快照有免费额度；universes 全市场池、部分市场数据需 Starter 及以上，直连同样受此限制。 时间是毫秒时间戳：timestamp、start_time、end_time 均为毫秒。 常见问题（FAQ） Q：REST 直连和 SDK 有什么差别？ A：数据完全一样。SDK 额外做了：行式 DataFrame 转换、批量自动分批并发、内置重试（429/5xx/超时）、名称本地缓存、类型化异常。直连要自己实现这些工程细节。 Q：Authorization: Bearer 能用吗？ A：AlphaFeed 用的是 x-api-key 请求头，不是 Bearer Token，别混用。 Q：Go/Java/Excel 能用吗？ A：能。只要能发带自定义头的 HTTPS GET/POST 并解析 JSON 就行；Excel/Power BI 用 Power Query 的 Web.Contents 加 x-api-key 头也可以。 Q：需要付费吗？ A：免费额度可直连体验单标的历史与快照；全市场池、部分市场需 Starter 及以上，详见官网定价页。 小结 AlphaFeed 的能力不绑定 Python——它本质是一套 x-api-key 鉴权、/v1 路径、data 包裹的标准 REST API。认准这三点，你可以用 cURL 秒验连通、用任意语言直连，把 A 股/美股/港股数据接进你现有的技术栈。想省掉重试、分批、DataFrame 转换这些工程活，再上 Python SDK 即可。 参考 AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 引言：满心欢喜等着数钱，结果被“闷杀”？ 在A股摸爬滚打过的家人们都知道，最让人肾上腺素飙升的莫过于抓到涨停板。本以为次日会继续“一字”躺赢，结果不少人等来的却是开盘大幅低开，或者冲高无力后的反手杀跌。这种从云端掉进冰窟窿的落差，常让新手在恐慌中乱了方寸。 其实，涨停后的回调究竟是主力的“诱多陷阱”还是“洗盘良机”？这背后的K线迷阵，藏着主力最真实的操作底牌。看不懂次日走势，你抓到的可能不是金元宝，而是烫手的山芋。 核心逻辑：涨停回调的“降维打击” 主力费尽心思拉出涨停，次日却走得垂头丧气，这可不是在做慈善给散户抬轿子。 “涨停回调。是主力资金常用的手段之一，目的可能是防止散户资金离场。” 内行人都明白，主力通过低开低走，利用人性中“追涨杀跌”的弱点，让场外资金在视觉上产生该股“转弱”的错觉，从而达成“视觉遗忘”。这种战术能有效降低后续二波拉升时的跟风阻力，是典型的战略性修正。 形态一：低开高走——主力的“困兽斗”与自救 如果你发现股票低开，但随后有资金护盘向上探路，这通常是主力在“自救”。 底层逻辑： 前一交易日的主力可能也被自己拉出的高位筹码给“套”住了，或者洗盘用力过猛导致部分“打板族”没跑掉。主力必须拉升一段空间，才能腾挪出货。 核心观察位： 盯死前一交易日的收盘价。 出局信号： 如果反弹始终摸不到昨天的收盘价，说明上方抛压重得惊人，主力带不动了。一旦放量突破昨收，大阳线甚至连板可期；反之，若摸高乏力，赶紧离场，别陪主力一起“站岗”。 形态二：高开高走——急需援军的“接力赛” 这种走势往往意味着主力手里的筹码并不足以支撑接下来的独舞，他们急需市场的“合力”。 底层逻辑： 主力在通过高开释放信号，吸引场外资金进场。若援军踊跃，这股气势就能顶住。 核心观察位： 观察股价是否能在涨停位附近封死。 出局信号： 1.如果股价触及涨停却反复开板（炸板），这就是市场出现了严重的“分歧”，资金合力不足。 2.如果股价在高位横盘不动，说明买盘已经枯竭。这时候别犹豫，落袋为安是最稳妥的选择，否则第三天极大概率会被大幅低开直接“闷杀”。 形态三：一字板——极致的控盘肌肉秀 这是市场上最强悍的形态，开盘即巅峰，主力压根不想给场外的人低价上车的机会。 底层逻辑： 展现了主力志在必得的拉升决心和绝对的控盘权。 核心观察位： 买盘封单数量的动态变化。 出局信号： 盯紧封单。如果封单在盘中像融化的冰雪一样骤减，且炸板后半天补不回来，说明主力撤单了，抛压已经失控，必须第一时间清仓。 形态四：高开低走——防守反击的止损预警 最怕的就是这种开盘满堂彩、随后一路阴跌的走势，这是典型的“杀猪盘”前兆。 底层逻辑： 主力在高位已经完成了部分派发，剩下的就是利用散户的惯性思维进行最后收割。 核心观察位： 严格执行“3%止损原则”。 出局信号： 1.如果从开盘高位回落超过3个点，且跌破了昨天的收盘价，逻辑已经破坏，必须执行止损。 2.观察第一波反弹：如果反弹连今日的开盘价都收不回来，这就是明确的止盈（Take Profit）离场信号，绝不恋战。 形态五：平开高走——依靠共识的“群众运动” 主力控盘能力有限，完全看市场的情绪和合力。 底层逻辑： 这种票没有“一言九鼎”的主力，全靠散户和游资的共鸣。 核心观察位： 量能的持续性。 出局信号： 平开高走的票最容易留下长长的上影线。一旦你发现上涨显得“吃力”，量能跟不上，说明合力已经透支。趁着还没跳水，果断撤离。 形态六：大幅低开/跌停——妖股的“生死时速” 这种情况通常发生在前期涨翻天的“妖股”身上，是风险集中引爆的至暗时刻。 底层逻辑： 情绪极度透支后的崩盘，或者是突发利空导致的踩踏。 核心观察位： 有没有大单“翘板”（巨量买盘强行打开跌停）。 出局信号： 如果有人翘板，或许还有翻红的一线生机，供短线高手博弈。但如果你看到跌停板上封单如山，且毫无动静，那次日大概率还会“地板”开盘。这属于生死局，普通玩家别去肉身试错。 总结：在喧嚣中读懂筹码的“语言” 老道提醒：股票交易不是简单的看涨看跌，而是要透过K线看清背后的资金博弈。涨停只是主力的名片，次日的走势才是主力的心声。在实战中，你要建立自己的策略库，而不是被一个涨停板冲昏了头脑。 在主力精心布置的K线迷阵中，你是那个看清底牌的人吗？如果你还没想清楚，欢迎多看几遍今天的私房笔记。 
