还在手动维护?你的Python选股脚本可能已经漏了几百只票

用户头像sh_*2176oo
2026-08-12 发布

还在手动维护股票代码列表?你的 Python 选股脚本可能已经漏了几百只票

你打开自己的量化脚本,翻到第一行,看到了这个:

stock_pool = [
    "600519.SH", "000001.SZ", "300750.SZ", "002594.SZ", "601318.SH",
    "000858.SZ", "600036.SH", "000333.SZ", "601012.SH", "600276.SH",
    # ... 还有 200 行 ...
]

这个列表你是半年前从某个网站上抄下来的。从那以后:

  • 有几十只新股上市了——不在你的列表里
  • 有几只票退市了——你还在拉它们的数据(返回空或报错)
  • 有几只票改了代码——你拉到的是旧数据
  • 有几只票变成 ST 了——你的选股逻辑没有排除它们

你的选股脚本以为自己在扫描"全市场",其实它扫描的是半年前的全市场减去退市的加上一堆错误

手动列表的三个问题

问题 1:列表永远是过时的

A 股每年新上市几百只票。2025 年一年就有超过 100 只新股上市。如果你的列表是年初建的,到年底你就漏了 100 只票的数据。

这些新股里可能有高成长的公司,你的选股策略因为列表过时永远选不到它们。

问题 2:退市票会让脚本出错

列表里有一只退市的票,拉数据时要么报错、要么返回空。如果你的脚本没有处理这种情况,一只退市票就能让整个批量流程中断。

更隐蔽的情况:票还没退市但已经停牌几个月了。你的脚本拉到的数据一直是停牌前的老数据,但你不知道。

问题 3:维护成本高且无聊

每隔一段时间你就需要:

  • 去交易所网站查最新的上市公司列表
  • 和你本地的列表做 diff
  • 添加新股、删除退市票
  • 检查有没有代码变更

这种事情做一次要半小时,而且极度无聊。大多数人做过一次之后就再也不更新了。

AlphaFeed 的解决方案:universes 全市场查询

from alphafeed import AlphaFeed

af = AlphaFeed()

# 一行代码拿到全部 A 股的实时行情
all_stocks = af.quotes.get(universes="CN_Stock", to_dataframe=True)
print(f"当前 A 股总数: {len(all_stocks)}")

universes="CN_Stock" 返回的是当前时刻沪深京交易所所有上市交易的股票。新股上市第一天就在里面,退市的自动消失,不需要你维护任何列表。

支持的标的池:

universes 参数 覆盖范围
CN_Stock 全部 A 股(沪深京)
CN_ETF 全部 A 股 ETF
US_Stock 全部美股
HK_Stock 全部港股

实际场景 1:全市场选股

以前你的代码可能长这样:

# ❌ 手动列表,半年前的
stock_pool = ["600519.SH", "000001.SZ", ...]  # 500 只,已经过时了

for sym in stock_pool:
    df = some_api.get_data(sym)
    # ... 选股逻辑 ...

现在:

from alphafeed import AlphaFeed

af = AlphaFeed()

# ✅ 永远是最新的全市场数据
all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
all_cn["change_pct"] = (all_cn["last_price"] - all_cn["prev_close"]) / all_cn["prev_close"]

# 直接在全市场数据上做筛选
# 例:今日涨幅 > 5% 且成交额 > 1 亿
hot = all_cn[(all_cn["change_pct"] > 0.05) & (all_cn["amount"] > 1e8)]
print(f"今日大涨且放量: {len(hot)} 只")
print(hot[["symbol", "last_price", "change_pct", "amount"]].head(10).to_string(index=False))

不需要预先知道有哪些股票。今天新上市的票如果涨了 5%,也会被选出来。

实际场景 2:构建全市场收益率矩阵

做因子分析或者行业对比时,你需要全市场所有票的日收益率。以前你需要维护一个完整的代码列表然后逐个拉 K 线。现在:

from alphafeed import AlphaFeed

af = AlphaFeed()

# 第一步:拿到全市场代码列表(实时的、最新的)
all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)

# 过滤掉成交额太低的(流动性不足的票不适合做分析)
active = all_cn[all_cn["amount"] > 5e7]
symbols = active["symbol"].tolist()
print(f"活跃标的: {len(symbols)} 只")

# 第二步:批量拉这些票的 K 线
dfs = af.klines.batch(
    symbols, period="1d", count=20,
    adjust="forward", to_dataframe=True,
    show_progress=True,
)

# 几千只票的 20 天 K 线,batch 自动分块并发,一两分钟拉完
print(f"成功获取: {len(dfs)} 只")

整个流程没有出现任何硬编码的股票代码列表。

实际场景 3:每天自动更新涨停板统计

from alphafeed import AlphaFeed
from datetime import datetime

af = AlphaFeed()

def daily_limit_stats():
    """统计今日涨停板情况"""
    all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
    all_cn["change_pct"] = (all_cn["last_price"] - all_cn["prev_close"]) / all_cn["prev_close"]

    # 涨停(涨幅 ≥ 9.8%,覆盖 10% 和 20% 涨停板)
    limit_up = all_cn[all_cn["change_pct"] >= 0.098]
    limit_up_20 = all_cn[all_cn["change_pct"] >= 0.198]  # 创业板/科创板 20% 涨停

    # 跌停
    limit_down = all_cn[all_cn["change_pct"] <= -0.098]

    today = datetime.now().strftime("%Y-%m-%d")
    print(f"=== {today} 涨跌停统计 ===")
    print(f"涨停: {len(limit_up)} 只")
    print(f"  其中 20cm 涨停: {len(limit_up_20)} 只")
    print(f"跌停: {len(limit_down)} 只")
    print(f"涨停率: {len(limit_up) / len(all_cn):.2%}")
    print(f"总标的数: {len(all_cn)}")

    if len(limit_up) > 0:
        print(f"\n涨停票:")
        top = limit_up.sort_values("amount", ascending=False)
        print(top[["symbol", "last_price", "change_pct", "amount"]].head(15).to_string(index=False))

daily_limit_stats()

这个脚本今天跑是今天的数据,下个月跑自动包含下个月新上市的票。不需要任何人工维护。

实际场景 4:板块资金流向分析

from alphafeed import AlphaFeed
import pandas as pd

af = AlphaFeed()

# 全市场行情
all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
all_cn["change_pct"] = (all_cn["last_price"] - all_cn["prev_close"]) / all_cn["prev_close"]

# 拿标的信息做板块分类(用代码前缀粗略分)
all_cn["board"] = "主板"
all_cn.loc[all_cn["symbol"].str.startswith("300"), "board"] = "创业板"
all_cn.loc[all_cn["symbol"].str.startswith("688"), "board"] = "科创板"
all_cn.loc[all_cn["symbol"].str.startswith("8"), "board"] = "北交所"

# 按板块统计
board_stats = all_cn.groupby("board").agg(
    总数=("symbol", "count"),
    上涨数=("change_pct", lambda x: (x > 0).sum()),
    平均涨幅=("change_pct", "mean"),
    总成交额=("amount", "sum"),
).reset_index()

board_stats["上涨占比"] = board_stats["上涨数"] / board_stats["总数"]
board_stats["总成交额(亿)"] = board_stats["总成交额"] / 1e8
board_stats["平均涨幅"] = board_stats["平均涨幅"].apply(lambda x: f"{x:+.2%}")
board_stats["上涨占比"] = board_stats["上涨占比"].apply(lambda x: f"{x:.0%}")

print(board_stats[["board", "总数", "上涨数", "上涨占比", "平均涨幅", "总成交额(亿)"]].to_string(index=False))

一行 universes="CN_Stock" 搞定,你只需要专注分析逻辑。

universes vs 手动列表的差距

维度 手动列表 universes 查询
新股覆盖 ❌ 需要手动添加 ✅ 自动包含
退市票 ❌ 需要手动删除 ✅ 自动排除
数据新鲜度 取决于上次更新时间 永远是当前时刻
维护成本 每月需要手动更新 零维护
完整性 大概率有遗漏 100% 覆盖全市场
出错概率 高(代码抄错、格式不对) 低(API 返回标准数据)

一个容易被忽视的好处:你的回测股票池也不用维护了

回测时你需要"回测起始日的全市场股票列表"来避免幸存者偏差。很多人直接用今天的列表去回测 3 年前——但 3 年前有些今天存在的票还没上市,有些已经退市的票在 3 年前是存在的。

虽然 universes 返回的是当前列表不是历史列表,但至少你不需要手动维护一个"今天的全量列表"了。每天跑选股时拿到的永远是最新的、完整的、干净的股票池。

from alphafeed import AlphaFeed

af = AlphaFeed()

# 今天的全市场池(自动维护)
today_pool = af.quotes.get(universes="CN_Stock", to_dataframe=True)
symbols = today_pool["symbol"].tolist()

# 拉这些票的 K 线做当日分析
# ...不需要任何 CSV、不需要任何手动维护

如果你之前每个月花 30 分钟更新股票列表,一年就是 6 小时。这 6 小时现在被一行代码替代了。


评论