Overall Statistics |
Total Trades 279 Average Win 2.83% Average Loss -0.79% Compounding Annual Return 129.904% Drawdown 21.100% Expectancy 1.571 Net Profit 509.943% Sharpe Ratio 2.966 Probabilistic Sharpe Ratio 98.738% Loss Rate 44% Win Rate 56% Profit-Loss Ratio 3.60 Alpha 0.686 Beta 0.137 Annual Standard Deviation 0.287 Annual Variance 0.082 Information Ratio -0.564 Tracking Error 0.613 Treynor Ratio 6.194 Total Fees $10303.77 Estimated Strategy Capacity $11000000.00 Lowest Capacity Asset BTCUSDT 18N |
#$@$@ ---------------------------------------------------------------- #$@$@ ------------- Author; Emmanuel Azubuike --------------- #$@$@ ---------------------------------------------------------------- class ScanWin(): def __init__(self, wT, wL): self.window = None self.winLength = wL if (wT is "int"):self.window = RollingWindow[int](wL) elif (wT is "bool"):self.window = RollingWindow[bool](wL) elif (wT is "float"):self.window = RollingWindow[float](wL) elif (wT is "TradeBar"):self.window = RollingWindow[TradeBar](wL) def isAbove(self, series): return (self.window[0] > series[0]) def isBelow(self, series): return (self.window[0] < series[0]) def Add(self,value): self.window.Add(value) def IsReady(self): return (self.window is not None) and \ (self.window.Count >= self.winLength) def __getitem__(self, index): return self.window[index]
#$@$@ ---------------------------------------------------------------- #$@$@ ------------- Author; Emmanuel Azubuike --------------- #$@$@ ---------------------------------------------------------------- from AlgorithmImports import * import operator import math from ScanWin import * class EMACrossoverUniverse(QCAlgorithm): #$@$@ ---------------------------------------------------------------- def Initialize(self): self.InitA() self.InitB() self.Initc() self.Initd() #$@$@ ---------------------------------------------------------------- def Initd(self): self.SetStartDate(2020, 1, 1) self.SetCash(100000) self.SetBenchmark(Symbol.Create("BTCUSDT", SecurityType.Crypto, Market.Binance)) def Initc(self): self.UniverseSettings.Resolution = Resolution.Daily self.symDataDict = { } self.UniverseTickers = ["SOLUSDT", "LUNAUSDT", "ADAUSDT", "BTCUSDT"] universeSymbols = [] for symbol in self.UniverseTickers: universeSymbols.append(Symbol.Create(symbol, SecurityType.Crypto, Market.Binance)) self.SetUniverseSelection(ManualUniverseSelectionModel(universeSymbols)) #$@$@ -------------------- def InitA(self): self.emaSlowPeriod = int(self.GetParameter('emaSlowPeriod')) self.emaFastPeriod = int(self.GetParameter('emaFastPeriod')) self.warmupPeriod = self.emaSlowPeriod self.maxExposurePct = float(self.GetParameter("maxExposurePct"))/100 self.maxHoldings = int(self.GetParameter("maxHoldings")) self.minAmountToBeInvested = int(self.GetParameter("minAmountToBeInvested")) #$@$@ -------------------- def InitB(self): self.symbol = "BTCUSDT" self.SetBrokerageModel(BrokerageName.Binance, AccountType.Cash) self.SetAccountCurrency("USDT") self.AddCrypto(self.symbol, Resolution.Daily) self.EnableAutomaticIndicatorWarmUp = True self.SetWarmUp(timedelta(self.warmupPeriod)) self.SelectedSymbolsAndWeights = {} #$@$@ ------------------------------------------------------------ @property def PAC(self): numHoldings = len([x.Key for x in self.Portfolio if self.IsInvested(x.Key)]) return numHoldings >= self.maxHoldings #$@$@ ------------------------------------------------ def OnData(self, dataSlice): #$@$@ loop through the symbols in the slice for symbol in dataSlice.Keys: if symbol in self.symDataDict: symbolData = self.symDataDict[symbol] symbolData.OnSymbolData(self.Securities[symbol].Price, dataSlice[symbol]) if self.IsInvested(symbol): symbolData.MOP() else: if(not self.PAC): if( symbolData.ESF() ): self.ONP(symbolData.symbol) symbolData.OnNewPositionOpened() #$@$@ ----------------------------------------------------- def ReHo(self, rebalanceCurrHoldings=False): for symbol in self.SelectedSymbolsAndWeights: symbolWeight = round(1/len(self.SelectedSymbolsAndWeights),4) self.SetWH(symbol,symbolWeight) sortedSymbolsAndWeights = {k: v for k, v in sorted(self.SelectedSymbolsAndWeights.items(), key=lambda item: item[1], reverse=True)} for symbol in sortedSymbolsAndWeights: self.SetSymbolHoldings(symbol) def SetSymbolHoldings(self, symbol): adjustedWeight = self.SelectedSymbolsAndWeights[symbol] cash = self.Portfolio.TotalPortfolioValue - self.Portfolio.TotalHoldingsValue percent = adjustedWeight * self.maxExposurePct cost = self.Portfolio.TotalPortfolioValue * percent orderMsg = f"{symbol} | alloc. ({round(adjustedWeight*100,2)}% adjusted) " if (cost > cash): percent = self.GetTruncatedValue(cash / self.Portfolio.TotalPortfolioValue, 3) if(self.Portfolio[symbol].Invested): orderMsg = f"[Re-Balancing] {orderMsg}" else: orderMsg = f"[NEW Addition] {orderMsg}" self.SetHoldings(symbol, percent, tag=orderMsg) #$@$@ --------–--------–--------–--------–--------–--------–--------–--------– def SetWH(self,symbol,symbolWeight): if( self.Portfolio.Invested ): TCB = sum( [x.Value.HoldingsCost for x in self.Portfolio if x.Value.Invested] ) else: TCB = 0.0 CA = self.Portfolio.TotalPortfolioValue - self.Portfolio.TotalHoldingsValue # CA = self.Portfolio.CashBook["USDT"].Amount WB = TCB + CA ATI = WB * symbolWeight if(self.Portfolio[symbol].Invested): profitPct = self.Portfolio[symbol].UnrealizedProfitPercent adjustedATI = ATI * (1 + profitPct) adjustedWeight = adjustedATI / self.Portfolio.TotalPortfolioValue else: adjustedWeight = ATI / self.Portfolio.TotalPortfolioValue symbolWeight = self.GetTruncatedValue(symbolWeight,3) adjustedWeight = self.GetTruncatedValue(adjustedWeight,3) self.SelectedSymbolsAndWeights[symbol] = adjustedWeight #$@$@ ------------------------------------------------- def ONP(self, symbol): self.SelectedSymbolsAndWeights[symbol] = 0 self.ReHo() #$@$@ ----------------------------------------------------- def Exp(self, symbol, exitMsg=""): profitPct = round(self.Securities[symbol].Holdings.UnrealizedProfitPercent,2) self.Liquidate(symbol, tag=f"SELL {symbol.Value} ({profitPct}% profit) [{exitMsg}]") self.SelectedSymbolsAndWeights.pop(symbol) return #$@$@ ------------------------------------------------------ def OnSecuritiesChanged(self, changes): for security in changes.AddedSecurities: symbol = security.Symbol if( symbol in self.UniverseTickers and \ symbol not in self.symDataDict.keys()): self.symDataDict[symbol] = SymbolData(symbol, self) def GetTruncatedValue(self, value, decPlaces): truncFactor = 10.0 ** decPlaces return math.trunc(value * truncFactor) / truncFactor def IsInvested(self, symbol): return self.Portfolio[symbol].Invested and self.Portfolio[symbol].Quantity * self.Securities[symbol].Price > self.minAmountToBeInvested class SymbolData(): def __init__(self, theSymbol, algo): #$@$@ Algo / Symbol / Price reference self.algo = algo self.symbol = theSymbol self.lastPrice = 0 self.price = 0 self.Init2() #$@$@ ---------------------------------------- def Init2(self): self.indicators = { 'EMA_FAST' : self.algo.EMA(self.symbol,self.algo.emaFastPeriod,Resolution.Daily), 'EMA_SLOW' : self.algo.EMA(self.symbol,self.algo.emaSlowPeriod,Resolution.Daily)} for key, indicator in self.indicators.items(): self.algo.WarmUpIndicator(self.symbol, indicator, Resolution.Minute) self.emaFastWindow = ScanWin("float", 2) self.emaSlowWindow = ScanWin("float", 2) self.lastPriceWindow = ScanWin("float", 2) #$@$@ ---------------------------------------- def OnSymbolData(self, lastKnownPrice, tradeBar): self.lastPrice = lastKnownPrice self.UpdateRW() #$@$@ ---------------------------------------- def UpdateRW(self): self.emaFastWindow.Add(self.indicators['EMA_FAST'].Current.Value) self.emaSlowWindow.Add(self.indicators['EMA_SLOW'].Current.Value) self.lastPriceWindow.Add(self.lastPrice) #$@$@ ---------------------------------------- def IsReady(self): return (self.indicators['EMA_FAST'].IsReady and self.indicators['EMA_SLOW'].IsReady) #$@$@ ---------------------------------------- def ESF(self): if( self.IsReady() ): if( self.emaFastWindow.isAbove(self.emaSlowWindow) and \ self.lastPriceWindow.isAbove(self.emaFastWindow) ): return True return False #$@$@ ---------------------------------------- def EXFD(self): if( self.IsReady() ): if ( self.lastPriceWindow.isBelow(self.emaSlowWindow) or \ self.emaSlowWindow.isAbove(self.emaFastWindow) ): return True return False #$@$@ --------------------------------------------------------- def OnNewPositionOpened(self): return #$@$@ ----------------------------------------------------------------------------- def MOP(self): if(self.EXFD()): self.Exp(exitMsg="Exit Signal Fired") #$@$@ ---------------------------------------- def Exp(self, exitMsg): self.algo.Exp(self.symbol, exitMsg)