# Allan Lin - EMA (Py) v0.01
# (Note: This was heavily copied from public)
# (Spend a ton of time working on testing and graphing)
import clr
from clr import AddReference
clr.AddReference("System")
clr.AddReference("QuantConnect.Algorithm")
clr.AddReference("QuantConnect.Indicators")
clr.AddReference("QuantConnect.Common")
from System import *
from QuantConnect import *
from QuantConnect.Data import *
from QuantConnect.Algorithm import *
from QuantConnect.Indicators import *
import decimal as d
import numpy as np
class CryptoWarmupMovingAverageCross(QCAlgorithm):
def Initialize(self):
# If Backtesting (Use this data)
self.SetStartDate(2017, 10, 1) # Set Start Date
self.SetEndDate(2017, 12, 12) # Set End Date
self.SetCash(10000) # Set Strategy Cash
# Default Values
fast_period = 12
slow_period = 26
resolutionTime = Resolution.Hour
self.__coin = "BTCUSD"
# Set brokerage we are using: GDAX for cryptos
self.SetBrokerageModel(BrokerageName.GDAX, AccountType.Cash)
# Avoid Fees?
# Ref: https://www.quantconnect.com/forum/discussion/3040/fees-on-limit-orders-on-gdax
DefaultOrderProperties = GDAXOrderProperties()
DefaultOrderProperties.PostOnly = True
# Set crypto to BTC at 1 hour resolution
self.AddCrypto(self.__coin, resolutionTime)
# create a fast exponential moving average at daily resolution
self.fast = self.EMA(self.__coin, fast_period, resolutionTime)
# create a slow exponential moving average at daily resolution
self.slow = self.EMA(self.__coin, slow_period, resolutionTime)
# "slow_period + 1" because rolling window waits for one to fall off the back to be considered ready
# History method returns a dict with a pandas.DataFrame
history = self.History([self.__coin], slow_period + 1)
# prints out the tail of the dataframe
self.Log(str(history.loc[self.__coin].tail()))
self.Log(str(history.loc[self.__coin].head()))
# Populate warmup data
for index, row in history.loc[self.__coin].iterrows():
self.fast.Update(index, row["close"])
self.slow.Update(index, row["close"])
# Log warmup status
self.Log("FAST {0} READY. Samples: {1}".format("IS" if self.fast.IsReady else "IS NOT", self.fast.Samples))
self.Log("SLOW {0} READY. Samples: {1}".format("IS" if self.slow.IsReady else "IS NOT", self.slow.Samples))
# Adding the EMA Plot
emaPlot = Chart("EMA")
emaPlot.AddSeries(Series("SLOW", SeriesType.Line, 0))
emaPlot.AddSeries(Series("FAST", SeriesType.Line, 0))
emaPlot.AddSeries(Series("CROSS", SeriesType.Scatter, 0))
emaPlot.AddSeries(Series("PRICE", SeriesType.Line, 0))
self.AddChart(emaPlot)
usdPlot = Chart("USD")
usdPlot.AddSeries(Series("BUY", SeriesType.Scatter, 0))
usdPlot.AddSeries(Series("SELL", SeriesType.Scatter, 0))
usdPlot.AddSeries(Series("CASH", SeriesType.Line, 0))
self.AddChart(usdPlot)
self.previous = None
def OnData(self, data):
# wait for our slow ema to fully initialize
if not self.slow.IsReady:
return
# only once per Hour (temporary remove)
if self.previous is not None and self.Time == self.previous:
return
# define a small tolerance on our checks to avoid bouncing
tolerance = 0.00015
# Determine number of BTC held
cash = self.Portfolio.Cash
price = self.Securities[self.__coin].Price
holdings = self.Portfolio[self.__coin].Quantity
# Log stats
#self.Log("{0} Holding {1} BTCUSD @ {2}".format(self.Time, holdings, holdings * price))
# Plot Points
self.Plot("EMA", "SLOW", self.slow.Current.Value)
self.Plot("EMA", "FAST", self.fast.Current.Value)
self.Plot("USD", "PRICE", price)
self.Plot("USD", "CASH", cash)
# we only want to go long if we're currently short or flat
if holdings <= 0.0001:
# if the fast is greater than the slow, we'll go long
if self.fast.Current.Value > self.slow.Current.Value * d.Decimal(1 + tolerance):
self.Log("BUY >> {0}".format(price))
self.Plot("USD", "BUY", price)
self.Plot("EMA", "CROSS", price)
# Market Buys = Fees.
#self.SetHoldings(self.__coin, 1)
# Trying to Cut Cost (by doing a limit order)
canBuyNumCoins = float(cash) / float(price)
self.LimitOrder(self.__coin, canBuyNumCoins, price, "buy")
# we only want to liquidate if we're currently long
# if the fast is less than the slow we'll liquidate our long
if holdings > 0.0001 and self.fast.Current.Value < self.slow.Current.Value:
self.Log("SELL >> {0}".format(price))
self.Plot("USD", "SELL", price)
self.Plot("EMA", "CROSS", price)
# We won't use market sells (for same reason).
self.SetHoldings(self.__coin, 0)
self.Liquidate(self.__coin)
#self.Sell(self.__coin, holdings)
#self.Order(self.__coin, holdings, OrderType.StopMarket)
#self.StopLimitOrder(self.__coin, holdings, price, "sell")
self.previous = self.Time