Bollinger Mean Reversion on a Two-Leg Price Spread
Summary
This Python strategy constructs a spread from two instrument prices weighted by configurable leg ratios. It updates the spread periodically and, after collecting enough observations, calculates a rolling mean and standard deviation over a 20-observation window. The upper and lower bands are set two standard deviations from the mean. When the spread reaches the upper band, it targets a short position in the first leg and a long position in the second; at the lower band, it targets the reverse. Positions are closed when the spread returns to the mean.
The implementation maintains target positions for both legs and submits orders using bar close prices adjusted by a configurable price offset. It requires both legs’ bar data and cancels outstanding orders as each bar update is processed. The document provides implementation code but no historical test, performance metrics, or method for selecting related instruments. Equal default leg ratios do not account for different price scales, hedge ratios, or changing relationships, so the spread’s mean-reverting behavior is assumed rather than demonstrated.
Key ideas
- The strategy calculates a weighted price difference between two instruments and trades deviations from its rolling average.
- Its entry bands are two rolling standard deviations above and below the mean, using a 20-observation window.
- An upper-band signal shorts the first leg and buys the second, while a lower-band signal reverses those positions.
- Both legs are targeted to flat when the spread returns to its mean.
- The code provides no evidence that a chosen pair is cointegrated or that the spread reliably reverts.
Tags
Full text
# PairTradingStrategy
# PairTradingStrategy
## Source (MIT)
```python
from typing import List, Dict
from datetime import datetime
import numpy as np
from howtrader.app.portfolio_strategy import StrategyTemplate, StrategyEngine
from howtrader.trader.utility import BarGenerator
from howtrader.trader.object import TickData, BarData
class PairTradingStrategy(StrategyTemplate):
""""""
author = "用Python的交易员"
price_add = 5
boll_window = 20
boll_dev = 2
fixed_size = 1
leg1_ratio = 1
leg2_ratio = 1
leg1_symbol = ""
leg2_symbol = ""
current_spread = 0.0
boll_mid = 0.0
boll_down = 0.0
boll_up = 0.0
parameters = [
"price_add",
"boll_window",
"boll_dev",
"fixed_size",
"leg1_ratio",
"leg2_ratio",
]
variables = [
"leg1_symbol",
"leg2_symbol",
"current_spread",
"boll_mid",
"boll_down",
"boll_up",
]
def __init__(
self,
strategy_engine: StrategyEngine,
strategy_name: str,
vt_symbols: List[str],
setting: dict
):
""""""
super().__init__(strategy_engine, strategy_name, vt_symbols, setting)
self.bgs: Dict[str, BarGenerator] = {}
self.targets: Dict[str, int] = {}
self.last_tick_time: datetime = None
self.spread_count: int = 0
self.spread_data: np.array = np.zeros(100)
# Obtain contract info
self.leg1_symbol, self.leg2_symbol = vt_symbols
def on_bar(bar: BarData):
""""""
pass
for vt_symbol in self.vt_symbols:
self.targets[vt_symbol] = 0
self.bgs[vt_symbol] = BarGenerator(on_bar)
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bars(1)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
if (
self.last_tick_time
and self.last_tick_time.minute != tick.datetime.minute
):
bars = {}
for vt_symbol, bg in self.bgs.items():
bars[vt_symbol] = bg.generate()
self.on_bars(bars)
bg: BarGenerator = self.bgs[tick.vt_symbol]
bg.update_tick(tick)
self.last_tick_time = tick.datetime
def on_bars(self, bars: Dict[str, BarData]):
""""""
self.cancel_all()
# Return if one leg data is missing
if self.leg1_symbol not in bars or self.leg2_symbol not in bars:
return
# Calculate current spread
leg1_bar = bars[self.leg1_symbol]
leg2_bar = bars[self.leg2_symbol]
# Filter time only run every 5 minutes
if (leg1_bar.datetime.minute + 1) % 5:
return
self.current_spread = (
leg1_bar.close_price * self.leg1_ratio - leg2_bar.close_price * self.leg2_ratio
)
# Update to spread array
self.spread_data[:-1] = self.spread_data[1:]
self.spread_data[-1] = self.current_spread
self.spread_count += 1
if self.spread_count <= self.boll_window:
return
# Calculate boll value
buf: np.array = self.spread_data[-self.boll_window:]
std = buf.std()
self.boll_mid = buf.mean()
self.boll_up = self.boll_mid + self.boll_dev * std
self.boll_down = self.boll_mid - self.boll_dev * std
# Calculate new target position
leg1_pos = self.get_pos(self.leg1_symbol)
if not leg1_pos:
if self.current_spread >= self.boll_up:
self.targets[self.leg1_symbol] = -1
self.targets[self.leg2_symbol] = 1
elif self.current_spread <= self.boll_down:
self.targets[self.leg1_symbol] = 1
self.targets[self.leg2_symbol] = -1
elif leg1_pos > 0:
if self.current_spread >= self.boll_mid:
self.targets[self.leg1_symbol] = 0
self.targets[self.leg2_symbol] = 0
else:
if self.current_spread <= self.boll_mid:
self.targets[self.leg1_symbol] = 0
self.targets[self.leg2_symbol] = 0
# Execute orders
for vt_symbol in self.vt_symbols:
target_pos = self.targets[vt_symbol]
current_pos = self.get_pos(vt_symbol)
pos_diff = target_pos - current_pos
volume = abs(pos_diff)
bar = bars[vt_symbol]
if pos_diff > 0:
price = bar.close_price + self.price_add
if current_pos < 0:
self.cover(vt_symbol, price, volume)
else:
self.buy(vt_symbol, price, volume)
elif pos_diff < 0:
price = bar.close_price - self.price_add
if current_pos > 0:
self.sell(vt_symbol, price, volume)
else:
self.short(vt_symbol, price, volume)
self.put_event()
```Shown in full with attribution under the source's licence. Licence: MIT
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.