from Alphas.BasePairsTradingAlphaModel import BasePairsTradingAlphaModel
from datetime import timedelta
from scipy.stats import pearsonr
import numpy as np
import pandas as pd
class PairsTradingAlpha(BasePairsTradingAlphaModel):
''' 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")
for security in changes.AddedSecurities:
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.symbolsfrom Execution.ImmediateExecutionModel import ImmediateExecutionModel
from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel
from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity
from SmallCapGrowthStocks import SmallCapGrowthStocks
from PairsTradingAlpha import PairsTradingAlpha
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.AddEquity("SPY", Resolution.Minute)
self.AddAlpha(PairsTradingAlpha(3, Resolution.Daily))
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)
