价格是持仓的结果，曲线是原因。

2020 年 4 月，WTI 原油期货结算价跌至 -37.63 美元/桶。那一天，全世界的财经媒体都在播报"负油价"这个史无前例的事件。但很少有人注意到另一件事：全球最大的原油 ETF——USO（United States Oil Fund），在 2020 年上半年原油价格上涨的背景下，净值却下跌了超过 36%。

原油价格涨了，USO 为什么亏钱？

答案藏在期货合约的"展期"机制里。USO 不持有实物原油，它持有的是近月期货合约。当近月合约即将到期时，USO 必须卖掉近月合约、转而买入远月合约。这个"换月"的动作，在期货术语里叫展期（Roll）。在 2020 年 4 月，全球原油市场处于深度期货溢价（Contango）结构——远月合约比近月合约贵得多。每次展期，USO 都在"高卖低买"，净值被系统性侵蚀。

这就是期货市场特有的收益来源：展期收益（Roll Yield）。它不是来自标的价格的涨跌，而是来自期货曲线结构本身。

理解展期收益的机制，不仅是套利玩家的必备技能，也是大宗商品量化策略的核心课题——CTA 基金的多空混合、商品指数的被动投资、甚至宏观对冲基金的期限结构交易，都在以不同方式捕获这一收益来源。

本文将系统拆解三个问题：

1. 期货曲线如何产生展期收益？
2. 如何量化展期结构并设计捕获策略？
3. 如何在历史数据上回测展期策略？

全文包含完整的数据获取、曲线计算和回测代码，核心数据接口使用 TickDB 的 K 线接口（覆盖 10 年级别清洗对齐的期货历史数据）。

---

一、期货曲线的基础结构

1.1 近月合约与远月合约

每个期货品种在同一时刻都有多个不同到期月份的合约在交易。以 WTI 原油为例，2026 年 4 月时市场上同时存在：

• CL2026M（2026 年 6 月到期）
• CL2026Q（2026 年 9 月到期）
• CL2026Z（2026 年 12 月到期）

同一品种不同到期合约的价格，按时间排列，就构成了期货曲线（Futures Curve）。

期货曲线有两种基本形态：

曲线形态	名称	特征	对多头持仓的影响
Contango（期货溢价）	近低远高	远月价格 > 近月价格	每次展期产生负收益（买贵卖便宜）
Backwardation（现货溢价）	近高远低	远月价格 < 近月价格	每次展期产生正收益（买便宜卖贵）

数学上可以这样定义曲线的陡峭程度：

$$
\text{Spread}t = \frac{F{\text{near},t} - F_{\text{far},t}}{F_{\text{near},t}} \times \frac{365}{T_{\text{days}}}
$$

其中 $T_{\text{days}}$ 是近月与远月合约之间的到期天数差，$\text{Spread}$ 的单位是年化百分比。这个指标衡量的是曲线的平坦或陡峭程度，数值越大表示曲线越陡。

1.2 为什么期货曲线不是平的

期货曲线的结构反映了市场对未来供需的预期，这是理解展期收益的前提。

Contango 结构的成因：

• 储存成本高（原油需要油罐或油轮储存，成本显著）
• 市场预期未来供需宽松（增产预期、需求下降预期）
• 金融机构多头展期需求旺盛，形成买压推高远月

Backwardation 结构的成因：

• 短期供给紧张（地缘冲突、OPEC 减产、管道故障）
• 现货溢价补偿（持有实物者不愿交货）
• 市场预期供需最终会再平衡

关键洞察：期货曲线不是随机的，它的形态本身携带了关于未来供需的"市场预期"。展期策略的本质，是判断这种预期与现实之间的偏差。

---

二、展期收益的量化机制

2.1 展期收益的数学定义

假设你在时刻 $t$ 买入近月合约，价格为 $F_t$。经过一天后（时刻 $t+1$），你有两个变化需要处理：

1. 价格变动收益：近月合约价格从 $F_t$ 变为 $F_{t+1}$
2. 展期收益：近月合约即将到期，你需要将仓位从近月合约滚动到远月合约

设近月合约到期前的剩余天数为 $N$，展期后的远月合约为 $F_t^{\text{next}}$，则展期收益率为：

$$
r_{\text{roll},t} = \frac{F_t^{\text{next}} - F_t}{F_t} \times \frac{1}{N}
$$

当 $F_t^{\text{next}} > F_t$（Contango）时，$r_{\text{roll}}$ 为负；当 $F_t^{\text{next}} < F_t$（Backwardation）时，$r_{\text{roll}}$ 为正。

总持仓收益则由两部分组成：

$$
r_{\text{total},t} = \underbrace{\frac{F_{t+1} - F_t}{F_t}}{\text{价格变动收益}} + \underbrace{r{\text{roll},t}}_{\text{展期收益}}
$$

这就是为什么在 Contango 市场中，即便原油价格上涨，持有多头期货的基金净值可能仍然下跌——价格变动收益被展期损失抵消了。

2.2 展期窗口与展期时机

在实际操作中，展期不是发生在合约到期的最后一刻。基金管理人会提前开始展期，以避免流动性枯竭。常见的展期策略有三类：

展期策略	描述	优点	缺点
固定日期展期	每月固定日期（如到期前 5-10 天）展期	简单可预测	忽略曲线结构变化
持仓期优化展期	根据历史数据寻找最优展期窗口	收益最大化	参数敏感，存在过拟合风险
结构信号展期	当曲线转为 Backwardation 时延迟展期	动态适应市场	可能错过最佳展期点

对于量化策略来说，结构信号展期是最值得深入研究的方向。它将展期时机从"日历问题"转化为"市场结构判断问题"。

---

三、展期结构监控：数据获取与指标计算

3.1 数据获取

实现展期策略的第一步是构建期货曲线数据。由于期货品种繁多（不同月份合约、不同交易所），获取完整且对齐的历史数据是一个工程挑战。

以下是使用 TickDB 获取原油期货 K 线数据的生产级代码示例。TickDB 提供了覆盖 10 年级别的期货历史 K 线数据，可用于计算合约间价差和展期指标：

import os
import time
import json
import requests
from datetime import datetime, timedelta
from typing import Optional, List, Dict
import pandas as pd


class FuturesCurveData:
    """
    期货曲线数据获取器
    从 TickDB 获取多合约 K 线数据，计算展期结构指标
    
    ⚠️ 注意事项：
    - 需设置 TICKDB_API_KEY 环境变量
    - 期货品种格式因交易所而异，请参考 TickDB symbol 规范
    - 高频请求需遵守限频规则（code:3001 + Retry-After）
    """

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("请设置 TICKDB_API_KEY 环境变量")
        
        self.base_url = "https://api.tickdb.ai/v1"
        self.headers = {"X-API-Key": self.api_key}
        self._session = requests.Session()
        self._session.headers.update(self.headers)
        
        # 请求计数器，用于限频监控
        self._request_count = 0
        self._last_request_time = time.time()

    def _handle_rate_limit(self, response: requests.Response):
        """处理限频响应（code: 3001）"""
        if response.status_code == 429 or (response.text and "3001" in response.text):
            retry_after = int(response.headers.get("Retry-After", 5))
            print(f"⚠️ 触发限频，等待 {retry_after} 秒...")
            time.sleep(retry_after)
            return True
        return False

    def get_kline(
        self, 
        symbol: str, 
        interval: str = "1d", 
        start_time: Optional[int] = None,
        end_time: Optional[int] = None,
        limit: int = 1000
    ) -> pd.DataFrame:
        """
        获取 K 线数据
        
        参数:
            symbol: 品种代码（如 CL.M26, CL.U26）
            interval: K 线周期（1m, 5m, 1h, 1d, 1w）
            start_time: 起始时间戳（毫秒）
            end_time: 结束时间戳（毫秒）
            limit: 单次最大获取数量
        
        返回:
            DataFrame，列为 [timestamp, open, high, low, close, volume]
        """
        url = f"{self.base_url}/market/kline"
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit,
        }
        if start_time:
            params["start"] = start_time
        if end_time:
            params["end"] = end_time

        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = self._session.get(
                    url, 
                    params=params, 
                    timeout=(3.05, 10)
                )
                
                if self._handle_rate_limit(response):
                    continue
                    
                response.raise_for_status()
                data = response.json()
                
                self._request_count += 1
                
                if data.get("code") != 0:
                    raise RuntimeError(f"API 错误 {data.get('code')}: {data.get('message')}")
                
                klines = data.get("data", {}).get("klines", [])
                
                df = pd.DataFrame(klines)
                if not df.empty:
                    df["timestamp"] = pd.to_datetime(df["t"], unit="ms")
                    df = df[["timestamp", "o", "h", "l", "c", "v"]]
                    df.columns = ["timestamp", "open", "high", "low", "close", "volume"]
                
                return df

            except requests.exceptions.Timeout:
                print(f"⏱️ 请求超时（尝试 {attempt + 1}/{max_retries}）")
                time.sleep(2 ** attempt)
            except requests.exceptions.RequestException as e:
                print(f"❌ 请求失败: {e}")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                else:
                    raise

        return pd.DataFrame()

    def get_near_next_contracts(
        self,
        base_symbol: str,
        current_date: datetime
    ) -> tuple:
        """
        获取近月合约和远月合约代码
        实际项目中应结合交易所公布的合约日历表
        此处为简化示例
        
        返回: (近月合约代码, 远月合约代码)
        """
        # 简化逻辑：实际使用时建议维护一份合约日历
        # 格式: CL.M26 (近月), CL.U26 (次近月)
        month_map = {
            1:"F", 2:"G", 3:"H", 4:"J", 5:"K", 6:"M",
            7:"N", 8:"Q", 9:"U", 10:"V", 11:"X", 12:"Z"
        }
        
        year = current_date.year % 100
        month = current_date.month
        
        near_code = f"{base_symbol}.{month_map[month]}{year}"
        next_month = month + 1 if month < 12 else 1
        next_year = year if month < 12 else (year + 1) % 100
        next_code = f"{base_symbol}.{month_map[next_month]}{next_year}"
        
        return near_code, next_code


# 使用示例
if __name__ == "__main__":
    client = FuturesCurveData()
    
    # 获取近月合约过去 90 天数据
    end_ts = int(datetime.now().timestamp() * 1000)
    start_ts = int((datetime.now() - timedelta(days=90)).timestamp() * 1000)
    
    near, next_contract = client.get_near_next_contracts("CL", datetime.now())
    print(f"近月合约: {near}, 次近月合约: {next_contract}")
    
    df_near = client.get_kline(near, "1d", start_ts, end_ts)
    df_next = client.get_kline(next_contract, "1d", start_ts, end_ts)
    
    print(f"近月数据: {len(df_near)} 条, 时间范围: {df_near['timestamp'].min()} ~ {df_near['timestamp'].max()}")

3.2 展期结构指标计算

获取到近月和远月合约数据后，可以计算三个核心指标：

import numpy as np


class RollYieldAnalyzer:
    """
    展期收益分析器
    基于近月/远月合约 K 线数据，计算展期结构指标
    """

    def __init__(self, near_df: pd.DataFrame, next_df: pd.DataFrame):
        """
        参数:
            near_df: 近月合约 K 线 DataFrame
            next_df: 远月合约 K 线 DataFrame
        """
        self.near = near_df.set_index("timestamp").sort_index()
        self.next = next_df.set_index("timestamp").sort_index()
        
        # 内连接对齐时间
        merged = self.near.join(
            self.next[["close"]], 
            how="inner", 
            rsuffix="_next"
        )
        self.data = merged[["close", "close_next"]].dropna()

    def calc_basis_spread(self) -> pd.Series:
        """
        计算基差展期（年化）
        
        公式: Spread = (Near - Next) / Near * (365 / days_to_expiry)
        返回: 年化基差展期序列
        """
        # 简化：假设近月与次近月之间固定 30 天
        days_interval = 30
        basis = (self.data["close"] - self.data["close_next"]) / self.data["close"]
        annualized = basis * (365 / days_interval)
        return annualized

    def calc_implied_roll(self) -> pd.Series:
        """
        计算隐含展期收益率（每日）
        
        公式: r_roll = (F_next - F_near) / F_near * (1 / days_interval)
        """
        days_interval = 30
        roll = (self.data["close_next"] - self.data["close"]) / self.data["close"]
        daily_roll = roll / days_interval
        return daily_roll

    def detect_curve_regime(self, threshold: float = 0.005) -> pd.Series:
        """
        检测曲线形态（Contango / Backwardation）
        
        参数:
            threshold: 年化展期阈值（默认 ±0.5%），用于过滤噪音
        
        返回:
            Series: 1 = Backwardation, -1 = Contango, 0 = 中性
        """
        spread = self.calc_basis_spread()
        regime = pd.Series(0, index=spread.index)
        regime[spread > threshold] = 1      # Backwardation
        regime[spread < -threshold] = -1     # Contango
        return regime

    def rolling_regime_stats(self, window: int = 20) -> pd.DataFrame:
        """
        计算滚动窗口内的展期结构统计
        
        参数:
            window: 滚动窗口天数
        
        返回:
            DataFrame: 包含展期均值、标准差、当前形态等统计量
        """
        spread = self.calc_basis_spread()
        regime = self.detect_curve_regime()
        
        stats = pd.DataFrame({
            "annualized_spread": spread,
            "daily_roll_yield": self.calc_implied_roll() * 100,  # 转为百分比
            "regime": regime,
        })
        
        rolling_stats = pd.DataFrame({
            "spread_mean": spread.rolling(window).mean() * 100,
            "spread_std": spread.rolling(window).std() * 100,
            "contango_ratio": (regime == -1).rolling(window).mean(),
            "backwardation_ratio": (regime == 1).rolling(window).mean(),
        }, index=spread.index)
        
        return pd.concat([stats, rolling_stats], axis=1)

---

四、展期收益捕获策略：设计与回测

4.1 策略核心逻辑

基于以上分析，我们设计一个基于曲线结构的动态展期策略。策略的核心思想是：

当市场处于 Backwardation（现货溢价）时，延迟展期以捕获正展期收益；当市场处于 Contango（期货溢价）时，提前展期以减少负展期暴露。

具体规则如下：

市场形态	曲线信号	展期行为
Backwardation（展期收益为正）	年化展期 > +1%	延迟展期，持有近月合约更长时间
Contango（展期收益为负）	年化展期 < -1%	提前展期，减少近月持有
中性	年化展期在 ±1% 之间	按固定节奏展期

4.2 展期策略回测框架

import numpy as np
import pandas as pd
from typing import Optional, Dict, List
from dataclasses import dataclass


@dataclass
class RollSignal:
    """展期信号数据结构"""
    timestamp: pd.Timestamp
    near_price: float
    next_price: float
    annualized_spread: float  # 年化基差
    regime: int               # 1=Backwardation, -1=Contango, 0=中性
    signal: str               # "hold_near" / "roll_early" / "roll_normal"


class RollYieldBacktester:
    """
    展期策略回测引擎
    
    功能:
    1. 模拟期货多头持仓期间的展期操作
    2. 计算展期收益对总收益的贡献
    3. 对比固定日期展期 vs 曲线信号展期两种策略
    
    ⚠️ 回测局限性：
    - 未考虑实际交易中的滑点和流动性冲击
    - 展期成本假设为固定的合约价差
    - 未考虑保证金利息和资金占用成本
    """

    def __init__(
        self,
        near_df: pd.DataFrame,
        next_df: pd.DataFrame,
        roll_interval_days: int = 5,
        backwardation_threshold: float = 0.01,
        contango_threshold: float = -0.01,
    ):
        """
        参数:
            near_df: 近月合约 K 线
            next_df: 远月合约 K 线
            roll_interval_days: 固定展期策略下的展期周期（天）
            backwardation_threshold: Backwardation 阈值（年化）
            contango_threshold: Contango 阈值（年化）
        """
        self.analyzer = RollYieldAnalyzer(near_df, next_df)
        self.roll_interval = roll_interval_days
        self.bt_threshold = backwardation_threshold
        self.ct_threshold = contango_threshold
        
        self._prepare_data()

    def _prepare_data(self):
        """合并数据并计算信号"""
        self.stats = self.analyzer.rolling_regime_stats(window=20)
        
        # 生成展期信号
        self.signals: List[RollSignal] = []
        
        for ts, row in self.stats.iterrows():
            spread = row["annualized_spread"]
            near_price = row["close"]
            next_price = row["close_next"]
            
            # 展期信号决策
            if spread > self.bt_threshold:
                signal = "hold_near"  # Backwardation，持有近月
            elif spread < self.ct_threshold:
                signal = "roll_early"  # Contango，提前展期
            else:
                signal = "roll_normal"
            
            self.signals.append(RollSignal(
                timestamp=ts,
                near_price=near_price,
                next_price=next_price,
                annualized_spread=spread,
                regime=row["regime"],
                signal=signal,
            ))

        self.signal_df = pd.DataFrame(self.signals)

    def simulate_fixed_roll(self) -> pd.DataFrame:
        """
        模拟固定日期展期策略
        
        假设每 N 天（roll_interval）展期一次，
        展期成本 = (远月价格 - 近月价格) / 近月价格
        """
        df = self.signal_df.copy()
        df["days_since_roll"] = np.arange(len(df)) % self.roll_interval
        
        # 模拟展期操作标记
        df["roll_executed"] = df["days_since_roll"] == 0
        
        # 展期收益率（仅在展期日计算）
        df["roll_return"] = np.where(
            df["roll_executed"],
            (df["next_price"] - df["near_price"]) / df["near_price"],
            0.0
        )
        
        # 累计复利展期收益
        df["cumulative_roll_return"] = (1 + df["roll_return"]).cumprod()
        
        return df

    def simulate_signal_roll(self) -> pd.DataFrame:
        """
        模拟信号驱动展期策略
        
        展期决策根据曲线形态动态调整：
        - hold_near: 不展期，等待 Backwardation 消退
        - roll_early: 立即展期，减少 Contango 暴露
        - roll_normal: 按固定节奏展期
        """
        df = self.signal_df.copy()
        
        roll_returns = []
        position_held = True
        
        for i, row in df.iterrows():
            signal = row["signal"]
            near_price = row["near_price"]
            next_price = row["next_price"]
            
            if signal == "hold_near":
                # 不展期，但记录近月合约每日收益率
                daily_return = 0.0  # 展期日之外，展期收益为 0
                roll_returns.append(daily_return)
            elif signal == "roll_early" and position_held:
                # 提前展期，计算展期成本
                roll_cost = (next_price - near_price) / near_price
                roll_returns.append(roll_cost)
                position_held = False
            else:
                # 正常展期或已展期
                roll_returns.append(0.0)
        
        df["signal_roll_return"] = roll_returns
        df["cumulative_signal_roll"] = (1 + pd.Series(roll_returns)).cumprod()
        
        return df

    def run_backtest(self) -> Dict:
        """
        运行完整回测，对比两种策略
        
        返回:
            包含各策略收益指标的字典
        """
        fixed = self.simulate_fixed_roll()
        signal = self.simulate_signal_roll()
        
        # 计算近月合约自身价格收益（作为基准）
        near_returns = fixed["near_price"].pct_change().fillna(0)
        cumulative_near = (1 + near_returns).cumprod()
        
        # 总收益 = 近月价格变动 + 展期收益
        fixed_total = cumulative_near * fixed["cumulative_roll_return"]
        signal_total = cumulative_near * signal["cumulative_signal_roll"]
        
        # 核心指标计算
        def calc_metrics(cumulative_curve: pd.Series, name: str) -> Dict:
            returns = cumulative_curve.pct_change().dropna()
            
            total_return = (cumulative_curve.iloc[-1] - 1) * 100
            annual_return = total_return / (len(cumulative_curve) / 252)  # 假设 252 交易日
            
            if returns.std() > 0:
                sharpe = annual_return / (returns.std() * np.sqrt(252))
            else:
                sharpe = 0.0
            
            peak = cumulative_curve.expanding().max()
            drawdown = (cumulative_curve - peak) / peak * 100
            max_drawdown = drawdown.min()
            
            return {
                "策略": name,
                "累计收益率(%)": round(total_return, 2),
                "年化收益率(%)": round(annual_return, 2),
                "夏普比率": round(sharpe, 2),
                "最大回撤(%)": round(max_drawdown, 2),
            }
        
        results = {
            "基准（仅持有多头近月）": calc_metrics(cumulative_near, "基准"),
            "固定日期展期": calc_metrics(fixed_total, "固定展期"),
            "曲线信号展期": calc_metrics(signal_total, "信号展期"),
        }
        
        # 展期收益分解
        fixed_roll_total = (fixed["cumulative_roll_return"].iloc[-1] - 1) * 100
        signal_roll_total = (signal["cumulative_signal_roll"].iloc[-1] - 1) * 100
        near_total = (cumulative_near.iloc[-1] - 1) * 100
        
        results["展期收益分解"] = {
            "近月价格变动收益(%)": round(near_total, 2),
            "固定展期策略贡献(%)": round(fixed_roll_total, 2),
            "信号展期策略贡献(%)": round(signal_roll_total, 2),
            "两种展期策略差异(%)": round(signal_roll_total - fixed_roll_total, 2),
        }
        
        return results

    def generate_report(self) -> str:
        """生成文本回测报告"""
        results = self.run_backtest()
        
        report_lines = [
            "=" * 60,
            "展期策略回测报告",
            "=" * 60,
            "",
            "一、策略表现对比",
            "-" * 40,
        ]
        
        for strategy, metrics in results.items():
            if strategy == "展期收益分解":
                report_lines.append("")
                report_lines.append("二、展期收益分解")
                report_lines.append("-" * 40)
                for k, v in metrics.items():
                    report_lines.append(f"  {k}: {v}")
            else:
                report_lines.append(f"\n【{strategy}】")
                for k, v in metrics.items():
                    if k != "策略":
                        report_lines.append(f"  {k}: {v}")
        
        report_lines.append("")
        report_lines.append("三、回测参数")
        report_lines.append("-" * 40)
        report_lines.append(f"  回测区间: {self.stats.index.min().date()} ~ {self.stats.index.max().date()}")
        report_lines.append(f"  固定展期周期: 每 {self.roll_interval} 天")
        report_lines.append(f"  Backwardation 阈值: 年化 {self.bt_threshold*100}%")
        report_lines.append(f"  Contango 阈值: 年化 {self.ct_threshold*100}%")
        
        report_lines.append("")
        report_lines.append("=" * 60)
        report_lines.append("⚠️ 回测局限性说明：")
        report_lines.append("  以上结果基于历史数据模拟，未完全考虑：")
        report_lines.append("  - 实际展期的流动性冲击和滑点成本")
        report_lines.append("  - 保证金利息和资金占用成本")
        report_lines.append("  - 合约切换时的价格跳空")
        report_lines.append("  - 极端行情下的展期困难")
        report_lines.append("  建议在实际使用前进行更严格的模拟和实盘验证。")
        report_lines.append("=" * 60)
        
        return "\n".join(report_lines)

4.3 回测使用示例

# 使用 TickDB 数据运行回测
if __name__ == "__main__":
    client = FuturesCurveData()
    
    # 获取 3 年历史数据
    end_ts = int(datetime.now().timestamp() * 1000)
    start_ts = int((datetime.now() - timedelta(days=365 * 3)).timestamp() * 1000)
    
    # 近月和次近月合约代码（示例）
    # 实际使用时请根据期货合约日历替换为真实合约代码
    near_symbol = "CL.M26"      # WTI 原油近月
    next_symbol = "CL.U26"      # WTI 原油次近月
    
    print(f"正在获取 {near_symbol} 和 {next_symbol} 历史数据...")
    
    df_near = client.get_kline(near_symbol, "1d", start_ts, end_ts)
    df_next = client.get_kline(next_symbol, "1d", start_ts, end_ts)
    
    if df_near.empty or df_next.empty:
        print("❌ 数据获取失败，请检查品种代码是否正确")
        print("💡 访问 tickdb.ai 查看支持的期货品种列表")
    else:
        print(f"✅ 近月数据 {len(df_near)} 条，远月数据 {len(df_next)} 条")
        
        # 运行回测
        backtester = RollYieldBacktester(
            near_df=df_near,
            next_df=df_next,
            roll_interval_days=5,
            backwardation_threshold=0.01,
            contango_threshold=-0.01,
        )
        
        print(backtester.generate_report())

---

五、关键风险与局限性

展期策略在理论上看似优雅，但实际执行中面临多个维度的风险：

5.1 展期成本风险

在深度 Contango 市场中，每次展期都意味着实质性损失。2020 年上半年的原油市场，年化 Contango 幅度一度超过 40%，即便原油价格上涨超过 30%，多头期货的总收益仍然是负的。

量化处理方式：在策略层面设定最大可接受展期成本阈值，超过阈值时停止开仓或转为做空展期。

5.2 流动性风险

期货合约在临近到期时流动性急剧下降。实际展期时，可能面临：

• 买卖价差急剧扩大
• 无法以理想价格完成展期
• 持仓量（Open Interest）不足以容纳既有仓位

量化处理方式：展期前监控合约的持仓量和买卖盘深度，在流动性充足时提前分批展期。

5.3 曲线结构突变风险

期货曲线形态可能在短时间内发生剧烈变化。例如，地缘政治事件导致原油供给预期骤变，Contango 在数小时内变为 Backwardation。依赖历史均值回复的策略可能在这种情形下失效。

量化处理方式：引入波动率过滤器，当市场不确定性极高时降低仓位。

5.4 滚动收益率的均值回复陷阱

历史数据显示，展期收益率（年化基差）具有明显的均值回复特征——极端 Backwardation 之后通常会回归中性，反之亦然。但这并不意味着简单的"均值回复策略"就能盈利。均值回复的速度和幅度都有高度不确定性，参数过拟合是实盘中的常见问题。

---

六、实际部署考量

将上述回测框架部署到生产环境，还需要补充以下工程要素：

组件	说明	实现要点
合约日历管理	自动识别当前主力合约和次主力合约	需对接交易所公布的合约到期日历
实时曲线监控	在交易时段实时追踪近远月价差变化	WebSocket 推送 + 阈值告警
分批展期执行	避免单次大量展期的流动性冲击	将总仓位分 3-5 批次，每日执行一部分
滑点估算模型	基于历史买卖价差估算实际展期成本	建议在计算展期收益时额外扣除 0.02-0.05% 作为缓冲
保证金管理	期货杠杆下保证金不足会强制平仓	建议仓位不超过总资金的 20%，预留足够保证金缓冲

---

结语

期货市场的展期收益，本质上是市场对未来供需预期的一种定价。理解它，不是为了找到稳赚不赔的圣杯，而是为了更诚实地面对期货多头策略的真实收益来源。

三个核心认知：

1. Contango 不是敌人，Backwardation 不是朋友——它们的幅度和持续时间决定了展期策略的盈亏，而非曲线形态本身。
2. 展期收益可以被量化，但无法被消除——无论采用哪种展期策略，曲线结构对持仓收益的影响始终存在。差别在于，你是被动接受还是主动管理。
3. 策略的竞争壁垒在于数据质量和执行效率——当所有人都用同一套曲线信号时，微小的执行差异（展期时机、滑点控制）就成为决定性因素。

---

下一步行动

如果你希望亲手复现本文的回测结果：

1. 访问 tickdb.ai 注册（免费，无需信用卡）
2. 在控制台生成 API Key
3. 设置环境变量 TICKDB_API_KEY，复制本文代码即可运行

如果你关注的是更细粒度的期货数据（分钟级 K 线、成交量加权价格等），TickDB 的 K 线接口支持 1 分钟到 1 周多种时间周期，可用于高频展期结构分析和盘口动态监控。

如果你习惯用 AI 辅助开发，在 AI 助手中搜索安装 tickdb-market-data SKILL，可直接用自然语言查询期货合约数据并生成分析图表。

---

本文不构成任何投资建议。期货交易涉及杠杆，存在重大亏损风险。展期策略的过往表现不代表未来收益，实盘部署前请充分评估流动性和保证金风险。市场有风险，投资需谨慎。