| Overall Statistics |
|
Total Trades 288 Average Win 0.03% Average Loss -0.03% Compounding Annual Return 28.106% Drawdown 3.500% Expectancy 0.283 Net Profit 3.732% Sharpe Ratio 1.72 Probabilistic Sharpe Ratio 60.442% Loss Rate 40% Win Rate 60% Profit-Loss Ratio 1.13 Alpha -0 Beta 0.555 Annual Standard Deviation 0.115 Annual Variance 0.013 Information Ratio -1.592 Tracking Error 0.1 Treynor Ratio 0.357 Total Fees $299.53 Estimated Strategy Capacity $66000000.00 Lowest Capacity Asset IAU T5QHGBE8K3TX |
# region imports
from AlgorithmImports import *
from tslearn.barycenters import softdtw_barycenter
from tslearn.clustering import TimeSeriesKMeans
import joblib
# endregion
class TslearnExampleAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2022, 7, 4)
self.SetCash(100000)
tickers = ["SPY", "QQQ", "DIA",
"AAPL", "MSFT", "TSLA",
"IEF", "TLT", "SHV", "SHY",
"GLD", "IAU", "SLV",
"USO", "XLE", "XOM"]
self.symbols = [self.AddEquity(ticker, Resolution.Daily).Symbol for ticker in tickers]
training_length = 252
self.training_data = {}
history = self.History(self.symbols, training_length, Resolution.Daily).unstack(0).close
for symbol in self.symbols:
self.training_data[symbol] = RollingWindow[float](training_length)
for close_price in history[symbol]:
self.training_data[symbol].Add(close_price)
self.model = TimeSeriesKMeans(n_clusters=6, # We have 6 main groups
metric="dtw")
self.Train(self.my_training_method)
self.Train(self.DateRules.Every(DayOfWeek.Sunday), self.TimeRules.At(8,0), self.my_training_method)
def get_features(self):
close_price = pd.DataFrame({symbol: list(data)[::-1] for symbol, data in self.training_data.items()})
log_price = np.log(close_price)
log_normal_price = (log_price - log_price.mean()) / log_price.std()
return log_normal_price
def my_training_method(self):
features = self.get_features()
self.model.fit(features.T.values)
def OnData(self, slice: Slice) -> None:
for symbol in self.symbols:
if symbol in slice.Bars:
self.training_data[symbol].Add(slice.Bars[symbol].Close)
features = self.get_features()
self.labels = self.model.predict(features.T.values)
for i in set(self.labels):
assets_in_cluster = features.columns[[n for n, k in enumerate(self.labels) if k == i]]
size = 1/6/len(assets_in_cluster)
self.SetHoldings([PortfolioTarget(symbol, size) for symbol in assets_in_cluster])
def OnEndOfAlgorithm(self):
model_key = "model"
file_name = self.ObjectStore.GetFilePath(model_key)
self.model.to_hdf5(file_name + ".hdf5")