| Overall Statistics |
|
Total Orders 418 Average Win 0.37% Average Loss -0.38% Compounding Annual Return 0.389% Drawdown 12.400% Expectancy 0.034 Start Equity 100000 End Equity 101959.31 Net Profit 1.959% Sharpe Ratio -0.788 Sortino Ratio -0.789 Probabilistic Sharpe Ratio 0.644% Loss Rate 48% Win Rate 52% Profit-Loss Ratio 0.97 Alpha -0.033 Beta -0.015 Annual Standard Deviation 0.043 Annual Variance 0.002 Information Ratio -0.727 Tracking Error 0.15 Treynor Ratio 2.192 Total Fees $915.03 Estimated Strategy Capacity $15000000.00 Lowest Capacity Asset XLK RGRPZX100F39 Portfolio Turnover 3.50% Drawdown Recovery 233 |
#region imports
from AlgorithmImports import *
from scipy import stats
from math import floor
from datetime import timedelta
from collections import deque
import itertools as it
from decimal import Decimal
#endregion
# https://quantpedia.com/Screener/Details/12
class PairsTradingAlgorithm(QCAlgorithm):
def initialize(self):
self.set_start_date(self.end_date - timedelta(5*365))
self.set_cash(100000)
# Define some parameters.
self._threshold = 2
formation_period = 252
# Add a set of Equities to trade as pairs.
tickers = [
'XLK', 'QQQ', 'BANC', 'BBVA', 'BBD', 'BCH', 'BLX', 'BSBR', 'BSAC', 'SAN',
'CIB', 'BXS', 'BAC', 'BOH', 'BMO', 'BK', 'BNS', 'BKU', 'BBT','NBHC', 'OFG',
'BFR', 'CM', 'COF', 'C', 'VLY', 'WFC', 'WAL', 'WBK','RBS', 'SHG', 'STT', 'STL',
'SMFG', 'STI'
]
for ticker in tickers:
equity = self.add_equity(ticker, Resolution.DAILY)
equity.session.size = formation_period
# Create all possible pairs of the stocks.
self._pairs = list(it.combinations(self.securities.values(), 2))
self._sorted_pairs = []
# Add a Scheduled Event to sort the pairs by their similarities.
self.schedule.on(self.date_rules.month_start("SPY"), self.time_rules.midnight, self._rank_pairs)
self.set_warm_up(2*formation_period)
def on_data(self, data):
if self.is_warming_up or not data.bars:
return
for equity_a, equity_b in self._sorted_pairs:
history_a = self._get_history(equity_a)
history_b = self._get_history(equity_b)
# Calculate the spread of the two price series.
spread = np.array(history_a) - np.array(history_b)
mean = np.mean(spread)
std = np.std(spread)
ratio = equity_a.price / equity_b.price
# When the pair prices have diverged, open a long-short position.
quantity = int(self.calculate_order_quantity(equity_a, 0.1))
if spread[-1] > mean + self._threshold * std:
if not equity_a.invested and not equity_b.invested:
self.sell(equity_a, quantity)
self.buy(equity_b, floor(ratio*quantity))
elif spread[-1] < mean - self._threshold * std:
if not equity_a.invested and not equity_b.invested:
self.sell(equity_b, quantity)
self.buy(equity_a, floor(ratio*quantity))
# When the prices revert back, close the position.
elif equity_a.invested and equity_b.invested:
self.liquidate(equity_a)
self.liquidate(equity_b)
def _rank_pairs(self):
# Every 6 months, update the pairs we trade.
if self.time.month in [1, 7]:
previous_pairs = self._sorted_pairs
distances = {}
for equity_a, equity_b in self._pairs:
# Get the trailing prices of each Equity.
if not (equity_a.session.is_ready and equity_b.session.is_ready):
continue
history_a = self._get_history(equity_a)
history_b = self._get_history(equity_b)
# Calculate the sum of squared deviations between the two
# normalized price series.
norm_a = np.array(history_a)/history_a[0]
norm_b = np.array(history_b)/history_b[0]
distances[(equity_a, equity_b)] = sum((norm_a - norm_b)**2)
# Select the 4 pairs with the smallest distance to be the pairs we trade.
self._sorted_pairs = sorted(distances, key=lambda x: distances[x])[:4]
# Liquidate the previous pairs that didn't make it into `self._sorted_pairs`
for pair in previous_pairs:
if pair not in self._sorted_pairs:
for equity in pair:
self.liquidate(equity)
def _get_history(self, equity):
return [bar.close for bar in equity.session][::-1]