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 -5.978 Tracking Error 0.09 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset |
class SymbolData: def __init__(self, algorithm, symbol): self.algorithm = algorithm self.symbol = symbol # Daily consolidator definition and subscription to data self.daily_consolidator = TradeBarConsolidator(timedelta(days=1)) self.algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.daily_consolidator) self.daily_consolidator.DataConsolidated += self.OnDailyBar # signal resolution consolidator definition and subscription to data self.consolidator = TradeBarConsolidator(timedelta(minutes=30)) self.algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.consolidator) self.consolidator.DataConsolidated += self.OnBar # self.algorithm.ResolveConsolidator # minute consolidator definition and subscription to data self.minute_consolidator = TradeBarConsolidator(timedelta(minutes=1)) self.algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.minute_consolidator) self.minute_consolidator.DataConsolidated += self.OnMinuteBar # Volume daily SMA self.vol_sma = SimpleMovingAverage(20) # 30 min resolution - 10 period SMA for price self.sma = SimpleMovingAverage(10) self.algorithm.RegisterIndicator(self.symbol, self.sma, self.consolidator) self.minute_bars = RollingWindow[TradeBar](30) self.bars = RollingWindow[TradeBar](3) self.WarmUpIndicators() def WarmUpIndicators(self): # returns a dataframe history = self.algorithm.History(self.symbol, 20, Resolution.Daily) for bar in history.itertuples(): time = bar.Index[1] open = bar.open high = bar.high low = bar.low close = bar.close volume = bar.volume min_bar = TradeBar(time, self.symbol, open, high, low, close, volume) self.sma.Update(time, close) self.vol_sma.Update(time, volume) #self.atr.Update(trade_bar) self.minute_bars.Add(min_bar) def OnDailyBar(self, sender, bar): # Updates volume sma with latest daily bar data self.vol_sma.Update(bar.EndTime, bar.Volume) def OnBar(self, sender, bar): # Saves signal resolution bars self.bars.Add(bar) def OnMinuteBar(self, sender, bar): # Saves minute bars self.minute_bars.Add(bar) @property def IsReady(self): return self.vol_sma.IsReady and self.sma.IsReady and self.minute_bars.IsReady and \ self.bars.IsReady def KillConsolidators(self): self.algorithm.SubscriptionManager.RemoveConsolidator(self.symbol, self.consolidator) self.algorithm.SubscriptionManager.RemoveConsolidator(self.symbol, self.minute_consolidator) self.algorithm.SubscriptionManager.RemoveConsolidator(self.symbol, self.daily_consolidator) @property def RecentDollarVolume(self): dollar_volume = 0 for bar in list(self.minute_bars): dollar_volume += bar.Volume * bar.Close return dollar_volume @property def CandlePatternSignal(self): # Close > Open # Wick < .5*Body # Body > 1.02*Open most_recent_bar = self.bars[0] close = most_recent_bar.Close open = most_recent_bar.Open high = most_recent_bar.High wick = high - close body = close - open return close > open and wick < 0.5 * body and body > 1.02 * open @property def UnusualVolume(self): vol_sma = self.vol_sma.Current.Value recent_volume = self.bars[0].Volume return recent_volume > 3 * vol_sma def UnusualVolumeSignal(self): volLim = bool(self.RecentDollarVolume > 500000) if volLim and self.CandlePatternSignal and self.UnusualVolume: return True return False #return self.RecentDollarVolume > 500000 and self.CandlePatternSignal and self.UnusualVolume
class TradeManagement: def __init__(self, algorithm, symbol): self.algorithm = algorithm self.symbol = symbol self.entry_limit_ticket = None self.stop_loss = None self.take_profit = None def LimitOrder(self, quantity, limit_price): # self.entry_limit_ticket = self.algorithm.LimitOrder(...) # self.stop_loss = .. # .... pass def CheckAndUpdate(self): current_market_price = self.algorithm.Securities[self.symbol].Price if not self.ActivePosition: return # stop loss # take profit # self.algorithm.Liquidate def CancelEntryOrder(self): if self.entry_limit_ticket is not None: self.entry_limit_ticket.Cancel() self.stop_loss = None self.take_profit = None @property def ActivePosition(self): return self.algorithm.Portfolio[self.symbol].Invested def GetPositionSize(self): portfolio_value = self.algorithm.Portfolio.TotalPortfolioValue #...
from SymbolData import * class LogicalRedOrangeGoshawk(QCAlgorithm): def Initialize(self): self.SetStartDate(2021, 4, 1) self.SetEndDate(2021, 5, 7) self.SetCash(100000) self.benchmark = "SPY" self.AddEquity("SPY", Resolution.Minute) self.AddEquity("ELTK", Resolution.Minute) self.AddUniverse(self.SelectCoarse) self.UniverseSettings.Resolution = Resolution.Minute self.symbols = {} self.Schedule.On(self.DateRules.EveryDay(self.benchmark), self.TimeRules.AfterMarketOpen(self.benchmark, 30), self.Rebalance) def Rebalance(self): self.Debug(f"Universe Size... {len(self.symbols)}") for symbol, symbol_data in self.symbols.items(): if not symbol_data.IsReady: self.Debug("we aint ready") continue if symbol_data.UnusualVolumeSignal: self.Debug(f"Unusual Volume Signal: {symbol_data.UnusualVolumeSignal}") self.Debug(f"Bar Volume: {symbol_data.RecentDollarVolume}") self.Debug(f"Candle Pattern: {symbol_data.CandlePatternSignal}") self.Debug(f"Unusual Volume: {symbol_data.UnusualVolume}") self.Debug(f"{self.Time} -- {symbol} has a signal!") def SelectCoarse(self, coarse): filtered_by_price = [c for c in coarse if c.AdjustedPrice >= 4 and c.AdjustedPrice <= 8] return [c.Symbol for c in filtered_by_price] def OnSecuritiesChanged(self, changes): for security in changes.AddedSecurities: symbol = security.Symbol if symbol not in self.symbols and symbol.Value != self.benchmark and symbol.Value == 'ELTK': self.symbols[symbol] = SymbolData(self, symbol) self.Debug(symbol) for security in changes.RemovedSecurities: symbol = security.Symbol if symbol in self.symbols: symbol_data = self.symbols.pop(symbol, None) symbol_data.KillConsolidators()