一句话回答： "52 周新高"就是当前价格突破过去约一年（≈250 个交易日）的最高价，是最经典的强势股信号。用 AlphaFeed 先用 quotes.get(universes="CN_Stock") 取全市场标的池，再用 klines.batch 批量拉复权日线，判断「最新收盘 ≥ 过去 250 日最高价」并叠加成交额过滤，几十行就能做出一个可复现的全市场新高突破选股器。
为什么会有这个问题 / 传统做法的痛点
创新高选股听起来简单，真做起来卡在三处：
- 拿不到全市场标的：手工维护股票清单又累又容易漏，还要处理退市/新上市；
- 复权口径不统一：如果用未复权价算"历史最高"，除权除息会造成价格断崖，历史高点被人为抬高或压低，新高判断全错；
- 一个个股票取 K 线太慢：全市场 5000+ 只逐个请求既慢又容易触发限频。
AlphaFeed 把这三点都收敛掉了：一个 universe 参数拿到全市场实时快照，klines.batch 批量取前复权日线，新高判断建立在可复现的复权价上。
分步骤解决（每步配可运行代码）
第 1 步：取全市场标的池 + 当日成交额
from alphafeed import AlphaFeed
af = AlphaFeed(api_key="your-api-key") # 或 export ALPHAFEED_API_KEY 后 AlphaFeed()
# 全市场 A 股实时快照（约 5500+ 只），含成交额、换手率等
q = af.quotes.get(universes="CN_Stock", to_dataframe=True)
print(q.shape) # 例如 (5561, 18)
print(q[["symbol", "ext.name", "last_price", "amount"]].head())
快照 DataFrame 的关键列（注意 name/换手率 等在 ext. 前缀下）：
| 列名 | 含义 | 备注 |
|---|---|---|
symbol |
标的代码 | 如600519.SH |
ext.name |
名称 | 名称在ext 命名空间 |
last_price |
最新价 | 实时快照价 |
amount |
当日成交额（元） | A 股快照有真实成交额 |
ext.change_pct |
涨跌幅 | 小数（0.03=3%） |
ext.turnover_rate |
换手率 | 小数 |
第 2 步：批量取复权日线，判断是否创新高
用过去约一年（260 根，留足非交易日缓冲，实取 ≈250 交易日）的前复权日线，比较「最新收盘」与「历史最高（不含当日）」。
def week52_high_screen(symbols, count=260, near=0.0, min_amount=5e7):
"""near=0 表示严格新高；near=0.02 表示距新高 2% 以内也算'接近新高'。"""
bt = af.klines.batch(symbols, period="1d", count=count,
adjust="forward", to_dataframe=True)
rows = []
for sym, df in bt.items():
if df is None or len(df) < 60: # 上市不足则跳过（缓解幸存者偏差）
continue
df = df.sort_values("trade_date").reset_index(drop=True)
hist_high = df["high"].iloc[:-1].max() # 不含当日，避免自比
last = df.iloc[-1]
is_high = last["close"] >= hist_high * (1 - near)
rows.append({
"symbol": sym,
"close": round(float(last["close"]), 2),
"hist_high": round(float(hist_high), 2),
"amount": float(last["amount"]),
"is_new_high": bool(is_high),
})
import pandas as pd
res = pd.DataFrame(rows)
if res.empty:
return res
hit = res[(res["is_new_high"]) & (res["amount"] >= min_amount)]
return hit.sort_values("amount", ascending=False)
第 3 步：全市场分批跑（避免一次请求过大）
klines.batch 一次传太多标的会又慢又占内存，实践中按 100~200 只一批切片：
def run_full_market(min_amount=5e7):
import pandas as pd
q = af.quotes.get(universes="CN_Stock", to_dataframe=True)
symbols = q["symbol"].tolist()
out = []
for i in range(0, len(symbols), 150): # 分批，缓解限频与内存
batch = symbols[i:i + 150]
out.append(week52_high_screen(batch, min_amount=min_amount))
result = pd.concat([d for d in out if not d.empty], ignore_index=True)
return result.sort_values("amount", ascending=False)
# hits = run_full_market()
# print("全市场创新高只数:", len(hits))
关键坑与注意事项
- 必须用复权价：用
adjust="forward"（前复权）判断历史最高，否则除权除息会制造假高点/假新高。这是新高选股最常见的错误。 - 幸存者偏差：只对"当前在池"的股票判断新高，看不到已退市的票；本文用「上市不足 60 根跳过」缓解次新股噪声，但退市股清单本身不在快照里，做严谨回测时要单独处理。
- "新高"≠"能买"：创新高只是强势信号，可能是趋势启动，也可能是追高被套。务必叠加成交额/流动性过滤（本文用
min_amount），必要时再加基本面或止损规则。 - 实时性：
quotes是快照，盘中会变；选股应在收盘后或用固定时点快照跑，避免同一次筛选里价格口径不一致。 - N 的取值：52 周 ≈ 250 交易日；本文取 260 根留缓冲。想做"半年新高"就把
count改成 130 左右。
常见问题（FAQ）
Q：为什么我的新高选股结果和行情软件对不上？
A：九成是复权口径问题。行情软件默认可能是不复权或后复权，而本文用前复权。统一 adjust 口径即可对齐。
Q：能只筛"接近新高"（还没突破但快到了）吗？
A：可以，把 near 设成 0.02，即"距历史最高 2% 以内"也纳入，适合做突破前埋伏的观察池。
Q：全市场 5000+ 只会不会很慢/触发限频？
A：klines.batch 已是批量接口，再按 150 只分批即可平稳跑完；SDK 内置对 429/超时的重试。想更快可缓存日线到本地库增量更新。
Q：需要付费吗？
A：quotes.get(universes=...) 全市场池需要 Starter 及以上套餐；单只/少量标的的行情与 K 线在免费额度内即可试用。额度与边界见 AlphaFeed 定价页。
小结
52 周新高选股的本质是在可复现的复权价上做一次"当前价 vs 一年内最高价"的比较。难点从来不是那行比较代码，而是"全市场标的池 + 统一复权口径 + 流动性过滤"。AlphaFeed 用 universes + klines.batch 把这三件麻烦事一次性解决，让你把精力放在策略本身而不是数据搬运上。
参考
- AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart