| Overall Statistics |
|
Total Trades 22 Average Win 0.00% Average Loss 0.00% Compounding Annual Return -0.018% Drawdown 0.000% Expectancy -0.869 Net Profit -0.006% Sharpe Ratio -0.252 Probabilistic Sharpe Ratio 21.742% Loss Rate 90% Win Rate 10% Profit-Loss Ratio 0.31 Alpha 0 Beta -0.001 Annual Standard Deviation 0.001 Annual Variance 0 Information Ratio 0.264 Tracking Error 0.471 Treynor Ratio 0.176 Total Fees $22.00 Estimated Strategy Capacity $1200000000.00 |
from SymbolData import SymbolData
from TradeManagement import TradeManagement
class CryingBlueFlamingo(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2020, 1, 1) # Set Start Date
self.SetCash(100000) # Set Strategy Cash
self.benchmark = "SPY"
self.AddEquity(self.benchmark)
self.AddUniverse(self.CoarseSelection)
self.UniverseSettings.Resolution = Resolution.Minute
self.universe_size = 100
self.symbol_data = {}
self.trade_managers = {}
self.Schedule.On(self.DateRules.EveryDay(self.benchmark), self.TimeRules.AfterMarketOpen(self.benchmark, 15), self.Rebalance)
def Rebalance(self):
'''Fires everyday 15 minutes after market open'''
for symbol, symbol_data in self.symbol_data.items():
if not symbol_data.IsReady:
continue
signal = self.CalculateSignal(symbol_data)
if signal:
trade_manager = self.trade_managers[symbol]
trade_manager.CreateEntry(-1)
def OnData(self, data):
for symbol, trade_manager in self.trade_managers.items():
if not self.Portfolio[symbol].Invested:
continue
current_price = self.Securities[symbol].Price
stop_loss = trade_manager.stop_loss
take_profit = trade_manager.take_profit
if current_price > stop_loss or current_price < take_profit:
trade_manager.Liquidate()
def CalculateSignal(self, symbol_data):
sma = symbol_data.sma.Current.Value
atr = symbol_data.atr.Current.Value
bars = symbol_data.bar_window
latest_daily_bar = bars[0]
# checking if the high of the latest daily bar is greater than the high
# of all following bars
uptrend = all([latest_daily_bar.High > bar.High for bar in list(bars)[1:]])
red_bar = latest_daily_bar.Close < latest_daily_bar.Open
body = abs(latest_daily_bar.Open - latest_daily_bar.Close)
tail = abs(latest_daily_bar.Close - latest_daily_bar.Low)
head = abs(latest_daily_bar.High - latest_daily_bar.Open)
hanging_man = (tail > 2 * body) and (head < 0.3 * body)
# latest_market_price
price = self.Securities[symbol_data.symbol].Price
above_sma = price > sma
signal = red_bar and uptrend and hanging_man and above_sma
return signal
def CoarseSelection(self, coarse):
# list of ~8500 stocks (coarse data)
# coarse is a list of CoarseFundamental objects
# Descending order
sorted_by_liquidity = sorted(coarse, key=lambda c:c.DollarVolume, reverse=True)
most_liquid_coarse = sorted_by_liquidity[:self.universe_size]
# needs to return a list of Symbol object
most_liquid_symbols = [c.Symbol for c in most_liquid_coarse]
return most_liquid_symbols
def OnSecuritiesChanged(self, changes):
'''Fires after universe selection if there are any changes'''
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data and symbol.Value != self.benchmark:
self.symbol_data[symbol] = SymbolData(self, symbol)
self.trade_managers[symbol] = TradeManagement(self, symbol)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data:
symbol_data_object = self.symbol_data.pop(symbol, None)
symbol_data_object.KillConsolidators()
if symbol in self.trade_managers:
self.trade_managers.pop(symbol, None)class TradeManagement:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.days_active = 0
self.entry_price = None
self.stop_loss = None
self.take_profit = None
def CreateEntry(self, quantity):
# initial entry market order
self.algorithm.MarketOrder(self.symbol, quantity)
current_price = self.algorithm.Securities[self.symbol].Price
atr = self.algorithm.symbol_data[self.symbol].atr.Current.Value
self.entry_price = current_price
self.stop_loss = self.entry_price + 0.5 * atr
self.take_profit = self.entry_price - 1 * atr
self.algorithm.Debug(f"Entering {self.symbol}...Entry Price: {current_price}, Take Profit: {self.take_profit}, StopLoss: {self.stop_loss}")
def Liquidate(self):
self.algorithm.Debug(f"Liquidating.. {self.symbol}....{self.algorithm.Securities[self.symbol].Price}")
self.algorithm.Liquidate(self.symbol)
self.entry_price = None
self.stop_loss = None
self.take_profit = None
self.days_active = 0class SymbolData:
'''Containers to hold relevant data for each symbol'''
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
# defines consolidator and then registers to receive data
self.daily_consolidator = TradeBarConsolidator(timedelta(days=1))
self.algorithm.SubscriptionManager.AddConsolidator(symbol, self.daily_consolidator)
self.daily_consolidator.DataConsolidated += self.OnDailyBar
# 1. instantiantes a SimpleMovingAverage object
# 2. subscribes it to receive data
self.sma = SimpleMovingAverage(20)
self.algorithm.RegisterIndicator(symbol, self.sma, self.daily_consolidator)
self.atr = AverageTrueRange(1)
self.algorithm.RegisterIndicator(symbol, self.atr, self.daily_consolidator)
# holds recent bars
self.bar_window = RollingWindow[TradeBar](5)
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
trade_bar = TradeBar(time, self.symbol, open, high, low, close, volume)
self.sma.Update(time, close)
self.atr.Update(trade_bar)
self.bar_window.Add(trade_bar)
def OnDailyBar(self, sender, bar):
'''Fires each time our daily_consolidator produces a bar
that bar is passed in through the bar parameter'''
# save that bar to our rolling window
self.bar_window.Add(bar)
def KillConsolidators(self):
self.algorithm.SubscriptionManager.RemoveConsolidator(self.symbol, self.daily_consolidator)
def IsReady(self):
return self.sma.IsReady and self.atr.IsReady and self.bar_window.IsReady