| Overall Statistics |
|
Total Trades 252 Average Win 1.27% Average Loss -1.12% Compounding Annual Return 26.953% Drawdown 9.600% Expectancy 0.253 Net Profit 40.218% Sharpe Ratio 1.401 Probabilistic Sharpe Ratio 64.728% Loss Rate 41% Win Rate 59% Profit-Loss Ratio 1.13 Alpha 0.163 Beta 0.414 Annual Standard Deviation 0.136 Annual Variance 0.018 Information Ratio 0.849 Tracking Error 0.148 Treynor Ratio 0.459 Total Fees $466.20 Estimated Strategy Capacity $90000000.00 Lowest Capacity Asset VX XZBZTQ4K6PYH |
""" Steps: 1) Copy backtest files and run back tests for exact same - time periods - # contracts (1) / comparable contract size Resources: Extract portfolio NAV from strategy equity chart : https://www.quantconnect.com/forum/discussion/2817/how-to-get-backtest-results-into-research/p1 Compare live algo vs backtest : https://www.quantconnect.com/forum/discussion/7606/a-new-reconciliation-metric/p1 """
#region imports
from AlgorithmImports import *
#endregion
import decimal
import pandas as pd
from datetime import datetime
from io import StringIO
from datetime import datetime, timedelta
import pickle
from math import floor
# page w/ all Micro symbols - https://www.quantconnect.com/datasets/algoseek-us-futures
"""
Algo Overview
Intraday momentum strategy that uses the momentum gauge with the following chars.
- Uses the mg difference - postive or negative to decide whether to short or long
- Uses overnight price gap to further qualify that the above is a continuing trend
Algo Deployment Steps
- Record previous prod code in file
- Increment VERSION
- Write RELEASE_NOTES or set to empty string ""
- Ensure GO_LIVE_CLOSING_OVERWRITE_VALUE is set to most recent correct value
- Deploy Algo
"""
# To Do:
# Modify deployment steps if self.LiveMode works as expected.
# Determine if we want to add back the old ratcheting system vs START_TAKE_PROFIT_POINTS_HIGHER
# Modify text messages to indicate when a ratchet occurs when multiple ratches take place i.e. "You've made a lot of money x2"
# Add custom rounding class for brokerage perecision - "To meet brokerage precision requirements, order StopPrice was rounded to 3670.00 from 3669.9165125"
# Setup to use the continous contract
# Can we switch from a hardcoded margin requirment to a dynamic self.initialMargin for holding daily? NOT overnight
# Customized profit / loss day ?? Keep it generalized, likely not going to do
# Adding tight stop loss after 2pm
# --------------------------------------------------------- #
# -- Release Configurations -- #
RELEASE_NOTES = "RELEASE NOTES: Fixed bug, forgot to change log level to minimal"
VERSION = 3.09
# -- User Configurations -- #
RISK_PERCENTAGE = .08
LOG_LEVEL = "minimal"
NUMBERS = ["+12035831419", "+15059771023", "+12035215903"]
FIRST_NAME = "Alpha"
# -- Global Configurations -- #
# FUTURES_CONTRACT = Futures.Indices.SP500EMini
FUTURES_CONTRACT = Futures.Indices.MicroSP500EMini
EQUITY = "SPXL"
BENCHMARK = "SPXL"
GAP_MIN = .002
START_TAKE_PROFIT_POINTS = 25
PROFIT_POINTS = 10
START_TAKE_PROFIT_POINTS_HIGHER = 50
PROFIT_POINTS_HIGHER = 30
GO_LIVE_CLEAR = False
GO_LIVE_CLOSING_OVERWRITE = False
GO_LIVE_CLOSING_OVERWRITE_VALUE = 3821.25
GO_LIVE_CLOSING_VIEW = False
MAX_LOSS_POINTS_AS_PERCENTAGE = .0087
MARGIN_REQ_PER_CONTRACT = 1800
# https://www.interactivebrokers.com/en/trading/margin-futures-fops.php
PRICE_PER_POINT = 5
class gapMomentum(QCAlgorithm):
closeObjectStoreKey = "close_object"
def Initialize(self):
""" Initialize -- note SP500 MG start date data is 2018-11-28 """
# -- Preset Times -- #
# # Opto Set
# self.SetStartDate(2021, 1, 1)
# self.SetEndDate(2022, 3, 25)
# # Long Set
# self.SetStartDate(2021, 1, 1)
# self.SetEndDate(2022, 5, 15)
# self.SetStartDate(2021, 1, 1)
# self.SetEndDate(2022, 4, 1)
# 1 Year
# self.SetStartDate(2021, 3, 25)
# self.SetEndDate(2022, 3, 25)
# Debug margin req
# self.SetStartDate(2020, 3, 3)
# self.SetEndDate(2020, 4, 25)
# Current Set
self.SetStartDate(2021, 1, 1)
self.SetEndDate(2022, 6, 1)
# -- Optimizations Used -- #
start_take_profit_points = self.GetParameter("start_take_profit_points")
self.start_take_profit_points = START_TAKE_PROFIT_POINTS if start_take_profit_points is None else float(start_take_profit_points)
profit_points = self.GetParameter("profit_points")
self.profit_points = PROFIT_POINTS if profit_points is None else float(profit_points)
gap_min = self.GetParameter("gap_min")
self.gapMin = GAP_MIN if gap_min is None else float(gap_min)
risk_percentage = self.GetParameter("risk_percentage")
self.risk_percentage = RISK_PERCENTAGE if risk_percentage is None else float(risk_percentage)
max_loss_as_percentage = self.GetParameter("max_loss_as_percentage")
self.max_loss_as_percentage = MAX_LOSS_POINTS_AS_PERCENTAGE if max_loss_as_percentage is None else float(max_loss_as_percentage)
start_take_profit_points_higher = self.GetParameter("start_take_profit_points_higher")
self.start_take_profit_points_higher = START_TAKE_PROFIT_POINTS_HIGHER if start_take_profit_points_higher is None else float(start_take_profit_points_higher)
profit_points_higher = self.GetParameter("profit_points_higher")
self.profit_points_higher = PROFIT_POINTS_HIGHER if profit_points_higher is None else float(profit_points_higher)
# -- Futures & Equity Setup & Go Live Clear -- #
if GO_LIVE_CLEAR or GO_LIVE_CLOSING_OVERWRITE or GO_LIVE_CLOSING_VIEW:
self.SetStartDate(2022, 2, 14)
self.SetEndDate(2022, 2, 16)
self.SetCash(30000)
# Setup Futures Contract
self.futuresContract = self.AddFuture(FUTURES_CONTRACT)
self.futuresContract.SetFilter(5, 150)
self.contracts = {}
self.oi_contract = None
# Brokerage Model
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage)
# Add SPXL for market hours
self.equity = self.AddEquity(EQUITY, Resolution.Minute).Symbol
# Initial dwonload for back-test
url = self.Download(address="https://www.dropbox.com/s/6w6ip6fvh2q7ciw/SP500_MG_Real_prod.csv?dl=1", headers={"Cache-Control": "no-cache", "Pragma": "no-cache"})
self.MG_df = pd.read_csv(StringIO(url))
self.MG_df['Date'] = self.MG_df['Date'].apply(Utils.dateModFromString)
self.MG_df['Date'] = self.MG_df['Date'].apply(Utils.dateModToFormat)
# Capture Closing price
self.Schedule.On(self.DateRules.EveryDay(EQUITY), self.TimeRules.BeforeMarketClose(self.equity, 0), self.captureClosingPrice)
# Scheduled Entry
self.Schedule.On(self.DateRules.EveryDay(EQUITY), self.TimeRules.AfterMarketOpen(self.equity, 0), self.openPosition)
# Initilization Vars
self.closingPrice = None
self.priceGap = None
self.price = None
self.positionType = None
self.openingPrice = 0
self.security = None
self.gapDown = None
self.gapUp = None
self.closePrice = None
self.positionType = None
self.boughtIn = False
self.sitOut = False
self.textProfit = True
self.closePriceRH = None
self.closePriceMP = None
self.fillPrice = None
self.stopLossTriggered = False
self.stopLossTriggeredHigher = False
self.SetWarmup(1, Resolution.Daily)
if self.LiveMode:
self.Log("Initialized Algorithim -- Version: " + str(VERSION))
for number in NUMBERS:
self.Notify.Sms(number, "Initialized Algorithim -- Version: " + str(VERSION))
if RELEASE_NOTES != "":
self.Notify.Sms(number, RELEASE_NOTES)
# RELEASE_NOTES
def OnData(self, data):
""" Sorts Futures Contracts by OpenInterest """
self.data = data
if GO_LIVE_CLOSING_VIEW:
value = self.ObjectStore.Read(self.closeObjectStoreKey)
deserialized_value = bytes(self.ObjectStore.ReadBytes(self.closeObjectStoreKey))
self.closingPrice = pickle.loads(deserialized_value)
self.Debug("Current closingPrice stored: " + str(self.closingPrice))
return
if GO_LIVE_CLOSING_OVERWRITE:
self.ObjectStore.SaveBytes(self.closeObjectStoreKey, pickle.dumps(GO_LIVE_CLOSING_OVERWRITE_VALUE))
self.Debug("Manually stored closingPrice: " + str(GO_LIVE_CLOSING_OVERWRITE_VALUE))
return
if GO_LIVE_CLEAR:
self.ObjectStore.Delete(self.closeObjectStoreKey)
return
# Iterates through futures contracts
for chain in self.data.FutureChains:
# self.Log("Current Security At OnData: "+ str(self.security))
contracts = sorted([k for k in chain.Value if k.OpenInterest > 0],
key=lambda k : k.OpenInterest, reverse=True)
# self.Log("Contracts At Ondata: " + str(contracts))
if not contracts:
# self.Log("Continuing at OnData, no contracts")
continue
# Chooses Futures Contract with most open interest
self.oi_contract = contracts[0]
# Uses Futures Contract with most Open Interest
self.symbol = self.oi_contract.Symbol
if self.symbol not in self.contracts:
self.contracts[self.symbol] = SymbolData(self, self.symbol)
self.contracts[self.symbol].consolidator.DataConsolidated += self.IntradayBars
# self.Log("Setting Consolidator at Intraday")
# self.Debug(" Debug for chain ")
self.security = self.Securities[self.symbol]
self.price = self.security.Price
# self.Log("Current Security At OnData: " + str(self.security))
# self.Log("Current Price For Security At OnData: " + str(self.price))
self.initialMargin = self.security.BuyingPowerModel.InitialOvernightMarginRequirement # [RHINSEN] - do we need this?
self.closePriceMP = self.price
def openPosition(self):
""" Opens Position Short or Long """
self.Log("At OpenPosition")
# Return for hardcoding / clearing
if GO_LIVE_CLEAR == True or GO_LIVE_CLOSING_OVERWRITE == True or GO_LIVE_CLOSING_VIEW == True:
return
# Catch if OnData didn't run first, if it hasn't, set security
# if self.security is None: # Remove to set secruity evertytime
if self.security is None:
for chain in self.data.FutureChains:
contracts = sorted([k for k in chain.Value if k.OpenInterest > 0],
key=lambda k : k.OpenInterest, reverse=True)
self.Log("Contracts: " + str(contracts))
if not contracts:
self.Log("Continuing at openPosition -- No Contracts")
continue
# Chooses Futures Contract with most open interest
self.oi_contract = contracts[0]
# Uses Futures Contract with most Open Interest
if self.symbol not in self.contracts:
self.contracts[self.symbol] = SymbolData(self, self.symbol)
self.contracts[self.symbol].consolidator.DataConsolidated += self.IntradayBars
self.Log("Set Consolidator At Open")
self.security = self.Securities[self.symbol]
self.price = self.security.Price
self.Log("Current Security At openPosition: " + str(self.security))
self.Log("Current Price For Security At openPosition: " + str(self.price))
# Read in stored value for closePrice to closingPrice when not warming up
if not self.IsWarmingUp:
if self.ObjectStore.ContainsKey(self.closeObjectStoreKey):
value = self.ObjectStore.Read(self.closeObjectStoreKey)
deserialized_value = bytes(self.ObjectStore.ReadBytes(self.closeObjectStoreKey))
self.closingPrice = pickle.loads(deserialized_value)
self.Log("Set closingPrice At openPosition From Saved Value")
else:
if self.LiveMode:
self.closingPrice = GO_LIVE_CLOSING_OVERWRITE_VALUE
self.Log("Set closingPrice At openPosition From Global Value")
# Calculate price gap for MG
if self.price is not None and self.closingPrice is not None and self.priceGap is None:
self.Log("Setting PriceGap at openPosition")
# Needed for a new day of trading
self.startInvestment = self.Portfolio.TotalPortfolioValue
self.boughtIn = False
self.sitOut = False
self.textProfit = True
self.fillPrice = None
self.keepPrice = self.price
self.openingPrice = self.price
# Price Gap Logic
self.priceGap = self.price - self.closingPrice
self.gapUp = self.closingPrice * (1+self.gapMin)
self.gapDown = self.closingPrice * (1-self.gapMin)
self.Log("PriceGap set at openPosition- Old Closing Price: " + str(self.closingPrice) + " -- Open Price: " + str(self.price) + " -- Price Gap: " + str(self.priceGap) + " -- gap Up: " + str(self.gapUp) + " -- gap down: " + str(self.gapDown))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, "Set at openPosition- Old Closing Price: " + str(self.closingPrice) + " -- Open Price: " + str(self.price) + " -- Price Gap: " + str(self.priceGap) + " -- gap Up: " + str(self.gapUp) + " -- gap down: " + str(self.gapDown))
# Cause a reinitilization of the file download every open position if LIVE
if self.LiveMode:
url = self.Download(address="https://www.dropbox.com/s/6w6ip6fvh2q7ciw/SP500_MG_Real_prod.csv?dl=1", headers={"Cache-Control": "no-cache", "Pragma": "no-cache"})
self.MG_df = pd.read_csv(StringIO(url))
# Grab last known entry, which should be yesterday
self.dayUsed = self.MG_df.iloc[-1][0]
self.MG_positive = float(self.MG_df.iloc[-1][1])
self.MG_negative = float(self.MG_df.iloc[-1][2])
self.MG_difference = float(self.MG_df.iloc[-1][3])
self.MG_daily_change = float(self.MG_df.iloc[-1][4])
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, str(self.Time) + " -- Using Day: " + str(self.dayUsed) + " -- MG Difference: " + str(self.MG_difference) + " -- MG Daily Change: " + str(self.MG_daily_change))
# self.Debug(str(self.Time) + " opening Portfolio Margin: " + str(self.Portfolio.MarginRemaining))
def captureClosingPrice(self):
""" Capture closing price """
# Closing Price information [MPURCELL]
if not self.IsWarmingUp:
self.closePriceRH = self.price
if self.closePriceRH is not None:
self.closePrice = self.closePriceRH
self.ObjectStore.SaveBytes(self.closeObjectStoreKey, pickle.dumps(self.closePrice))
self.priceGap = None
self.Log('Used RH close price of: ' + str(self.closePrice))
if LOG_LEVEL == "info":
for number in NUMBERS:
self.Notify.Sms(number, "Closing Price: " + str(self.closePrice))
elif self.closePriceMP is not None:
self.closePrice = self.closePriceMP
self.ObjectStore.SaveBytes(self.closeObjectStoreKey, pickle.dumps(self.closePrice))
self.priceGap = None
self.Log('Used MP close price of: ' + str(self.closePrice))
if LOG_LEVEL == "info":
for number in NUMBERS:
self.Notify.Sms(number, "Closing Price: " + str(self.closePrice))
else:
self.closePrice = GO_LIVE_CLOSING_OVERWRITE_VALUE
self.Log('Used Global Close Price: ' + str(self.closePrice))
self.ObjectStore.SaveBytes(self.closeObjectStoreKey, pickle.dumps(self.closePrice))
self.priceGap = None
def IntradayBars(self, sender, bar):
""" Creates IntraDay Bars """
if not self.oi_contract or self.oi_contract.Symbol != bar.Symbol:
return
# Check if Today's date in MG File for back-testing
try:
current_time = f'{self.Time.year}-{self.Time.month}-{self.Time.day}'
todayidxList = self.MG_df.index[self.MG_df['Date']==current_time].tolist()
todayIDX = todayidxList[0]
self.todaysValues = True
except:
self.todaysValues = False
# Use Yesterday's numbers to prevent look-ahead bias
if self.todaysValues == True:
self.dayUsed = self.MG_df.loc[todayIDX-1][0]
self.MG_positive = float(self.MG_df.loc[todayIDX-1][1])
self.MG_negative = float(self.MG_df.loc[todayIDX-1][2])
self.MG_difference = float(self.MG_df.loc[todayIDX-1][3])
self.MG_daily_change = float(self.MG_df.loc[todayIDX-1][4])
# Time Logic
current_time = self.Time.strftime("%H:%M")
current_time_as_date = datetime.strptime(current_time, '%H:%M').time()
# This is a back-up if open position somehow doesn't get the priceGap, an OK use of a `==`
if current_time_as_date >= datetime.strptime('9:30', '%H:%M').time() and current_time_as_date <= datetime.strptime('10:00', '%H:%M').time():
if self.priceGap is None and not self.IsWarmingUp:
self.Log("Setting PriceGap at Intraday")
if self.ObjectStore.ContainsKey(self.closeObjectStoreKey):
value = self.ObjectStore.Read(self.closeObjectStoreKey)
deserialized_value = bytes(self.ObjectStore.ReadBytes(self.closeObjectStoreKey))
self.closingPrice = pickle.loads(deserialized_value)
self.Log("Set closingPrice At Intraday From Saved Value")
else:
if self.LiveMode:
self.closingPrice = GO_LIVE_CLOSING_OVERWRITE_VALUE
self.Log("Set closingPrice At Intraday From Overwrite Value")
# Needed for a new day of trading
self.startInvestment = self.Portfolio.TotalPortfolioValue
self.boughtIn = False
self.sitOut = False
self.textProfit = True
self.fillPrice = None
self.keepPrice = self.price
self.openingPrice = self.price
# Price Gap Logic
self.priceGap = self.price - self.closingPrice
self.gapUp = self.closingPrice * (1+self.gapMin)
self.gapDown = self.closingPrice * (1-self.gapMin)
self.Log("PriceGap Set at Intraday Bars == - Old Closing Price: " + str(self.closingPrice) + " -- Open Price: " + str(self.price) + " -- Price Gap: " + str(self.priceGap) + " -- gap Up: " + str(self.gapUp) + " -- gap down: " + str(self.gapDown))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, "Set at == - Old Closing Price: " + str(self.closingPrice) + " -- Open Price: " + str(self.price) + " -- Price Gap: " + str(self.priceGap) + " -- gap Up: " + str(self.gapUp) + " -- gap down: " + str(self.gapDown))
if self.LiveMode:
url = self.Download(address="https://www.dropbox.com/s/6w6ip6fvh2q7ciw/SP500_MG_Real_prod.csv?dl=1", headers={"Cache-Control": "no-cache", "Pragma": "no-cache"})
self.MG_df = pd.read_csv(StringIO(url))
# Grab last known entry, which should be yesterday
self.dayUsed = self.MG_df.iloc[-1][0]
self.MG_positive = float(self.MG_df.iloc[-1][1])
self.MG_negative = float(self.MG_df.iloc[-1][2])
self.MG_difference = float(self.MG_df.iloc[-1][3])
self.MG_daily_change = float(self.MG_df.iloc[-1][4])
self.Log(str(self.Time) + " -- Using Day: " + str(self.dayUsed) + " -- MG Difference: " + str(self.MG_difference) + " -- MG Daily Change: " + str(self.MG_daily_change))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, str(self.Time) + " -- Using Day: " + str(self.dayUsed) + " -- MG Difference: " + str(self.MG_difference) + " -- MG Daily Change: " + str(self.MG_daily_change))
# Begin to allow buy in process when price gap is exceeded
if current_time_as_date >= datetime.strptime('9:30', '%H:%M').time() and current_time_as_date <= datetime.strptime('16:55', '%H:%M').time() and self.sitOut == False:
# For Initial BuyIn
if not self.security.Invested and self.boughtIn == False and self.sitOut == False and self.priceGap and self.Securities[EQUITY].Exchange.ExchangeOpen == True and not self.IsWarmingUp:
"""
It's not enough to wish, dream, hope. Even children know this. We must set sail into the sea of uncertainty.
We must meet fear face-to-face. We must take our dreams as maps for a greater journey.
Dreams, to come true, need a good story. So go live one.
— Vironika Tugaleva
"""
# -- Long -- #
if float(self.MG_daily_change) > 0:
self.positionType = 'Long'
# Calculate cutOffPrice using percentage instead of static number
self.max_loss_points = self.price * self.max_loss_as_percentage
self.cutOffPrice = self.price - self.max_loss_points
if self.price > self.gapUp:
self.Log("Investing Long at Intraday")
self.contractsToTradeMaxByRisk = floor( (self.Portfolio.TotalPortfolioValue * self.risk_percentage) / (self.max_loss_points * PRICE_PER_POINT))
self.contractsToTradeMaxByEquity = floor( self.Portfolio.TotalPortfolioValue / MARGIN_REQ_PER_CONTRACT )
if self.contractsToTradeMaxByEquity < self.contractsToTradeMaxByRisk:
self.contractsToTrade = self.contractsToTradeMaxByEquity
else:
self.contractsToTrade = self.contractsToTradeMaxByRisk
self.MarketOrder(self.symbol, 1)
# self.Debug(str(self.Time) + " Buy price: " + str(self.price) + " long contractsToTrade: " + str(self.contractsToTrade) + " portfolio value: " + str(self.Portfolio.TotalPortfolioValue) )
# self.Debug(str(self.Time) + " risk%: " + str(self.risk_percentage) + " risk $: " + str(self.contractsToTrade * self.max_loss_points * 5) + " max_loss_pts: " +str(self.max_loss_points))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, str(FIRST_NAME) + ", IntraDay Gap Momentum Playing Long Today at algo price: " + str(self.price))
elif LOG_LEVEL == "minimal":
self.Notify.Sms(number, str(FIRST_NAME) + ", IntraDay Gap Momentum Playing Long Today")
self.boughtIn = True
self.stopLossTriggered = False
self.stopLossTriggeredHigher = False
self.keepPrice = self.price
## Global SL
self.stopMarketTicket = self.StopMarketOrder(self.symbol, -1, self.cutOffPrice)
# -- Short -- #
elif float(self.MG_daily_change) < 0:
self.positionType = 'Short'
# Calculate cutOffPrice using percentage instead of static number
self.max_loss_points = self.price * self.max_loss_as_percentage
self.cutOffPrice = self.price + self.max_loss_points
if self.price < self.gapDown:
self.Log("Investing Short at Intraday")
self.contractsToTradeMaxByRisk = floor( (self.Portfolio.TotalPortfolioValue * self.risk_percentage) / (self.max_loss_points * PRICE_PER_POINT))
self.contractsToTradeMaxByEquity = floor( self.Portfolio.TotalPortfolioValue / MARGIN_REQ_PER_CONTRACT )
if self.contractsToTradeMaxByEquity < self.contractsToTradeMaxByRisk:
self.contractsToTrade = self.contractsToTradeMaxByEquity
else:
self.contractsToTrade = self.contractsToTradeMaxByRisk
self.MarketOrder(self.symbol, -1)
# self.Debug(str(self.Time) + " Sell price: " + str(self.price) + " short contractsToTrade: " + str(self.contractsToTrade) + " portfolio value: " + str(self.Portfolio.TotalPortfolioValue))
# self.Debug(str(self.Time) + " risk%: " + str(self.risk_percentage) + " risk $: " + str(self.contractsToTrade * self.max_loss_points * 5) + " max_loss_pts: " +str(self.max_loss_points))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, str(FIRST_NAME) + ", IntraDay Gap Momentum Playing Short Today at algo price: " + str(self.price))
elif LOG_LEVEL == "minimal":
self.Notify.Sms(number, str(FIRST_NAME) + ", IntraDay Gap Momentum Playing Short Today")
self.boughtIn = True
self.stopLossTriggered = False
self.stopLossTriggeredHigher = False
self.keepPrice = self.price
## Global SL
self.stopMarketTicket = self.StopMarketOrder(self.symbol, 1, self.cutOffPrice)
# -- Dynamic Exit -- #
if self.security.Invested and self.stopLossTriggered == False:
# For Long, set a take profit after PROFIT_BUFFER_POINTS exceeded
if self.positionType == 'Long':
if self.price >= (self.keepPrice + self.start_take_profit_points):
# Update Global SL
self.stopLossTriggered = True
self.keepPoints = self.keepPrice + self.profit_points
updateSettings = UpdateOrderFields()
updateSettings.StopPrice = decimal.Decimal(round(self.keepPoints, 2))
response = self.stopMarketTicket.Update(updateSettings)
# Capture Profit Estimation
self.estimatedProfit = int(abs((round(self.keepPoints, 2) - self.fillPrice) * self.contractsToTrade * PRICE_PER_POINT))
self.Debug("Estimated profit Long: " + str(self.estimatedProfit))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, "TP StopLoss Triggered At: " + str(self.price))
self.Notify.Sms(number, "You will be making money today")
elif LOG_LEVEL == "minimal":
self.Notify.Sms(number, "You have safely locked in an estimated $" + str(self.estimatedProfit) + " in profit")
# For Short, set a take profit after PROFIT_BUFFER_POINTS exceeded
if self.positionType == 'Short':
if self.price <= (self.keepPrice - self.start_take_profit_points):
# Update Global SL
self.stopLossTriggered = True
self.keepPoints = self.keepPrice - self.profit_points
updateSettings = UpdateOrderFields()
updateSettings.StopPrice = decimal.Decimal(round(self.keepPoints, 2))
response = self.stopMarketTicket.Update(updateSettings)
# Capture Profit Estimation
self.estimatedProfit = int(abs((round(self.keepPoints, 2) - self.fillPrice) * self.contractsToTrade * PRICE_PER_POINT))
self.Debug("Estimated profit Short: " + str(self.estimatedProfit))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, "TP StopLoss Triggered At: " + str(self.price))
self.Notify.Sms(number, "You will be making money today")
elif LOG_LEVEL == "minimal":
self.Notify.Sms(number, "You have safely locked in an estimated $" + str(self.estimatedProfit) + " in profit")
# Check if higher take profit threshold is met & raise the dynamic Exit
if self.security.Invested and self.stopLossTriggeredHigher == False:
if self.positionType == 'Long':
if self.price >= (self.keepPrice + self.start_take_profit_points_higher):
# Update Global SL
self.stopLossTriggeredHigher = True
self.keepPoints = self.keepPrice + self.profit_points_higher
updateSettings = UpdateOrderFields()
updateSettings.StopPrice = decimal.Decimal(round(self.keepPoints, 2))
response = self.stopMarketTicket.Update(updateSettings)
# Capture Profit Estimation
self.estimatedProfit = int(abs((round(self.keepPoints, 2) - self.fillPrice) * self.contractsToTrade * PRICE_PER_POINT))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, "Higher TP StopLoss Triggered At: " + str(self.price))
self.Notify.Sms(number, "You will be making ALOT of money today, " + str(FIRST_NAME))
elif LOG_LEVEL == "minimal":
self.Notify.Sms(number, "You have safely locked in an estimated $" + str(self.estimatedProfit) + " in profit")
# self.Debug(str(self.Time) + " Higher TP start - short algo price: " + str(self.price) + " keepPoints higher: " + str(self.keepPoints) + " keepPrice: " +str(self.keepPrice))
if self.positionType == 'Short':
if self.price <= (self.keepPrice - self.start_take_profit_points_higher):
# Update Global SL
self.stopLossTriggeredHigher = True
self.keepPoints = self.keepPrice - self.profit_points_higher
updateSettings = UpdateOrderFields()
updateSettings.StopPrice = decimal.Decimal(round(self.keepPoints, 2))
response = self.stopMarketTicket.Update(updateSettings)
# Capture Profit Estimation
self.estimatedProfit = int(abs((round(self.keepPoints, 2) - self.fillPrice) * self.contractsToTrade * PRICE_PER_POINT))
for number in NUMBERS:
if LOG_LEVEL == "info":
self.Notify.Sms(number, "Higher TP StopLoss Triggered At: " + str(self.price))
self.Notify.Sms(number, "You will be making ALOT of money today, " + str(FIRST_NAME))
elif LOG_LEVEL == "minimal":
self.Notify.Sms(number, "You have safely locked in an estimated $" + str(self.estimatedProfit) + " in profit")
# Check whether max_loss_stop is exceeded, if so sit out
if self.positionType == 'Long':
if self.price <= self.cutOffPrice:
self.sitOut = True
if self.security.Invested:
gapMomentum.closePosition(self)
if self.positionType == 'Short':
if self.price >= self.cutOffPrice:
self.sitOut = True
if self.security.Invested:
gapMomentum.closePosition(self)
# Liquadate always before end of day
if self.security.Invested and current_time_as_date >= datetime.strptime('16:55', '%H:%M').time():
# self.Debug('Liqudated Close of Day ' + str(self.positionType) + ': ' + str(self.Time))
gapMomentum.closePosition(self)
# Bought in, and have closed position
if not self.security.Invested and self.boughtIn == True and self.textProfit == True:
self.textProfit = False
# Get Daily Profit
profit = round((self.Portfolio.TotalPortfolioValue - self.startInvestment), 2)
# self.Debug('Daily Profit: '+str(profit))
if profit >= 0:
gains = 'Profit'
else:
gains = 'Loss'
# Msg
if LOG_LEVEL in ['minimal', 'info']:
for number in NUMBERS:
self.Notify.Sms(number, "IntraDay Gap Momentum Sold " + str(self.positionType) + " Today for " + str(gains) + ": $" + str(profit))
def closePosition(self):
""" close position """
### Liquadate and Text
self.Liquidate()
# -- Clean up Stop Losses -- #
# Gets Open Tickets related to stop loss
open_tickets = self.Transactions.GetOpenOrderTickets()
# Parses through open tickets looking for stop loss
stop_tickets = [ticket for ticket in open_tickets if ticket.OrderType == OrderType.StopMarket]
# If More then one active stop loss ticket, cancel current stop losses, create ticket
if len(stop_tickets) > 1 :
for ticket in stop_tickets:
ticket.Cancel('Canceled stop loss tickets')
create_ticket = True
def OnSecuritiesChanged(self, changes):
""" QC function for when futures contract changes """
for security in changes.RemovedSecurities:
self.symbol = security.Symbol
symbolData = self.contracts.get(self.symbol, None)
def OnOrderEvent(self, orderevent):
""" Grabs price from initial fill event """
order = self.Transactions.GetOrderById(orderevent.OrderId)
if orderevent.Status == OrderStatus.Filled:
self.fillPrice = orderevent.FillPrice
class SymbolData:
""" For data consolidation with 5 minute bars"""
def __init__(self, algorithm, symbol):
periods = {
'1D': timedelta(1),
'INTRADAY': timedelta(minutes=5)
}
self.consolidator = TradeBarConsolidator(periods['INTRADAY'])
# if algorithm.LiveMode:
message = str(type(self.consolidator))
algorithm.Log(message)
# Used for Intraday Function
self.atr = AverageTrueRange(14, MovingAverageType.Simple)
algorithm.RegisterIndicator(symbol, self.atr, self.consolidator)
class Utils:
""" Utility Functions """
@staticmethod
def dateModFromString(x):
return datetime.strptime(x, '%Y-%m-%d')
@staticmethod
def dateModToFormat(x):
return x.strftime("%Y-%-m-%-d")
"""
Algo Objective:
Trade VIX futures intraday using previous day's direction and MG for direction.
According to VIX Research:
Days up Avg: 1.67
Days up median: 1.0
Days down Avg: 2.53
Days down median: 2.0
MG_dailyChange predicts
VIX updays after MG_daily_change < 0: 44%
VIX downdays after MG_daily_change > 0: 68%
Both conclude that predicting down days is easier than predicting up days.
try adding TP pts similar to Gap Mo. PROD
"""
"""
Notes / Inferences:
Shorting vix works much better than going long (as predicted). Going Long only has positive return if using MG_daily_change only
MG_daily_change adds a lot of benefit to the short side - only a little to the long side.
MG_difference does the worst
Very little difference in CAGR between MG_daily_change only and MG_daily_change with days in a row (in_a_row has better PSR, info ratio, but/and trades 23% less)
I could put $ into shorting now.
- Long does better when the previously day was a down day for the VIX.
"""
# Next Steps
# Add TP pts
# add SL
"""
From 2021 - June 2022
VX Range Avg: 0.63
VX Range Median: .40
"""
# region imports
from AlgorithmImports import *
import statistics
from datetime import datetime
from io import StringIO
from datetime import datetime, timedelta
# endregion
FUTURES_CONTRACT = Futures.Indices.VIX
EQUITY = "SPY"
class CryingFluorescentYellowBuffalo(QCAlgorithm):
def Initialize(self):
# Debug Set
# self.SetStartDate(2022,3,1)
# self.SetEndDate(2022,6,10)
# Long Set
self.SetStartDate(2021,1,1)
self.SetEndDate(2022,6,1)
self.SetCash(50000)
self.equity = self.AddEquity(EQUITY, Resolution.Minute)
self.vix = self.AddData(CBOE, "VIX").Symbol
# Setup Futures Contract
self.futuresContract = self.AddFuture(FUTURES_CONTRACT, dataNormalizationMode = DataNormalizationMode.Raw, dataMappingMode = DataMappingMode.LastTradingDay, contractDepthOffset = 0)
# Initial dwonload for back-test
url = self.Download(address="https://www.dropbox.com/s/6w6ip6fvh2q7ciw/SP500_MG_Real_prod.csv?dl=1", headers={"Cache-Control": "no-cache", "Pragma": "no-cache"})
self.MG_df = pd.read_csv(StringIO(url))
self.MG_df['Date'] = self.MG_df['Date'].apply(Utils.dateModFromString)
self.MG_df['Date'] = self.MG_df['Date'].apply(Utils.dateModToFormat)
self.Schedule.On(self.DateRules.EveryDay(EQUITY), self.TimeRules.AfterMarketOpen(EQUITY, 1), self.dayStart)
self.Schedule.On(self.DateRules.EveryDay(EQUITY), self.TimeRules.BeforeMarketClose(EQUITY, 0), self.dayEnd)
# Iniatialzed Vars
self.MG_daily_change = None
self.previouslyUp = False
def OnData(self, data: Slice):
self.data = data
if data.ContainsKey("VIX"):
self.vixPriceOpen = float(data['VIX'].Open)
self.vixPriceClose = float(data['VIX'].Close)
self.futuresPrice = self.futuresContract.Price
def dayStart(self):
# Check if Today's date in MG File for back-testing
try:
current_time = f'{self.Time.year}-{self.Time.month}-{self.Time.day}'
todayidxList = self.MG_df.index[self.MG_df['Date']==current_time].tolist()
todayIDX = todayidxList[0]
self.todaysValues = True
except:
self.todaysValues = False
# Use Yesterday's numbers to prevent look-ahead bias
if self.todaysValues == True:
self.dayUsed = self.MG_df.loc[todayIDX-1][0]
self.MG_positive = float(self.MG_df.loc[todayIDX-1][1])
self.MG_negative = float(self.MG_df.loc[todayIDX-1][2])
self.MG_difference = float(self.MG_df.loc[todayIDX-1][3])
self.MG_daily_change = float(self.MG_df.loc[todayIDX-1][4])
if self.MG_daily_change is not None and self.vixPriceOpen is not None and self.vixPriceClose is not None:
# self.Debug(str(self.Time) + " Vix prev Open: " + str(self.vixPriceOpen) + " Vix prev Close: " + str(self.vixPriceClose) + " MG_daily_change: " + str(self.MG_daily_change))
# -- Short --
if self.MG_daily_change > 0 and self.vixPriceOpen > self.vixPriceClose:
self.MarketOrder(self.futuresContract.Mapped, -1)
self.Debug(str(self.Time) + " sell price: " + str(self.futuresPrice))
# -- Long --
# if self.MG_daily_change < 0 and self.previouslyUp == False:
# self.MarketOrder(self.futuresContract.Mapped, 1)
# self.Debug(str(self.Time) + " sell price: " + str(self.futuresPrice))
def dayEnd(self):
if self.Portfolio.Invested:
self.Liquidate()
self.Debug(str(self.Time) + " close price: " + str(self.futuresPrice))
if self.vixPriceOpen < self.vixPriceClose:
self.previouslyUp = True
self.Debug(str(self.Time) + " self.previouslyUp: True" )
else:
self.previouslyUp = False
self.Debug(str(self.Time) + " self.previouslyUp: False" )
class Utils:
""" Utility Functions """
@staticmethod
def dateModFromString(x):
return datetime.strptime(x, '%Y-%m-%d')
@staticmethod
def dateModToFormat(x):
return x.strftime("%Y-%-m-%-d")