【量化系统从零构建 #14】择时(上)：均线交叉·多均线 + MACD金叉死叉 + 动量反转

【量化系统从零构建 #14】择时(上)：均线交叉·多均线 + MACD金叉死叉 + 动量反转

系列：《量化系统从零构建》｜连载项目 · 纯 GET 取数 · 仅依赖 requests
适用：想解决「单标的何时进出」的读者；在 #10/#11 指标上生成买卖信号：均线交叉、MACD 金叉死叉、动量反转，不依赖任何行情终端，也不发起网络请求。

1. 你将得到什么

• ma_cross：快慢均线交叉信号，金叉 1 / 死叉 -1 / 其余 0，前 slow-1 为 None。
• macd_cross：DIF/DEA 交叉信号，金叉 1 / 死叉 -1。
• momentum_reversal：动量反转预警，超买 -1 / 超卖 1，作为反向参考。
• 本篇交付：单标的择时信号三件套，#15 把它们和资金面/情绪组合成综合择时。

2. 本篇取数约定

本篇是策略层，不发起网络请求，吃 #09 复权后的 close 序列（或外部传入）。指标计算纯 Python，不引 numpy/pandas。

3. 信号定义

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):
"""快慢均线交叉：1 金叉 / -1 死叉 / 0 其它；前 slow-1 为 None。"""
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 交叉：1 金叉 / -1 死叉 / 0 其它；前 slow-1 为 None。"""
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 momentum_reversal(close, window=20, thr=0.3):
"""动量反转：近 window 收益超 thr 为超买(-1)，低于 -thr 为超卖(1)，否则 0。"""
out = []
for i in range(len(close)):
if i < window:
out.append(None); continue
ret = close[i] / close[i – window] – 1
out.append(–1 if ret > thr else (1 if ret < –thr else 0))
return out

def run_check():
# 合成：先涨后跌再涨，非真实行情
close = [10,11,12,13,14,15,16,17,18,19,20,19,18,17,16,15,16,17,18,19,20,21,22,23,24]
sig = ma_cross(close, 5, 20)
assert len(sig) == len(close)
assert all(v is None for v in sig[:19]) # 前 slow-1 无信号
# 针对性验证 ma_cross 能识别金叉/死叉（短序列 fast=2 slow=3）
sc = ma_cross([1, 2, 3, 4, 5, 4, 3, 2, 3, 4, 5], 2, 3)
assert 1 in sc and –1 in sc
mc = macd_cross(close)
assert len(mc) == len(close)
assert 1 in mc and –1 in mc # MACD 应出现金叉与死叉
mr = momentum_reversal(close, window=20)
assert mr[20] is not None and mr[–1] is not None
print("校验通过")

if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "–check":
run_check()
else:
print("本篇基于 adj_close 序列计算，不发起网络；用 #09 的 clean 输出喂入即生成信号。")

5. 跑通示例

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

6. 坑与注意事项

7. 小结与下篇预告

本篇交出单标的择时三件套（均线 / MACD / 动量反转）。下一步把它们和资金面、市场情绪组合成综合择时。

下一篇计划写 #15《择时(下)：北向资金跟随·涨跌停连板情绪·综合择时打分》：把 #14 信号叠加北向资金流与涨跌停情绪，做多信号投票的综合择时打分。

8. 免责声明

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