加密永续合约资金费率均值回归：把多空拥挤度做成可交易信号

import numpy as np
import pandas as pd

np.random.seed(42)

def simulate_funding_rate(n_periods=1000, theta=0.15, mu=0.0001, sigma=0.0003):
    """
    用 OU 过程模拟资金费率序列
    theta: 均值回归速度
    mu:    长期均值（约 0.01%/8h，年化约 11%）
    sigma: 波动率
    """
    rate = np.zeros(n_periods)
    rate[0] = mu
    for t in range(1, n_periods):
        # OU 过程：dr = theta*(mu - r)dt + sigma*dW
        dr = theta * (mu - rate[t-1]) + sigma * np.random.randn()
        rate[t] = rate[t-1] + dr
    return rate

# 每 8 小时一个周期，1000 个周期约等于 333 天
funding = simulate_funding_rate(1000)

# 同步构造一条 BTC 永续价格路径（与资金费率负相关：高费率后倾向回落）
def simulate_price(funding, s0=30000, base_vol=0.02):
    n = len(funding)
    price = np.zeros(n)
    price[0] = s0
    for t in range(1, n):
        # 资金费率极端时，下一期收益倾向反向修正
        reversion_drift = -8.0 * (funding[t-1] - funding.mean())
        shock = base_vol * np.random.randn()
        ret = reversion_drift + shock
        price[t] = price[t-1] * (1 + ret)
    return price

price = simulate_price(funding)

df = pd.DataFrame({
    'funding_rate': funding,
    'price': price
})
df['funding_ma'] = df['funding_rate'].rolling(30).mean()
df['funding_std'] = df['funding_rate'].rolling(30).std()
print(df.describe())

def build_signals(df, z_entry=1.5, z_exit=0.3):
    """
    Z-Score 均值回归信号
    Z > z_entry  : 多头过度拥挤 -> 做空
    Z < -z_entry : 空头过度拥挤 -> 做多
    |Z| < z_exit : 平仓
    """
    df = df.copy()
    df['z_score'] = (df['funding_rate'] - df['funding_ma']) / df['funding_std']

    position = np.zeros(len(df))
    for t in range(1, len(df)):
        z = df['z_score'].iloc[t]
        prev = position[t-1]

        if np.isnan(z):
            position[t] = 0
            continue

        if prev == 0:
            if z > z_entry:
                position[t] = -1   # 费率过高，做空永续
            elif z < -z_entry:
                position[t] = 1    # 费率过低，做多永续
            else:
                position[t] = 0
        else:
            # 持仓中，Z 回归到中枢附近则平仓
            if abs(z) < z_exit:
                position[t] = 0
            else:
                position[t] = prev

    df['position'] = position
    return df

df = build_signals(df)
print(df[['funding_rate', 'z_score', 'position']].tail(20))

def backtest(df, fee=0.0004):
    df = df.copy()
    df['price_ret'] = df['price'].pct_change().fillna(0)

    # 仓位在下一期生效（execute-on-i+1）
    df['exec_position'] = df['position'].shift(1).fillna(0)

    # 价格盈亏
    df['pnl_price'] = df['exec_position'] * df['price_ret']

    # 资金费率盈亏：做空永续（position=-1）时，若费率为正则收取费率
    # 收益 = -position_sign * funding_rate（空头在正费率时收钱）
    df['pnl_funding'] = -df['exec_position'] * df['funding_rate']

    # 交易成本：仓位变化时收取
    df['trade'] = df['exec_position'].diff().abs().fillna(0)
    df['cost'] = df['trade'] * fee

    df['pnl'] = df['pnl_price'] + df['pnl_funding'] - df['cost']
    df['equity'] = (1 + df['pnl']).cumprod()
    return df

result = backtest(df)

# 评估指标（跳过 warmup 段）
warmup = 30
evaluate = result.iloc[warmup:].copy()

total_return = evaluate['equity'].iloc[-1] / evaluate['equity'].iloc[0] - 1
periods_per_year = 3 * 365          # 每天 3 个 8h 周期
n_years = len(evaluate) / periods_per_year
ann_return = (1 + total_return) ** (1 / n_years) - 1
sharpe = evaluate['pnl'].mean() / evaluate['pnl'].std() * np.sqrt(periods_per_year)

# 最大回撤
cummax = evaluate['equity'].cummax()
drawdown = (evaluate['equity'] - cummax) / cummax
max_dd = drawdown.min()

n_trades = int(evaluate['trade'].sum() / 2)

print(f"总收益:    {total_return:.2%}")
print(f"年化收益:  {ann_return:.2%}")
print(f"夏普比率:  {sharpe:.2f}")
print(f"最大回撤:  {max_dd:.2%}")
print(f"交易次数:  {n_trades}")