您好，登录后才能下订单哦！
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
# 如何使用Python实现一个简单的商品期货布林指标突破策略
## 一、策略概述
### 1.1 布林带指标简介
布林带(Bollinger Bands)是由约翰·布林格(John Bollinger)在1980年代发明的技术分析工具，由三条线组成：
- 中轨(Middle Band)：N日移动平均线(通常为20日SMA)
- 上轨(Upper Band)：中轨 + k倍标准差(通常k=2)
- 下轨(Lower Band)：中轨 - k倍标准差
该指标通过计算价格的波动区间，反映市场的超买超卖状态。当价格突破上下轨时，往往预示着趋势的开始或反转。
### 1.2 突破策略原理
本策略的基本逻辑是：
1. 当价格**收盘价突破上轨**时，做多入场
2. 当价格**收盘价跌破下轨**时，做空入场
3. 当价格**回归中轨**时，平仓离场
## 二、环境准备
### 2.1 所需Python库
```python
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import tushare as ts # 获取金融数据的库
import talib # 技术指标计算库
以螺纹钢期货(RB9999)为例，获取历史数据：
# 使用tushare pro获取数据(需要注册获取token)
pro = ts.pro_api('your_token_here')
# 获取螺纹钢主力合约日线数据
df = pro.fut_daily(ts_code="RB9999.SHF", start_date="20200101", end_date="20231231")
df = df.sort_values('trade_date').reset_index(drop=True)
def calculate_bollinger_bands(data, window=20, k=2):
"""
计算布林带指标
参数:
data: 价格序列
window: 移动平均窗口(默认20)
k: 标准差倍数(默认2)
返回:
DataFrame包含中轨、上轨、下轨
"""
data['MA'] = data['close'].rolling(window=window).mean()
data['STD'] = data['close'].rolling(window=window).std()
data['Upper'] = data['MA'] + k * data['STD']
data['Lower'] = data['MA'] - k * data['STD']
return data[['MA', 'Upper', 'Lower']]
upper, middle, lower = talib.BBANDS(
df['close'].values,
timeperiod=20,
nbdevup=2,
nbdevdn=2,
matype=0
)
df['Upper'] = upper
df['MA'] = middle
df['Lower'] = lower
def generate_signals(df):
"""
生成交易信号
1 表示做多，-1 表示做空，0 表示无信号
"""
df['signal'] = 0
# 突破上轨做多
df.loc[(df['close'] > df['Upper']) & (df['close'].shift(1) <= df['Upper'].shift(1)), 'signal'] = 1
# 突破下轨做空
df.loc[(df['close'] < df['Lower']) & (df['close'].shift(1) >= df['Lower'].shift(1)), 'signal'] = -1
# 回归中轨平仓
df.loc[
((df['close'] <= df['MA']) & (df['close'].shift(1) > df['MA'].shift(1))) |
((df['close'] >= df['MA']) & (df['close'].shift(1) < df['MA'].shift(1))),
'signal'
] = 0
return df
def manage_positions(df):
"""
管理持仓状态
"""
df['position'] = 0 # 持仓方向：1多仓，-1空仓，0无仓
current_position = 0
for i in range(len(df)):
# 如果有信号且当前无持仓
if df.at[i, 'signal'] != 0 and current_position == 0:
current_position = df.at[i, 'signal']
# 如果出现平仓信号
elif df.at[i, 'signal'] == 0 and current_position != 0:
current_position = 0
df.at[i, 'position'] = current_position
return df
def calculate_returns(df):
"""
计算策略收益
"""
df['daily_return'] = df['close'].pct_change()
df['strategy_return'] = df['position'].shift(1) * df['daily_return']
df['cumulative_return'] = (1 + df['strategy_return']).cumprod()
return df
def analyze_performance(df):
"""
分析策略表现
"""
total_return = df['cumulative_return'].iloc[-1] - 1
annual_return = (1 + total_return) ** (252/len(df)) - 1
max_drawdown = (df['cumulative_return'].cummax() - df['cumulative_return']).max()
sharpe_ratio = df['strategy_return'].mean() / df['strategy_return'].std() * np.sqrt(252)
print(f"总收益率: {total_return*100:.2f}%")
print(f"年化收益率: {annual_return*100:.2f}%")
print(f"最大回撤: {max_drawdown*100:.2f}%")
print(f"夏普比率: {sharpe_ratio:.2f}")
plt.figure(figsize=(14, 7))
plt.plot(df['close'], label='Price', alpha=0.5)
plt.plot(df['MA'], label='Middle Band', linestyle='--')
plt.plot(df['Upper'], label='Upper Band', linestyle='--', color='red')
plt.plot(df['Lower'], label='Lower Band', linestyle='--', color='green')
plt.fill_between(df.index, df['Upper'], df['Lower'], color='grey', alpha=0.1)
plt.title('Bollinger Bands')
plt.legend()
plt.show()
plt.figure(figsize=(14, 7))
plt.plot(df['close'], label='Price', alpha=0.5)
# 标记买入信号
plt.scatter(
df.index[df['signal'] == 1],
df['close'][df['signal'] == 1],
marker='^', color='green', label='Buy Signal'
)
# 标记卖出信号
plt.scatter(
df.index[df['signal'] == -1],
df['close'][df['signal'] == -1],
marker='v', color='red', label='Sell Signal'
)
plt.title('Trading Signals')
plt.legend()
plt.show()
plt.figure(figsize=(14, 7))
plt.plot(df['cumulative_return'], label='Strategy')
plt.plot((1 + df['daily_return']).cumprod() - 1, label='Buy & Hold')
plt.title('Cumulative Returns')
plt.legend()
plt.show()
可以通过网格搜索寻找最优参数组合：
def optimize_parameters(df):
results = []
for window in range(10, 50, 5):
for k in [1.5, 2, 2.5]:
temp_df = calculate_bollinger_bands(df.copy(), window, k)
temp_df = generate_signals(temp_df)
temp_df = manage_positions(temp_df)
temp_df = calculate_returns(temp_df)
total_return = temp_df['cumulative_return'].iloc[-1] - 1
max_drawdown = (temp_df['cumulative_return'].cummax() - temp_df['cumulative_return']).max()
results.append({
'window': window,
'k': k,
'return': total_return,
'max_drawdown': max_drawdown
})
return pd.DataFrame(results)
可考虑加入以下改进： 1. 结合趋势过滤器(如200日均线方向) 2. 加入成交量确认 3. 设置止损止盈机制
def add_risk_management(df, stop_loss=0.05, take_profit=0.1):
"""
添加止损止盈逻辑
"""
df['position'] = 0
current_position = 0
entry_price = 0
for i in range(len(df)):
# 开仓逻辑
if df.at[i, 'signal'] != 0 and current_position == 0:
current_position = df.at[i, 'signal']
entry_price = df.at[i, 'close']
# 止损止盈逻辑
elif current_position != 0:
pct_change = (df.at[i, 'close'] - entry_price) / entry_price
if (current_position == 1 and (pct_change <= -stop_loss or pct_change >= take_profit)) or \
(current_position == -1 and (pct_change >= stop_loss or pct_change <= -take_profit)):
current_position = 0
# 中轨平仓逻辑
elif df.at[i, 'signal'] == 0 and current_position != 0:
current_position = 0
df.at[i, 'position'] = current_position
return df
# 完整策略实现
def run_bollinger_strategy(data, window=20, k=2, stop_loss=None, take_profit=None):
# 计算布林带
data = calculate_bollinger_bands(data, window, k)
# 生成信号
data = generate_signals(data)
# 添加风险管理
if stop_loss and take_profit:
data = add_risk_management(data, stop_loss, take_profit)
else:
data = manage_positions(data)
# 计算收益
data = calculate_returns(data)
return data
# 运行策略
result = run_bollinger_strategy(df.copy(), window=20, k=2, stop_loss=0.03, take_profit=0.06)
analyze_performance(result)
本文实现了一个基于布林带突破的商品期货交易策略，主要特点包括： 1. 使用Python实现了布林带指标计算 2. 构建了完整的信号生成和持仓管理系统 3. 加入了基础的回测和绩效评估功能 4. 提出了策略优化和风险控制的改进方向
实际应用中需要注意： - 历史回测不能完全预测未来表现 - 需考虑交易成本和滑点影响 - 建议在模拟盘充分测试后再投入实盘
通过不断优化参数和加入更多过滤条件，可以进一步提高策略的稳定性和盈利能力。 “`
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。