Overall Statistics |
Total Trades 880 Average Win 0.20% Average Loss -0.09% Compounding Annual Return 2.330% Drawdown 3.900% Expectancy 1.046 Net Profit 54.650% Sharpe Ratio 0.843 Probabilistic Sharpe Ratio 19.899% Loss Rate 38% Win Rate 62% Profit-Loss Ratio 2.28 Alpha 0.009 Beta 0.265 Annual Standard Deviation 0.019 Annual Variance 0 Information Ratio -0.293 Tracking Error 0.033 Treynor Ratio 0.062 Total Fees $2195.45 Estimated Strategy Capacity $250000.00 Lowest Capacity Asset FLTR UW2H9E1WXILH |
import math import sys import numpy as np import statistics from time import sleep ''' Institutional Bond ''' class DeterminedLightBrownAlpaca(QCAlgorithm): #List<SymbolData> SymbolData = new List<SymbolData>(); def Initialize(self): # LIVE TRADING if self.LiveMode: self.Debug("Trading Live!") self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin) # Group Trading # Use a default FA Account Group with an Allocation Method self.DefaultOrderProperties = InteractiveBrokersOrderProperties() # account group created manually in IB/TWS self.DefaultOrderProperties.FaGroup = "Institutional Bond" # supported allocation methods are: EqualQuantity, NetLiq, AvailableEquity, PctChange self.DefaultOrderProperties.FaMethod = "AvailableEquity" # set a default FA Allocation Profile # Alex: I commented the following line out, since it would "reset" the previous settings #self.DefaultOrderProperties = InteractiveBrokersOrderProperties() # allocation profile created manually in IB/TWS # self.DefaultOrderProperties.FaProfile = "TestProfileP" #Algo Start self.SetCash(100000) self.SetStartDate(2003, 1, 1) #self.SetEndDate(2007, 1, 1) self.SetBenchmark("AGG") self.SymbolData = [] self.Keeps = int(self.GetParameter("Keeps")) self.CashReservePct = float(self.GetParameter("CashReservePct")) self.WeightRa = float(self.GetParameter("WeightRa")) self.WeightRb = float(self.GetParameter("WeightRb")) self.WeightV = float(self.GetParameter("WeightV")) self.FilterSMA = int(self.GetParameter("FirstSMA")) self.LookbackRa = self.GetParameter("LookbackRa") self.LookbackRb = self.GetParameter("LookbackRb") self.LookbackV = self.GetParameter("LookbackV") self.Url = "" self.RebalancePeriod = self.GetParameter("RebalancePeriod") self.GrowthSymbols = [] self.SafetySymbol = self.AddEquity("SHY", Resolution.Daily).Symbol # List<string> GrowthSymbols = new List<String>{ # // safety symbol # "SHY" # }; self.first = True self.LastRotationTime = self.Time #self.LastRotationTime = DateTime.Now self.SetGrowthSymbols() self.Debug("WeightRa: " + str(self.WeightRa)) self.Debug("WeightRb: " + str(self.WeightRb)) self.Debug("WeightV: " + str(self.WeightV)) self.Debug("SymbolUrl: " + str(self.Url)) self.Debug("RebalancePeriod: " + str(self.RebalancePeriod)) periodADays = self.DaysFromLookback(self.LookbackRa) periodBDays = self.DaysFromLookback(self.LookbackRb) periodVolDays = self.DaysFromLookback(self.LookbackV) #self.Debug("LookbackRa: {0} ({1} days)", str(self.LookbackRa), str(periodADays)) #self.Debug("LookbackRb: {0} ({1} days)", str(self.LookbackRb), str(periodBDays)) #self.Debug("LookbackV: {0} ({1} days)", str(self.LookbackV), str(periodVolDays)) for symbol in self.GrowthSymbols: self.Debug("adding symbol to universe: " + symbol); symbolObject = self.AddSecurity(SecurityType.Equity, symbol, Resolution.Daily).Symbol # TODO - add parameters for lookback periods periodAPerf = self.MOMP(symbol, periodADays, Resolution.Daily) periodBPerf = self.MOMP(symbol, periodBDays, Resolution.Daily) #// (252/12) * x months ''' FIXME ''' #// FIXME -The moving average is calculated using the closing price from the last day of each month(e.g. a 10 month moving average has 10 datapoints) #filterSMA = self.SMA(symbol, 21*self.FilterSMA, Resolution.Daily) ''' Update this indicator every month ''' #filterSMA = SimpleMovingAverage(21*self.FilterSMA) filterSMA = SimpleMovingAverage(self.FilterSMA) consolidator = TradeBarConsolidator(CalendarType.Monthly) self.RegisterIndicator(symbolObject, filterSMA, consolidator) #// we use log returns instead of definition in replay std = StandardDeviation(periodVolDays) annualizationFactor = float(math.sqrt(252)) logReturns = self.LOGR(symbol, periodVolDays, Resolution.Daily) #CompositeIndicator<IndicatorDataPoint> volatility = std.Of(logReturns).Times(annualizationFactor); volatility_ = IndicatorExtensions.Of(std, logReturns) volatility = IndicatorExtensions.Times(volatility_, annualizationFactor) #sleep(1) sData = SymbolData(symbol, periodAPerf, periodBPerf, volatility, filterSMA) self.SymbolData.append(sData) # SymbolData.Add(new SymbolData # { # Symbol = symbol, # ReturnA = periodAPerf, # ReturnB = periodBPerf, # Volatility = volatility, # FilterSMA = filterSMA # }); #// assumption we don't look back more than a year self.SetWarmup(timedelta(days=365)) self.lastDay = -1 self.newWeek = True self.newMonth = True self.counter = 2 self.newBiWeek = True self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol self.Schedule.On(self.DateRules.MonthStart("SPY"), self.TimeRules.At(0, 0), self.TrackMonth) self.Schedule.On(self.DateRules.WeekStart("SPY"), self.TimeRules.At(0, 0), self.TrackWeek) self.Schedule.On(self.DateRules.WeekStart("SPY"), self.TimeRules.At(0, 0), self.TrackBiWeekly) def TrackMonth(self): self.newMonth = True def TrackWeek(self): self.newWeek = True def TrackBiWeekly(self): if self.counter == 2: self.newBiWeek = True self.counter = 1 else: self.counter += 1 def DaysFromLookback(self, period): strAmount = period[0:len(period)-1] #strAmount = period.substr(0, len(period)-1) amount = np.int32(strAmount) #amount = int32.Parse(strAmount) multiplier = period.endswith("M") if multiplier == False: multiplier = 1 else: multiplier = 21 return multiplier * amount def SetGrowthSymbols(self): self.GrowthSymbols.append("BIL") self.GrowthSymbols.append("FLOT") self.GrowthSymbols.append("FLTR") self.GrowthSymbols.append("GBF") self.GrowthSymbols.append("GOVT") self.GrowthSymbols.append("GSY") self.GrowthSymbols.append("GVI") self.GrowthSymbols.append("IEI") self.GrowthSymbols.append("LQD") self.GrowthSymbols.append("MBB") self.GrowthSymbols.append("MINT") self.GrowthSymbols.append("SHM") self.GrowthSymbols.append("SHY") self.GrowthSymbols.append("SPIP") self.GrowthSymbols.append("SPTI") self.GrowthSymbols.append("STIP") self.GrowthSymbols.append("STPZ") self.GrowthSymbols.append("USIG") self.GrowthSymbols.append("VCSH") def GetEtfReplayRankings(self): #orderedReturnA = self.SymbolData.OrderByDescending(x.ReturnA) orderedReturnA = sorted(self.SymbolData, key = lambda x: x.ReturnA.Current.Value, reverse = True) #orderedReturnB = self.SymbolData.OrderByDescending(x.ReturnB) orderedReturnB = sorted(self.SymbolData, key = lambda x: x.ReturnB.Current.Value, reverse = True) #orderedVol = self.SymbolData.OrderBy(x.Volatility) orderedVol = sorted(self.SymbolData, key = lambda x: x.Volatility.Current.Value, reverse = False) rankings = {} #// add all the symbols for sd in self.SymbolData: rankings[sd.Symbol] = 0.0 for i in range(0, len(orderedReturnA), 1): current = orderedReturnA[i] rankings[current.Symbol] += i * self.WeightRa for i in range(0, len(orderedReturnB), 1): current = orderedReturnB[i] rankings[current.Symbol] += i * self.WeightRb for i in range(0, len(orderedVol), 1): current = orderedVol[i] rankings[current.Symbol] += i * self.WeightV return rankings def OnData(self, data): if self.IsWarmingUp: return self.Log("OnData slice: date=" + str(self.Time))# + ", cnt=" + data.Count); try: rebalance = False if self.first: rebalance = True self.first = False else: #// var delta = Time.Subtract(LastRotationTime); #// rebalance = delta > RotationInterval; rebalance = self.LastRotationTime.month != self.Time.month if (self.RebalancePeriod == "Quarterly"): #// January - 1 #// April - 4 #// July - 7 #// Oct - 10 rebalance = (rebalance and ((self.Time.month == 1) or (self.Time.month == 4) or (self.Time.month == 7) or (self.Time.month == 10))) if rebalance == True: self.Debug("Rebalancing Quarterly on " + str(self.Time)) elif (self.RebalancePeriod == "Daily"): if self.Time.day != self.lastDay: rebalance = True self.lastDay = self.Time.day self.Debug("Rebalancing Daily on " + str(self.Time)) else: rebalance = False elif (self.RebalancePeriod == "Weekly"): if self.newWeek == True: rebalance = True self.newWeek = False self.Debug("Rebalancing Weekly on " + str(self.Time)) else: rebalance = False elif (self.RebalancePeriod == "Bi-Monthly"): if self.newBiWeek == True: rebalance = True self.newBiWeek = False self.Debug("Rebalancing Bi Weekly on " + str(self.Time)) else: rebalance = False elif(self.RebalancePeriod == "Monthly"): if self.newMonth == True: rebalance = True self.newMonth = False self.Debug("Rebalancing Monthly on " + str(self.Time)) else: rebalance = False if rebalance: self.LastRotationTime = self.Time #// pick which one is best from growth and safety symbols #//var orderedObjScores = SymbolData.OrderByDescending(x => x.ObjectiveScore).ToList(); rankings = self.GetEtfReplayRankings() #orderedObjScores = self.SymbolData.OrderBy(rankings[x.Symbol]).ToList() orderedObjScores = sorted(self.SymbolData, key = lambda x: rankings[x.Symbol], reverse = False) nextHoldings = [] self.Debug("OrderedObjScores Count: " + str(len(orderedObjScores))) for i in range(0, self.Keeps, 1): currentSymbolData = orderedObjScores[i] lastClose = self.Securities[currentSymbolData.Symbol].Close maFilter = lastClose > currentSymbolData.FilterSMA.Current.Value if maFilter: #// this meets our investment criteria if self.Portfolio[self.SafetySymbol].Invested: self.Liquidate(self.SafetySymbol) nextHoldings.append(currentSymbolData.Symbol) else: if self.SafetySymbol not in nextHoldings: nextHoldings.append(self.SafetySymbol) #// FIXME - allocate to "Safety" symbol, right now saftey symbol is part of the growth symbols so it should bubble up but that doesn't match replay #else: # if not self.Portfolio[self.SafetySymbol].Invested: # self.Liquidate() # self.SetHoldings(self.SafetySymbol, 1) # break # if self.Portfolio[self.SafetySymbol].Invested: # break if self.Portfolio[self.SafetySymbol].Invested: return self.Log(">>NextPositions<<") for position in nextHoldings: self.Log("\t>>" + str(position)) if len(nextHoldings) == 0: #// goto 100% cash self.Log("LIQUIDATE>>CASH") self.Liquidate() else: for kvp in self.Portfolio.Securities.Values: #// liquidate anything we are currently invested in but not part of our next portfolio state if kvp.Invested and not (kvp.Symbol in nextHoldings): self.Log("LIQUIDATE>>" + str(kvp.Symbol)) self.Liquidate(kvp.Symbol) allocationPercentage = (1.0 - self.CashReservePct) / self.Keeps for symbol in nextHoldings: self.Log("BUY>>" + str(symbol)) self.SetHoldings([PortfolioTarget(symbol, allocationPercentage)]) except: print("Unexpected error:", sys.exc_info()[0]) raise #Error("OnData: " + ex.Message + "\r\n\r\n" + ex.StackTrace); class SymbolData: def __init__(self, symbol, ReturnA, ReturnB, Volatility, FilterSMA): self.Symbol = symbol self.ReturnA = ReturnA self.ReturnB = ReturnB self.Volatility = Volatility self.FilterSMA = FilterSMA