系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想把单标的信号升级成「市场级综合择时」的读者；数据由智兔数服提供，在 #14 信号上叠加北向资金流与涨跌停情绪，做多信号投票的综合打分，不依赖任何行情终端，也不发起网络请求。

1. 你将得到什么

• combine_timing：把均线 / MACD / 北向资金 / 涨跌停情绪四类信号按 ±1 投票，求和得综合分（正多负空）。
• market_timing：取最新一根 bar 的均线 + MACD 信号，叠加当日北向与情绪，输出综合择时分。
• 本篇交付：策略层择时闭环，综合分供 #16 回测与 #18 风控直接消费。

2. 本篇取数约定

本篇是策略层，不发起网络请求。均线 / MACD 吃 #09 的 close；北向资金来自 #07 money_flow（或 /ht/nbzj）；涨跌停计数来自 #31 类实时接口。数据来自 智兔数服（www.zhituapi.com）。本篇只做信号组合。

3. 综合择时打分

四类各 ±1 投票，综合分范围约 [-4, 4]；绝对值越大，方向越一致。

4. 核心模板函数

import sys


def _sma(series, n):
    out = []
    for i in range(len(series)):
        if i < n - 1:
            out.append(None)
        else:
            out.append(sum(series[i - n + 1:i + 1]) / n)
    return out


def _ema(series, n):
    if not series:
        return []
    k = 2 / (n + 1)
    out = [series[0]]
    for i in range(1, len(series)):
        out.append(series[i] * k + out[i - 1] * (1 - k))
    return out


def _macd(close, fast=12, slow=26, signal=9):
    ef, es = _ema(close, fast), _ema(close, slow)
    dif = [ef[i] - es[i] for i in range(len(close))]
    dea = _ema(dif, signal)
    return dif, dea


def ma_cross(close, fast=5, slow=20):
    f, s = _sma(close, fast), _sma(close, slow)
    out = []
    for i in range(len(close)):
        if f[i] is None or s[i] is None:
            out.append(None); continue
        if i == 0 or f[i - 1] is None or s[i - 1] is None:
            out.append(0); continue
        if f[i - 1] <= s[i - 1] and f[i] > s[i]:
            out.append(1)
        elif f[i - 1] >= s[i - 1] and f[i] < s[i]:
            out.append(-1)
        else:
            out.append(0)
    return out


def macd_cross(close, fast=12, slow=26, signal=9):
    dif, dea = _macd(close, fast, slow, signal)
    out = []
    for i in range(len(close)):
        if dif[i] is None or dea[i] is None:
            out.append(None); continue
        if i == 0 or dif[i - 1] is None or dea[i - 1] is None:
            out.append(0); continue
        if dif[i - 1] <= dea[i - 1] and dif[i] > dea[i]:
            out.append(1)
        elif dif[i - 1] >= dea[i - 1] and dif[i] < dea[i]:
            out.append(-1)
        else:
            out.append(0)
    return out


def combine_timing(ma_sig, macd_sig, north_flow, limit_up_cnt, limit_down_cnt):
    """四类信号 ±1 投票求和；正多负空。"""
    score = 0
    if ma_sig == 1: score += 1
    elif ma_sig == -1: score -= 1
    if macd_sig == 1: score += 1
    elif macd_sig == -1: score -= 1
    if north_flow > 0: score += 1
    elif north_flow < 0: score -= 1
    if limit_up_cnt > limit_down_cnt: score += 1
    elif limit_up_cnt < limit_down_cnt: score -= 1
    return score


def market_timing(close, north_flow, limit_up_cnt, limit_down_cnt, fast=5, slow=20):
    """取最新非 None 的均线/MACD 信号，叠加当日北向与情绪，返回综合分。"""
    ma = ma_cross(close, fast, slow)
    mc = macd_cross(close)
    last_ma = next((v for v in reversed(ma) if v is not None), 0)
    last_mc = next((v for v in reversed(mc) if v is not None), 0)
    return combine_timing(last_ma, last_mc, north_flow, limit_up_cnt, limit_down_cnt)


def run_check():
    # 合成单边上行收盘价，非真实行情
    close = list(range(10, 31))   # 10..30 单调涨
    # 多头环境：北向流入 + 涨多跌少
    score_bull = market_timing(close, north_flow=1.2e8, limit_up_cnt=30, limit_down_cnt=5)
    assert isinstance(score_bull, int)
    # 空头环境：北向流出 + 跌多涨少
    score_bear = market_timing(close, north_flow=-1.0e8, limit_up_cnt=5, limit_down_cnt=40)
    assert score_bear < score_bull        # 第二个更空
    # 全多信号应得高分
    assert combine_timing(1, 1, 1.0e8, 20, 5) == 4
    print("校验通过")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--check":
        run_check()
    else:
        print("本篇基于信号序列与资金面/情绪组合，不发起网络；market_timing(close, north, up, down) 即得综合分。")

5. 跑通示例

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

6. 坑与注意事项

1. 信号取「最新非 None」：实盘用 next(reversed(...)) 拿最近信号，别拿历史交叉当当前指令。
2. 北向/情绪是日频：这两类信号当天只有一个值，与分钟级均线信号粒度不同，组合时注意对齐。
3. 综合分不是仓位：score=3 不代表满仓，#18 风控再把它映射成仓位比例。
4. 投票等权简单：可给均线/MACD 更高权重，权重写在 combine_timing 里即可。

7. 小结与下篇预告

本篇把单标的信号升级成市场级综合择时：四类信号投票，综合分正多负空。策略层择时闭环。

下一篇计划写 #16《回测(上)：向量化回测引擎·成本模型》：用综合择时/选股信号驱动一笔可复现的回测，先解决「向量化计算 + 手续费/滑点成本」。

8. 免责声明

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

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

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

把代码里的 你的智兔token 换成你拿到的真实 token，上面的脚本就能直接打印综合择时打分数据。