Overall Statistics
Total Trades
209
Average Win
5.18%
Average Loss
-2.01%
Compounding Annual Return
4957.904%
Drawdown
37.700%
Expectancy
1.001
Net Profit
513.472%
Sharpe Ratio
25.198
Probabilistic Sharpe Ratio
98.750%
Loss Rate
44%
Win Rate
56%
Profit-Loss Ratio
2.58
Alpha
21.531
Beta
0.652
Annual Standard Deviation
0.853
Annual Variance
0.728
Information Ratio
27.598
Tracking Error
0.781
Treynor Ratio
33.005
Total Fees
$1177.03
Estimated Strategy Capacity
$220000.00
Lowest Capacity Asset
BTCUSD E3
from io import StringIO
from datetime import datetime
import pandas as pd
import numpy as np
from scipy import optimize

class CasualFluorescentYellowCaterpillar(QCAlgorithm):

    def Initialize(self):
        #Backtest dates
        self.SetStartDate(2021, 4, 19)  # Set Start Date
        #self.SetEndDate(2021, 9, 18)
        
        #Algorithm cash
        self.SetCash(4000)
        
        #Symbols to be traded
        self.symbols = ['BTCUSD','ETHUSD', 'XRPUSD', 'SOLUSD', 'LUNAUSD']
        self.resolution = Resolution.Hour
        
        #Parameters
        self.window = int(self.GetParameter("window"))
        self.rebalance = int(self.GetParameter("rebalance"))
        self.t_count = -1
        
        #Additional variables
        #List to store previous weights
        self.last_w = [0 for i in range(len(self.symbols))]
        self.use_last = True
        self.rolling = [RollingWindow[float](self.window) for symbol in self.symbols]
        
        self.initializeSymbols()
        self.initRollingWindow()
        
        self.Rf=0 # April 2019 average risk  free rate of return in USA approx 3%
        annRiskFreeRate = self.Rf/100
        self.r0 = (np.power((1 + annRiskFreeRate),  (1.0 / 360.0)) - 1.0) * 100 
        self.portfolioSize = len(self.symbols)
        
        self.SetBrokerageModel(BrokerageName.Bitfinex)
        self.SetBenchmark("BTCUSD")
        return
    
    def initializeSymbols(self):
        self.symbols_objects = []
        for symbol in self.symbols:
            data = self.AddCrypto(symbol, self.resolution, Market.Bitfinex)
            data.SetFeeModel(CustomFeeModel(self))
            self.symbols_objects.append(data.Symbol)
        return
    
    def initRollingWindow(self):
        c = 0
        for symbol in self.symbols:
            df = pd.DataFrame()
            while df.empty:
                df = self.History(self.Symbol(symbol), self.window)
                d = df['close'].to_list()
                for x in d:
                    self.rolling[c].Add(x)
            c += 1
        return
    
    def OnData(self, data):
        self.t_count += 1
        if self.t_count % self.rebalance == 0:
            self.SpecificTime()
            
        c = 0
        for symbol in self.symbols:
            self.rolling[c].Add(data[symbol].Close)
            c+=1
        
        return
    
    def SpecificTime(self):
        #Check the len of the rolling windows
        flag = True
        for roll in self.rolling:
            l = [i for i in roll][::-1]
            self.Log(len(l))
            if len(l) < self.window:
                flag = False
        if not flag:
            return
        #try:
        
        Ri = []
        c = 0
        for symbol in self.symbols:
            Ri.append([i for i in self.rolling[c]][::-1])
            c+=1
        Ri = np.array(Ri).transpose()
        self.Log(Ri)
        Ri = StockReturnsComputing(Ri, self.window, self.portfolioSize)
        Ei = np.mean(Ri, axis = 0)
        #except:
        #    self.Log(str("Error during data extraction"))
        #    return
        
        cov = np.cov(Ri, rowvar=False)
            
        #initialization
        xOptimal =[]
        minRiskPoint = []
        expPortfolioReturnPoint =[]
        maxSharpeRatio = 0
        
        #compute maximal Sharpe Ratio and optimal weights
        result = MaximizeSharpeRatioOptmzn(Ei, cov, self.r0, self.portfolioSize)
        xOptimal.append(result.x)
        
        w = list(xOptimal[0])
        w = [ 0 if wx < 0.0000001 else wx for wx in w ]
        
        self.Debug(w)
        
        if not self.use_last:
            self.Liquidate()
        targets = []
        for i in range(len(w)):
            currency = self.symbols[i]
            if not self.use_last:
                self.SetHoldings(currency, w[i])
            else:
                targets.append(PortfolioTarget(currency, 0.75*w[i]))
        if self.use_last:
            self.SetHoldings(targets)
        return
    
    def OnEndOfAlgorithm(self):
        self.Liquidate()
        return

# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.002
        return OrderFee(CashAmount(fee, "USD"))
        
def MaximizeSharpeRatioOptmzn(MeanReturns, CovarReturns, RiskFreeRate, PortfolioSize):
    
    # define maximization of Sharpe Ratio using principle of duality
    def  f(x, MeanReturns, CovarReturns, RiskFreeRate, PortfolioSize):
        funcDenomr = np.sqrt(np.matmul(np.matmul(x, CovarReturns), x.T) )
        funcNumer = np.matmul(np.array(MeanReturns),x.T)-RiskFreeRate
        func = -(funcNumer / funcDenomr)
        return func

    #define equality constraint representing fully invested portfolio
    def constraintEq(x):
        A=np.ones(x.shape)
        b=1
        constraintVal = np.matmul(A,x.T)-b 
        return constraintVal
    
    #define bounds and other parameters
    xinit=np.repeat(0.33, PortfolioSize)
    cons = ({'type': 'eq', 'fun':constraintEq})
    lb = 0
    ub = 1
    bnds = tuple([(lb,ub) for x in xinit])
    
    #invoke minimize solver
    opt = optimize.minimize (f, x0 = xinit, args = (MeanReturns, CovarReturns,\
                             RiskFreeRate, PortfolioSize), method = 'SLSQP',  \
                             bounds = bnds, constraints = cons, tol = 10**-3)
    
    return opt
    
def StockReturnsComputing(StockPrice, Rows, Columns):
    
    StockReturn = np.zeros([Rows-1, Columns])
    for j in range(Columns):        # j: Assets
        for i in range(Rows-1):     # i: Daily Prices
            StockReturn[i,j]=((StockPrice[i+1, j]-StockPrice[i,j])/StockPrice[i,j])*100

    return StockReturn
from io import StringIO
from datetime import datetime
import pandas as pd
import numpy as np
from scipy import optimize 

class CasualFluorescentYellowCaterpillar(QCAlgorithm):

    def Initialize(self):
        #Backtest dates
        self.SetStartDate(2021, 4, 19)  # Set Start Date
        self.SetEndDate(2021, 4, 22)
        
        #Algorithm cash
        self.SetCash(4000)
        
        #Symbols to be traded
        self.resolution = Resolution.Hour
        self.symbols = ['BTCUSD','ETHUSD', 'XRPUSD', 'SOLUSD', 'LUNAUSD']
        self.initializeSymbols()
        
        #Parameters
        self.window = int(self.GetParameter("window"))
        self.rebalance = int(self.GetParameter("rebalance"))
        self.days_count = -1
        
        self.Rf=0 # April 2019 average risk  free rate of return in USA approx 3%
        annRiskFreeRate = self.Rf/100
        self.r0 = (np.power((1 + annRiskFreeRate),  (1.0 / 360.0)) - 1.0) * 100 
        self.portfolioSize = len(self.symbols)
        
        #Additional variables
        #List to store previous weights
        self.last_w = [0 for i in range(len(self.symbols))]
        self.use_last = True
        self.rolling = [RollingWindow[float](self.window) for symbol in self.symbols]
        self.initRollingWindow()
        
        self.SetBrokerageModel(BrokerageName.Bitfinex)
        self.SetBenchmark("BTCUSD")
        return
    
    def initializeSymbols(self):
        self.symbols_objects = []
        for symbol in self.symbols:
            data = self.AddCrypto(symbol, self.resolution, Market.Bitfinex)
            data.SetFeeModel(CustomFeeModel(self))
            self.symbols_objects.append(data.Symbol)
        return
    
    def initRollingWindow(self):
        c = 0
        for symbol in self.symbols:
            df = pd.DataFrame()
            while df.empty:
                df = self.History(self.Symbol(symbol), self.window)
                d = df['close'].to_list()
                for x in d:
                    self.rolling[c].Add(x)
            c += 1
        return
    
    def OnData(self, data):
        c = 0
        for symbol in self.symbols:
            self.rolling[c].Add(data[symbol].Close)
            c+=1
        self.days_count += 1
        if self.days_count % self.rebalance == 0 or self.days_count == 0:
            self.SpecificTime()
        return
    
    def SpecificTime(self):
        try:
            if self.window == 0:
                self.window = self.warm_up_count
            
            Ri = []
            for symbol in self.symbols:
                df = pd.DataFrame()
                while df.empty:
                    df = self.History(self.Symbol(symbol), self.window)
                    self.Debug(df.index)
                    d = df['close'].to_list()
                Ri.append(d)
            Ri = np.array(Ri).transpose()
            #self.Log(Ri)
            Ri = StockReturnsComputing(Ri, self.window, self.portfolioSize)
            Ei = np.mean(Ri, axis = 0)
            #self.Log(Ei)
        except:
            self.Log(str("Error during data extraction"))
            return
            
        cov = np.cov(Ri, rowvar=False)
            
        #initialization
        xOptimal =[]
        minRiskPoint = []
        expPortfolioReturnPoint =[]
        maxSharpeRatio = 0
        
        #compute maximal Sharpe Ratio and optimal weights
        result = MaximizeSharpeRatioOptmzn(Ei, cov, self.r0, self.portfolioSize)
        xOptimal.append(result.x)
        
        w = list(xOptimal[0])
        w = [ 0 if wx < 0.0000001 else wx for wx in w ]
        
        self.Debug(w)
        
        if not self.use_last:
            self.Liquidate()
        targets = []
        for i in range(len(w)):
            currency = self.symbols[i]
            if not self.use_last:
                self.SetHoldings(currency, w[i])
            else:
                targets.append(PortfolioTarget(currency, 0.9*w[i]))
        if self.use_last:
            self.SetHoldings(targets)
        return
    
    def OnEndOfAlgorithm(self):
        self.Liquidate()
        return

# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.002
        return OrderFee(CashAmount(fee, "USD"))
        
def MaximizeSharpeRatioOptmzn(MeanReturns, CovarReturns, RiskFreeRate, PortfolioSize):
    
    # define maximization of Sharpe Ratio using principle of duality
    def  f(x, MeanReturns, CovarReturns, RiskFreeRate, PortfolioSize):
        funcDenomr = np.sqrt(np.matmul(np.matmul(x, CovarReturns), x.T) )
        funcNumer = np.matmul(np.array(MeanReturns),x.T)-RiskFreeRate
        func = -(funcNumer / funcDenomr)
        return func

    #define equality constraint representing fully invested portfolio
    def constraintEq(x):
        A=np.ones(x.shape)
        b=1
        constraintVal = np.matmul(A,x.T)-b 
        return constraintVal
    
    #define bounds and other parameters
    xinit=np.repeat(0.33, PortfolioSize)
    cons = ({'type': 'eq', 'fun':constraintEq})
    lb = 0
    ub = 1
    bnds = tuple([(lb,ub) for x in xinit])
    
    #invoke minimize solver
    opt = optimize.minimize (f, x0 = xinit, args = (MeanReturns, CovarReturns,\
                             RiskFreeRate, PortfolioSize), method = 'SLSQP',  \
                             bounds = bnds, constraints = cons, tol = 10**-3)
    
    return opt
    
def StockReturnsComputing(StockPrice, Rows, Columns):
    
    StockReturn = np.zeros([Rows-1, Columns])
    for j in range(Columns):        # j: Assets
        for i in range(Rows-1):     # i: Daily Prices
            StockReturn[i,j]=((StockPrice[i+1, j]-StockPrice[i,j])/StockPrice[i,j])*100

    return StockReturn
from io import StringIO
from datetime import datetime
import pandas as pd
import numpy as np
from scipy import optimize 

class CasualFluorescentYellowCaterpillar(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2021, 4, 19)  # Set Start Date
        #self.SetEndDate(2021, 7, 2)
        self.SetCash(4000)
        
        self.symbols = ['BTCUSD','ETHUSD', 'XRPUSD', 'SOLUSD', 'LUNAUSD']
        
        self.warm_up = len(self.symbols)
        self.warm_up_count = 1
        self.window = 8
        self.rebalance = 8
        self.resolution = Resolution.Hour
        self.last_w = [0 for i in range(len(self.symbols))]
        self.use_last = True
        
        #Initialize symbols
        self.initializeSymbols()
        
        #self.Ris = self.Ris.to_numpy() #.transpose()
        
        #set risk free asset rate of return
        self.Rf=0 # April 2019 average risk  free rate of return in USA approx 3%
        annRiskFreeRate = self.Rf/100
        
        #compute daily risk free rate in percentage
        self.r0 = (np.power((1 + annRiskFreeRate),  (1.0 / 360.0)) - 1.0) * 100 
        
        self.portfolioSize = len(self.symbols)
        
        self.SetBrokerageModel(BrokerageName.Bitfinex)
        
        self.SetBenchmark("BTCUSD")
        
        self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(6, 0), self.SpecificTime)
        self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(14, 0), self.SpecificTime)
        self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(22, 0), self.SpecificTime)
        return
    
    def initializeSymbols(self):
        self.symbols_objects = []
        for symbol in self.symbols:
            data = self.AddCrypto(symbol, self.resolution, Market.Bitfinex)
            data.SetFeeModel(CustomFeeModel(self))
            self.symbols_objects.append(data.Symbol)
        return
    
    def OnData(self, data):
        if self.warm_up_count < self.warm_up:
            self.warm_up_count += 1
            return
        self.warm_up_count += 1
        
        if self.warm_up_count % self.rebalance == 0:
            try:
                if self.window == 0:
                    self.window = self.warm_up_count
                
                Ri = []
                for symbol in self.symbols:
                    df = pd.DataFrame()
                    while df.empty:
                        df = self.History(self.Symbol(symbol), self.window)
                        d = df['close'].to_list()
                    Ri.append(d)
                Ri = np.array(Ri).transpose()
                Ri = StockReturnsComputing(Ri, self.window, self.portfolioSize)
                Ei = np.mean(Ri, axis = 0)
            except:
                self.Log(str("Error during data extraction"))
                return
                
            cov = np.cov(Ri, rowvar=False)
                
            #initialization
            xOptimal =[]
            minRiskPoint = []
            expPortfolioReturnPoint =[]
            maxSharpeRatio = 0
            
            #compute maximal Sharpe Ratio and optimal weights
            result = MaximizeSharpeRatioOptmzn(Ei, cov, self.r0, self.portfolioSize)
            xOptimal.append(result.x)
            
            w = list(xOptimal[0])
            w = [ 0 if wx < 0.0000001 else wx for wx in w ]
            
            self.Debug(w)
            
            if not self.use_last:
                self.Liquidate()
            targets = []
            for i in range(len(w)):
                currency = self.symbols[i]
                if not self.use_last:
                    self.SetHoldings(currency, w[i])
                else:
                    targets.append(PortfolioTarget(currency, 0.9*w[i]))
            if self.use_last:
                self.SetHoldings(targets)
        return
    
    def OnEndOfAlgorithm(self):
        self.Liquidate()
        try:
            #Ri = self.Ris[0:self.warm_up_count, :]
            #Ei = np.mean(Ri, axis = 0)
            if self.window == 0:
                    self.window = self.warm_up_count
            Ri = []
            for symbol in self.symbols:
                df = pd.DataFrame()
                while df.empty:
                    df = self.History(self.Symbol(symbol), self.window)
                    d = df['close'].to_list()
                Ri.append(d)
            Ri = np.array(Ri).transpose()
            Ri = StockReturnsComputing(Ri, self.window, self.portfolioSize)
            Ei = np.mean(Ri, axis = 0)
            #self.Debug(Ei)
        except:
            self.Log(str("Error during data extraction"))
            return
            
        cov = np.cov(Ri, rowvar=False)
            
        #initialization
        xOptimal =[]
        minRiskPoint = []
        expPortfolioReturnPoint =[]
        maxSharpeRatio = 0
        
        #compute maximal Sharpe Ratio and optimal weights
        result = MaximizeSharpeRatioOptmzn(Ei, cov, self.r0, self.portfolioSize)
        xOptimal.append(result.x)
        
        w = list(xOptimal[0])
        w = [ 0 if wx < 0.0000001 else wx for wx in w ]
        self.Debug(w)

# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.002
        return OrderFee(CashAmount(fee, "USD"))
        
def MaximizeSharpeRatioOptmzn(MeanReturns, CovarReturns, RiskFreeRate, PortfolioSize):
    
    # define maximization of Sharpe Ratio using principle of duality
    def  f(x, MeanReturns, CovarReturns, RiskFreeRate, PortfolioSize):
        funcDenomr = np.sqrt(np.matmul(np.matmul(x, CovarReturns), x.T) )
        funcNumer = np.matmul(np.array(MeanReturns),x.T)-RiskFreeRate
        func = -(funcNumer / funcDenomr)
        return func

    #define equality constraint representing fully invested portfolio
    def constraintEq(x):
        A=np.ones(x.shape)
        b=1
        constraintVal = np.matmul(A,x.T)-b 
        return constraintVal
    
    #define bounds and other parameters
    xinit=np.repeat(0.33, PortfolioSize)
    cons = ({'type': 'eq', 'fun':constraintEq})
    lb = 0
    ub = 1
    bnds = tuple([(lb,ub) for x in xinit])
    
    #invoke minimize solver
    opt = optimize.minimize (f, x0 = xinit, args = (MeanReturns, CovarReturns,\
                             RiskFreeRate, PortfolioSize), method = 'SLSQP',  \
                             bounds = bnds, constraints = cons, tol = 10**-3)
    
    return opt
    
def StockReturnsComputing(StockPrice, Rows, Columns):
    
    StockReturn = np.zeros([Rows-1, Columns])
    for j in range(Columns):        # j: Assets
        for i in range(Rows-1):     # i: Daily Prices
            StockReturn[i,j]=((StockPrice[i+1, j]-StockPrice[i,j])/StockPrice[i,j])*100

    return StockReturn