| Overall Statistics |
|
Total Trades 15 Average Win 2949.00% Average Loss -3.29% Compounding Annual Return 117.685% Drawdown 83.800% Expectancy 255.427 Net Profit 5471.716% Sharpe Ratio 1.716 Probabilistic Sharpe Ratio 66.199% Loss Rate 71% Win Rate 29% Profit-Loss Ratio 896.49 Alpha 1.141 Beta -0.185 Annual Standard Deviation 0.651 Annual Variance 0.424 Information Ratio 1.463 Tracking Error 0.676 Treynor Ratio -6.027 Total Fees $1331849.66 Estimated Strategy Capacity $5000000.00 Lowest Capacity Asset BTCUSD XJ |
import numpy as np
import pandas as pd
from sklearn.linear_model import RidgeClassifier
from AlgorithmImports import *
class MachineLearningAlgo(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2016, 5, 2)
self.SetEndDate(2021, 6, 30)
self.SetCash(1000000)
self.AddEquity("SPY", Resolution.Daily)
self.SetBenchmark("SPY")
self.SetBrokerageModel(BrokerageName.AlphaStreams)
self.SetExecution(ImmediateExecutionModel())
self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())
self.AddCrypto("BTCUSD", Resolution.Daily)
self.ticker = "BTCUSD"
self.AddUniverseSelection(ManualUniverseSelectionModel(self.ticker))
self.lookback = 30
self.SetWarmup(self.lookback)
self.Close_rolling = RollingWindow[float](self.lookback)
self.Volume_rolling = RollingWindow[float](self.lookback)
self.fast_volume_LWMA_indicator = self.LWMA(self.ticker, 5, Resolution.Daily)
self.slow_volume_LWMA_indicator = self.LWMA(self.ticker, 20, Resolution.Daily)
self.RSI_rolling = RollingWindow[float](self.lookback)
self.RSI_indicator = self.RSI(self.ticker, 25, Resolution.Daily)
self.Trend_rolling = RollingWindow[float](self.lookback)
self.trLWMA_indicator = self.LWMA(self.ticker, 15, Resolution.Daily)
self.ROC_indicator = self.ROC(self.ticker, 1, Resolution.Daily)
self.AD_rolling = RollingWindow[float](self.lookback)
self.AD_indicator = self.AD(self.ticker, Resolution.Daily)
self.STO_rolling = RollingWindow[float](self.lookback)
self.STO_indicator = self.STO(self.ticker, 14, 14, 3, Resolution.Daily)
self.KAMA_rolling = RollingWindow[float](self.lookback)
self.KAMA_indicator = self.KAMA(self.ticker, 25, Resolution.Daily)
self.AddAlpha(MachineLearningAlphaModel(self.ticker,
self.Close_rolling, self.Volume_rolling,
self.fast_volume_LWMA_indicator, self.slow_volume_LWMA_indicator,
self.RSI_rolling, self.RSI_indicator,
self.Trend_rolling, self.trLWMA_indicator, self.ROC_indicator,
self.AD_rolling, self.AD_indicator,
self.STO_rolling, self.STO_indicator,
self.KAMA_rolling, self.KAMA_indicator))
class MachineLearningAlphaModel:
def __init__(self, ticker, closew, volumew,
fast_volume_lwmai, slow_volume_lwmai,
rsiw, rsii, trendw, trlwmai, roci,
adw, adi, stow, stoi, kamaw, kamai):
self.period = timedelta(30)
self.ticker = ticker
self.Close_rolling = closew
self.Volume_rolling= volumew
self.fast_volume_LWMA_indicator = fast_volume_lwmai
self.slow_volume_LWMA_indicator = slow_volume_lwmai
self.RSI_rolling = rsiw
self.RSI_indicator = rsii
self.Trend_rolling = trendw
self.trLWMA_indicator = trlwmai
self.ROC_indicator = roci
self.AD_rolling = adw
self.AD_indicator = adi
self.STO_rolling = stow
self.STO_indicator = stoi
self.KAMA_rolling = kamaw
self.KAMA_indicator = kamai
def GetMLModel(self):
self.MLModel = 0
self.MLModel = RidgeClassifier(random_state=18)
def Update(self, algorithm, data):
insights = []
if data.Bars.ContainsKey(self.ticker) and not algorithm.IsWarmingUp and self.RSI_indicator.IsReady \
and self.fast_volume_LWMA_indicator.IsReady and self.fast_volume_LWMA_indicator.IsReady\
and self.trLWMA_indicator.IsReady and self.AD_indicator.IsReady\
and self.STO_indicator.IsReady and self.KAMA_indicator.IsReady:
self.Close_rolling.Add(data[self.ticker].Close)
self.fast_volume_LWMA_indicator.Update(data.Bars[self.ticker].EndTime, data.Bars[self.ticker].Volume)
self.slow_volume_LWMA_indicator.Update(data.Bars[self.ticker].EndTime, data.Bars[self.ticker].Volume)
self.Volume_rolling.Add(self.fast_volume_LWMA_indicator.Current.Value / self.slow_volume_LWMA_indicator.Current.Value)
self.RSI_rolling.Add(self.RSI_indicator.Current.Value)
self.ROC_indicator.Update(data[self.ticker].EndTime, self.trLWMA_indicator.Current.Value)
self.Trend_rolling.Add(self.ROC_indicator.Current.Value)
self.AD_rolling.Add(self.AD_indicator.Current.Value)
self.STO_rolling.Add(self.STO_indicator.Current.Value)
self.KAMA_rolling.Add(self.KAMA_indicator.Current.Value)
if self.Close_rolling.IsReady and self.Volume_rolling.IsReady and self.RSI_rolling.IsReady \
and self.Trend_rolling.IsReady and self.AD_rolling.IsReady\
and self.STO_rolling.IsReady and self.KAMA_rolling.IsReady:
df1 = pd.DataFrame(self.Close_rolling, columns=["Close"]).reset_index(drop=True)
df2 = pd.DataFrame(self.Volume_rolling, columns=["Volume"]).reset_index(drop=True)
df3 = pd.DataFrame(self.RSI_rolling, columns=["RSI"]).reset_index(drop=True)
df4 = pd.DataFrame(self.Trend_rolling, columns=["Trend"]).reset_index(drop=True)
df5 = pd.DataFrame(self.AD_rolling, columns=["AD"]).reset_index(drop=True)
df6 = pd.DataFrame(self.STO_rolling, columns=["STO"]).reset_index(drop=True)
df7 = pd.DataFrame(self.KAMA_rolling, columns=["KAMA"]).reset_index(drop=True)
self.df = pd.concat([df1, df2, df3, df4, df5, df6, df7], axis=1)
# calculate daily forward returns to be used to set Target / Signal
self.df['Return'] = np.log(self.df['Close'].shift(-1)/self.df['Close'])
self.df = self.df.dropna()
# set Signal / Target
self.df['Signal'] = 0
self.df.loc[self.df['Return'] > 0, 'Signal'] = 1
self.df.loc[self.df['Return'] < 0, 'Signal'] = -1
# set training data
self.X = self.df.drop(['Close', 'Return','Signal'], axis=1)
self.Y = self.df['Signal']
# align feature set & signal
self.Y, self.X = self.Y.align(self.X, axis=0, join='inner')
self.X_train = self.X[:-1]
self.Y_train = self.Y[:-1]
self.X_train.replace([np.inf, -np.inf], np.nan, inplace=True)
self.Y_train.replace([np.inf, -np.inf], np.nan, inplace=True)
drops = []
[drops.append(i) for i in range(self.X_train.shape[0]) if self.X_train.iloc[i].isnull().any()]
[drops.append(i) for i in range(self.Y_train.shape[0]) if self.Y_train.iloc[i] == np.nan and i not in drops]
self.X_train.drop(index=self.X_train.index[drops], inplace=True)
self.Y_train.drop(index=self.Y_train.index[drops], inplace=True)
if self.X_train.empty or self.Y_train.empty: return []
# fit / train ML model
self.GetMLModel()
self.MLModel.fit(self.X_train, self.Y_train)
# predict next day signal using today's values of feature set
self.X_today = self.X.iloc[-1]
# self.X_today is Series, so convert to numpy array
self.X_today = self.X_today.to_numpy()
# reshape self.X_today because it only has 1 day's sample
self.X_today = self.X_today.reshape(1,-1)
# Y_predict will take predicted signal
self.Y_predict = self.Y.iloc[-1]
try:
self.Y_predict = self.MLModel.predict(self.X_today)
except: return []
# set insight based on predicted signal
if self.Y_predict == 1:
insights.append(Insight(self.ticker, self.period, InsightType.Price, InsightDirection.Up, 1, None))
elif self.Y_predict == -1:
insights.append(Insight(self.ticker, self.period, InsightType.Price, InsightDirection.Down, 1, None))
else:
insights.append(Insight(self.ticker, self.period, InsightType.Price, InsightDirection.Flat, 1, None))
return insights
def OnSecuritiesChanged(self, algorithm, changes):
self.changes = changes