Log-Price Spread Thresholds for Two-Instrument Statistical Arbitrage
Summary
This statistical-arbitrage demo tracks closing prices for two instruments, takes their logarithms, and uses the difference as a spread signal. When the spread crosses configured positive or negative thresholds, it opens opposing legs: short the first instrument and long the second for one direction, and the reverse for the other. Larger moves trigger closure, while a return toward the center is intended to close open positions. The example processes bar data and checks positions on ticks, with a routine to respond when the number of open positions suggests one leg may be missing.
The code is an illustrative example, not a validated strategy: it reports no historical performance, costs, hedge-ratio estimation, or evidence that the selected instruments remain cointegrated. Threshold units and conditions merit careful review; the configured values are small fractions while one exit condition compares the absolute spread directly with the threshold. The bar handler also names specific futures contracts, limiting portability. Real implementation would need robust synchronization, order-state handling, and safeguards against unhedged exposure.
Key ideas
- The demo uses the difference between two log prices as its spread measure.
- It opens opposite positions in the instruments when the spread exceeds directional thresholds.
- It attempts to close positions after larger moves or when the spread returns toward the center.
- A position check tries to address cases where only one leg appears open.
- The example lacks performance evidence and requires review of threshold logic, instrument selection, and order handling.
Tags
Full text
# StatArb
# StatArb
statistics arbitrage demo
## Source (Apache-2.0)
```python
#!/usr/bin/env python
# encoding: utf-8
import logging
import time
import numpy as np
from collections import deque
from gmsdk import *
from math import log
eps = 1e-6
class StatArb(StrategyBase):
'''
statistics arbitrage demo
'''
def __init__(self, *args, **kwargs):
super(StatArb, self).__init__(*args, **kwargs)
logging.basicConfig(format='%(asctime)s - %(levelname)s: %(message)s')
self.tick_size = self.config.getfloat('ss', 'tick_size') or 0.2
self.threshold = self.config.getfloat('ss', 'sigma') or 2.34
self.significant_diff = self.threshold * 0.0015 ## 3/4 sigma
self.stop_lose_threshold = self.threshold * 0.002 ## 2 * sigma
self.trade_exchange_a = self.config.get('ss', 'trade_exchange_a') or 'CFFEX'
self.trade_secid_a = self.config.get('ss', 'trade_secid_a')
self.trade_unit_a = self.config.get('ss', 'trade_unit_a') or 1
self.last_price_a = 0.0
self.trade_exchange_b = self.config.get('ss', 'trade_exchange_b') or 'CFFEX'
self.trade_secid_b = self.config.get('ss', 'trade_secid_b')
self.trade_unit_b = self.config.get('ss', 'trade_unit_b') or 1
self.last_price_b = 0.0
self.pos_side_up = False
self.pos_side_down = False
self.window_size = 20
self.close_buffer_symbol_a = deque(maxlen=self.window_size)
self.close_buffer_symbol_b = deque(maxlen=self.window_size)
self.at_risk = 0
self.bar_type = self.config.get('ss', 'bar_type')
def on_tick(self, tick):
if tick.sec_id == self.trade_secid_a:
self.last_price_a = tick.last_price
elif tick.sec_id == self.trade_secid_b:
self.last_price_b = tick.last_price
self.check_position()
def on_bar(self, bar):
if bar.bar_type == 15:
#print (bar.sec_id == 'IF1704')
if bar.sec_id == 'IF1703':
a = 1
#print ('bar')
self.close_buffer_symbol_a.append(bar.close)
elif bar.sec_id == 'IF1704': #self.trade_secid_b:
b = 1
a = 1
#print (bar.close)
#print (bar.sec_id == self.trade_secid_a)
self.close_buffer_symbol_b.append(bar.close)
if a == 1 and b == 1:
self.algo_action()
#print ('action')
def open_side_up(self):
self.open_short(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
self.open_long(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
self.pos_side_up = True
def close_side_up(self):
self.close_short(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
self.close_long(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
self.pos_side_up = False
def open_side_down(self):
self.open_long(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
self.open_short(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
self.pos_side_down = True
def close_side_down(self):
self.close_long(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
self.close_short(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
self.pos_side_down = False
def algo_action(self):
# type: () -> object
latest_a = self.close_buffer_symbol_a.pop()
lna = log(latest_a)
latest_b = self.close_buffer_symbol_b.pop()
lnb = log(latest_b)
diff = lna - lnb
#print (diff)
#print(self.stop_lose_threshold)
if diff > self.stop_lose_threshold:
self.close_side_up()
#print ('a')
elif diff > self.significant_diff and diff < self.stop_lose_threshold:
self.open_side_up()
#print ('b')
elif diff < - self.stop_lose_threshold:
self.close_side_down()
#print ('c')
elif diff < - self.significant_diff and diff > - self.stop_lose_threshold:
self.open_side_down()
#print ('d')
elif abs(diff) < self.threshold:
if self.pos_side_up:
self.close_side_up()
if self.pos_side_down:
self.close_side_down()
def check_position(self):
""" TODO: check if one leg position and close it """
ps = self.get_positions()
count = len(ps)
if count % 2 != 0:
self.at_risk += 1
## if more than 4 tick data passed, need to force quit
if self.at_risk < 4:
return
for p in ps:
if self.pos_side_up:
if p.side == OrderSide_Ask:
self.close_short(p.exchange, p.sec_id, self.last_price_a, p.volume)
elif p.side == OrderSide_Bid:
self.close_long(p.exchange, p.sec_id, self.last_price_b, p.volume)
if self.pos_side_down:
if p.side == OrderSide_Ask:
self.close_short(p.exchange, p.sec_id, self.last_price_b, p.volume)
elif p.side == OrderSide_Bid:
self.close_long(p.exchange, p.sec_id, self.last_price_a, p.volume)
else:
self.at_risk = 0
if __name__ == '__main__':
#import pdb; pdb.set_trace()
dm = StatArb(config_file='strategy_sa.ini')
ret = dm.run()
print("Statistics Arbitrage: ", dm.get_strerror(ret))
```Shown in full with attribution under the source's licence. Licence: Apache-2.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.