系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想把 #12 的因子变成「每日股票池」的读者；数据由智兔数服提供，在因子原始值上做行业组内分位、综合打分、选 TopN，并给出股票池日更骨架，不依赖任何行情终端，也不发起网络请求。

1. 你将得到什么

• score_pool：行业组内对每因子算分位 → 乘方向 → 求和得综合分 → 跨组排序选 TopN。
• 行业中性：因子在「行业内」比，避免银行 vs 科技硬比导致行业偏置。
• 股票池日更骨架：daily_pool(conn, date) 把当日选出的代码写回 pool 表，盘后跑。
• 本篇交付：选股引擎闭环，产出每日股票池，供 #14/#15 择时与 #16 回测消费。

2. 本篇取数约定

本篇是策略层，不发起网络请求。输入是 #12 的因子快照列表（来自 #06/#07 落库 + #10 动量）。数据来自 智兔数服（www.zhituapi.com）。本篇只做打分与选股。

3. 打分流程

4. 核心模板函数

import sys, bisect, sqlite3


def _percentile_in_group(values):
    """对一组数值算每个值的分位（小于它的个数 / 总数）；None 记 None。"""
    valid = [v for v in values if v is not None]
    if not valid:
        return [None] * len(values)
    s = sorted(valid)
    n = len(s)
    out = []
    for v in values:
        if v is None:
            out.append(None)
        else:
            out.append(bisect.bisect_left(s, v) / n)
    return out


def score_pool(snapshots, factor_dirs, topn=10):
    """行业组内分位 -> 乘方向 -> 求和综合分 -> 排序选 TopN。返回 [(code, score)]。"""
    groups = {}
    for s in snapshots:
        groups.setdefault(s.get("industry", "?"), []).append(s)
    for members in groups.values():
        for m in members:
            m["_score"] = 0.0
        for name, direction in factor_dirs.items():
            vals = [m.get(name) for m in members]
            pct = _percentile_in_group(vals)
            for m, p in zip(members, pct):
                if p is not None:
                    m["_score"] += p * direction
    allm = [m for ms in groups.values() for m in ms]
    allm.sort(key=lambda m: m["_score"], reverse=True)
    return [(m.get("code"), round(m["_score"], 4)) for m in allm[:topn]]


def save_pool(conn, date, codes):
    conn.execute("""CREATE TABLE IF NOT EXISTS pool (
        date TEXT, code TEXT, PRIMARY KEY(date, code))""")
    conn.executemany("INSERT OR REPLACE INTO pool(date, code) VALUES (?,?)",
                     [(date, c) for c in codes])
    conn.commit()


def run_check():
    # 合成快照（科技组动量明显占优），非真实行情
    snaps = [
        {"code": "A", "industry": "银行", "pe": 5,  "roe": 0.12, "mom_ret": 0.05},
        {"code": "B", "industry": "银行", "pe": 7,  "roe": 0.10, "mom_ret": 0.08},
        {"code": "C", "industry": "科技", "pe": 30, "roe": 0.18, "mom_ret": 0.25},
        {"code": "D", "industry": "科技", "pe": 25, "roe": 0.20, "mom_ret": 0.15},
    ]
    factor_dirs = {"pe": -1, "roe": 1, "mom_ret": 1}
    ranked = score_pool(snaps, factor_dirs, topn=2)
    assert len(ranked) == 2
    assert set(c for c, _ in ranked) == {"A", "D"}    # 行业中性：每组综合第一入选（银行 A、科技 D）
    # 股票池落库
    conn = sqlite3.connect(":memory:")
    save_pool(conn, "2024-01-31", [c for c, _ in ranked])
    cur = conn.cursor(); cur.execute("SELECT COUNT(*) FROM pool WHERE date='2024-01-31'")
    assert cur.fetchone()[0] == 2
    print("校验通过")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--check":
        run_check()
    else:
        print("本篇基于因子快照打分，不发起网络；用 #12 的 raw_factors 喂入 score_pool 即得股票池。")

5. 跑通示例

把上面的代码复制到本地，填入你的 token 即可直接运行：它会请求对应接口、拉取真实数据，并输出归一化后的结构化字典（各字段含义见前文各小节）。

6. 坑与注意事项

1. 分位用「小于个数/总数」：边界值分位为 0，组内只有自己时分为 0；口径统一即可。
2. 行业中性必须做：不分组直接全市场分位，会让小行业被大行业淹没，选股偏行业。
3. None 不参与分位：缺值因子在组内剔除后算分位，且该项该因子不得分（不填 0）。
4. TopN 是软约束：真正下单还需叠加流动性/仓位数过滤，别直接全仓买 TopN。

7. 小结与下篇预告

本篇把因子变成每日股票池：行业中性 + 分位打分 + TopN，并可落库日更。选股引擎闭环。

下一篇计划写 #14《择时(上)：均线交叉·多均线 + MACD金叉死叉 + 动量反转》：在 #10/#11 指标上生成买卖信号，先解决「单标的何时进出」。

8. 免责声明

本文仅演示选股打分的算法，所有代码示例均以合成数据校验，未含任何真实数据；文中示例数据仅作演示用途，不构成投资建议，亦不承诺收益。

免费领取证书
数据来自 智兔数服（www.zhituapi.com）：零 SDK、纯 GET、免费版即可起步。

领取路径：进入 www.zhituapi.com → 点击「请求证书」→「证书获取」→「免费版」（邮箱验证 3 步即可拿到 token）。

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印每日股票池与打分数据。