【量化系统从零构建 #15】择时(下)：北向资金跟随·涨跌停连板情绪·综合择时打分

【量化系统从零构建 #15】择时(下)：北向资金跟随·涨跌停连板情绪·综合择时打分

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

1. 你将得到什么

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

2. 本篇取数约定

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

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. 坑与注意事项

7. 小结与下篇预告

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

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

8. 免责声明

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