vipg
227 lines
7.2 KiB
Python
227 lines
7.2 KiB
Python
"""
|
||
跨期价差均值回归回测：支持任意合约月份组合。
|
||
"""
|
||
import sqlite3
|
||
from collections import defaultdict
|
||
from statistics import mean, stdev
|
||
|
||
DB_PATH = "/Users/vipg/Documents/futures-data-warehouse/db/futures.db"
|
||
|
||
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_spread(month_a, month_b, cross_year=False):
|
||
"""
|
||
构建两合约月份的价差序列。
|
||
month_a, month_b: '01','05','09'
|
||
cross_year: True 表示 year_b = year_a + 1（如 09-01 跨年）
|
||
返回 [(date, ts_a, ts_b, p_a, p_b, spread)]
|
||
"""
|
||
data = []
|
||
for d in dates:
|
||
items = by_date[d]
|
||
for ts_code in list(items.keys()):
|
||
if not ts_code.endswith(".ZCE") or not ts_code.startswith("FG"):
|
||
continue
|
||
yr, mon = ts_code[2:4], ts_code[4:6]
|
||
if mon == month_a:
|
||
yr_b = str(int(yr) + 1) if cross_year else yr
|
||
ts_b = f"FG{yr_b}{month_b}.ZCE"
|
||
if ts_b in items:
|
||
p_a = items[ts_code]
|
||
p_b = items[ts_b]
|
||
spread = p_b - p_a
|
||
data.append((d, ts_code, ts_b, p_a, p_b, spread))
|
||
return data
|
||
|
||
|
||
def backtest_spread(spread_data, window=60, entry_z=2.0, exit_z=0.5, max_hold=20):
|
||
equity = 1.0
|
||
position = 0
|
||
entry_spread = 0
|
||
entry_p_a = 0
|
||
entry_date = ""
|
||
hold_days = 0
|
||
trades = []
|
||
daily_eq = []
|
||
|
||
for i, (d, ts_a, ts_b, p_a, p_b, spread_val) in enumerate(spread_data):
|
||
if i < window:
|
||
daily_eq.append((d, equity))
|
||
continue
|
||
|
||
hist = [spread_data[j][5] for j in range(i - window, i)]
|
||
roll_mean = mean(hist)
|
||
roll_std = stdev(hist)
|
||
|
||
signal = 0
|
||
if spread_val > roll_mean + entry_z * roll_std:
|
||
signal = -1 # 价差过大 → 空价差（空远月、多近月）
|
||
elif spread_val < roll_mean - entry_z * roll_std:
|
||
signal = 1
|
||
|
||
should_exit = False
|
||
if position != 0:
|
||
hold_days += 1
|
||
if hold_days >= max_hold:
|
||
should_exit = True
|
||
if position == 1 and spread_val >= roll_mean - exit_z * roll_std:
|
||
should_exit = True
|
||
elif position == -1 and spread_val <= roll_mean + exit_z * roll_std:
|
||
should_exit = True
|
||
|
||
if position != 0 and should_exit:
|
||
if position == 1:
|
||
pnl = (spread_val - entry_spread) / entry_p_a
|
||
else:
|
||
pnl = (entry_spread - spread_val) / entry_p_a
|
||
equity *= (1 + pnl)
|
||
trades.append((entry_date, d, entry_spread, spread_val, pnl, position))
|
||
position = 0
|
||
hold_days = 0
|
||
|
||
if position == 0 and signal != 0:
|
||
position = signal
|
||
entry_spread = spread_val
|
||
entry_p_a = p_a
|
||
entry_date = d
|
||
hold_days = 0
|
||
|
||
daily_eq.append((d, equity))
|
||
|
||
if position != 0 and spread_data:
|
||
d, _, _, p_a, _, spread_val = spread_data[-1]
|
||
if position == 1:
|
||
pnl = (spread_val - entry_spread) / entry_p_a
|
||
else:
|
||
pnl = (entry_spread - spread_val) / entry_p_a
|
||
equity *= (1 + pnl)
|
||
trades.append((entry_date, d, entry_spread, spread_val, pnl, position))
|
||
|
||
return trades, daily_eq
|
||
|
||
|
||
def print_results(trades, daily_eq, label):
|
||
if not trades:
|
||
print(f"\n{label}: 无交易")
|
||
return
|
||
wins = [t for t in trades if t[4] > 0]
|
||
losses = [t for t in trades if t[4] <= 0]
|
||
total_ret = daily_eq[-1][1] - 1.0 if daily_eq else 0
|
||
wr = len(wins) / len(trades) if trades else 0
|
||
|
||
returns = []
|
||
for i in range(1, len(daily_eq)):
|
||
if daily_eq[i-1][1] > 0:
|
||
returns.append(daily_eq[i][1] / daily_eq[i-1][1] - 1)
|
||
avg_ret = mean(returns) if returns else 0
|
||
std_ret = stdev(returns) if len(returns) > 1 else 1
|
||
sharpe = (avg_ret / std_ret) * (252 ** 0.5) if std_ret > 0 else 0
|
||
|
||
peak = 1.0
|
||
mdd = 0.0
|
||
for _, e in daily_eq:
|
||
if e > peak: peak = e
|
||
dd = (peak - e) / peak
|
||
if dd > mdd: mdd = dd
|
||
|
||
print(f"\n{label}")
|
||
print(f" {'' if trades else '无'}交易次数: {len(trades)}")
|
||
print(f" 总收益率: {total_ret:+.2%}")
|
||
print(f" 夏普比率: {sharpe:.2f}")
|
||
print(f" 最大回撤: {mdd:.2%}")
|
||
print(f" 胜率: {wr:.0%}")
|
||
if losses:
|
||
avg_w = mean(t[4] for t in wins)
|
||
avg_l = mean(t[4] for t in losses)
|
||
print(f" 盈亏比: {abs(avg_w/avg_l):.2f}")
|
||
|
||
|
||
# ── 分析所有有效组合 ──────────────────────────
|
||
|
||
pairs = [
|
||
("01", "05", False, "同一年 01-05"),
|
||
("05", "09", False, "同一年 05-09"),
|
||
("09", "01", True, "跨年 09-01"),
|
||
]
|
||
|
||
# 先看各组合的统计特征
|
||
print("=" * 60)
|
||
print("各合约组合价差统计")
|
||
print("=" * 60)
|
||
for ma, mb, cross, label in pairs:
|
||
data = build_spread(ma, mb, cross)
|
||
n = len(data)
|
||
vals = [r[5] for r in data]
|
||
pos = sum(1 for v in vals if v > 0)
|
||
avg_s = mean(vals)
|
||
std_s = stdev(vals)
|
||
print(f"\n{label:>12} {n:>5}天 均值{avg_s:>+7.1f} σ{std_s:>6.1f} 正{pos:>4}({pos/n:.0%})")
|
||
|
||
# 回测各组合
|
||
print("\n" + "=" * 60)
|
||
print("均值回归回测 (window=60 entry=2σ exit=0.5σ max_hold=20)")
|
||
print("=" * 60)
|
||
for ma, mb, cross, label in pairs:
|
||
data = build_spread(ma, mb, cross)
|
||
trades, eq = backtest_spread(data, window=60, entry_z=2.0, exit_z=0.5, max_hold=20)
|
||
print_results(trades, eq, label)
|
||
|
||
# 当前可交易组合 (202607)
|
||
print("\n" + "=" * 60)
|
||
print("当前可交易组合分析 (FG2609, FG2701, FG2705)")
|
||
print("=" * 60)
|
||
|
||
today = "20260717"
|
||
items = by_date.get(today, {})
|
||
for ts in ["FG2609.ZCE", "FG2701.ZCE", "FG2705.ZCE"]:
|
||
if ts in items:
|
||
print(f" {ts}: {items[ts]}")
|
||
else:
|
||
print(f" {ts}: 无当日数据")
|
||
|
||
# 当前配对及统计
|
||
pairs_now = [
|
||
("FG2609.ZCE", "FG2701.ZCE", "09-01(跨年)"),
|
||
("FG2701.ZCE", "FG2705.ZCE", "01-05(同一年)"),
|
||
("FG2609.ZCE", "FG2705.ZCE", "09-05(跨年)"),
|
||
]
|
||
|
||
for ts_a, ts_b, label in pairs_now:
|
||
if ts_a not in items or ts_b not in items:
|
||
print(f"\n{label}: {ts_a} 或 {ts_b} 无数据")
|
||
continue
|
||
|
||
spread_now = items[ts_b] - items[ts_a]
|
||
print(f"\n{label}: {ts_a} vs {ts_b}")
|
||
print(f" 当前价差: {spread_now:+.1f}")
|
||
|
||
# 找历史均值
|
||
for ma, mb, cross, _ in pairs:
|
||
if (ts_a[4:6] == ma and ts_b[4:6] == mb) or \
|
||
(ts_a[4:6] == mb and ts_b[4:6] == ma and cross):
|
||
data = build_spread(ma, mb, cross)
|
||
if len(data) < 60:
|
||
continue
|
||
vals = [r[5] for r in data]
|
||
# 近 60 天
|
||
recent = vals[-60:]
|
||
avg_r = mean(recent)
|
||
std_r = stdev(recent)
|
||
z = (spread_now - avg_r) / std_r if std_r else 0
|
||
print(f" 近60日均值: {avg_r:.1f} σ: {std_r:.1f} z值: {z:.2f}")
|
||
if abs(z) >= 2:
|
||
print(f" → 偏离 {z:.0f}σ，触发信号！")
|
||
else:
|
||
print(f" → 未触发 (需 |z|>=2)")
|