| Overall Statistics |
|
Total Trades 1830 Average Win 0.17% Average Loss -0.18% Compounding Annual Return 3.234% Drawdown 12.100% Expectancy 0.320 Net Profit 80.130% Sharpe Ratio 0.622 Probabilistic Sharpe Ratio 3.579% Loss Rate 32% Win Rate 68% Profit-Loss Ratio 0.93 Alpha 0.026 Beta 0.018 Annual Standard Deviation 0.044 Annual Variance 0.002 Information Ratio -0.451 Tracking Error 0.173 Treynor Ratio 1.547 Total Fees $30179.74 Estimated Strategy Capacity $9300000.00 Lowest Capacity Asset CLY UIB7ED20G0V9 |
import datetime
import math
import sys
import numpy as np
import statistics
from time import sleep
class DeterminedLightBrownAlpaca(QCAlgorithm):
def Initialize(self):
self.SetCash(1000000)
self.SetStartDate(2003, 1, 1)
#self.SetEndDate(2007, 1, 1)
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 = ["AGZ","BIL","HYG","IEF","IEI","IGIB","IGLB","JNK","LQD","MBB","MINT","SHY","TBF","TBX","TIP","TLT"]
self.SafetySymbol = self.AddEquity("SHY", Resolution.Daily).Symbol
# List<string> GrowthSymbols = new List<String>{
# // safety symbol
# "SHY"
# };
self.rebalance = True
self.LastRotationTime = self.Time
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)
for symbol in self.GrowthSymbols:
self.Debug("adding symbol to universe: " + symbol);
symbolObject = self.AddSecurity(SecurityType.Equity, symbol, Resolution.Daily).Symbol
sData = SymbolData(self,symbolObject, periodADays, periodBDays, periodVolDays, self.FilterSMA)
self.SymbolData.append(sData)
#// 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 not multiplier :
multiplier = 1
else:
multiplier = 21
return multiplier * amount
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);
if not self.rebalance:
#// var delta = Time.Subtract(LastRotationTime);
#// rebalance = delta > RotationInterval;
self.rebalance = self.LastRotationTime.month != self.Time.month
if (self.RebalancePeriod == "Quarterly"):
#// January - 1
#// April - 4
#// July - 7
#// Oct - 10
self.rebalance = (self.rebalance and
((self.Time.month == 1) or
(self.Time.month == 4) or
(self.Time.month == 7) or
(self.Time.month == 10)))
if self.rebalance:
self.Debug("Rebalancing Quarterly on " + str(self.Time))
elif (self.RebalancePeriod == "Daily"):
if self.Time.day != self.lastDay:
self.rebalance = True
self.lastDay = self.Time.day
self.Debug("Rebalancing Daily on " + str(self.Time))
else:
self.rebalance = False
elif (self.RebalancePeriod == "Weekly"):
if self.newWeek:
self.rebalance = True
self.newWeek = False
self.Debug("Rebalancing Weekly on " + str(self.Time))
else:
self.rebalance = False
elif (self.RebalancePeriod == "Bi-Monthly"):
if self.newBiWeek:
self.rebalance = True
self.newBiWeek = False
self.Debug("Rebalancing Bi Weekly on " + str(self.Time))
else:
self.rebalance = False
elif(self.RebalancePeriod == "Monthly"):
if self.newMonth:
self.rebalance = True
self.newMonth = False
self.Debug("Rebalancing Monthly on " + str(self.Time))
else:
self.rebalance = False
if self.rebalance:
self.rebalance = False
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]
#if currentSymbolData.Symbol not in data.Bars:continue
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)])
class SymbolData:
def __init__(self,algo, symbol, periodADays, periodBDays, periodVolDays, FilterSMA):
self.Symbol = symbol
self.algo = algo
self.ReturnA = self.algo.MOMP(symbol, periodADays, Resolution.Daily)
self.ReturnB = self.algo.MOMP(symbol, periodBDays, Resolution.Daily)
#self.Volatility = Volatility
self.FilterSMA = SimpleMovingAverage(FilterSMA)
consolidator = TradeBarConsolidator(CalendarType.Monthly)
self.algo.RegisterIndicator(self.Symbol, self.FilterSMA, consolidator)
std = StandardDeviation(periodVolDays)
annualizationFactor = float(math.sqrt(252))
logReturns = self.algo.LOGR(symbol, periodVolDays, Resolution.Daily)
#CompositeIndicator<IndicatorDataPoint> volatility = std.Of(logReturns).Times(annualizationFactor);
volatility_ = IndicatorExtensions.Of(std, logReturns)
self.Volatility = IndicatorExtensions.Times(volatility_, annualizationFactor)