Overall Statistics Total Trades 7777 Average Win 0.45% Average Loss -0.47% Compounding Annual Return 5.930% Drawdown 30.500% Expectancy 0.062 Net Profit 171.508% Sharpe Ratio 0.517 Probabilistic Sharpe Ratio 0.608% Loss Rate 46% Win Rate 54% Profit-Loss Ratio 0.95 Alpha 0.04 Beta 0.06 Annual Standard Deviation 0.086 Annual Variance 0.007 Information Ratio -0.193 Tracking Error 0.172 Treynor Ratio 0.741 Total Fees $15068.15 Estimated Strategy Capacity$29000000.00 Lowest Capacity Asset LYB UQRGJ93635GL
# https://quantpedia.com/strategies/pairs-trading-with-stocks/
#
# The investment universe consists of stocks from NYSE, AMEX, and NASDAQ, while illiquid stocks are removed from the investment universe. Cumulative
# total return index is then created for each stock (dividends included), and the starting price during the formation period is set to $1 (price normalization). # Pairs are formed over twelve months (formation period) and are then traded in the next six-month period (trading period). The matching partner for each stock # is found by looking for the security that minimizes the sum of squared deviations between two normalized price series. Top 20 pairs with the smallest historical # distance measure are then traded, and a long-short position is opened when pair prices have diverged by two standard deviations, and the position is closed # when prices revert. # # QC implementation changes: # - Universe consists of top 500 most liquid US stocks with price > 5$.
#   - Maximum number of pairs traded at one time is set to 5.

import numpy as np
import itertools as it

def Initialize(self):
self.SetStartDate(2005, 1, 1)
self.SetCash(100000)

# Daily price data.
self.history_price = {}
self.period = 12 * 21

# Equally weighted brackets.

self.sorted_pairs = []

self.coarse_count = 500
self.month = 6
self.selection_flag = True
self.UniverseSettings.Resolution = Resolution.Daily

self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)

def OnSecuritiesChanged(self, changes):
security.SetFeeModel(CustomFeeModel(self))
security.SetLeverage(5)

for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.history_price:
del self.history_price[symbol]

symbols = [x for x in self.history_price.keys() if x != self.symbol]
self.symbol_pairs = list(it.combinations(symbols, 2))

# minimize the sum of squared deviations
distances = {}
for pair in self.symbol_pairs:
distances[pair] = self.Distance(self.history_price[pair], self.history_price[pair])

if len(distances) != 0:
self.sorted_pairs = [x for x in sorted(distances.items(), key = lambda x: x)[:20]]

self.Liquidate()

def CoarseSelectionFunction(self, coarse):
# Update the rolling window every day.
for stock in coarse:
symbol = stock.Symbol

if symbol in self.history_price:

if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False

selected = sorted([x for x in coarse if x.HasFundamentalData and x.Price > 5 and x.Market == 'usa'],
key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]

# Warmup price rolling windows.
for stock in selected:
symbol = stock.Symbol

if symbol in self.history_price:
continue

self.history_price[symbol] = RollingWindow[float](self.period)
history = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet")
continue
closes = history.loc[symbol].close
for time, close in closes.iteritems():

return [x.Symbol for x in selected if self.history_price[x.Symbol].IsReady]

def OnData(self, data):
if self.sorted_pairs is None: return

pairs_to_remove = []

for pair in self.sorted_pairs:
# Calculate the spread of two price series.
price_a = [x for x in self.history_price[pair]]
price_b = [x for x in self.history_price[pair]]
norm_a = np.array(price_a) / price_a[-1]
norm_b = np.array(price_b) / price_b[-1]

# Long-short position is opened when pair prices have diverged by two standard deviations.
if actual_spread > mean + 2*std or actual_spread < mean - 2*std:
# open new position for pair, if there's place for it.
symbol_a = pair
symbol_b = pair
a_price_norm = norm_a
b_price_norm = norm_b
a_price = price_a
b_price = price_b

# a stock's price > b stock's price
if a_price_norm > b_price_norm:
long_q = traded_portfolio_value / b_price    # long b stock
short_q = -traded_portfolio_value / a_price  # short a stock
if self.Securities.ContainsKey(symbol_a) and self.Securities.ContainsKey(symbol_b) and \
self.Securities[symbol_a].Price != 0 and self.Securities[symbol_a].IsTradable and \
self.MarketOrder(symbol_a, short_q)
self.MarketOrder(symbol_b, long_q)

# b stock's price > a stock's price
else:
if self.Securities.ContainsKey(symbol_a) and self.Securities.ContainsKey(symbol_b) and \
self.Securities[symbol_a].Price != 0 and self.Securities[symbol_a].IsTradable and \
self.MarketOrder(symbol_a, long_q)
self.MarketOrder(symbol_b, short_q)

# The position is closed when prices revert back.
else:
# make opposite order to opened position
pairs_to_remove.append(pair)

for pair in pairs_to_remove:

def Distance(self, price_a, price_b):
# Calculate the sum of squared deviations between two normalized price series.
price_a = [x for x in price_a]
price_b = [x for x in price_b]

norm_a = np.array(price_a) / price_a[-1]
norm_b = np.array(price_b) / price_b[-1]
return sum((norm_a - norm_b)**2)

def Selection(self):
if self.month == 6:
self.selection_flag = True

self.month += 1
if self.month > 12:
self.month = 1

# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))