第三十章：综合实战项目——从数据获取到实盘部署的全流程实战

终于走到这一步了。

前面二十九章，我们聊了基差交易的原理、统计套利的模型、风险管理的细节。但说实话，光有理论就像手里有把好刀却不知道怎么用。这一章，咱们就把所有东西串起来，亲手搭一个完整的基差交易系统。

我个人习惯是，做这种综合项目，先画一张大图，把整个流程看清楚。来，先看看这个系统的整体架构：

第一步：数据获取——没有数据，一切都是空谈

做基差交易，最核心的数据就两样：期货价格和现货价格。但实际落地时，你会发现事情没那么简单。

数据来源怎么选？

• 期货数据：我一般用 CTP 接口或者新浪/腾讯的免费行情。实盘用 CTP，回测用本地存储的 tick 数据。
• 现货数据：这个比较头疼。如果是股指期货，用中证指数公司的数据。商品期货的话，得找现货报价平台，比如生意社、我的钢铁网。
• 持仓成本数据：保证金利率、仓储费、交割费，这些得从交易所官网扒。

来看一段数据获取的代码骨架：

import ccxt
import pandas as pd
from datetime import datetime, timedelta

class DataFetcher:
    def __init__(self, symbol_spot='BTC/USDT', symbol_future='BTC-USD-240329'):
        self.exchange = ccxt.binance()
        self.spot_symbol = symbol_spot
        self.future_symbol = symbol_future
    
    def fetch_spot_price(self):
        """获取现货最新价"""
        ticker = self.exchange.fetch_ticker(self.spot_symbol)
        return ticker['last']
    
    def fetch_future_price(self):
        """获取期货最新价"""
        ticker = self.exchange.fetch_ticker(self.future_symbol)
        return ticker['last']
    
    def fetch_historical_data(self, days=30):
        """获取历史数据用于回测"""
        since = self.exchange.parse8601(
            (datetime.now() - timedelta(days=days)).isoformat()
        )
        ohlcv = self.exchange.fetch_ohlcv(
            self.spot_symbol, '1h', since=since
        )
        df = pd.DataFrame(ohlcv, columns=['timestamp','open','high','low','close','volume'])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        return df

第二步：基差计算引擎——核心中的核心

基差 = 期货价格 - 现货价格。就这么简单？嗯，理论上是。但实际中，你得考虑持仓成本。

完整的基差公式应该是：

基差 = 期货价格 - 现货价格 - 持仓成本

其中持仓成本 = 资金占用成本 + 仓储费 + 保险费 - 便利收益

你想想看，如果忽略持仓成本，你看到的基差可能全是「假信号」。我记得有一次，螺纹钢的基差看起来很大，结果一算持仓成本，发现那点价差连资金利息都覆盖不了。

class BasisCalculator:
    def __init__(self, interest_rate=0.04, storage_cost=0, convenience_yield=0):
        self.interest_rate = interest_rate  # 年化利率
        self.storage_cost = storage_cost    # 仓储费（按天）
        self.convenience_yield = convenience_yield  # 便利收益
    
    def cost_of_carry(self, days_to_expiry):
        """计算持仓成本"""
        financing_cost = self.interest_rate * (days_to_expiry / 365)
        storage = self.storage_cost * days_to_expiry
        return financing_cost + storage - self.convenience_yield
    
    def basis(self, future_price, spot_price, days_to_expiry):
        """计算调整后的基差"""
        carry = self.cost_of_carry(days_to_expiry)
        raw_basis = future_price - spot_price
        adjusted_basis = raw_basis - carry * spot_price
        return adjusted_basis

第三步：信号生成——什么时候进场，什么时候跑

基差算出来了，然后呢？你得有个规则来判断「现在该不该交易」。

我常用的信号策略：

1. 阈值法：基差超过历史均值 ± 2 倍标准差时，开仓；回归到均值时平仓。
2. 统计套利法：用协整检验确认期货和现货有长期均衡关系，然后做配对交易。
3. 机器学习法：用 LSTM 预测基差的未来走势，这个我后面会单独讲。

class SignalGenerator:
    def __init__(self, entry_z=2.0, exit_z=0.5):
        self.entry_z = entry_z
        self.exit_z = exit_z
    
    def generate_signal(self, basis_series):
        """生成交易信号"""
        mean = basis_series.mean()
        std = basis_series.std()
        current_basis = basis_series.iloc[-1]
        
        z_score = (current_basis - mean) / std
        
        if z_score > self.entry_z:
            return 'short_basis'  # 基差过大，做空基差（卖期货买现货）
        elif z_score < -self.entry_z:
            return 'long_basis'   # 基差过小，做多基差（买期货卖现货）
        elif abs(z_score) < self.exit_z:
            return 'close'        # 回归，平仓
        else:
            return 'hold'         # 持有不动

第四步：风控与执行——活下来比赚钱更重要

信号有了，但你不能无脑下单。风控是基差交易的命门。

我的风控清单：

• 单笔最大亏损：不超过总资金的 2%。
• 最大持仓时间：基差交易一般持有 3-10 天，超过 15 天强制平仓。
• 流动性检查：买卖盘口深度不够的品种，坚决不碰。
• 极端行情熔断：当基差波动超过 5 倍标准差时，暂停交易 1 小时。

执行层面，我建议用「限价单 + 冰山委托」的方式。为什么？因为基差交易通常涉及双边开仓，如果全用市价单，滑点能吃掉你一半利润。

class RiskManager:
    def __init__(self, max_loss_pct=0.02, max_hold_days=15):
        self.max_loss_pct = max_loss_pct
        self.max_hold_days = max_hold_days
        self.positions = {}
    
    def check_entry(self, signal, capital, current_price):
        """检查是否允许开仓"""
        position_value = capital * 0.1  # 单笔占用10%资金
        stop_loss_price = current_price * (1 - self.max_loss_pct)
        
        return {
            'allowed': True,
            'position_size': position_value // current_price,
            'stop_loss': stop_loss_price
        }
    
    def check_exit(self, position, days_held, current_pnl):
        """检查是否需要强制平仓"""
        if days_held > self.max_hold_days:
            return 'time_exit'
        if current_pnl / position['value'] < -self.max_loss_pct:
            return 'stop_loss'
        return 'hold'

第五步：实盘部署——从代码到真金白银

这一步，很多人会栽跟头。代码在回测里跑得风生水起，一上实盘就各种幺蛾子。

部署架构我推荐这样：

1. 数据层：用 Redis 缓存实时行情，MySQL 存历史数据。
2. 计算层：用 Celery 做定时任务，每 5 分钟计算一次基差。
3. 执行层：对接交易所 API，用异步框架（如 asyncio）处理下单。
4. 监控层：用 Grafana + Prometheus 做实时看板，微信告警。

# 部署脚本示例（简化版）
import schedule
import time
from datetime import datetime

def job():
    """每5分钟执行一次的主循环"""
    # 1. 获取数据
    spot = fetcher.fetch_spot_price()
    future = fetcher.fetch_future_price()
    
    # 2. 计算基差
    days_to_expiry = (expiry_date - datetime.now()).days
    adjusted_basis = calculator.basis(future, spot, days_to_expiry)
    
    # 3. 生成信号
    signal = signal_gen.generate_signal(basis_history)
    
    # 4. 风控检查
    if risk_mgr.check_entry(signal, capital, future):
        # 5. 执行交易
        executor.place_order(signal, spot, future)
    
    # 6. 记录日志
    logger.info(f"Basis: {adjusted_basis:.2f}, Signal: {signal}")

# 定时任务
schedule.every(5).minutes.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

第六步：回测与优化——别被漂亮曲线骗了

系统搭好了，第一件事不是实盘，而是回测。

回测时我特别关注三个指标：

指标	含义	我的及格线
夏普比率	风险调整后收益	> 1.5
最大回撤	最惨的时候亏多少	< 15%
胜率	赚钱的交易占比	> 55%
盈亏比	平均盈利 / 平均亏损	> 1.5

如果回测曲线太漂亮，比如年化 50% 以上、最大回撤不到 5%，那你得警惕了。大概率是过拟合，或者数据泄露。我见过太多人拿着回测曲线来问我「这个策略能不能实盘」，我一看，好家伙，用了未来函数。

写在最后

好了，一个完整的基差交易系统就这么搭起来了。从数据获取到实盘部署，每一步都有坑，但每一步也都有解法。

说实话，做量化交易这么多年，我最大的体会是：系统本身并不复杂，复杂的是你对市场的理解和对风险的敬畏。基差交易看起来是数学问题，实际上是对市场结构、资金流向、交割规则的深刻理解。

嗯，这一章的内容就到这里。代码你可以直接拿去用，但记得根据自己的品种和资金量做调整。毕竟，别人的鞋再好看，也不一定合你的脚。