"""
主力合约 1月/5月/9月 两两价差走势图 + 均值回归交易信号。
输出 HTML 图表，浏览器打开即可查看。
用法: python3 spread_chart.py
"""
import sqlite3
import json
import os
from collections import defaultdict
from statistics import mean, stdev
DB_PATH = os.path.join(os.path.dirname(__file__), "db", "futures.db")
OUT_PATH = os.path.join(os.path.dirname(__file__), "spread_chart.html")
WINDOW = 60
ENTRY_Z = 2.0
def build_spreads():
"""构建三组价差序列：01-05, 05-09, 01-09（同年配对）。"""
conn = sqlite3.connect(DB_PATH)
daily = conn.execute(
"SELECT ts_code, trade_date, close FROM daily WHERE close IS NOT NULL AND close != '' ORDER BY trade_date"
).fetchall()
conn.close()
by_date = defaultdict(dict)
for ts_code, d, close in daily:
by_date[d][ts_code] = float(close)
dates = sorted(by_date.keys())
def build(month_a, month_b, cross_year=False):
per_date = {}
for d in dates:
items = by_date[d]
best_yr = -1
best = None
for ts_code in items:
if not ts_code.endswith(".ZCE") or not ts_code.startswith("FG"):
continue
yr_str, mon = ts_code[2:4], ts_code[4:6]
if mon == month_a:
yr = int(yr_str)
yr_b_str = str(yr + 1) if cross_year else yr_str
ts_b = f"FG{yr_b_str}{month_b}.ZCE"
if ts_b in items and yr > best_yr:
best_yr = yr
spread = items[ts_b] - items[ts_code]
best = (d, spread)
if best:
per_date[best[0]] = best[1]
sorted_dates = sorted(per_date.keys())
return sorted_dates, [per_date[d] for d in sorted_dates]
return {
"01-05": build("01", "05"),
"05-09": build("05", "09"),
"01-09": build("01", "09"),
}
def compute_signals(spread_data):
"""计算每个价差的滚动 z-score 和交易信号。"""
month_map = {
"01-05": ("01", "05"),
"05-09": ("05", "09"),
"01-09": ("01", "09"),
}
signals = {}
for label, (dates, values) in spread_data.items():
if len(values) < WINDOW + 1:
signals[label] = None
continue
ma, mb = month_map[label]
recent = values[-WINDOW:]
roll_mean = mean(recent)
roll_std = stdev(recent)
latest = values[-1]
z = (latest - roll_mean) / roll_std if roll_std > 0 else 0
if z > ENTRY_Z:
action = f"做空{mb}，做多{ma}"
elif z < -ENTRY_Z:
action = f"做多{mb}，做空{ma}"
else:
action = "观望"
z_dates = dates[WINDOW:]
z_values = []
for i in range(WINDOW, len(values)):
h = values[i - WINDOW : i]
m = mean(h)
s = stdev(h)
z_values.append((values[i] - m) / s if s > 0 else 0)
signals[label] = {
"latest": latest,
"latest_date": dates[-1],
"mean": round(roll_mean, 1),
"std": round(roll_std, 1),
"z_score": round(z, 2),
"action": action,
"count": len(values),
"z_dates": z_dates,
"z_values": z_values,
}
return signals
def build_chart_data(spread_data, signals):
"""构建前端所需的数据结构。"""
all_dates_set = set()
for dates, _ in spread_data.values():
all_dates_set.update(dates)
all_dates = sorted(all_dates_set)
colors = {
"01-05": {"line": "#06b6d4", "area": "rgba(6,182,212,0.08)"},
"05-09": {"line": "#22c55e", "area": "rgba(34,197,94,0.08)"},
"01-09": {"line": "#f59e0b", "area": "rgba(245,158,11,0.08)"},
}
# 主图数据集
datasets = []
for label, (dates, values) in spread_data.items():
date_map = dict(zip(dates, values))
series = [date_map.get(d) for d in all_dates]
c = colors[label]
datasets.append({
"label": label,
"data": series,
"borderColor": c["line"],
"backgroundColor": c["area"],
"borderWidth": 1.5,
"pointRadius": 0,
"fill": True,
"tension": 0.1,
})
# z-score 数据集
z_datasets = []
for label, sig in signals.items():
if sig is None:
continue
date_map = dict(zip(sig["z_dates"], sig["z_values"]))
series = [date_map.get(d) for d in all_dates]
c = colors[label]
z_datasets.append({
"label": label,
"data": series,
"borderColor": c["line"],
"backgroundColor": "transparent",
"borderWidth": 1,
"pointRadius": 0,
"tension": 0.1,
})
# 统计卡片
stat_cards = []
for label, (dates, values) in spread_data.items():
if not values:
continue
pos = sum(1 for v in values if v > 0)
neg = sum(1 for v in values if v < 0)
stat_cards.append({
"label": label,
"color": colors[label]["line"],
"count": len(values),
"avg": round(sum(values) / len(values), 1),
"max": max(values),
"min": min(values),
"pos": pos,
"neg": neg,
})
# 信号卡片
sig_cards = []
for label, sig in signals.items():
if sig is None:
continue
if "做空" in sig["action"]:
sig_type = "bear"
sig_color = "#ef4444"
elif "做多" in sig["action"]:
sig_type = "bull"
sig_color = "#22c55e"
else:
sig_type = "neutral"
sig_color = "#94a3b8"
sig_cards.append({
"label": label,
"color": colors[label]["line"],
"sig_type": sig_type,
"sig_color": sig_color,
"action": sig["action"],
"latest": sig["latest"],
"mean": sig["mean"],
"std": sig["std"],
"z_score": sig["z_score"],
"latest_date": sig["latest_date"],
})
return {
"all_dates": all_dates,
"datasets": datasets,
"z_datasets": z_datasets,
"stat_cards": stat_cards,
"sig_cards": sig_cards,
}
TEMPLATE = r"""
FG 主力合约价差走势
玻璃期货 — 主力合约价差分析
1月 / 5月 / 9月 两两价差（同年配对，远月减近月） | 均值回归信号（60日滚动，2σ 入场）
"""
def generate_html(chart_data):
"""渲染 HTML。"""
payload = json.dumps({
"allDates": chart_data["all_dates"],
"datasets": chart_data["datasets"],
"zDatasets": chart_data["z_datasets"],
"statCards": chart_data["stat_cards"],
"sigCards": chart_data["sig_cards"],
})
html = TEMPLATE.replace("__DATA__", payload)
with open(OUT_PATH, "w") as f:
f.write(html)
print(f"已生成: {OUT_PATH}")
if __name__ == "__main__":
data = build_spreads()
sigs = compute_signals(data)
chart_data = build_chart_data(data, sigs)
generate_html(chart_data)