Python期货量化交易：甲醇与白糖均线策略实战

前言

期货量化交易的第一步是什么？不是写策略，而是获取数据+搭建回测框架。

本文用Python实现两个期货策略：

1. 甲醇策略：三均线 + ATR过滤
2. 白糖策略：均线 + 季节性规律

---

数据获取：AkShare

AkShare 是开源的Python金融数据接口库，支持期货数据：

import akshare as ak

# 获取甲醇主力合约日线数据
df = ak.futures_main_sina(symbol='MA0')

# 获取白糖主力合约
df = ak.futures_main_sina(symbol='SR0')

MA0是甲醇主力连续合约，SR0是白糖主力连续合约。

---

甲醇策略：三均线 + ATR过滤

策略逻辑

# 三条均线
MA_short = MA(10)   # 短期均线
MA_long = MA(30)    # 长期均线
MA_filter = MA(60)  # 趋势过滤均线

# 信号生成
多头：MA_short > MA_long 且 close > MA_filter
空头：MA_short < MA_long 且 close < MA_filter

核心思想：

• 短期均线上穿长期均线 → 趋势向上
• 价格在60日均线上方 → 确认多头趋势
• ATR用于衡量波动率（辅助止损）

代码实现

def calculate_ma_signals(data, short_ma=10, long_ma=30, filter_ma=60):
    data['MA_short'] = data['close'].rolling(short_ma).mean()
    data['MA_long'] = data['close'].rolling(long_ma).mean()
    data['MA_filter'] = data['close'].rolling(filter_ma).mean()
    
    # ATR计算
    data['high_low'] = data['high'] - data['low']
    data['high_close'] = np.abs(data['high'] - data['close'].shift())
    data['low_close'] = np.abs(data['low'] - data['close'].shift())
    data['TR'] = data[['high_low', 'high_close', 'low_close']].max(axis=1)
    data['ATR'] = data['TR'].rolling(14).mean()
    
    # 信号：1=多头，-1=空头，0=无仓位
    data['Signal'] = 0
    long_cond = (data['MA_short'] > data['MA_long']) & (data['close'] > data['MA_filter'])
    short_cond = (data['MA_short'] < data['MA_long']) & (data['close'] < data['MA_filter'])
    data.loc[long_cond, 'Signal'] = 1
    data.loc[short_cond, 'Signal'] = -1
    
    return data

---

白糖策略：均线 + 季节性

白糖策略逻辑

白糖有明显的季节性规律——每年5-8月是消费旺季（夏季饮料需求），价格往往上涨。

def calculate_sr_signals(data, ma1=5, ma2=20, ma3=60, bull_start=5, bull_end=8):
    data['MA1'] = data['close'].rolling(ma1).mean()
    data['MA2'] = data['close'].rolling(ma2).mean()
    data['MA3'] = data['close'].rolling(ma3).mean()
    
    # 季节性判断：5-8月为多头季节
    data['Month'] = data.index.month
    data['Season_Bull'] = data['Month'].between(bull_start, bull_end)
    
    # 信号：三条均线多头排列 + 处于多头季节
    data['Signal'] = 0
    long_cond = (data['close'] > data['MA1']) & (data['close'] > data['MA2']) & \
                (data['close'] > data['MA3']) & (data['Season_Bull'])
    short_cond = (data['close'] < data['MA1']) & (data['close'] < data['MA2']) & \
                 (data['close'] < data['MA3']) & (~data['Season_Bull'])
    data.loc[long_cond, 'Signal'] = 1
    data.loc[short_cond, 'Signal'] = -1
    
    return data

核心思想：

• 三条均线（5/20/60日）全部向上排列 → 强趋势
• 5-8月旺季 → 做多为主
• 非旺季 + 均线空头 → 做空

---

回测引擎

def backtest_strategy(data):
    # 信号延迟一天执行（避免未来函数）
    data['Position'] = data['Signal'].shift()
    
    # 日收益率
    data['Return'] = data['close'].pct_change()
    
    # 策略收益 = 持仓 × 日收益
    data['Strategy_Return'] = data['Position'] * data['Return']
    
    # 累计收益
    data['Cum_Return'] = (1 + data['Strategy_Return']).cumprod()
    
    return data

关键细节：Signal.shift() 确保信号延迟一天执行，避免”用未来数据做决策”的错误。

---

绩效评估

def evaluate_strategy(data):
    cum_return = data['Cum_Return'].iloc[-1] - 1
    annual_return = (cum_return + 1) ** (252/len(data)) - 1
    
    # 最大回撤
    data['Peak'] = data['Cum_Return'].cummax()
    data['Drawdown'] = (data['Cum_Return'] - data['Peak']) / data['Peak']
    max_drawdown = data['Drawdown'].min()
    
    # 胜率
    winning_trades = data[data['Strategy_Return'] > 0]
    win_rate = len(winning_trades) / max(1, len(data[data['Signal'] != 0]))
    
    # 盈亏比
    avg_win = winning_trades['Strategy_Return'].mean()
    avg_loss = data[data['Strategy_Return'] < 0]['Strategy_Return'].mean()
    profit_ratio = abs(avg_win / avg_loss) if avg_loss != 0 else np.nan
    
    return {
        '累计收益': cum_return,
        '年化收益': annual_return,
        '最大回撤': max_drawdown,
        '胜率': win_rate,
        '盈亏比': profit_ratio
    }

---

主程序

if __name__ == "__main__":
    # 甲醇策略
    ma_df = get_main_contract('MA0')
    ma_df = calculate_ma_signals(ma_df)
    ma_df = backtest_strategy(ma_df)
    ma_perf = evaluate_strategy(ma_df)
    print("甲醇策略绩效:", ma_perf)
    
    # 白糖策略
    sr_df = get_main_contract('SR0')
    sr_df = calculate_sr_signals(sr_df)
    sr_df = backtest_strategy(sr_df)
    sr_perf = evaluate_strategy(sr_df)
    print("白糖策略绩效:", sr_perf)

---

策略对比

指标	甲醇策略	白糖策略
核心逻辑	3均线+ATR	均线+季节性
适合品种	趋势性强的品种	有季节性规律的品种
信号频率	中等	较低（旺季集中）
风险控制	ATR止损	季节性过滤

---

完整代码

import pandas as pd
import numpy as np
import akshare as ak

def get_main_contract(symbol, start_date=None, end_date=None):
    if start_date and end_date:
        df = ak.futures_main_sina(symbol=symbol, start_date=start_date, end_date=end_date)
    else:
        df = ak.futures_main_sina(symbol=symbol)
    df.rename(columns={
        '日期': 'date', '开盘价': 'open', '最高价': 'high',
        '最低价': 'low', '收盘价': 'close', '成交量': 'volume',
        '持仓量': 'open_interest', '动态结算价': 'settle',
    }, inplace=True)
    df['date'] = pd.to_datetime(df['date'])
    df.set_index('date', inplace=True)
    return df.sort_index()

def calculate_ma_signals(data, short_ma=10, long_ma=30, filter_ma=60):
    data['MA_short'] = data['close'].rolling(short_ma).mean()
    data['MA_long'] = data['close'].rolling(long_ma).mean()
    data['MA_filter'] = data['close'].rolling(filter_ma).mean()
    data['TR'] = data[['high', 'low', 'close']].apply(
        lambda x: max(x['high']-x['low'], abs(x['high']-x['close']), abs(x['low']-x['close'])), axis=1)
    data['ATR'] = data['TR'].rolling(14).mean()
    data['Signal'] = 0
    data.loc[(data['MA_short']>data['MA_long'])&(data['close']>data['MA_filter']), 'Signal'] = 1
    data.loc[(data['MA_short']<data['MA_long'])&(data['close']<data['MA_filter']), 'Signal'] = -1
    return data

def calculate_sr_signals(data, ma1=5, ma2=20, ma3=60):
    data['MA1'] = data['close'].rolling(ma1).mean()
    data['MA2'] = data['close'].rolling(ma2).mean()
    data['MA3'] = data['close'].rolling(ma3).mean()
    data['Season_Bull'] = data.index.month.between(5, 8)
    data['Signal'] = 0
    data.loc[(data['close']>data['MA1'])&(data['close']>data['MA2'])&(data['close']>data['MA3'])&(data['Season_Bull']), 'Signal'] = 1
    data.loc[(data['close']<data['MA1'])&(data['close']<data['MA2'])&(data['close']<data['MA3'])&(~data['Season_Bull']), 'Signal'] = -1
    return data

def backtest_strategy(data):
    data['Position'] = data['Signal'].shift()
    data['Return'] = data['close'].pct_change()
    data['Strategy_Return'] = data['Position'] * data['Return']
    data['Cum_Return'] = (1 + data['Strategy_Return']).cumprod()
    return data

def evaluate_strategy(data):
    cum_return = data['Cum_Return'].iloc[-1] - 1
    annual_return = (cum_return + 1) ** (252/len(data)) - 1
    data['Peak'] = data['Cum_Return'].cummax()
    data['Drawdown'] = (data['Cum_Return'] - data['Peak']) / data['Peak']
    max_drawdown = data['Drawdown'].min()
    win_rate = len(data[data['Strategy_Return']>0]) / max(1, len(data[data['Signal']!=0]))
    avg_win = data[data['Strategy_Return']>0]['Strategy_Return'].mean()
    avg_loss = data[data['Strategy_Return']<0]['Strategy_Return'].mean()
    return {'累计收益': cum_return, '年化收益': annual_return, '最大回撤': max_drawdown, '胜率': win_rate, '盈亏比': abs(avg_win/avg_loss)}

if __name__ == "__main__":
    ma_df = get_main_contract('MA0')
    ma_df = calculate_ma_signals(ma_df)
    ma_df = backtest_strategy(ma_df)
    print("甲醇策略绩效:", evaluate_strategy(ma_df))
    
    sr_df = get_main_contract('SR0')
    sr_df = calculate_sr_signals(sr_df)
    sr_df = backtest_strategy(sr_df)
    print("白糖策略绩效:", evaluate_strategy(sr_df))

---

总结

模块	关键点
数据获取	AkShare futures_main_sina
信号生成	均线交叉+趋势过滤+季节性
回测引擎	shift(1) 避免未来函数
绩效评估	年化收益、最大回撤、胜率、盈亏比

改进建议：

• 加入手续费和滑点
• 用ATR做动态止损
• 多品种组合分散风险
• 加入仓位管理（凯利公式）

---

代码文件：F:\202507\akshare_ma_future.py，106行完整实现。

Stay tuned! 🚀