Overall Statistics |
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -0.764 Tracking Error 0.242 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset |
from typing import Dict, List from AlgorithmImports import * from datetime import timedelta class ExampleAlpha(QCAlgorithm): bar_period = timedelta(hours=4) window_size = 52 resolution = Resolution.Hour market = Market.GDAX def Initialize(self): if not self.LiveMode: self.DebugMode = True self.SetCash(10000.00) self.SetStartDate(2020, 1, 1) self.SetEndDate(2021, 6, 1) crypto_tickers = ["BTCUSD", "ETHUSD"] symbols=[Symbol.Create(x, SecurityType.Crypto, self.market) for x in crypto_tickers] self.UniverseSettings.Resolution = self.resolution if self.market == Market.GDAX: self.UniverseSettings.Leverage = 1 elif self.market == Market.Kraken: self.UniverseSettings.Leverage = 3 self.SetUniverseSelection(ManualUniverseSelectionModel(symbols)) self.SetAlpha(ExampleAlphaModel(self, self.bar_period, self.window_size, self.resolution)) self.SetPortfolioConstruction(InsightWeightingPortfolioConstructionModel()) self.SetExecution(ImmediateExecutionModel()) self.SetRiskManagement(NullRiskManagementModel()) class IndicatorTwo(PythonIndicator): @property def ready(self) -> bool: return self.tradebar_window.Count > self.WarmUpPeriod def __init__(self): self.name = "IndicatorTwo" self.Period = 52 self.Value = 0.0 PythonIndicator.__init__(self.name, self.Period) self.WarmUpPeriod = 52 self._obv = OnBalanceVolume() self._slow_length = 13 self._fast_length = 7 self._ema_slow = ExponentialMovingAverage("ema_slow", self._slow_length) self._ema_fast = ExponentialMovingAverage("ema_fast", self._fast_length) self.tradebar_window = RollingWindow[TradeBar](52) self.fast_ema_window = RollingWindow[float](52) self.slow_ema_window = RollingWindow[float](52) self.obv_window = RollingWindow[float](52) def Update(self, input: TradeBar) -> bool: self.tradebar_window.Add(input) self._obv.Update(input) self._ema_slow.Update(input.EndTime, self._obv.Current.Value) self._ema_fast.Update(input.EndTime, self._obv.Current.Value) self.fast_ema_window.Add(self._ema_fast.Current.Value) self.slow_ema_window.Add(self._ema_slow.Current.Value) self.obv_window.Add(self._obv.Current.Value) self.Value = self._obv.Current.Value return self.ready class IndicatorOne(PythonIndicator): def ready(self) -> bool: self.tradebar_window.Count > self.WarmUpPeriod def __init__(self): PythonIndicator.__init__("IndicatorOne", 20) self.WarmUpPeriod = 20 self._obv = OnBalanceVolume() self._slow_length = 13 self._fast_length = 7 self._ema_slow = ExponentialMovingAverage("ema_slow", self._slow_length) self._ema_fast = ExponentialMovingAverage("ema_fast", self._fast_length) self.tradebar_window = RollingWindow[TradeBar](20) self.fast_ema_window = RollingWindow[float](20) #TODO self.slow_ema_window = RollingWindow[float](20) #TODO self.obv_window = RollingWindow[float](20) #TODO self.Value = 0.0 def Update(self, input: TradeBar) -> bool: self.tradebar_window.Add(input) self._obv.Update(input) self._ema_slow.Update(input.EndTime, self._obv.Current.Value) self._ema_fast.Update(input.EndTime, self._obv.Current.Value) self.fast_ema_window.Add(self._ema_fast.Current.Value) self.slow_ema_window.Add(self._ema_slow.Current.Value) self.obv_window.Add(self._obv.Current.Value) self.Value = self._obv.Current.Value return self.ready class SymbolData: @property def ready(self) -> bool: return self.tradebar_window.IsReady and self.indicator_one.ready and self.indicator_two.ready # return self.tradebar_window.IsReady and self.indicator_one.ready #uncomment @property def is_band_positive(self) -> bool: return self.indicator_one.fast_ema_window[0] > self.indicator_one.slow_ema_window[0] and self.indicator_two.fast_ema_window[0] > self.indicator_two.slow_ema_window[0] # return self.indicator_one.fast_ema_window[0] > self.indicator_one.slow_ema_window[0] def __init__(self, algorithm: QCAlgorithm, security: Security, bar_period: timedelta, window_size: int, resolution: Resolution): self.algorithm = algorithm self.security = security self.symbol = security.Symbol self.indicator_one = IndicatorOne() self.indicator_two = IndicatorTwo() self.bar_period = bar_period self.window_size = window_size self.previous_direction = InsightDirection.Down self.resolution = resolution self.tradebar_window = RollingWindow[TradeBar](window_size) self.consolidator = TradeBarConsolidator(bar_period) self.consolidator.DataConsolidated += self.consolidator_handler self.sample_count = 0 self.warming_up = True self.consolidator_ready = False # Warmup # total_time = Time.Multiply(self.bar_period, float(self.window_size)) # algorithm.Debug(total_time) history = algorithm.History(self.symbol, total_time, self.resolution) # algorithm.Debug(history) if history.empty: algorithm.Debug("HIsToRY IS EmpTy") #TODO: handle error symbol_history = history.loc[self.symbol] for idx, row in symbol_history.iterrows(): # algorithm.Debug(idx) # algorithm.Debug(row) tradebar = TradeBar( idx, self.symbol, row["open"], row["high"], row["low"], row["close"], row["volume"], self.bar_period ) self.consolidator.Update(tradebar) self.warming_up = False def hook_up_consolidator(self, algorithm: QCAlgorithm): algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.consolidator) self.consolidator_ready = True algorithm.Debug("Add consolidator") def consolidator_handler(self, sender, tradebar): self.indicator_one.Update(tradebar) self.indicator_two.Update(tradebar) self.tradebar_window.Add(tradebar) class ExampleAlphaModel(AlphaModel): def __init__(self, algorithm: QCAlgorithm, bar_period: timedelta, window_size: int, resolution: Resolution): self.symbol_data_keyed_by_symbol: Dict[Symbol, SymbolData] = {} self.bar_period = bar_period self.window_size = window_size self.resolution = resolution self.sample_count = 0 def Update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]: insights = [] for key, symbol_data in self.symbol_data_keyed_by_symbol.items(): if symbol_data.security.Price == 0: continue if not symbol_data.ready: continue if not symbol_data.warming_up and not symbol_data.consolidator_ready: symbol_data.hook_up_consolidator(algorithm) continue if symbol_data.consolidator_ready: direction = InsightDirection.Up if symbol_data.is_band_positive else InsightDirection.Down if direction != symbol_data.previous_direction: new_insight = Insight(symbol_data.security.Symbol, timedelta(minutes=1), InsightType.Price, direction) new_insight.Weight = 0.5 insights.append(new_insight) symbol_data.previous_direction = direction return insights def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges): for added_security in changes.AddedSecurities: symbol_data = SymbolData(algorithm, added_security, self.bar_period, self.window_size, self.resolution) algorithm.Debug(symbol_data) self.symbol_data_keyed_by_symbol[added_security.Symbol] = symbol_data for removed_security in changes.RemovedSecurities: symbol_data = self.symbol_data_keyed_by_symbol.pop(removed_security.Symbol, None) if symbol_data is not None: algorithm.SubscriptionManager.RemoveConsolidator(removed_security.Symbol, symbol_data.consolidator)