基金数据实战：从净值到持仓的全链路量化（第 5 篇）：基金历史MA：均线排列与选基信号

一、前言

上一篇（第 4 篇）我们用 jj/lskx 拿到了基金 K 线的开高低收，其中收盘价 c 是后续一切趋势与乖离计算的基石。但单根 K 线只能看「当下」，要看「趋势方向」还差一步：均线（Moving Average）。

本篇组合两个接口：

jj/lsma：历史均线序列，逐交易日回传 ma3/ma5/ma10/…/ma250 共 10 条均线，用于做排列、金叉、粘合等时序判断；
jj/zxma：最新均线单根，结构同上，用于实时取当下均线快照。

二者结合，我们就能纯 Python 自研「多头/空头排列」「金叉/死叉」「乖离率」「均线粘合度」等趋势指标，并把排列与金叉状态收敛成「选基信号」，为第 10 篇量化选基提前埋点。

衔接说明：本篇是「模块二·基金历史与均线」的核心篇，第 4 篇解决了「K 线与收盘价 c」，本篇解决「均线趋势」，下一篇（第 6 篇）将进一步把股票技术指标迁移到基金自研。

二、接口字段解读

2.1 历史均线 jj/lsma

https://api.biyingapi.com/jj/lsma/{基金代码}/{分时级别}/{您的licence}

字段类型含义

t	string	交易时间（yyyy-MM-dd）
ma3	number	3 日均线（无则为 null）
ma5	number	5 日均线（无则为 null）
ma10	number	10 日均线（无则为 null）
ma15	number	15 日均线（无则为 null）
ma20	number	20 日均线（无则为 null）
ma30	number	30 日均线（无则为 null）
ma60	number	60 日均线（无则为 null）
ma120	number	120 日均线（无则为 null）
ma200	number	200 日均线（无则为 null）
ma250	number	250 日均线（无则为 null）

重要：该接口只返回 t 与 10 条均线，不返回收盘价 c。要计算乖离率所需的收盘价，必须回到第 4 篇的 jj/lskx 取 c 字段配合使用——本篇绝不虚构 c/close/open/vol 等字段。

2.2 最新均线 jj/zxma

https://api.biyingapi.com/jj/zxma/{基金代码}/{分时级别}/{您的licence}

字段与 jj/lsma 完全相同（仅 t 与 ma3..ma250 共 11 个），区别在返回形态为单根最新快照（dict 或单元素列表）。

字段类型含义

t	string	交易时间（yyyy-MM-dd）
ma3	number	3 日均线（无则为 null）
ma5	number	5 日均线（无则为 null）
ma10	number	10 日均线（无则为 null）
ma15	number	15 日均线（无则为 null）
ma20	number	20 日均线（无则为 null）
ma30	number	30 日均线（无则为 null）
ma60	number	60 日均线（无则为 null）
ma120	number	120 日均线（无则为 null）
ma200	number	200 日均线（无则为 null）
ma250	number	250 日均线（无则为 null）

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

接口只给裸均线，以下全部用 Python 手写：

多头排列（自研）：ma5 > ma10 > ma20 > ma60（严格递减）记为多头排列，返回布尔与参与值；技术上代表短期成本持续高于中长期，趋势向上。

空头排列（自研）：ma5 < ma10 < ma20 < ma60（严格递增）记为空头排列，趋势向下。

均线金叉 / 死叉（自研）：以短周期（如 ma5）与长周期（如 ma20）为例，用前后两根 bar 的位置关系判定——短上穿长（前 diff ≤ 0 且后 diff > 0）为金叉；短下穿长（前 diff ≥ 0 且后 diff < 0）为死叉；首根无前值跳过。

均线乖离率（自研）：(c – maN) / maN * 100%。其中 c 来自第 4 篇 jj/lskx 的收盘价，本篇 ma 接口不含 c；c 与 ma60 的偏离表示中长期趋势乖离，正为强势超买、负为弱势超卖。

均线粘合度（自研）：取 ma5/ma10/ma20 三条，计算 std ÷ mean，值越小越粘合，代表横盘蓄势、方向选择临近。

null 值处理：均线为 null 时所有计算先判空，避免 None 参与数值运算导致异常。

选基信号（收口）：把「多头排列 + 金叉」作为选基强弱信号输出（强/中/弱），为第 10 篇量化选基铺垫。

四、数据表设计

CREATE TABLE IF NOT EXISTS fund_ma (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dm TEXT, level TEXT, t TEXT,
ma3 REAL, ma5 REAL, ma10 REAL, ma15 REAL, ma20 REAL,
ma30 REAL, ma60 REAL, ma120 REAL, ma200 REAL, ma250 REAL,
collected_at TEXT,
UNIQUE(dm, level, t)
);

CREATE TABLE IF NOT EXISTS fund_ma_signal (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dm TEXT, level TEXT, t TEXT,
is_bull INTEGER, is_bear INTEGER, cross INTEGER,
bias_ma60 REAL, adhesion REAL,
select_signal TEXT,
collected_at TEXT,
UNIQUE(dm, level, t)
);

fund_ma 存裸均线序列（落在允许字段清单内）；fund_ma_signal 存本篇自研的排列/金叉/乖离/粘合/选基信号，均为衍生列。

五、完整可运行代码

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

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

# ———- 日志初始化 ———-
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(url, timeout=15, max_retry=3):
for attempt in range(1, max_retry + 1):
try:
resp = requests.get(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):
# 基金代码按官方原值透传，不做交易所后缀拼接
return code

# ———- 历史均线序列 ———-
def fetch_lsma(code, level):
code = with_exchange_suffix(code)
url = f"{API_BASE}/jj/lsma/{code}/{level}/{LICENCE}"
data = biying_api_get_retry(url)
if isinstance(data, list):
return data
if isinstance(data, dict):
return data.get("data", [data])
return []

# ———- 最新均线单根 ———-
def fetch_zxma(code, level):
code = with_exchange_suffix(code)
url = f"{API_BASE}/jj/zxma/{code}/{level}/{LICENCE}"
data = biying_api_get_retry(url)
if isinstance(data, list) and data:
return data[0] if isinstance(data[0], dict) else {}
if isinstance(data, dict):
return data
return {}

# ———- 收盘价 c 来自第4篇 jj/lskx（本篇 ma 接口不含 c） ———-
def fetch_lskx_c(code, level):
code = with_exchange_suffix(code)
url = f"{API_BASE}/jj/lskx/{code}/{level}/{LICENCE}"
data = biying_api_get_retry(url)
if isinstance(data, list):
return data
if isinstance(data, dict):
return data.get("data", [data])
return []

# ———- 数值判空工具 ———-
def to_num(v):
return pd.to_numeric(v, errors="coerce")

# ———- 多头排列判定（自研） ———-
def is_bull_arrangement(row):
# ma5 > ma10 > ma20 > ma60（严格递减）
ma5 = to_num(row.get("ma5"))
ma10 = to_num(row.get("ma10"))
ma20 = to_num(row.get("ma20"))
ma60 = to_num(row.get("ma60"))
if any(pd.isna(x) for x in (ma5, ma10, ma20, ma60)):
return {"is_bull": False, "ma5": None, "ma10": None,
"ma20": None, "ma60": None}
is_bull = bool(ma5 > ma10 > ma20 > ma60)
return {"is_bull": is_bull,
"ma5": float(ma5), "ma10": float(ma10),
"ma20": float(ma20), "ma60": float(ma60)}

# ———- 空头排列判定（自研） ———-
def is_bear_arrangement(row):
# ma5 < ma10 < ma20 < ma60（严格递增）
ma5 = to_num(row.get("ma5"))
ma10 = to_num(row.get("ma10"))
ma20 = to_num(row.get("ma20"))
ma60 = to_num(row.get("ma60"))
if any(pd.isna(x) for x in (ma5, ma10, ma20, ma60)):
return {"is_bear": False, "ma5": None, "ma10": None,
"ma20": None, "ma60": None}
is_bear = bool(ma5 < ma10 < ma20 < ma60)
return {"is_bear": is_bear,
"ma5": float(ma5), "ma10": float(ma10),
"ma20": float(ma20), "ma60": float(ma60)}

# ———- 金叉/死叉判定（自研） ———-
def cross_signal(short_prev, short_cur, long_prev, long_cur):
# 1=金叉（短上穿长），-1=死叉（短下穿长），0=无
if None in (short_prev, short_cur, long_prev, long_cur):
return 0
prev_diff = short_prev – long_prev
cur_diff = short_cur – long_cur
if prev_diff <= 0 < cur_diff:
return 1
if prev_diff >= 0 > cur_diff:
return –1
return 0

# ———- 乖离率（自研，c 来自 lskx） ———-
def bias_rate(c, ma_n):
# c: 第4篇 jj/lskx 收盘价；ma_n: 本篇 lsma/zxma 均线
if c is None or ma_n is None or ma_n == 0:
return None
return round((c – ma_n) / ma_n * 100, 2)

# ———- 均线粘合度（自研） ———-
def adhesion(ma5, ma10, ma20):
# std / mean，越小越粘合
vals = [to_num(ma5), to_num(ma10), to_num(ma20)]
if any(pd.isna(x) for x in vals):
return None
arr = np.array(vals, dtype=float)
if arr.mean() == 0:
return None
return round(float(arr.std() / arr.mean()), 4)

# ———- 选基信号（收口，自研） ———-
def select_signal(is_bull, cross):
if is_bull and cross == 1:
return "强"
if is_bull:
return "中"
return "弱"

# ———- 序列处理：逐 bar 计算信号 ———-
def process_ma_series(ma_list, c_map):
signals = []
prev = None
for row in ma_list:
t = row.get("t")
bull = is_bull_arrangement(row)
bear = is_bear_arrangement(row)
cross = 0
if prev is not None:
cross = cross_signal(
to_num(prev.get("ma5")), to_num(row.get("ma5")),
to_num(prev.get("ma20")), to_num(row.get("ma20"))
)
c = c_map.get(t) # c 来自 jj/lskx
bias_ma60 = bias_rate(c, to_num(row.get("ma60")))
adh = adhesion(row.get("ma5"), row.get("ma10"), row.get("ma20"))
sig = select_signal(bull["is_bull"], cross)
signals.append({
"t": t,
"is_bull": bull["is_bull"],
"is_bear": bear["is_bear"],
"cross": cross,
"bias_ma60": bias_ma60,
"adhesion": adh,
"select_signal": sig,
})
prev = row
return signals

# ———- 落库：裸均线 ———-
def save_fund_ma(dm, level, ma_list):
if not ma_list:
return
conn = sqlite3.connect(DB_PATH)
try:
rows = []
for r in ma_list:
rows.append({
"dm": dm, "level": level, "t": r.get("t"),
"ma3": to_num(r.get("ma3")),
"ma5": to_num(r.get("ma5")),
"ma10": to_num(r.get("ma10")),
"ma15": to_num(r.get("ma15")),
"ma20": to_num(r.get("ma20")),
"ma30": to_num(r.get("ma30")),
"ma60": to_num(r.get("ma60")),
"ma120": to_num(r.get("ma120")),
"ma200": to_num(r.get("ma200")),
"ma250": to_num(r.get("ma250")),
"collected_at": time.strftime("%Y-%m-%d %H:%M:%S"),
})
pd.DataFrame(rows).to_sql("fund_ma", conn, if_exists="append", index=False)
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uk_fma ON fund_ma(dm, level, t)")
conn.commit()
except Exception as e:
logger.error(f"落库 fund_ma 失败：{str(e)}")
finally:
conn.close()

# ———- 落库：自研信号 ———-
def save_fund_ma_signal(dm, level, signals):
if not signals:
return
conn = sqlite3.connect(DB_PATH)
try:
rows = []
for s in signals:
rows.append({
"dm": dm, "level": level, "t": s.get("t"),
"is_bull": int(s.get("is_bull", False)),
"is_bear": int(s.get("is_bear", False)),
"cross": s.get("cross", 0),
"bias_ma60": s.get("bias_ma60"),
"adhesion": s.get("adhesion"),
"select_signal": s.get("select_signal"),
"collected_at": time.strftime("%Y-%m-%d %H:%M:%S"),
})
pd.DataFrame(rows).to_sql("fund_ma_signal", conn, if_exists="append", index=False)
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uk_fmas ON fund_ma_signal(dm, level, t)")
conn.commit()
except Exception as e:
logger.error(f"落库 fund_ma_signal 失败：{str(e)}")
finally:
conn.close()

# ———- 演示入口 ———-
def demo(codes=None, level="d"):
if codes is None:
codes = ["000001"] # 示例基金代码，级别 d
for code in codes:
ma_list = fetch_lsma(code, level)
# c 来自第4篇 jj/lskx，按 t 对齐
lskx = fetch_lskx_c(code, level)
c_map = {x.get("t"): to_num(x.get("c"))
for x in lskx if isinstance(x, dict)}
signals = process_ma_series(ma_list, c_map)
print(f"\\n=== {code}（级别 {level}）===")
if signals:
last = signals[–1]
print(f"最新交易日 {last['t']} | 多头排列 {last['is_bull']} | "
f"空头排列 {last['is_bear']} | 金叉死叉 {last['cross']} | "
f"ma60乖离率 {last['bias_ma60']}% | 粘合度 {last['adhesion']} | "
f"选基信号 {last['select_signal']}")
print("（示例占位：以上为接口字段经自研计算的逻辑演示，非真实成交数据）")
save_fund_ma(code, level, ma_list)
save_fund_ma_signal(code, level, signals)
print("\\n【仅为数据演示，不构成投资建议】已采集均线并落库自研信号")

if __name__ == "__main__":
demo()

六、业务关键点

接口不含收盘价 c：lsma/zxma 只回传 t 与均线，乖离率所需的 c 必须回查第 4 篇 jj/lskx，本篇任何代码与字段表都不虚构 c/close/open/vol。

返回形态要判空：lsma 为序列（list），zxma 可能为 dict 或单元素列表，代码需兼容两种形态；首根 bar 无前值，金叉/死叉判定必须跳过，否则 prev.get 会抛 AttributeError。

null 防御是常态：新基金或上市不足 250 日的基金，ma120/ma200/ma250 等常为 null，所有数值运算前先用 to_num + isna 判空，避免 None 参与 >/</std 计算。

均线严格递减/递增：多头/空头排列用的是严格不等号，若相邻两条均线相等（粘合临界）则不计入排列，避免把「走平」误判为趋势。

时间戳对齐：lsma 的 t 与 lskx 的 t 均为 yyyy-MM-dd，落库用 (dm, level, t) 唯一索引，便于把均线与收盘价按交易日精确关联。

七、拓展练习

把「多头排列 + 金叉」的基金标记为「强」选基信号，批量扫描自选池，统计强信号占比，作为第 10 篇量化选基的初筛。

用 ma60 乖离率对基金做分档（如 >5% 超买、<-5% 超卖），结合粘合度观察「粘合后发散」是否对应后续净值突破。

八、下篇预告

下一篇《基金技术指标：把股票指标迁移到基金（自研）》将基于本篇的均线 ma 与第4篇的 K 线 o/h/l/c，用纯 Python 自研基金的「KDJ 随机指标」「BOLL 布林带」「MACD」三大技术指标——因为官方只给了均线和裸 K 线，真正的 KDJ/BOLL/MACD 必须自己用收盘价序列算出来，这是基金量化区别于股票篇（官方直接给指标）的核心自研战场。

九、免责申明

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

基金数据实战：从净值到持仓的全链路量化（第 5 篇）：基金历史MA：均线排列与选基信号