| Overall Statistics |
|
Total Trades 20 Average Win 6.13% Average Loss -2.32% Compounding Annual Return -17.411% Drawdown 22.600% Expectancy -0.160 Net Profit -6.192% Sharpe Ratio -0.307 Probabilistic Sharpe Ratio 19.510% Loss Rate 77% Win Rate 23% Profit-Loss Ratio 2.64 Alpha 0.118 Beta -0.187 Annual Standard Deviation 0.291 Annual Variance 0.085 Information Ratio -1.281 Tracking Error 0.938 Treynor Ratio 0.479 Total Fees $2461.32 Estimated Strategy Capacity $110000000000000.00 Lowest Capacity Asset ETHUSD E3 |
#region imports
from AlgorithmImports import *
#endregion
import pandas as pd
import numpy as np
import math
import datetime
class JumpingBlueSalamander(QCAlgorithm):
def Initialize(self):
# Set Start Date
self.SetStartDate(2022, 1, 15)
# self.SetEndDate(2022, 1, 20)
# Set Strategy Cash
self.SetCash(100000)
# Set brokerage model
self.SetBrokerageModel(BrokerageName.Bitfinex, AccountType.Margin)
# Add Crypto
btc = self.AddCrypto("BTCUSD", Resolution.Hour)
# Set margin
btc.BuyingPowerModel = SecurityMarginModel(3.3)
# Symbol
self.BTC_symbol = btc.Symbol
# Create 4-hour consolidator
four_hour = TradeBarConsolidator(timedelta(hours=4))
# Register "FourHourHandler" to receive 4-hour consolidated bars
four_hour.DataConsolidated += self.FourHourHandler
# Subscribe our 4-hour consolidator object to be automatically updated with 4-hour bars
self.SubscriptionManager.AddConsolidator(self.BTC_symbol, four_hour)
# RSI
self.relative_strength = RelativeStrengthIndex(14)
# Simple moving average
self.simple_moving_average = SimpleMovingAverage(9)
# 2-period EMA
self.ema_two = ExponentialMovingAverage(2)
# 5-period EMA
self.ema_five = ExponentialMovingAverage(5)
# 8-period EMA
self.ema_eight = ExponentialMovingAverage(8)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.relative_strength, four_hour)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.simple_moving_average, four_hour)
# History
history = self.History([self.BTC_symbol], 1000, Resolution.Hour)
# Loc history
history = history.loc[self.BTC_symbol]
# Four-hour bar storage
self.storage = {"open":[],"high":[],"low":[],"close":[],"volume":[]}
# Check
self.check = False
# Loop through history
for time, row in history.iterrows():
# Create tradebar with history data
bar = TradeBar(time, self.BTC_symbol, row.open, row.high, row.low, row.close, row.volume)
# Update 4-hour consolidator
four_hour.Update(bar)
# Check false
self.check = True
# Difference EMA
self.EMA_difference = 0
def OnData(self, data: Slice):
pass
def FourHourHandler(self, sender, bar):
# Four-hour bar storage
self.storage["open"].append(bar.Open)
# Four-hour bar storage
self.storage["high"].append(bar.High)
# Four-hour bar storage
self.storage["low"].append(bar.Low)
# Four-hour bar storage
self.storage["close"].append(bar.Close)
# Four-hour bar storage
self.storage["volume"].append(bar.Volume)
# If more than 100 data points stored
if len(self.storage["close"]) > 100:
# Four-hour bar storage
self.storage["open"] = self.storage["open"][-100:]
# Four-hour bar storage
self.storage["high"] = self.storage["high"][-100:]
# Four-hour bar storage
self.storage["low"] = self.storage["low"][-100:]
# Four-hour bar storage
self.storage["close"] = self.storage["close"][-100:]
# Four-hour bar storage
self.storage["volume"] = self.storage["volume"][-100:]
# Count
count = 0
# Convert storage into dataframe
dataframe = pd.DataFrame(self.storage)
# Fisher transform
fish_value = self.Fisher_Transform_Indicator(dataframe)
# Current fish value
current_fish = fish_value[-1]
# If current fish greater than SMA
if current_fish > self.simple_moving_average.Current.Value:
# Count +1
count += 1
# If current fish greater than previous fish
if current_fish > fish_value[-2]:
# Count +1
count += 1
# Squeeze momentum
count += self.Squeeze_Momentum_Indicator(dataframe)
# Update EMAs
self.ema_two.Update(self.Time, count)
self.ema_five.Update(self.Time, count)
self.ema_eight.Update(self.Time, count)
# If check is True
if self.check == True:
# If EMA difference is 0
if self.EMA_difference == 0:
# Get ema difference
self.EMA_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Else
else:
# Get previous ema difference
previous_ema_difference = self.EMA_difference
# Get current ema difference
current_ema_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Update
self.EMA_difference = current_ema_difference
# If current ema difference is positive and previous is negative
if current_ema_difference > 0 and previous_ema_difference < 0:
# If long
if self.Portfolio[self.BTC_symbol].IsShort:
# Liquidate
self.Liquidate()
# If current close greater than 2-period ema
if bar.Close > self.ema_two.Current.Value:
# If RSI greater than 50
if self.relative_strength.Current.Value > 50:
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Long
self.MarketOrder(self.BTC_symbol, value)
# If current ema difference is negative and previous is positive
if current_ema_difference < 0 and previous_ema_difference > 0:
# If long
if self.Portfolio[self.BTC_symbol].IsLong:
# Liquidate
self.Liquidate()
# If current close greater than 2-period ema
if bar.Close < self.ema_two.Current.Value:
# If RSI greater than 50
if self.relative_strength.Current.Value > 50:
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Short
self.MarketOrder(self.BTC_symbol, -value)
def Fisher_Transform_Indicator(self, dataframe):
# Dataframe
df = dataframe
window = 10
df["minLowPrice"] = df['low'].rolling(window = window).min()
df["maxHighPrice"] = df['high'].rolling(window = window).max()
df["mid_price"] = (df["low"] + df["high"])/2
df["minLowPrice"] = df["minLowPrice"].fillna(0)
df["maxHighPrice"] = df["maxHighPrice"].fillna(0)
diffRatio = 0.33
# diff calculation
x = []
for index, row in df.iterrows():
if row.mid_price == 0 or row.minLowPrice == 0 or row.maxHighPrice == 0:
x.append(0)
else:
diff = (row.mid_price - row.minLowPrice)/(row.maxHighPrice - row.minLowPrice) - 0.5
diff = 2 * diff
diff = diffRatio * diff + (1 - diffRatio) * x[-1]
x.append(diff)
y = []
for i in x:
if i > 0.99:
y.append(0.999)
elif i < -0.99:
y.append(-0.999)
else:
y.append(i)
# Fish calculation
z = []
for i in y:
fish = np.log((1.0 + i)/(1.0 - i))
fish = 0.5 * fish + 0.5 * fish
z.append(fish)
df["fish"] = z
j = z[-2:]
return j
def Squeeze_Momentum_Indicator(self, dataframe):
count = 0
# Dataframe
df = dataframe
# parameter setup
length = 20
mult = 2
length_KC = 20
mult_KC = 1.5
# calculate BB
m_avg = df['close'].rolling(window=length).mean()
m_std = df['close'].rolling(window=length).std(ddof=0)
df['upper_BB'] = m_avg + mult * m_std
df['lower_BB'] = m_avg - mult * m_std
# calculate true range
df['tr0'] = abs(df["high"] - df["low"])
df['tr1'] = abs(df["high"] - df["close"].shift())
df['tr2'] = abs(df["low"] - df["close"].shift())
df['tr'] = df[['tr0', 'tr1', 'tr2']].max(axis=1)
# calculate KC
range_ma = df['tr'].rolling(window=length_KC).mean()
df['upper_KC'] = m_avg + range_ma * mult_KC
df['lower_KC'] = m_avg - range_ma * mult_KC
# calculate bar value
highest = df['high'].rolling(window = length_KC).max()
lowest = df['low'].rolling(window = length_KC).min()
m1 = (highest + lowest)/2
df['value'] = (df['close'] - (m1 + m_avg)/2)
fit_y = np.array(range(0,length_KC))
df['value'] = df['value'].rolling(window = length_KC).apply(lambda x:
np.polyfit(fit_y, x, 1)[0] * (length_KC-1) +
np.polyfit(fit_y, x, 1)[1], raw=True)
# check for 'squeeze'
df['squeeze_on'] = (df['lower_BB'] > df['lower_KC']) & (df['upper_BB'] < df['upper_KC'])
df['squeeze_off'] = (df['lower_BB'] < df['lower_KC']) & (df['upper_BB'] > df['upper_KC'])
# lists
value_list = df["value"].to_list()
squeeze_list = df["squeeze_on"].to_list()
# Count
if value_list[-1] > value_list[-2]:
count += 1
if value_list[-2] > value_list[-3]:
count += 1
if value_list[-3] > value_list[-4]:
count += 1
if value_list[-1] > 0:
if squeeze_list[-1] == True:
count += 0.5
elif value_list[-1] < 0:
if squeeze_list[-1] == True:
count -= 0.5
return count
#region imports
from AlgorithmImports import *
#endregion
import pandas as pd
import numpy as np
import math
import datetime
class JumpingBlueSalamander(QCAlgorithm):
def Initialize(self):
# # # # # # # # # # # # # # # # # # # # User input section # # # # # # # # # # # # # # # # # # # #
# # # General backtest settings
# Set Start Date
self.SetStartDate(2021, 5, 10)
# Set End Date
# self.SetEndDate(2022, 1, 20)
# Set Strategy Cash
self.SetCash(100000)
# # # Indicator periods
# RSI period
self.RSI_period = 14
# Simple moving average period
self.SMA_period = 9
# Average true range period
self.ATR_period = 9
# Fastest EMA period
self.fastest_ema_period = 2
# Medium EMA period
self.mid_ema_period = 5
# Slowest EMA period
self.slow_ema_period = 8
# # # Fisher count parameters # # #
# Current fish greater than SMA count +=
self.current_fish_greater_than_SMA_add_count = 1
# Current fish greater than previous fish value count +=
self.current_fish_greater_than_previous_fish_add_count = 1
# # # Squeeze momentum count parameters # # #
# If current value greater than previous count +=
self.current_greater_than_previous_value = 1
# If previous value greater than one before count +=
self.previous_greater_than_one_before_value = 1
# If value -2 greater than value -3 count +=
self.value_two_greater_than_value_three = 1
# If value greater than 0 and squeeze on count +=
self.value_greater_than_zero_squeeze_on_count = 0.5
# If value less than 0 and squeeze on count -=
self.value_less_than_zero_squeeze_on_count = 0.5
# # # Liquidation percentages
# Trailing stop loss %
self.trailing_stop_loss_percent = 0.1
# Hard stop loss %
self.hard_stop_loss_percent = 5
# # # SMS phone number (Needs to include country code)
self.SMS_phone_number = "+1"
# # # # # # # # # # # # # # # # # # # # End user input section # # # # # # # # # # # # # # # # # # # #
# Set brokerage model
self.SetBrokerageModel(BrokerageName.Bitfinex, AccountType.Margin)
# Add Crypto
btc = self.AddCrypto("BTCUSD", Resolution.Hour)
# Set margin
btc.BuyingPowerModel = SecurityMarginModel(3.3)
# Symbol
self.BTC_symbol = btc.Symbol
# Create 4-hour consolidator
four_hour = TradeBarConsolidator(timedelta(hours=4))
# Register "FourHourHandler" to receive 4-hour consolidated bars
four_hour.DataConsolidated += self.FourHourHandler
# Subscribe our 4-hour consolidator object to be automatically updated with 4-hour bars
self.SubscriptionManager.AddConsolidator(self.BTC_symbol, four_hour)
# RSI
self.relative_strength = RelativeStrengthIndex(self.RSI_period)
# Simple moving average
self.simple_moving_average = SimpleMovingAverage(self.SMA_period)
# ATR
self.average_true_range = AverageTrueRange(self.ATR_period)
# 2-period EMA
self.ema_two = ExponentialMovingAverage(self.fastest_ema_period)
# 5-period EMA
self.ema_five = ExponentialMovingAverage(self.mid_ema_period)
# 8-period EMA
self.ema_eight = ExponentialMovingAverage(self.slow_ema_period)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.relative_strength, four_hour)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.simple_moving_average, four_hour)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.average_true_range, four_hour)
# History
history = self.History([self.BTC_symbol], 1000, Resolution.Hour)
# Loc history
history = history.loc[self.BTC_symbol]
# Four-hour bar storage
self.storage = {"open":[],"high":[],"low":[],"close":[],"volume":[]}
# Check
self.check = False
# Loop through history
for time, row in history.iterrows():
# Create tradebar with history data
bar = TradeBar(time, self.BTC_symbol, row.open, row.high, row.low, row.close, row.volume)
# Update 4-hour consolidator
four_hour.Update(bar)
# Check false
self.check = True
# Difference EMA
self.EMA_difference = 0
# ATR at time order submitted
self.ATR_value_when_order_submitted = None
# Trailing stop loss tracker
self.trailing_stop_loss_tracker = None
# Trailing ATR tracker
self.trailing_ATR_tracker = None
# Just submitted order tracker
self.just_submitted_order = False
def OnData(self, data: Slice):
pass
def FourHourHandler(self, sender, bar):
# Four-hour bar storage
self.storage["open"].append(bar.Open)
# Four-hour bar storage
self.storage["high"].append(bar.High)
# Four-hour bar storage
self.storage["low"].append(bar.Low)
# Four-hour bar storage
self.storage["close"].append(bar.Close)
# Four-hour bar storage
self.storage["volume"].append(bar.Volume)
# If more than 100 data points stored
if len(self.storage["close"]) > 100:
# Four-hour bar storage
self.storage["open"] = self.storage["open"][-100:]
# Four-hour bar storage
self.storage["high"] = self.storage["high"][-100:]
# Four-hour bar storage
self.storage["low"] = self.storage["low"][-100:]
# Four-hour bar storage
self.storage["close"] = self.storage["close"][-100:]
# Four-hour bar storage
self.storage["volume"] = self.storage["volume"][-100:]
# Count
count = 0
# Convert storage into dataframe
dataframe = pd.DataFrame(self.storage)
# Fisher transform
fish_value = self.Fisher_Transform_Indicator(dataframe)
# Current fish value
current_fish = fish_value[-1]
# If current fish greater than SMA
if current_fish > self.simple_moving_average.Current.Value:
# Count
count += self.current_fish_greater_than_SMA_add_count
# If current fish greater than previous fish
if current_fish > fish_value[-2]:
# Count
count += self.current_fish_greater_than_previous_fish_add_count
# Squeeze momentum
count += self.Squeeze_Momentum_Indicator(dataframe)
# Update EMAs
self.ema_two.Update(self.Time, count)
self.ema_five.Update(self.Time, count)
self.ema_eight.Update(self.Time, count)
# If check is True
if self.check == True:
# Just submitted order
self.just_submitted_order = False
# If EMA difference is 0
if self.EMA_difference == 0:
# Get ema difference
self.EMA_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Else
else:
# Get previous ema difference
previous_ema_difference = self.EMA_difference
# Get current ema difference
current_ema_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Update
self.EMA_difference = current_ema_difference
# If current ema difference is positive and previous is negative
if current_ema_difference > 0 and previous_ema_difference < 0:
# If short
if self.Portfolio[self.BTC_symbol].Quantity < -0.1:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because fast EMA crossed above slow EMA")
# If current close greater than 2-period ema
if bar.Close > self.ema_two.Current.Value:
# If RSI greater than 50
if self.relative_strength.Current.Value > 50:
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Long
self.MarketOrder(self.BTC_symbol, value)
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position opened because fast EMA crossed above slow EMA and current price greater than 2-period EMA")
# If current ema difference is negative and previous is positive
if current_ema_difference < 0 and previous_ema_difference > 0:
# If long
if self.Portfolio[self.BTC_symbol].Quantity > 0.1:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because fast EMA crossed below slow EMA")
# # If current close greater than 2-period ema
# if bar.Close < self.ema_two.Current.Value:
# If RSI greater than 50
if self.relative_strength.Current.Value > 50:
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Short
self.MarketOrder(self.BTC_symbol, -value)
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position opened because fast EMA crossed below slow EMA and current price lower than 2-period EMA")
# If long and not just submitted order
if self.Portfolio[self.BTC_symbol].Quantity > 0.1 and self.just_submitted_order == False:
# If long hard stop not triggered
if not self.long_liquidation_hard_stop_logic(bar.Close):
# If long ATR trailing stop not triggered
if not self.long_liquidation_ATR_trailing_stop_logic(bar.Close):
# Run trailing stop loss
self.long_liquidation_trailing_stop_loss(bar.Close)
# If short and not just submitted order
if self.Portfolio[self.BTC_symbol].Quantity < -0.1 and self.just_submitted_order == False:
# If short hard stop not triggered
if not self.short_liquidation_hard_stop_logic(bar.Close):
# If short ATR trailing stop not triggered
if not self.short_liquidation_ATR_trailing_stop_logic(bar.Close):
# Run trailing stop loss
self.short_liquidation_trailing_stop_loss(bar.Close)
def Fisher_Transform_Indicator(self, dataframe):
# Dataframe
df = dataframe
window = 10
df["minLowPrice"] = df['low'].rolling(window = window).min()
df["maxHighPrice"] = df['high'].rolling(window = window).max()
df["mid_price"] = (df["low"] + df["high"])/2
df["minLowPrice"] = df["minLowPrice"].fillna(0)
df["maxHighPrice"] = df["maxHighPrice"].fillna(0)
diffRatio = 0.33
# diff calculation
x = []
for index, row in df.iterrows():
if row.mid_price == 0 or row.minLowPrice == 0 or row.maxHighPrice == 0:
x.append(0)
else:
diff = (row.mid_price - row.minLowPrice)/(row.maxHighPrice - row.minLowPrice) - 0.5
diff = 2 * diff
diff = diffRatio * diff + (1 - diffRatio) * x[-1]
x.append(diff)
y = []
for i in x:
if i > 0.99:
y.append(0.999)
elif i < -0.99:
y.append(-0.999)
else:
y.append(i)
# Fish calculation
z = []
for i in y:
fish = np.log((1.0 + i)/(1.0 - i))
fish = 0.5 * fish + 0.5 * fish
z.append(fish)
df["fish"] = z
j = z[-2:]
return j
def Squeeze_Momentum_Indicator(self, dataframe):
count = 0
# Dataframe
df = dataframe
# parameter setup
length = 20
mult = 2
length_KC = 20
mult_KC = 1.5
# calculate BB
m_avg = df['close'].rolling(window=length).mean()
m_std = df['close'].rolling(window=length).std(ddof=0)
df['upper_BB'] = m_avg + mult * m_std
df['lower_BB'] = m_avg - mult * m_std
# calculate true range
df['tr0'] = abs(df["high"] - df["low"])
df['tr1'] = abs(df["high"] - df["close"].shift())
df['tr2'] = abs(df["low"] - df["close"].shift())
df['tr'] = df[['tr0', 'tr1', 'tr2']].max(axis=1)
# calculate KC
range_ma = df['tr'].rolling(window=length_KC).mean()
df['upper_KC'] = m_avg + range_ma * mult_KC
df['lower_KC'] = m_avg - range_ma * mult_KC
# calculate bar value
highest = df['high'].rolling(window = length_KC).max()
lowest = df['low'].rolling(window = length_KC).min()
m1 = (highest + lowest)/2
df['value'] = (df['close'] - (m1 + m_avg)/2)
fit_y = np.array(range(0,length_KC))
df['value'] = df['value'].rolling(window = length_KC).apply(lambda x:
np.polyfit(fit_y, x, 1)[0] * (length_KC-1) +
np.polyfit(fit_y, x, 1)[1], raw=True)
# check for 'squeeze'
df['squeeze_on'] = (df['lower_BB'] > df['lower_KC']) & (df['upper_BB'] < df['upper_KC'])
df['squeeze_off'] = (df['lower_BB'] < df['lower_KC']) & (df['upper_BB'] > df['upper_KC'])
# lists
value_list = df["value"].to_list()
squeeze_list = df["squeeze_on"].to_list()
# Count
if value_list[-1] > value_list[-2]:
count += self.current_greater_than_previous_value
if value_list[-2] > value_list[-3]:
count += self.previous_greater_than_one_before_value
if value_list[-3] > value_list[-4]:
count += self.value_two_greater_than_value_three
if value_list[-1] > 0:
if squeeze_list[-1] == True:
count += self.value_greater_than_zero_squeeze_on_count
elif value_list[-1] < 0:
if squeeze_list[-1] == True:
count -= self.value_less_than_zero_squeeze_on_count
return count
def long_liquidation_hard_stop_logic(self, close):
# If current price less than stop loss price
if (close / self.Portfolio[self.BTC_symbol].AveragePrice) < (1 - (self.hard_stop_loss_percent * 0.01)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because hard stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def long_liquidation_ATR_trailing_stop_logic(self, close):
# Check if trailing ATR tracker is None
if self.trailing_ATR_tracker is None:
# Update
self.trailing_ATR_tracker = close - (3.5 * self.average_true_range.Current.Value)
# Else
else:
# Check if current value greater than previous
if (close - (3.5 * self.average_true_range.Current.Value)) > self.trailing_ATR_tracker:
# Update
self.trailing_ATR_tracker = close - (3.5 * self.average_true_range.Current.Value)
# If close is lower than trailing ATR tracker
if close < self.trailing_ATR_tracker:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because ATR trailing stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def long_liquidation_trailing_stop_loss(self, close):
# If trailing stop loss not yet initiated
if self.trailing_stop_loss_tracker is None:
# If current price greater than 1.5 * ATR at order open + average price
if close > ((1.5 * self.ATR_value_when_order_submitted) + self.Portfolio[self.BTC_symbol].AveragePrice):
# Initiate trailing stop loss
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price greater than trailing stop loss
if close > self.trailing_stop_loss_tracker:
# Update
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price lower than trailing stop loss tracker by trailing stop loss percent
if (close / self.trailing_stop_loss_tracker) < (1 - (0.01 * self.trailing_stop_loss_percent)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because % trailing stop loss triggered")
def short_liquidation_hard_stop_logic(self, close):
# If current price greater than stop loss price
if (close / self.Portfolio[self.BTC_symbol].AveragePrice) > (1 + (self.hard_stop_loss_percent * 0.01)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because hard stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def short_liquidation_ATR_trailing_stop_logic(self, close):
# Check if trailing ATR tracker is None
if self.trailing_ATR_tracker is None:
# Update
self.trailing_ATR_tracker = close + (3.5 * self.average_true_range.Current.Value)
# Else
else:
# Check if current value less than previous
if (close + (3.5 * self.average_true_range.Current.Value)) < self.trailing_ATR_tracker:
# Update
self.trailing_ATR_tracker = close + (3.5 * self.average_true_range.Current.Value)
# If close is greater than trailing ATR tracker
if close > self.trailing_ATR_tracker:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because ATR trailing stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def short_liquidation_trailing_stop_loss(self, close):
# If trailing stop loss not yet initiated
if self.trailing_stop_loss_tracker is None:
# If current price less than average price - 1.5 * ATR at order open
if close < (self.Portfolio[self.BTC_symbol].AveragePrice - (1.5 * self.ATR_value_when_order_submitted)):
# Initiate trailing stop loss
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price less than trailing stop loss
if close < self.trailing_stop_loss_tracker:
# Update
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price greater than trailing stop loss tracker by trailing stop loss percent
if (close / self.trailing_stop_loss_tracker) > (1 + (0.01 * self.trailing_stop_loss_percent)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity)
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because % trailing stop loss triggered")
#region imports
from AlgorithmImports import *
#endregion
import pandas as pd
import numpy as np
import math
import datetime
class JumpingBlueSalamander(QCAlgorithm):
def Initialize(self):
# # # # # # # # # # # # # # # # # # # # User input section # # # # # # # # # # # # # # # # # # # #
# # # General backtest settings
# Set Start Date
self.SetStartDate(2021, 5, 10)
# Set End Date
# self.SetEndDate(2022, 1, 20)
# Set Strategy Cash
self.SetCash(100000)
# # # Indicator periods
# RSI period
self.RSI_period = int(self.GetParameter("RSI-period"))
# Simple moving average period
self.SMA_period = int(self.GetParameter("SMA-period"))
# Average true range period
self.ATR_period = int(self.GetParameter("ATR-period"))
# Fastest EMA period
self.fastest_ema_period = int(self.GetParameter("fastest-ema"))
# Medium EMA period
self.mid_ema_period = int(self.GetParameter("mid-ema"))
# Slowest EMA period
self.slow_ema_period = int(self.GetParameter("slow-ema"))
# # # Fisher count parameters # # #
# Current fish greater than SMA count +=
self.current_fish_greater_than_SMA_add_count = float(self.GetParameter("fisher-one"))
# Current fish greater than previous fish value count +=
self.current_fish_greater_than_previous_fish_add_count = float(self.GetParameter("fisher-two"))
# # # Squeeze momentum count parameters # # #
# If current value greater than previous count +=
self.current_greater_than_previous_value = float(self.GetParameter("squeeze-one"))
# If previous value greater than one before count +=
self.previous_greater_than_one_before_value = float(self.GetParameter("squeeze-two"))
# If value -2 greater than value -3 count +=
self.value_two_greater_than_value_three = float(self.GetParameter("squeeze-three"))
# If value greater than 0 and squeeze on count +=
self.value_greater_than_zero_squeeze_on_count = float(self.GetParameter("squeeze-four"))
# If value less than 0 and squeeze on count -=
self.value_less_than_zero_squeeze_on_count = float(self.GetParameter("squeeze-five"))
# # # Liquidation percentages
# Trailing stop loss %
self.trailing_stop_loss_percent = float(self.GetParameter("trailing-stop-loss-percent"))
# Hard stop loss %
self.hard_stop_loss_percent = float(self.GetParameter("hard-stop-percent"))
# # # SMS phone number (Needs to include country code)
self.SMS_phone_number = "+1"
# # # # # # # # # # # # # # # # # # # # End user input section # # # # # # # # # # # # # # # # # # # #
# Set brokerage model
self.SetBrokerageModel(BrokerageName.Bitfinex, AccountType.Margin)
# Add Crypto
btc = self.AddCrypto("BTCUSD", Resolution.Hour)
# Set margin
btc.BuyingPowerModel = SecurityMarginModel(3.3)
# Symbol
self.BTC_symbol = btc.Symbol
# Create 4-hour consolidator
four_hour = TradeBarConsolidator(timedelta(hours=4))
# Register "FourHourHandler" to receive 4-hour consolidated bars
four_hour.DataConsolidated += self.FourHourHandler
# Subscribe our 4-hour consolidator object to be automatically updated with 4-hour bars
self.SubscriptionManager.AddConsolidator(self.BTC_symbol, four_hour)
# RSI
self.relative_strength = RelativeStrengthIndex(self.RSI_period)
# Simple moving average
self.simple_moving_average = SimpleMovingAverage(self.SMA_period)
# ATR
self.average_true_range = AverageTrueRange(self.ATR_period)
# 2-period EMA
self.ema_two = ExponentialMovingAverage(self.fastest_ema_period)
# 5-period EMA
self.ema_five = ExponentialMovingAverage(self.mid_ema_period)
# 8-period EMA
self.ema_eight = ExponentialMovingAverage(self.slow_ema_period)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.relative_strength, four_hour)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.simple_moving_average, four_hour)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.average_true_range, four_hour)
# History
history = self.History([self.BTC_symbol], 1000, Resolution.Hour)
# Loc history
history = history.loc[self.BTC_symbol]
# Four-hour bar storage
self.storage = {"open":[],"high":[],"low":[],"close":[],"volume":[]}
# Check
self.check = False
# Loop through history
for time, row in history.iterrows():
# Create tradebar with history data
bar = TradeBar(time, self.BTC_symbol, row.open, row.high, row.low, row.close, row.volume)
# Update 4-hour consolidator
four_hour.Update(bar)
# Check false
self.check = True
# Difference EMA
self.EMA_difference = 0
# ATR at time order submitted
self.ATR_value_when_order_submitted = None
# Trailing stop loss tracker
self.trailing_stop_loss_tracker = None
# Trailing ATR tracker
self.trailing_ATR_tracker = None
# Just submitted order tracker
self.just_submitted_order = False
def OnData(self, data: Slice):
pass
def FourHourHandler(self, sender, bar):
# Four-hour bar storage
self.storage["open"].append(bar.Open)
# Four-hour bar storage
self.storage["high"].append(bar.High)
# Four-hour bar storage
self.storage["low"].append(bar.Low)
# Four-hour bar storage
self.storage["close"].append(bar.Close)
# Four-hour bar storage
self.storage["volume"].append(bar.Volume)
# If more than 100 data points stored
if len(self.storage["close"]) > 100:
# Four-hour bar storage
self.storage["open"] = self.storage["open"][-100:]
# Four-hour bar storage
self.storage["high"] = self.storage["high"][-100:]
# Four-hour bar storage
self.storage["low"] = self.storage["low"][-100:]
# Four-hour bar storage
self.storage["close"] = self.storage["close"][-100:]
# Four-hour bar storage
self.storage["volume"] = self.storage["volume"][-100:]
# Count
count = 0
# Convert storage into dataframe
dataframe = pd.DataFrame(self.storage)
# Fisher transform
fish_value = self.Fisher_Transform_Indicator(dataframe)
# Current fish value
current_fish = fish_value[-1]
# If current fish greater than SMA
if current_fish > self.simple_moving_average.Current.Value:
# Count
count += self.current_fish_greater_than_SMA_add_count
# If current fish greater than previous fish
if current_fish > fish_value[-2]:
# Count
count += self.current_fish_greater_than_previous_fish_add_count
# Squeeze momentum
count += self.Squeeze_Momentum_Indicator(dataframe)
# Update EMAs
self.ema_two.Update(self.Time, count)
self.ema_five.Update(self.Time, count)
self.ema_eight.Update(self.Time, count)
# If check is True
if self.check == True:
# Just submitted order
self.just_submitted_order = False
# If EMA difference is 0
if self.EMA_difference == 0:
# Get ema difference
self.EMA_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Else
else:
# Get previous ema difference
previous_ema_difference = self.EMA_difference
# Get current ema difference
current_ema_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Update
self.EMA_difference = current_ema_difference
# If current ema difference is positive and previous is negative
if current_ema_difference > 0 and previous_ema_difference < 0:
# If short
if self.Portfolio[self.BTC_symbol].Quantity < -0.1:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short position EMA liquidated")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because fast EMA crossed above slow EMA")
# If current close greater than 2-period ema
if bar.Close > self.ema_two.Current.Value:
# If RSI greater than 50
if self.relative_strength.Current.Value > 50:
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Long
self.MarketOrder(self.BTC_symbol, value, tag = "Long position EMA opened")
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position opened because fast EMA crossed above slow EMA and current price greater than 2-period EMA")
# If current ema difference is negative and previous is positive
if current_ema_difference < 0 and previous_ema_difference > 0:
# If long
if self.Portfolio[self.BTC_symbol].Quantity > 0.1:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long position EMA liquidated")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because fast EMA crossed below slow EMA")
# # If current close greater than 2-period ema
# if bar.Close < self.ema_two.Current.Value:
# If RSI greater than 50
if self.relative_strength.Current.Value > 50:
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Short
self.MarketOrder(self.BTC_symbol, -value, tag = "Short position EMA opened")
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position opened because fast EMA crossed below slow EMA and current price lower than 2-period EMA")
# If long and not just submitted order
if self.Portfolio[self.BTC_symbol].Quantity > 0.1 and self.just_submitted_order == False:
# If long hard stop not triggered
if not self.long_liquidation_hard_stop_logic(bar.Close):
# If long ATR trailing stop not triggered
if not self.long_liquidation_ATR_trailing_stop_logic(bar.Close):
# Run trailing stop loss
self.long_liquidation_trailing_stop_loss(bar.Close)
# If short and not just submitted order
if self.Portfolio[self.BTC_symbol].Quantity < -0.1 and self.just_submitted_order == False:
# If short hard stop not triggered
if not self.short_liquidation_hard_stop_logic(bar.Close):
# If short ATR trailing stop not triggered
if not self.short_liquidation_ATR_trailing_stop_logic(bar.Close):
# Run trailing stop loss
self.short_liquidation_trailing_stop_loss(bar.Close)
def Fisher_Transform_Indicator(self, dataframe):
# Dataframe
df = dataframe
window = 10
df["minLowPrice"] = df['low'].rolling(window = window).min()
df["maxHighPrice"] = df['high'].rolling(window = window).max()
df["mid_price"] = (df["low"] + df["high"])/2
df["minLowPrice"] = df["minLowPrice"].fillna(0)
df["maxHighPrice"] = df["maxHighPrice"].fillna(0)
diffRatio = 0.33
# diff calculation
x = []
for index, row in df.iterrows():
if row.mid_price == 0 or row.minLowPrice == 0 or row.maxHighPrice == 0:
x.append(0)
else:
diff = (row.mid_price - row.minLowPrice)/(row.maxHighPrice - row.minLowPrice) - 0.5
diff = 2 * diff
diff = diffRatio * diff + (1 - diffRatio) * x[-1]
x.append(diff)
y = []
for i in x:
if i > 0.99:
y.append(0.999)
elif i < -0.99:
y.append(-0.999)
else:
y.append(i)
# Fish calculation
z = []
for i in y:
fish = np.log((1.0 + i)/(1.0 - i))
fish = 0.5 * fish + 0.5 * fish
z.append(fish)
df["fish"] = z
j = z[-2:]
return j
def Squeeze_Momentum_Indicator(self, dataframe):
count = 0
# Dataframe
df = dataframe
# parameter setup
length = 20
mult = 2
length_KC = 20
mult_KC = 1.5
# calculate BB
m_avg = df['close'].rolling(window=length).mean()
m_std = df['close'].rolling(window=length).std(ddof=0)
df['upper_BB'] = m_avg + mult * m_std
df['lower_BB'] = m_avg - mult * m_std
# calculate true range
df['tr0'] = abs(df["high"] - df["low"])
df['tr1'] = abs(df["high"] - df["close"].shift())
df['tr2'] = abs(df["low"] - df["close"].shift())
df['tr'] = df[['tr0', 'tr1', 'tr2']].max(axis=1)
# calculate KC
range_ma = df['tr'].rolling(window=length_KC).mean()
df['upper_KC'] = m_avg + range_ma * mult_KC
df['lower_KC'] = m_avg - range_ma * mult_KC
# calculate bar value
highest = df['high'].rolling(window = length_KC).max()
lowest = df['low'].rolling(window = length_KC).min()
m1 = (highest + lowest)/2
df['value'] = (df['close'] - (m1 + m_avg)/2)
fit_y = np.array(range(0,length_KC))
df['value'] = df['value'].rolling(window = length_KC).apply(lambda x:
np.polyfit(fit_y, x, 1)[0] * (length_KC-1) +
np.polyfit(fit_y, x, 1)[1], raw=True)
# check for 'squeeze'
df['squeeze_on'] = (df['lower_BB'] > df['lower_KC']) & (df['upper_BB'] < df['upper_KC'])
df['squeeze_off'] = (df['lower_BB'] < df['lower_KC']) & (df['upper_BB'] > df['upper_KC'])
# lists
value_list = df["value"].to_list()
squeeze_list = df["squeeze_on"].to_list()
# Count
if value_list[-1] > value_list[-2]:
count += self.current_greater_than_previous_value
if value_list[-2] > value_list[-3]:
count += self.previous_greater_than_one_before_value
if value_list[-3] > value_list[-4]:
count += self.value_two_greater_than_value_three
if value_list[-1] > 0:
if squeeze_list[-1] == True:
count += self.value_greater_than_zero_squeeze_on_count
elif value_list[-1] < 0:
if squeeze_list[-1] == True:
count -= self.value_less_than_zero_squeeze_on_count
return count
def long_liquidation_hard_stop_logic(self, close):
# If current price less than stop loss price
if (close / self.Portfolio[self.BTC_symbol].AveragePrice) < (1 - (self.hard_stop_loss_percent * 0.01)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long hard stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because hard stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def long_liquidation_ATR_trailing_stop_logic(self, close):
# Check if trailing ATR tracker is None
if self.trailing_ATR_tracker is None:
# Update
self.trailing_ATR_tracker = close - (3.5 * self.average_true_range.Current.Value)
# Else
else:
# Check if current value greater than previous
if (close - (3.5 * self.average_true_range.Current.Value)) > self.trailing_ATR_tracker:
# Update
self.trailing_ATR_tracker = close - (3.5 * self.average_true_range.Current.Value)
# If close is lower than trailing ATR tracker
if close < self.trailing_ATR_tracker:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long ATR trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because ATR trailing stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def long_liquidation_trailing_stop_loss(self, close):
# If trailing stop loss not yet initiated
if self.trailing_stop_loss_tracker is None:
# If current price greater than 1.5 * ATR at order open + average price
if close > ((1.5 * self.ATR_value_when_order_submitted) + self.Portfolio[self.BTC_symbol].AveragePrice):
# Initiate trailing stop loss
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price greater than trailing stop loss
if close > self.trailing_stop_loss_tracker:
# Update
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price lower than trailing stop loss tracker by trailing stop loss percent
if (close / self.trailing_stop_loss_tracker) < (1 - (0.01 * self.trailing_stop_loss_percent)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because % trailing stop loss triggered")
def short_liquidation_hard_stop_logic(self, close):
# If current price greater than stop loss price
if (close / self.Portfolio[self.BTC_symbol].AveragePrice) > (1 + (self.hard_stop_loss_percent * 0.01)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short hard stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because hard stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def short_liquidation_ATR_trailing_stop_logic(self, close):
# Check if trailing ATR tracker is None
if self.trailing_ATR_tracker is None:
# Update
self.trailing_ATR_tracker = close + (3.5 * self.average_true_range.Current.Value)
# Else
else:
# Check if current value less than previous
if (close + (3.5 * self.average_true_range.Current.Value)) < self.trailing_ATR_tracker:
# Update
self.trailing_ATR_tracker = close + (3.5 * self.average_true_range.Current.Value)
# If close is greater than trailing ATR tracker
if close > self.trailing_ATR_tracker:
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short ATR trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because ATR trailing stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def short_liquidation_trailing_stop_loss(self, close):
# If trailing stop loss not yet initiated
if self.trailing_stop_loss_tracker is None:
# If current price less than average price - 1.5 * ATR at order open
if close < (self.Portfolio[self.BTC_symbol].AveragePrice - (1.5 * self.ATR_value_when_order_submitted)):
# Initiate trailing stop loss
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price less than trailing stop loss
if close < self.trailing_stop_loss_tracker:
# Update
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price greater than trailing stop loss tracker by trailing stop loss percent
if (close / self.trailing_stop_loss_tracker) > (1 + (0.01 * self.trailing_stop_loss_percent)):
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because % trailing stop loss triggered")
#region imports
from AlgorithmImports import *
#endregion
import pandas as pd
import numpy as np
import math
import datetime
class JumpingBlueSalamander(QCAlgorithm):
def Initialize(self):
# # # # # # # # # # # # # # # # # # # # User input section # # # # # # # # # # # # # # # # # # # #
# # # General backtest settings
# Set Start Date
self.SetStartDate(2020, 1, 1)
# Set End Date
self.SetEndDate(2020, 5, 1)
# Set Strategy Cash
self.SetCash(100000)
# # # Indicator periods
# RSI period
self.RSI_period = int(self.GetParameter("RSI-period"))
# Simple moving average period
self.SMA_period = int(self.GetParameter("SMA-period"))
# Average true range period
self.ATR_period = int(self.GetParameter("ATR-period"))
# Fastest EMA period
self.fastest_ema_period = int(self.GetParameter("fastest-ema"))
# Medium EMA period
self.mid_ema_period = int(self.GetParameter("mid-ema"))
# Slowest EMA period
self.slow_ema_period = int(self.GetParameter("slow-ema"))
# # # Fisher count parameters # # #
# Current fish greater than SMA count +=
self.current_fish_greater_than_SMA_add_count = float(self.GetParameter("fisher-one"))
# Current fish greater than previous fish value count +=
self.current_fish_greater_than_previous_fish_add_count = float(self.GetParameter("fisher-two"))
# # # Squeeze momentum count parameters # # #
# If current value greater than previous count +=
self.current_greater_than_previous_value = float(self.GetParameter("squeeze-one"))
# If previous value greater than one before count +=
self.previous_greater_than_one_before_value = float(self.GetParameter("squeeze-two"))
# If value -2 greater than value -3 count +=
self.value_two_greater_than_value_three = float(self.GetParameter("squeeze-three"))
# If value greater than 0 and squeeze on count +=
self.value_greater_than_zero_squeeze_on_count = float(self.GetParameter("squeeze-four"))
# If value less than 0 and squeeze on count -=
self.value_less_than_zero_squeeze_on_count = float(self.GetParameter("squeeze-five"))
# # # Liquidation percentages
# Trailing stop loss %
self.trailing_stop_loss_percent = float(self.GetParameter("trailing-stop-loss-percent"))
# Hard stop loss %
self.hard_stop_loss_percent = float(self.GetParameter("hard-stop-percent"))
# # # SMS phone number (Needs to include country code)
self.SMS_phone_number = "+1"
# # # # # # # # # # # # # # # # # # # # End user input section # # # # # # # # # # # # # # # # # # # #
# Set brokerage model
self.SetBrokerageModel(BrokerageName.Bitfinex, AccountType.Margin)
# Add Crypto
btc = self.AddCrypto("ETHUSD", Resolution.Minute)
# Set margin
btc.BuyingPowerModel = SecurityMarginModel(3.3)
# Symbol
self.BTC_symbol = btc.Symbol
# Create 4-hour consolidator
four_hour = TradeBarConsolidator(timedelta(hours=4))
# Register "FourHourHandler" to receive 4-hour consolidated bars
four_hour.DataConsolidated += self.FourHourHandler
# Subscribe our 4-hour consolidator object to be automatically updated with 4-hour bars
self.SubscriptionManager.AddConsolidator(self.BTC_symbol, four_hour)
# RSI
self.relative_strength = RelativeStrengthIndex(self.RSI_period)
# Simple moving average
self.simple_moving_average = SimpleMovingAverage(self.SMA_period)
# ATR
self.average_true_range = AverageTrueRange(self.ATR_period)
# 2-period EMA
self.ema_two = ExponentialMovingAverage(self.fastest_ema_period)
# 5-period EMA
self.ema_five = ExponentialMovingAverage(self.mid_ema_period)
# 8-period EMA
self.ema_eight = ExponentialMovingAverage(self.slow_ema_period)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.relative_strength, four_hour)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.simple_moving_average, four_hour)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.average_true_range, four_hour)
# Four-hour bar storage
self.storage = {"open":[],"high":[],"low":[],"close":[],"volume":[]}
# Difference EMA
self.EMA_difference = 0
# ATR at time order submitted
self.ATR_value_when_order_submitted = None
# Trailing stop loss tracker
self.trailing_stop_loss_tracker = None
# Trailing ATR tracker
self.trailing_ATR_tracker = None
# Just submitted order tracker
self.just_submitted_order = False
# # # # # Additions 8/21/2022 # # # # #
# Moving average
self.moving_average_rsi = SimpleMovingAverage(10)
# Create 2-day consolidator
two_day = TradeBarConsolidator(timedelta(days=2))
# Register "TwoDayHandler" to receive 2-day consolidated bars
two_day.DataConsolidated += self.TwoDayHandler
# Subscribe our 2-day consolidator object to be automatically updated with 2-day bars
self.SubscriptionManager.AddConsolidator(self.BTC_symbol, two_day)
# ADX
self.ADX_indicator_daily = AverageDirectionalIndex(14)
# RSI
self.RSI_indicator_daily = RelativeStrengthIndex(14)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.ADX_indicator_daily, two_day)
# Register indicator
self.RegisterIndicator(self.BTC_symbol, self.RSI_indicator_daily, two_day)
# RSI daily storage
self.RSI_daily_storage = []
# Market judgement
self.market_judgement = "None"
# Create 10-minute consolidator
ten_minute = TradeBarConsolidator(timedelta(minutes=10))
# Register "TenMinuteHandler" to receive 10-minute consolidated bars
ten_minute.DataConsolidated += self.TenMinuteHandler
# Subscribe our 10-minute consolidator object to be automatically updated with 10-minute bars
self.SubscriptionManager.AddConsolidator(self.BTC_symbol, ten_minute)
# Warmup
self.SetWarmUp(timedelta(days = 150))
def OnData(self, data: Slice):
pass
def TenMinuteHandler(self, sender, bar):
# If not warming up
if not self.IsWarmingUp:
# If long and not just submitted order
if self.Portfolio[self.BTC_symbol].Quantity > 0.1 and self.just_submitted_order == False and self.ATR_value_when_order_submitted is not None:
# If long hard stop not triggered
if not self.long_liquidation_hard_stop_logic(bar.Close):
# If long ATR trailing stop not triggered
if not self.long_liquidation_ATR_trailing_stop_logic(bar.Close):
# Run trailing stop loss
self.long_liquidation_trailing_stop_loss(bar.Close)
# If short and not just submitted order
if self.Portfolio[self.BTC_symbol].Quantity < -0.1 and self.just_submitted_order == False and self.ATR_value_when_order_submitted is not None:
# If short hard stop not triggered
if not self.short_liquidation_hard_stop_logic(bar.Close):
# If short ATR trailing stop not triggered
if not self.short_liquidation_ATR_trailing_stop_logic(bar.Close):
# Run trailing stop loss
self.short_liquidation_trailing_stop_loss(bar.Close)
def TwoDayHandler(self, sender, bar):
# If RSI daily is ready
if self.RSI_indicator_daily.IsReady:
# Update RSI daily storage
self.RSI_daily_storage.append(self.RSI_indicator_daily.Current.Value)
# If stored more than 14
if len(self.RSI_daily_storage) > 14:
# Cut
self.RSI_daily_storage = self.RSI_daily_storage[-14:]
# If ADX is ready
if self.ADX_indicator_daily.IsReady:
# Get minimum of RSI daliy storage
minimum_RSI = min(self.RSI_daily_storage)
# Get maximum of RSI daily storage
maximum_RSI = max(self.RSI_daily_storage)
# Bull
if (
# If ADX above 30
self.ADX_indicator_daily.Current.Value > 30 and
# If DI+ greater than DI-
self.ADX_indicator_daily.PositiveDirectionalIndex.Current.Value > self.ADX_indicator_daily.NegativeDirectionalIndex.Current.Value and
# If minimum RSI below 40
minimum_RSI < 40 and
# If Current RSI above 40
self.RSI_indicator_daily.Current.Value > 40
):
# Judgement
self.market_judgement = "bull"
# Bear
elif (
# If ADX above 30
self.ADX_indicator_daily.Current.Value > 30 and
# If DI+ less than DI-
self.ADX_indicator_daily.PositiveDirectionalIndex.Current.Value < self.ADX_indicator_daily.NegativeDirectionalIndex.Current.Value and
# If maximum RSI above 60
maximum_RSI > 60 and
# If Current RSI below 60
self.RSI_indicator_daily.Current.Value < 60
):
# Judgement
self.market_judgement = "bear"
# Neutral bull
elif (
# If ADX below 30
self.ADX_indicator_daily.Current.Value < 30 and
# If DI+ greater than DI-
self.ADX_indicator_daily.PositiveDirectionalIndex.Current.Value > self.ADX_indicator_daily.NegativeDirectionalIndex.Current.Value and
# If minimum RSI below 30
minimum_RSI < 30 and
# If Current RSI above 30
self.RSI_indicator_daily.Current.Value > 30
):
# Judgement
self.market_judgement = "neutral bull"
# Neutral bear
elif (
# If ADX below 30
self.ADX_indicator_daily.Current.Value < 30 and
# If DI+ less than DI-
self.ADX_indicator_daily.PositiveDirectionalIndex.Current.Value < self.ADX_indicator_daily.NegativeDirectionalIndex.Current.Value and
# If maximum RSI above 70
maximum_RSI > 70 and
# If Current RSI below 70
self.RSI_indicator_daily.Current.Value < 70
):
# Judgement
self.market_judgement = "neutral bear"
# Else
else:
# Judgement is None
self.market_judgement = "None"
def FourHourHandler(self, sender, bar):
# Four-hour bar storage
self.storage["open"].append(bar.Open)
# Four-hour bar storage
self.storage["high"].append(bar.High)
# Four-hour bar storage
self.storage["low"].append(bar.Low)
# Four-hour bar storage
self.storage["close"].append(bar.Close)
# Four-hour bar storage
self.storage["volume"].append(bar.Volume)
# If more than 100 data points stored
if len(self.storage["close"]) > 100:
# Four-hour bar storage
self.storage["open"] = self.storage["open"][-100:]
# Four-hour bar storage
self.storage["high"] = self.storage["high"][-100:]
# Four-hour bar storage
self.storage["low"] = self.storage["low"][-100:]
# Four-hour bar storage
self.storage["close"] = self.storage["close"][-100:]
# Four-hour bar storage
self.storage["volume"] = self.storage["volume"][-100:]
# Count
count = 0
# Convert storage into dataframe
dataframe = pd.DataFrame(self.storage)
# Fisher transform
fish_value = self.Fisher_Transform_Indicator(dataframe)
# Current fish value
current_fish = fish_value[-1]
# If current fish greater than SMA
if current_fish > self.simple_moving_average.Current.Value:
# Count
count += self.current_fish_greater_than_SMA_add_count
# If current fish greater than previous fish
if current_fish > fish_value[-2]:
# Count
count += self.current_fish_greater_than_previous_fish_add_count
# Squeeze momentum
count += self.Squeeze_Momentum_Indicator(dataframe)
# Update EMAs
self.ema_two.Update(self.Time, count)
self.ema_five.Update(self.Time, count)
self.ema_eight.Update(self.Time, count)
# If RSI is ready
if self.relative_strength.IsReady:
# Update moving average
self.moving_average_rsi.Update(self.Time, self.relative_strength.Current.Value)
# If not warming up
if not self.IsWarmingUp:
# Just submitted order
self.just_submitted_order = False
# If EMA difference is 0
if self.EMA_difference == 0:
# Get ema difference
self.EMA_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Else
else:
# Get previous ema difference
previous_ema_difference = self.EMA_difference
# Get current ema difference
current_ema_difference = self.ema_five.Current.Value - self.ema_eight.Current.Value
# Update
self.EMA_difference = current_ema_difference
# If current ema difference is positive and previous is negative
if current_ema_difference > 0 and previous_ema_difference < 0:
# If short
if self.Portfolio[self.BTC_symbol].Quantity < -0.1:
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short position EMA liquidated")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because fast EMA crossed above slow EMA")
# If current close greater than 2-period ema
if bar.Close > self.ema_two.Current.Value:
# If RSI greater than 50
if self.relative_strength.Current.Value > 50:
# If RSI greater than 10-period MA of RSI
if self.relative_strength.Current.Value > self.moving_average_rsi.Current.Value:
# If market judgement is bull
if self.market_judgement == "bull":
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Up, weight = 1)
# Emit insight
self.EmitInsights(insight)
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Long
self.MarketOrder(self.BTC_symbol, value, tag = "Bull position opened")
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Bull position opened")
# If market judgement is neutral bull
elif self.market_judgement == "neutral bull":
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Up, weight = 0.5)
# Emit insight
self.EmitInsights(insight)
# Value of order
value = (self.Portfolio.TotalPortfolioValue * 0.5) / bar.Close
# Long
self.MarketOrder(self.BTC_symbol, value, tag = "Neutral bull position opened")
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Neutral bull position opened")
# If current ema difference is negative and previous is positive
elif current_ema_difference < 0 and previous_ema_difference > 0:
# If long
if self.Portfolio[self.BTC_symbol].Quantity > 0.1:
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long position EMA liquidated")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because fast EMA crossed below slow EMA")
# If RSI below than 50
if self.relative_strength.Current.Value < 50:
# If RSI below 10-period MA of RSI
if self.relative_strength.Current.Value < self.moving_average_rsi.Current.Value:
# If market judgement is bear
if self.market_judgement == "bear":
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Down, weight = -1)
# Emit insight
self.EmitInsights(insight)
# Value of order
value = self.Portfolio.TotalPortfolioValue / bar.Close
# Short
self.MarketOrder(self.BTC_symbol, -value, tag = "Bear position opened")
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Bear position opened")
# If market judgement is neutral bear
elif self.market_judgement == "neutral bear":
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Down, weight = -0.5)
# Emit insight
self.EmitInsights(insight)
# Value of order
value = (self.Portfolio.TotalPortfolioValue * 0.5) / bar.Close
# Short
self.MarketOrder(self.BTC_symbol, -value, tag = "Neutral bear position opened")
# Update ATR
self.ATR_value_when_order_submitted = self.average_true_range.Current.Value
# Just submitted order
self.just_submitted_order = True
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Neutral bear position opened")
def Fisher_Transform_Indicator(self, dataframe):
# Dataframe
df = dataframe
window = 10
df["minLowPrice"] = df['low'].rolling(window = window).min()
df["maxHighPrice"] = df['high'].rolling(window = window).max()
df["mid_price"] = (df["low"] + df["high"])/2
df["minLowPrice"] = df["minLowPrice"].fillna(0)
df["maxHighPrice"] = df["maxHighPrice"].fillna(0)
diffRatio = 0.33
# diff calculation
x = []
for index, row in df.iterrows():
if row.mid_price == 0 or row.minLowPrice == 0 or row.maxHighPrice == 0:
x.append(0)
else:
diff = (row.mid_price - row.minLowPrice)/(row.maxHighPrice - row.minLowPrice) - 0.5
diff = 2 * diff
diff = diffRatio * diff + (1 - diffRatio) * x[-1]
x.append(diff)
y = []
for i in x:
if i > 0.99:
y.append(0.999)
elif i < -0.99:
y.append(-0.999)
else:
y.append(i)
# Fish calculation
z = []
for i in y:
fish = np.log((1.0 + i)/(1.0 - i))
fish = 0.5 * fish + 0.5 * fish
z.append(fish)
df["fish"] = z
j = z[-2:]
return j
def Squeeze_Momentum_Indicator(self, dataframe):
count = 0
# Dataframe
df = dataframe
# parameter setup
length = 20
mult = 2
length_KC = 20
mult_KC = 1.5
# calculate BB
m_avg = df['close'].rolling(window=length).mean()
m_std = df['close'].rolling(window=length).std(ddof=0)
df['upper_BB'] = m_avg + mult * m_std
df['lower_BB'] = m_avg - mult * m_std
# calculate true range
df['tr0'] = abs(df["high"] - df["low"])
df['tr1'] = abs(df["high"] - df["close"].shift())
df['tr2'] = abs(df["low"] - df["close"].shift())
df['tr'] = df[['tr0', 'tr1', 'tr2']].max(axis=1)
# calculate KC
range_ma = df['tr'].rolling(window=length_KC).mean()
df['upper_KC'] = m_avg + range_ma * mult_KC
df['lower_KC'] = m_avg - range_ma * mult_KC
# calculate bar value
highest = df['high'].rolling(window = length_KC).max()
lowest = df['low'].rolling(window = length_KC).min()
m1 = (highest + lowest)/2
df['value'] = (df['close'] - (m1 + m_avg)/2)
fit_y = np.array(range(0,length_KC))
df['value'] = df['value'].rolling(window = length_KC).apply(lambda x:
np.polyfit(fit_y, x, 1)[0] * (length_KC-1) +
np.polyfit(fit_y, x, 1)[1], raw=True)
# check for 'squeeze'
df['squeeze_on'] = (df['lower_BB'] > df['lower_KC']) & (df['upper_BB'] < df['upper_KC'])
df['squeeze_off'] = (df['lower_BB'] < df['lower_KC']) & (df['upper_BB'] > df['upper_KC'])
# lists
value_list = df["value"].to_list()
squeeze_list = df["squeeze_on"].to_list()
# Count
if value_list[-1] > value_list[-2]:
count += self.current_greater_than_previous_value
if value_list[-2] > value_list[-3]:
count += self.previous_greater_than_one_before_value
if value_list[-3] > value_list[-4]:
count += self.value_two_greater_than_value_three
if value_list[-1] > 0:
if squeeze_list[-1] == True:
count += self.value_greater_than_zero_squeeze_on_count
elif value_list[-1] < 0:
if squeeze_list[-1] == True:
count -= self.value_less_than_zero_squeeze_on_count
return count
def long_liquidation_hard_stop_logic(self, close):
# If current price less than stop loss price
if (close / self.Portfolio[self.BTC_symbol].AveragePrice) < (1 - (self.hard_stop_loss_percent * 0.01)):
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long hard stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because hard stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def long_liquidation_ATR_trailing_stop_logic(self, close):
# Check if trailing ATR tracker is None
if self.trailing_ATR_tracker is None:
# Update
self.trailing_ATR_tracker = close - (3.5 * self.average_true_range.Current.Value)
# Else
else:
# Check if current value greater than previous
if (close - (3.5 * self.average_true_range.Current.Value)) > self.trailing_ATR_tracker:
# Update
self.trailing_ATR_tracker = close - (3.5 * self.average_true_range.Current.Value)
# If close is lower than trailing ATR tracker
if close < self.trailing_ATR_tracker:
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long ATR trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because ATR trailing stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def long_liquidation_trailing_stop_loss(self, close):
# If trailing stop loss not yet initiated
if self.trailing_stop_loss_tracker is None:
# If current price greater than 1.5 * ATR at order open + average price
if close > ((1.5 * self.ATR_value_when_order_submitted) + self.Portfolio[self.BTC_symbol].AveragePrice):
# Initiate trailing stop loss
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price greater than trailing stop loss
if close > self.trailing_stop_loss_tracker:
# Update
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price lower than trailing stop loss tracker by trailing stop loss percent
if (close / self.trailing_stop_loss_tracker) < (1 - (0.01 * self.trailing_stop_loss_percent)):
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Long trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Long position liquidated because % trailing stop loss triggered")
def short_liquidation_hard_stop_logic(self, close):
# If current price greater than stop loss price
if (close / self.Portfolio[self.BTC_symbol].AveragePrice) > (1 + (self.hard_stop_loss_percent * 0.01)):
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short hard stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because hard stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def short_liquidation_ATR_trailing_stop_logic(self, close):
# Check if trailing ATR tracker is None
if self.trailing_ATR_tracker is None:
# Update
self.trailing_ATR_tracker = close + (3.5 * self.average_true_range.Current.Value)
# Else
else:
# Check if current value less than previous
if (close + (3.5 * self.average_true_range.Current.Value)) < self.trailing_ATR_tracker:
# Update
self.trailing_ATR_tracker = close + (3.5 * self.average_true_range.Current.Value)
# If close is greater than trailing ATR tracker
if close > self.trailing_ATR_tracker:
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short ATR trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because ATR trailing stop loss triggered")
# Return
return True
# Else
else:
# Return
return False
def short_liquidation_trailing_stop_loss(self, close):
# If trailing stop loss not yet initiated
if self.trailing_stop_loss_tracker is None:
# If current price less than average price - 1.5 * ATR at order open
if close < (self.Portfolio[self.BTC_symbol].AveragePrice - (1.5 * self.ATR_value_when_order_submitted)):
# Initiate trailing stop loss
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price less than trailing stop loss
if close < self.trailing_stop_loss_tracker:
# Update
self.trailing_stop_loss_tracker = close
# Else
else:
# If current price greater than trailing stop loss tracker by trailing stop loss percent
if (close / self.trailing_stop_loss_tracker) > (1 + (0.01 * self.trailing_stop_loss_percent)):
# Insight
insight = Insight.Price(self.BTC_symbol, timedelta(days=1), InsightDirection.Flat, weight = 0)
# Emit insight
self.EmitInsights(insight)
# Liquidate
self.MarketOrder(self.BTC_symbol, -self.Portfolio[self.BTC_symbol].Quantity, tag = "Short trailing stop triggered")
# Reset ATR
self.ATR_value_when_order_submitted = None
# Reset trailing stop loss
self.trailing_stop_loss_tracker = None
# Reset trailng ATR tracker
self.trailing_ATR_tracker = None
# Send SMS
self.Notify.Sms(self.SMS_phone_number, "Short position liquidated because % trailing stop loss triggered")