量化数据进阶：多维实战篇（第 8 篇）：MACD：金叉死叉与柱状动能

量化数据进阶：多维实战篇（第 8 篇）：MACD：金叉死叉与柱状动能

一、前言

上一篇《KDJ与BOLL：超买超卖与布林带宽》我们用 hsstock/history/kdj 与 hsstock/history/boll 补齐了「位置」这一维度——KDJ 回答「是不是超买超卖」，BOLL 回答「波动是不是压到极限」。但位置和方向的确认，都还缺少一个关键维度：动能。股价可以在低位横盘很久（KDJ 超卖、带宽收敛），却迟迟不启动，缺的就是动能的积累与释放。

本篇收束到本模块技术指标的最后一块拼图——趋势动能，使用接口：

• hsstock/history/macd：官方 MACD，一次返回时间 t 与 diff、dea、macd、ema12、ema26 五个字段。

一个字段串起三层信息：diff 与 dea 的上下穿构成金叉死叉（方向），macd 柱的正负与缩放构成柱状动能（强度），ema12 与 ema26 的差值提供快慢均线的交叉验证（确认）。三者叠起来，就形成「方向 + 强度 + 确认」的完整动能刻画。

需要特别提醒：该接口只返回指标值本身，不返回原始收盘价。因此教科书里「股价创新高、MACD 不创新高」的顶背离无法直接算——若要严格做价格与指标的背离，收盘价必须从其他接口（如历史 K 线类接口）另行获取后在本地按时间对齐。本文明确改用指标序列自身来实现：diff 创新高但 macd 柱绝对值较前高收缩，标记为「动能衰竭」。全文会反复强调这是指标层面的近似，不是严格意义上的价格背离，且不做超出文档的字段假设。

衔接说明：本篇是「模块三·历史技术指标」的第三篇，上一篇解决了「超买超卖与波动区间」，本篇解决「趋势动能」，也是本模块技术指标的最后一块拼图。下一篇将进入「自研指标」战场，彻底摆脱对官方指标接口的依赖。

二、接口字段解读

2.1 官方 MACD hsstock/history/macd

http://api.biyingapi.com/hsstock/history/macd/{代码.市场}/{级别}/{除权方式}/{LICENCE}?st={开始日期}&et={结束日期}&lt={返回条数}

路径参数：

查询参数：

返回字段（仅以下字段）：

说明：接口只给五个裸指标值，既不给原始收盘价，也不给 9 日 EMA 的周期参数。因此金叉死叉、零轴分区、柱状动能、动能衰竭、双线粘合度等，全部由这五个值手写推导。

三、自研衍生指标 / 业务逻辑

接口只给裸指标值，以下全部用 Python 手写：

四、数据表设计

CREATE TABLE IF NOT EXISTS stock_macd (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT, level TEXT, t TEXT,
diff REAL, dea REAL, macd REAL,
ema12 REAL, ema26 REAL,
collected_at TEXT,
UNIQUE(code, level, t)
);

— 可选：自研信号表
CREATE TABLE IF NOT EXISTS macd_signal (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT, level TEXT, t TEXT,
diff REAL, dea REAL, macd REAL, ema12 REAL, ema26 REAL,
cross INTEGER, — 自研：1 金叉 / -1 死叉 / 0 无信号 / NULL 首根
zone TEXT, — 自研：零轴上方多头区 / 零轴下方空头区 / 穿越区
bar_sign INTEGER, — 自研：1 红柱 / -1 绿柱 / 0 无柱
bar_streak INTEGER, — 自研：同号柱连续根数
bar_trend TEXT, — 自研：放大 / 缩小 / 持平
momentum_streak INTEGER, — 自研：|macd| 连续递增根数（动能持续）
exhaust INTEGER, — 自研：1 动能衰竭 / 0 否 / NULL 样本不足
stickiness REAL, — 自研：双线粘合度 |diff-dea|/|diff|
ema_gap REAL, — 自研：ema12 – ema26
ema_check TEXT, — 自研：多头共振 / 空头共振 / 快慢线分歧
UNIQUE(code, level, t)
);

说明：macd_signal 中除 code / level / t / diff / dea / macd / ema12 / ema26 外，其余列全部是本文自研衍生列，不是接口字段。

五、完整可运行代码

import requests
import logging
import time
import sqlite3
from datetime import datetime
import pandas as pd
import numpy as np

# ========== 全局配置 ==========
LICENCE = "你的licence"
DB_PATH = "quant.db"
LOG_FILE = "quant_collect.log"
API_BASE = "http://api.biyingapi.com"

# 接口返回的全部字段（严格对照官方文档，不增不减）
MACD_FIELDS = ["t", "diff", "dea", "macd", "ema12", "ema26"]

# 落库列（code / level 为本地维度字段，collected_at 为本地采集时间）
MACD_COLUMNS = ["code", "level", "t", "diff", "dea", "macd",
"ema12", "ema26", "collected_at"]

# 自研信号列
SIGNAL_COLUMNS = [
"code", "level", "t",
"diff", "dea", "macd", "ema12", "ema26",
"cross", "zone", "bar_sign", "bar_streak", "bar_trend",
"momentum_streak", "exhaust", "stickiness", "ema_gap", "ema_check",
]

MINUTE_LEVELS = ("5", "15", "30", "60")

# ———- 日志初始化 ———-
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
filemode="a"
)
logger = logging.getLogger(__name__)

# ———- 带重试 HTTP 请求（复用） ———-
def biying_api_get_retry(full_url, timeout=15, max_retry=3):
for attempt in range(1, max_retry + 1):
try:
resp = requests.get(full_url, timeout=timeout)
if resp.status_code == 200:
return resp.json()
logger.warning(f"HTTP状态码异常：{resp.status_code}，第{attempt}次重试")
except Exception as e:
logger.warning(f"网络请求异常，第{attempt}次重试，错误信息：{str(e)}")
time.sleep(2)
logger.error("达到最大重试次数，接口请求失败")
return []

# ———- 代码补市场后缀 ———-
def with_exchange_suffix(code):
code = str(code).strip()
if code.startswith("60") or code.startswith("68") or code.startswith("9"):
return code + ".SH"
if code.startswith("00") or code.startswith("30") or code.startswith("2"):
return code + ".SZ"
if code.startswith("8") or code.startswith("4"):
return code + ".BJ"
return code + ".SH"

# ———- 拉取官方 MACD ———-
def fetch_history_macd(code, level="d", adjust="n", start="", end="", limit=250):
if str(level) in MINUTE_LEVELS:
adjust = "n"
dm = with_exchange_suffix(code)
url = (f"{API_BASE}/hsstock/history/macd/{dm}/{level}/{adjust}/{LICENCE}"
f"?st={start}&et={end}&lt={limit}")
data = biying_api_get_retry(url)
if isinstance(data, dict):
data = data.get("data", [])
if not isinstance(data, list):
return []
return [row for row in data if isinstance(row, dict)]

# ———- pandas 清洗链 ———-
def clean_macd_frame(rows):
df = pd.DataFrame(rows)
if df.empty:
return df
for col in MACD_FIELDS:
if col not in df.columns:
df[col] = np.nan
df = df.reindex(columns=MACD_FIELDS)
df["t"] = pd.to_datetime(df["t"], errors="coerce")
for col in ("diff", "dea", "macd", "ema12", "ema26"):
df[col] = pd.to_numeric(df[col], errors="coerce")
df = df.dropna(subset=["t"])
df = df.sort_values("t")
df = df.drop_duplicates(subset=["t"], keep="last")
df = df.reset_index(drop=True)
return df

# ———- 自研指标 1：DIF/DEA 金叉死叉 ———-
def detect_macd_cross(prev_diff, prev_dea, curr_diff, curr_dea):
"""返回值：1 金叉 / -1 死叉 / 0 无信号 / None 首根无前值或数据缺失"""
if pd.isna(prev_diff) or pd.isna(prev_dea) or pd.isna(curr_diff) or pd.isna(curr_dea):
return None
if prev_diff <= prev_dea and curr_diff > curr_dea:
return 1
if prev_diff >= prev_dea and curr_diff < curr_dea:
return –1
return 0

# ———- 自研指标 2：零轴分区 ———-
def judge_zero_zone(diff, dea):
if pd.isna(diff) or pd.isna(dea):
return None
if diff > 0 and dea > 0:
return "零轴上方多头区"
if diff < 0 and dea < 0:
return "零轴下方空头区"
return "穿越区"

# ———- 自研指标 3：柱体符号 ———-
def judge_bar_sign(bar):
if pd.isna(bar):
return None
if bar > 0:
return 1
if bar < 0:
return –1
return 0

# ———- 自研指标 3：同号柱连续根数 ———-
def calc_bar_streak(signs, idx):
"""从当前根往回数，统计同号柱连续出现多少根"""
if idx < 0 or idx >= len(signs):
return None
curr = signs[idx]
if curr is None or pd.isna(curr) or curr == 0:
return 0
streak = 0
i = idx
while i >= 0:
prev = signs[i]
if prev is None or pd.isna(prev) or prev != curr:
break
streak += 1
i -= 1
return streak

# ———- 自研指标 4：柱体缩放 ———-
def judge_bar_trend(prev_bar, curr_bar, eps=0.0):
"""比较相邻两根柱体绝对值：放大 / 缩小 / 持平"""
if prev_bar is None or curr_bar is None:
return None
if pd.isna(prev_bar) or pd.isna(curr_bar):
return None
delta = curr_bar – prev_bar
if delta > eps:
return "放大"
if delta < –eps:
return "缩小"
return "持平"

# ———- 自研指标 5：柱体连续放大计数 ———-
def calc_momentum_streak(bars, idx):
"""从当前根往回数，|macd| 严格递增的连续根数（动能持续）"""
if idx < 0 or idx >= len(bars):
return None
curr = bars[idx]
if curr is None or pd.isna(curr):
return None
streak = 0
i = idx
while i > 0:
prev = bars[i – 1]
now = bars[i]
if prev is None or pd.isna(prev) or now is None or pd.isna(now):
break
if now > prev:
streak += 1
i -= 1
else:
break
return streak

# ———- 自研指标 6：动能衰竭（指标层面近似） ———-
def judge_exhaust(diffs, bars, idx, window=20):
"""1 动能衰竭 / 0 否 / None 样本不足
判定：窗口内 diff 创新高，但当前 |macd| 小于窗口内 |macd| 的最大值。
注意：接口不返回收盘价，这里是 diff 自身新高与柱值收缩的对比，非价格背离。
"""
if idx <= 0 or idx >= len(diffs):
return None
start = max(0, idx – window)
hist_diffs = [v for v in diffs[start:idx] if v is not None and not pd.isna(v)]
hist_bars = [v for v in bars[start:idx] if v is not None and not pd.isna(v)]
curr_diff = diffs[idx]
curr_bar = bars[idx]
if not hist_diffs or not hist_bars:
return None
if curr_diff is None or pd.isna(curr_diff) or curr_bar is None or pd.isna(curr_bar):
return None
if curr_diff <= 0 or max(hist_diffs) <= 0:
return 0
if curr_diff < max(hist_diffs):
return 0
return 1 if curr_bar < max(hist_bars) else 0

# ———- 自研指标 7：双线粘合度 ———-
def calc_stickiness(diff, dea):
"""|diff – dea| / |diff|，越小越接近变盘；diff 为 0 返回 None"""
if pd.isna(diff) or pd.isna(dea) or diff == 0:
return None
return round(abs(diff – dea) / abs(diff), 4)

# ———- 自研指标 8：ema12 / ema26 辅助验证 ———-
def calc_ema_gap(ema12, ema26):
if pd.isna(ema12) or pd.isna(ema26):
return None
return round(ema12 – ema26, 4)

def judge_ema_check(diff, ema_gap):
if pd.isna(diff) or ema_gap is None or pd.isna(ema_gap):
return None
if diff > 0 and ema_gap > 0:
return "多头共振"
if diff < 0 and ema_gap < 0:
return "空头共振"
return "快慢线分歧"

# ———- 逐根 bar 生成信号表 ———-
def build_signal_frame(df, exhaust_window=20):
if df.empty:
return pd.DataFrame(columns=SIGNAL_COLUMNS)
diffs = [pd.to_numeric(v, errors="coerce") for v in df["diff"].tolist()]
deas = [pd.to_numeric(v, errors="coerce") for v in df["dea"].tolist()]
macds = [pd.to_numeric(v, errors="coerce") for v in df["macd"].tolist()]
bars = [abs(v) if not pd.isna(v) else np.nan for v in macds]
signs = [judge_bar_sign(v) for v in macds]

records = []
for idx, (_, row) in enumerate(df.iterrows()):
ema_gap = calc_ema_gap(
pd.to_numeric(row.get("ema12"), errors="coerce"),
pd.to_numeric(row.get("ema26"), errors="coerce"),
)
prev_diff = diffs[idx – 1] if idx > 0 else np.nan
prev_dea = deas[idx – 1] if idx > 0 else np.nan
prev_bar = bars[idx – 1] if idx > 0 else None
records.append({
"t": row["t"],
"diff": diffs[idx],
"dea": deas[idx],
"macd": macds[idx],
"ema12": pd.to_numeric(row.get("ema12"), errors="coerce"),
"ema26": pd.to_numeric(row.get("ema26"), errors="coerce"),
"cross": detect_macd_cross(prev_diff, prev_dea, diffs[idx], deas[idx]),
"zone": judge_zero_zone(diffs[idx], deas[idx]),
"bar_sign": signs[idx],
"bar_streak": calc_bar_streak(signs, idx),
"bar_trend": judge_bar_trend(prev_bar, bars[idx]),
"momentum_streak": calc_momentum_streak(bars, idx),
"exhaust": judge_exhaust(diffs, bars, idx, window=exhaust_window),
"stickiness": calc_stickiness(diffs[idx], deas[idx]),
"ema_gap": ema_gap,
"ema_check": judge_ema_check(diffs[idx], ema_gap),
})
return pd.DataFrame(records)

# ———- 建表 ———-
def init_db():
conn = sqlite3.connect(DB_PATH)
try:
conn.execute("""
CREATE TABLE IF NOT EXISTS stock_macd (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT, level TEXT, t TEXT,
diff REAL, dea REAL, macd REAL,
ema12 REAL, ema26 REAL,
collected_at TEXT,
UNIQUE(code, level, t)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS macd_signal (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT, level TEXT, t TEXT,
diff REAL, dea REAL, macd REAL, ema12 REAL, ema26 REAL,
cross INTEGER, zone TEXT, bar_sign INTEGER,
bar_streak INTEGER, bar_trend TEXT,
momentum_streak INTEGER, exhaust INTEGER,
stickiness REAL, ema_gap REAL, ema_check TEXT,
UNIQUE(code, level, t)
)
""")
conn.commit()
finally:
conn.close()

# ———- 落库（INSERT OR REPLACE 幂等写入） ———-
def save_macd(code, level, df):
if df.empty:
return
conn = sqlite3.connect(DB_PATH)
try:
data = df.copy()
data.insert(0, "level", level)
data.insert(0, "code", str(code))
data["t"] = data["t"].astype(str)
data["collected_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
data = data[MACD_COLUMNS]
data.to_sql("tmp_macd", conn, if_exists="replace", index=False)
cols = ", ".join(MACD_COLUMNS)
conn.execute(f"INSERT OR REPLACE INTO stock_macd ({cols}) SELECT {cols} FROM tmp_macd")
conn.execute("DROP TABLE IF EXISTS tmp_macd")
conn.commit()
logger.info(f"落库MACD {code} {level} 共 {len(data)} 条")
except Exception as e:
logger.error(f"落库MACD失败：{str(e)}")
finally:
conn.close()

def save_signal(code, level, sig):
if sig.empty:
return
conn = sqlite3.connect(DB_PATH)
try:
data = sig.copy()
data.insert(0, "level", level)
data.insert(0, "code", str(code))
data["t"] = data["t"].astype(str)
data = data[SIGNAL_COLUMNS]
data.to_sql("tmp_signal", conn, if_exists="replace", index=False)
cols = ", ".join(SIGNAL_COLUMNS)
conn.execute(f"INSERT OR REPLACE INTO macd_signal ({cols}) SELECT {cols} FROM tmp_signal")
conn.execute("DROP TABLE IF EXISTS tmp_signal")
conn.commit()
logger.info(f"落库MACD信号 {code} {level} 共 {len(data)} 条")
except Exception as e:
logger.error(f"落库MACD信号失败：{str(e)}")
finally:
conn.close()

# ———- 演示入口 ———-
def demo(codes=None, level="d", start="20240101", end="20241231",
exhaust_window=20):
init_db()
if codes is None:
codes = ["600519", "000001", "300750"]
for code in codes:
rows = fetch_history_macd(code, level=level, adjust="n",
start=start, end=end, limit=250)
df = clean_macd_frame(rows)
if df.empty:
logger.warning(f"{code} MACD数据为空，跳过")
continue
sig = build_signal_frame(df, exhaust_window)
save_macd(code, level, df)
save_signal(code, level, sig)

last = sig.iloc[–1]
print(f"\\n=== {code} 级别 {level} ===")
print(f"最新时间 {last['t']}")
print(f"DIF(diff) {last['diff']} | DEA(dea) {last['dea']} | 柱(macd) {last['macd']}")
print(f"EMA12 {last['ema12']} | EMA26 {last['ema26']} | 差值 {last['ema_gap']} | {last['ema_check']}")
print(f"金叉死叉 {last['cross']}（1 金叉 / -1 死叉 / 0 无 / NULL 首根）")
print(f"零轴分区 {last['zone']} | 柱符号 {last['bar_sign']}（1 红柱 / -1 绿柱）")
print(f"同号连续 {last['bar_streak']} 根 | 柱体变化 {last['bar_trend']}")
print(f"动能持续 {last['momentum_streak']} 根 | 动能衰竭 {last['exhaust']}（1 衰竭 / 0 否 / NULL 样本不足）")
print(f"双线粘合度 {last['stickiness']}（越小越接近变盘）")
golden = int((sig["cross"] == 1).sum())
dead = int((sig["cross"] == –1).sum())
up_bars = int((sig["zone"] == "零轴上方多头区").sum())
down_bars = int((sig["zone"] == "零轴下方空头区").sum())
exhaust_bars = int(sig["exhaust"].fillna(0).sum())
print(f"区间内金叉 {golden} 次 | 死叉 {dead} 次")
print(f"零轴上方 {up_bars} 根 | 零轴下方 {down_bars} 根 | 动能衰竭 {exhaust_bars} 根")
print("\\n【仅为数据演示，不构成投资建议】已采集并落库 MACD 数据")

if __name__ == "__main__":
demo()

代码运行后，每只股票会依次打印：最新一根 bar 的时间、DIF/DEA/柱值、两条 EMA 及其差值与共振验证、金叉死叉标记、零轴分区、柱符号与同号连续根数、柱体缩放、动能持续根数、动能衰竭标记、双线粘合度，最后输出区间内的金叉/死叉次数与零轴分区、动能衰竭的 bar 根数。所有数值均由实际接口返回计算得出，本文不预设任何具体结果。

六、业务关键点

七、拓展练习

八、下篇预告

下一篇《自研指标：量价背离、换手异动与振幅》将把视角拉回「接口没给你、必须自己算」的战场——用 hsstock/real/time 的实时字段（最新价 p、昨收 yc、成交量 v、成交额 cje、换手率 tr、振幅 zf、涨跌幅 pc）自研「量价背离」「换手异动 Z-Score」「振幅异常」三个实战指标，彻底摆脱对官方指标接口的依赖。

九、免责申明

免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。