Overall Statistics Total Trades62Average Win0.47%Average Loss-0.54%Compounding Annual Return-35.142%Drawdown7.600%Expectancy-0.138Net Profit-2.460%Sharpe Ratio-2.464Probabilistic Sharpe Ratio18.456%Loss Rate54%Win Rate46%Profit-Loss Ratio0.88Alpha-0.261Beta-0.334Annual Standard Deviation0.131Annual Variance0.017Information Ratio-3.242Tracking Error0.156Treynor Ratio0.964Total Fees\$684.05
from Alphas.BasePairsTradingAlphaModel import BasePairsTradingAlphaModel
from datetime import timedelta
from scipy.stats import pearsonr
import numpy as np
import pandas as pd

''' This alpha model is designed to rank every pair combination by its pearson correlation
and trade the pair with the hightest correlation
This model generates alternating long ratio/short ratio insights emitted as a group'''

def __init__(self, lookback = 15,
resolution = Resolution.Minute,
threshold = 1,
minimumCorrelation = .5):
'''Initializes a new instance of the PearsonCorrelationPairsTradingAlphaModel class
Args:
lookback: lookback period of the analysis
resolution: analysis resolution
threshold: The percent [0, 100] deviation of the ratio from the mean before emitting an insight
minimumCorrelation: The minimum correlation to consider a tradable pair'''
super().__init__(lookback, resolution, threshold)
self.lookback = lookback
self.resolution = resolution
self.minimumCorrelation = minimumCorrelation
self.best_pair = ()

def Update(self, algorithm, data):
print("update!")
return super().Update(algorithm, data)

def OnSecuritiesChanged(self, algorithm, changes):
'''Event fired each time the we add/remove securities from the data feed.
Args:
algorithm: The algorithm instance that experienced the change in securities
changes: The security additions and removals from the algorithm'''

print("OnSecuritiesChanged")
self.Securities.append(security)

for security in changes.RemovedSecurities:
if security in self.Securities:
self.Securities.remove(security)
print("*** testing !!! ****")
symbols = [ x.Symbol for x in self.Securities ]

history = algorithm.History(symbols, self.lookback, self.resolution).close.unstack(level=0)

if not history.empty:

df = self.get_price_dataframe(history)
stop = len(df.columns)

corr = dict()

for i in range(0, stop):
for j in range(i+1, stop):
if (j, i) not in corr:
corr[(i, j)] = pearsonr(df.iloc[:,i], df.iloc[:,j])[0]

corr = sorted(corr.items(), key = lambda kv: kv[1])
if corr[-1][1] >= self.minimumCorrelation:
self.best_pair = (symbols[corr[-1][0][0]], symbols[corr[-1][0][1]])

super().OnSecuritiesChanged(algorithm, changes)

def HasPassedTest(self, algorithm, asset1, asset2):
'''Check whether the assets pass a pairs trading test
Args:
algorithm: The algorithm instance that experienced the change in securities
asset1: The first asset's symbol in the pair
asset2: The second asset's symbol in the pair
Returns:
True if the statistical test for the pair is successful'''
return self.best_pair is not None and self.best_pair == (asset1, asset2)

def get_price_dataframe(self, df):
timezones = { x.Symbol.Value: x.Exchange.TimeZone for x in self.Securities }

# Use log prices
df = np.log(df)

is_single_timeZone = len(set(timezones.values())) == 1

if not is_single_timeZone:
series_dict = dict()

for column in df:
# Change the dataframe index from data time to UTC time
to_utc = lambda x: Extensions.ConvertToUtc(x, timezones[column])
if self.resolution == Resolution.Daily:
to_utc = lambda x: Extensions.ConvertToUtc(x, timezones[column]).date()

data = df[[column]]
data.index = data.index.map(to_utc)
series_dict[column] = data[column]

df = pd.DataFrame(series_dict).dropna()

return (df - df.shift(1)).dropna()
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel

class SmallCapGrowthStocks(FundamentalUniverseSelectionModel):
'''
This module selects the most liquid stocks listed on the Nasdaq Stock Exchange.
'''

def __init__(self, filterFineData = True, universeSettings = None, securityInitializer = None):
'''Initializes a new default instance of the TechnologyUniverseModule'''
super().__init__(filterFineData, universeSettings, securityInitializer)
self.numberOfSymbolsCoarse = 1000
self.numberOfSymbolsFine = 100
self.dollarVolumeBySymbol = {}
self.symbols = []
self.lastMonth = -1

def SelectCoarse(self, algorithm, coarse):
'''
Performs a coarse selection:

-The stock must have fundamental data
-The stock must have positive previous-day close price
-The stock must have positive volume on the previous trading day
'''
if algorithm.Time.month == self.lastMonth:
return self.symbols

filtered = [x for x in coarse if x.HasFundamentalData and x.Volume > 0 and x.Price > 0]
sortedByDollarVolume = sorted(filtered, key = lambda x: x.DollarVolume, reverse=True)[:self.numberOfSymbolsCoarse]

self.symbols.clear()
self.dollarVolumeBySymbol.clear()
for x in sortedByDollarVolume:
self.symbols.append(x.Symbol)
self.dollarVolumeBySymbol[x.Symbol] = x.DollarVolume

return self.symbols

def SelectFine(self, algorithm, fine):
'''
Performs a fine selection for companies in the Morningstar Banking Sector
'''
if algorithm.Time.month == self.lastMonth:
return self.symbols
self.lastMonth = algorithm.Time.month

# Filter stocks
filteredFine = [x for x in fine if x.AssetClassification.StyleBox == StyleBox.SmallGrowth]

sortedByDollarVolume = []

# Sort stocks on dollar volume
sortedByDollarVolume = sorted(filteredFine, key = lambda x: self.dollarVolumeBySymbol[x.Symbol], reverse=True)

self.symbols = [x.Symbol for x in sortedByDollarVolume[:self.numberOfSymbolsFine]]

return self.symbols
from Execution.ImmediateExecutionModel import ImmediateExecutionModel
from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel
from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity
from SmallCapGrowthStocks import SmallCapGrowthStocks

class QuantumTransdimensionalProcessor(QCAlgorithm):

def Initialize(self):
self.SetStartDate(2019, 6, 29)  # Set Start Date
self.SetEndDate(2019,7,20)
self.SetCash(100000)  # Set Strategy Cash

self.SetExecution(ImmediateExecutionModel())

self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())

self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.01))

self.SetUniverseSelection(SmallCapGrowthStocks())

def OnData(self, data):
'''OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.
Arguments:
data: Slice object keyed by symbol containing the stock data
'''

# if not self.Portfolio.Invested:
#    self.SetHoldings("SPY", 1)