| Overall Statistics |
|
Total Trades 133 Average Win 3.70% Average Loss -1.92% Compounding Annual Return -44.872% Drawdown 22.100% Expectancy -0.156 Net Profit -19.348% Sharpe Ratio -0.864 Probabilistic Sharpe Ratio 7.807% Loss Rate 71% Win Rate 29% Profit-Loss Ratio 1.93 Alpha -0.236 Beta -0.37 Annual Standard Deviation 0.346 Annual Variance 0.119 Information Ratio -1.216 Tracking Error 0.385 Treynor Ratio 0.808 Total Fees $542.00 Estimated Strategy Capacity $360000.00 Lowest Capacity Asset SPY 326V9O0OPYZJA|SPY R735QTJ8XC9X Portfolio Turnover 3.68% |
# QuantConnect specific imports
from AlgorithmImports import *
# import QuantConnect as qc
###############################################################################
class CustomExerciseModel(DefaultExerciseModel):
# How top prevent the excercise from being triggered?
def OptionExercise(self, option: Option, order: OptionExerciseOrder):
order_event = OrderEvent(
order.Id,
option.Symbol,
Extensions.ConvertToUtc(option.LocalTime, option.Exchange.TimeZone),
OrderStatus.Filled,
Extensions.GetOrderDirection(order.Quantity),
0.0,
order.Quantity,
OrderFee.Zero,
"Tag"
)
order_event.IsAssignment = False
return
# return [ order_event ]###############################################################################
# Standard library imports
import datetime as DT
# from dateutil.parser import parse
# import decimal
# import numpy as np
# import pandas as pd
# import pytz
# from System.Drawing import Color
# import traceback
# QuantConnect specific imports
# import QuantConnect as qc
from AlgorithmImports import *
# Import from files
from notes_and_inputs import *
from symbol_data import SymbolData
###############################################################################
class CustomTradingStrategy(QCAlgorithm):
def Initialize(self):
"""Initialize algorithm."""
# Set backtest details
self.SetBacktestDetails()
# Add strategy variables required for the algo
self.AddStrategyVariables()
# Add instrument data to the algo
self.AddInstrumentData()
# Schedule functions
self.ScheduleFunctions()
#------------------------------------------------------------------------------
def SetBacktestDetails(self):
"""Set the backtest details."""
self.SetStartDate(START_DT.year, START_DT.month, START_DT.day)
if END_DATE:
self.SetEndDate(END_DT.year, END_DT.month, END_DT.day)
self.SetCash(CASH)
self.SetTimeZone(TIMEZONE)
# Setup trading framework
# Transaction and submit/execution rules will use IB models
# brokerages:
'''https://github.com/QuantConnect/Lean/blob/master/Common/Brokerages
/BrokerageName.cs'''
# account types: AccountType.Margin, AccountType.Cash
self.SetBrokerageModel(
BrokerageName.InteractiveBrokersBrokerage,
AccountType.Margin
)
# Configure all universe securities
# This sets the data normalization mode
# You can also set custom fee, slippage, fill, and buying power models
self.SetSecurityInitializer(
CustomSecurityInitializer(
self.BrokerageModel,
FuncSecuritySeeder(self.GetLastKnownPrices)
)
)
#------------------------------------------------------------------------------
def AddStrategyVariables(self):
"""Create required strategy variables."""
try:
self.RISK_PER_TRADE = float(self.GetParameter("RISK_PER_TRADE"))/100.0
except:
self.RISK_PER_TRADE = RISK_PER_TRADE
try:
self.MIN_DAYS_TO_EXPIRY = \
int(self.GetParameter("MIN_DAYS_TO_EXPIRY"))
except:
self.MIN_DAYS_TO_EXPIRY = MIN_DAYS_TO_EXPIRY
try:
i = int(self.GetParameter("BAR_MINUTES"))
self.BAR_MINUTES = BAR_MINUTES_OPTIONS[i]
except:
self.BAR_MINUTES = BAR_MINUTES
try:
self.START_TRADING_MINUTES_AFTER_OPEN = \
int(self.GetParameter("START_TRADING_MINUTES_AFTER_OPEN"))
except:
self.START_TRADING_MINUTES_AFTER_OPEN = \
START_TRADING_MINUTES_AFTER_OPEN
# try:
# self.MA_FAST_PERIOD = int(self.GetParameter("MA_FAST_PERIOD"))
# except:
# self.MA_FAST_PERIOD = MA_FAST_PERIOD
# try:
# self.MA_SLOW_PERIOD = int(self.GetParameter("MA_SLOW_PERIOD"))
# except:
# self.MA_SLOW_PERIOD = MA_SLOW_PERIOD
try:
self.OTM_STRIKES = int(self.GetParameter("OTM_STRIKES"))
except:
self.OTM_STRIKES = OTM_STRIKES
try:
self.MIN_TAKE_PROFIT_AMOUNT = \
float(self.GetParameter("MIN_TAKE_PROFIT_AMOUNT"))
except:
self.MIN_TAKE_PROFIT_AMOUNT = MIN_TAKE_PROFIT_AMOUNT
try:
self.STOP_LOSS_PERCENT = \
float(self.GetParameter("STOP_LOSS_PERCENT"))
except:
self.STOP_LOSS_PERCENT = STOP_LOSS_PERCENT
try:
self.TAKE_PROFIT_PERCENT = \
float(self.GetParameter("TAKE_PROFIT_PERCENT"))
except:
self.TAKE_PROFIT_PERCENT = TAKE_PROFIT_PERCENT
try:
self.TRAILING_STOP_PROFIT_TRIGGER = \
float(self.GetParameter("TRAILING_STOP_PROFIT_TRIGGER"))
except:
self.TRAILING_STOP_PROFIT_TRIGGER = TRAILING_STOP_PROFIT_TRIGGER
try:
self.TRAILING_STOP_PERCENT = \
float(self.GetParameter("TRAILING_STOP_PERCENT"))
except:
self.TRAILING_STOP_PERCENT = TRAILING_STOP_PERCENT
# # Verify MA periods - slow must be 2x or more fast
# if self.MA_SLOW_PERIOD < 2*self.MA_FAST_PERIOD:
# raise ValueError(
# f"Invalid MA Periods: fast={self.MA_FAST_PERIOD}, "
# f"slow={self.MA_SLOW_PERIOD}"
# )
#------------------------------------------------------------------------------
def AddInstrumentData(self):
"""Add instrument data to the algo."""
# Set data resolution based on input
if DATA_RESOLUTION == 'SECOND':
self.resolution = Resolution.Second
self.option_resolution = Resolution.Minute
elif DATA_RESOLUTION == 'MINUTE':
self.resolution = Resolution.Minute
self.option_resolution = Resolution.Minute
elif DATA_RESOLUTION == 'HOUR':
self.resolution = Resolution.Hour
self.option_resolution = Resolution.Hour
# Add data for the TICKER
self.ticker = self.AddEquity(
TICKER, self.resolution, extendedMarketHours=True
).Symbol
# Create and save the SymbolData instance
self.symbol_data = SymbolData(self, self.ticker)
# Save a link to the benchmark and set benchmark
self.bm = self.ticker
self.SetBenchmark(self.bm)
#------------------------------------------------------------------------------
def ScheduleFunctions(self):
"""Scheduling the functions required by the algo."""
# Schedule benchmark end of day event 5 minutes after the close
# Used to plot the benchmark on the equity curve
self.Schedule.On(
self.DateRules.EveryDay(self.bm),
self.TimeRules.BeforeMarketClose(self.bm, -5),
self.BenchmarkOnEndOfDay
)
#-------------------------------------------------------------------------------
def MyLog(self, message):
"""Add algo time to log if live trading. Otherwise just log message."""
if self.LiveMode:
self.Log(f'{self.Time}: {message}')
else:
self.Log(message)
#------------------------------------------------------------------------------
def OnData(self, data):
"""Event handler for new data pumped into the algo."""
# Check for new dividends or splits for the active securities
if data.Dividends.Count > 0 or data.Splits.Count > 0:
# Check for a dividend
symbol_object = self.symbol_data.symbol_object
if data.Dividends.get(symbol_object):
# Get the dividend info
dividend = data.Dividends[symbol_object].Distribution
# Get last 2 daily prices
hist = self.History([symbol_object], 2, Resolution.Daily)
price = hist.iloc[-1]['close'] # [-1] for last
previous_close = hist.iloc[0]['close'] # [0] for first
# Calculate the dividend adjustment factor
af = (previous_close-dividend)/previous_close
# Adjust the SymbolData class indicators
self.symbol_data.adjust_indicators(af)
ticker = symbol_object.Value
self.MyLog(
f"Adjusted {ticker} indicators for dividend={dividend}, "
f"with an adjustment factor={af}"
)
# Check for a split
if data.Splits.get(symbol_object):
# Make sure the split has occured and not just the warning
# split.Type == 0 for warning
# split.Type == 1 for split occured
if data.Splits[symbol_object].Type == 1:
split = data.Splits[symbol_object].SplitFactor
# Adjust the SymbolData class indicators
self.symbol_data.adjust_indicators(split, split=True)
ticker = symbol_object.Value
self.MyLog(
f"Adjusted {ticker} indicators for split={split}"
)
# Check if we are constantly looking for signals
if CONSTANT_SIGNAL_CHECK:
# Check for trading signals once the algo is no longer warming up
# trading is allowed and the indicators are all ready to be used
if not self.IsWarmingUp and self.symbol_data.indicators_ready \
and self.symbol_data.trading and data.get(TICKER) is not None:
self.symbol_data.check_for_signals()
#------------------------------------------------------------------------------
def BenchmarkOnEndOfDay(self):
"""Event handler for end of trading day for the benchmark."""
self.PlotBenchmarkOnEquityCurve()
#------------------------------------------------------------------------------
def OnEndOfAlgorithm(self):
"""Built-in event handler for end of the backtest."""
# Plot the benchmark buy and hold value on the equity curve chart
self.PlotBenchmarkOnEquityCurve(force_plot=True)
self.MyLog(f"End of Backtest")
#-------------------------------------------------------------------------------
def OnOrderEvent(self, orderEvent):
"""Built-in event handler for orders."""
# Catch option exercise order
if 'Automatic Exercise' in orderEvent.Message:
# Get the option
option = orderEvent.Symbol.Value
raise ValueError(f"QC Model automatically exercised {option}")
# Skip if not filled
if orderEvent.Status != OrderStatus.Filled:
if self.LiveMode:
self.MyLog(f"New order event: {orderEvent}")
return
# Call on_order_event for the SymbolData class
self.symbol_data.on_order_event(orderEvent)
#------------------------------------------------------------------------------
def PlotBenchmarkOnEquityCurve(self, force_plot=False):
"""Plot the benchmark buy & hold value on the strategy equity chart."""
# Initially set percent change to zero
pct_change = 0
# Get today's daily prices
# history algo on QC github shows different formats that can be used:
'''https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/
HistoryAlgorithm.py'''
hist = self.History([self.bm], timedelta(1), Resolution.Daily)
# Make sure hist df is not empty
if not hist.empty:
# Get today's closing price
price = hist.iloc[-1]['close']
try:
# Calculate the percent change since the first price
pct_change = (price-self.bm_first_price)/self.bm_first_price
except:
# We have not created the first price variable yet
# Get today's open and save as the first price
self.bm_first_price = hist.iloc[-1]['open']
# Log the benchmark's first price for reference
self.MyLog(f"Benchmark first price = {self.bm_first_price}")
# Calculate the percent change since the first price
pct_change = (price-self.bm_first_price)/self.bm_first_price
# Calculate today's ending value if we have the % change from the start
if pct_change != 0:
bm_value = round(CASH*(1+pct_change),2)
# Plot every PLOT_EVERY_DAYS days
try:
# We've previously created the counter, so increment it by 1
self.bm_plot_counter += 1
# same as: self.bm_plot_counter = self.bm_plot_counter + 1
except:
# We've not created the counter, so set it to 1
self.bm_plot_counter = 1
# Check if it's time to plot the benchmark value
if self.bm_plot_counter == PLOT_EVERY_DAYS or force_plot:
# Plot the benchmark's value to the Strategy Equity chart
# Plot function requires passing the chart name, series name,
# then the value to plot
self.Plot('Strategy Equity', 'Benchmark', bm_value)
# Plot the account leverage
account_leverage = self.Portfolio.TotalHoldingsValue \
/ self.Portfolio.TotalPortfolioValue
self.Plot('Leverage', 'Leverge', account_leverage)
# Reset counter to 0
self.bm_plot_counter = 0
# Log benchmark's ending price for reference
if force_plot:
self.MyLog(
f"Benchmark's first price = {self.bm_first_price}"
)
self.MyLog(f"Benchmark's final price = {price}")
self.MyLog(f"Benchmark buy & hold value = {bm_value}")
###############################################################################
class CustomSecurityInitializer(BrokerageModelSecurityInitializer):
def __init__(
self, brokerage_model: IBrokerageModel,
security_seeder: ISecuritySeeder) -> None:
super().__init__(brokerage_model, security_seeder)
def Initialize(self, security: Security) -> None:
"""
Define models to be used for securities as they are added to the
algorithm's universe.
"""
# First, call the superclass definition
# This method sets the reality models of each security using the
# default reality models of the brokerage model
super().Initialize(security)
# Define the data normalization mode
if DATA_MODE == 'RAW':
security.SetDataNormalizationMode(DataNormalizationMode.Raw)
else:
raise ValueError(f"Invalid DATA_MODE: ({DATA_MODE})")
# Define the fee model to use for the security
# security.SetFeeModel()
# Define the slippage model to use for the security
# security.SetSlippageModel()
# Define the fill model to use for the security
# security.SetFillModel()
# Define the buying power model to use for the security
# security.SetBuyingPowerModel()from AlgorithmImports import *
"""
SPY Option Scalping Strategy
Using Stochastics
Version 1.0.3
Platform: QuantConnect
By: Aaron Eller
www.excelintrading.com
aaron@excelintrading.com
Revision Notes:
1.0.0 (04/25/2023) - Initial.
1.0.1 (04/28/2023) - Added take profit inputs and logic.
- Added trailing stop inputs and logic.
- Added entry and entry filter inputs and logic.
- Added special entry at open inputs and logic.
- Added SR_TOLERANCE_PERCENT input and logic.
- Added ENTRY_AT_EXTREMES input and logic.
- Added LATE_ENTRY_HOLD_OVERNIGHT input and logic.
1.0.2 (05/02/2023) - Updated to handle in the money strike selection.
- Added logic to not enter if not on a new bar.
- Added ENTRY_EXTREME_FILTER_OVERBOUGHT and
ENTRY_EXTREME_FILTER_OVERSOLD inputs.
1.0.3 (05/05/2023) - Removed USE_STOP_LOSS as input. Now always True.
- Added RISK_PER_TRADE to replace PS_PCT.
- Changed ENTRY_CROSSES_AFTER_EXTREME_ZONES to allow
entry at the open (9:30) based on the last pre-market
session bar.
- Added INDICATOR_UPDATE_START_TIME and
INDICATOR_UPDATE_STOP_TIME inputs and logic.
- Added EXIT_ON_OPEN input and logic.
- Added NullOptionAssignmentModel required for holding
overnight with options that become deep in the money.
Notes:
- How to determine / handle "trending" days?
- Uses extended market hours for indicators.
- Only trades during the regular session.
- Consider a max number of trades (or losing trades) per day
Issue:
-Option excerise model -> makes difficult to trade options deep ITM
https://www.quantconnect.com/forum/discussion/14806/avoid-options-automatic-exercise/p1
https://github.com/QuantConnect/Lean/issues/6390
References:
-QC (Lean) Class List
https://lean-api-docs.netlify.app/annotated.html
"""
# Standard library imports
import datetime as DT
###############################################################################
# Backtest inputs
START_DATE = "01-01-2023" # must be in "MM-DD-YYYY" format
END_DATE = None # must be in "MM-DD-YYYY" format or None
CASH = 100000 # starting portfolio value
TIMEZONE = "US/Eastern" # e.g. "US/Eastern", "US/Central", "US/Pacific"
#-------------------------------------------------------------------------------
# DATA INPUTS
# Define the data resolution to be fed to the algorithm
# Must be "SECOND", "MINUTE", or "HOUR"
# NOTE: Will use 'MINUTE' for options if 'SECOND' is selected (for underlying)
DATA_RESOLUTION = 'MINUTE'
# Model uses raw data normalization mode which is required for options
# Set the underlying ticker symbol
TICKER = "SPY"
# Define the bar to trade (number of minutes)
BAR_MINUTES = 30 # can be algorithm parameter / focus on the 15m or larger bars as they look like the most promising
# BAR_MINUTES_OPTIONS = [1,2,3,4,5,6,10,12,13,15,20,30] # 12 options
# Turn on checking for signals constantly (not every BAR)
# If False, will only check for signals once per BAR_MINUTES close
# This mostly affects the trailing stop logic
CONSTANT_SIGNAL_CHECK = True
# Define the higher timeframe bar used for entry filtering
# HIGHER_BAR_MINUTES = 60 # 2, 3, 4, 5, 6, 10, 13, 15, 20, 30, 60
# Start trading minutes after the market open
START_TRADING_MINUTES_AFTER_OPEN = 0 # can be algorithm parameter
# Number of minutes prior to the close to stop trading
STOP_TRADING_MINUTES_BEFORE_CLOSE = 0
# Turn on exiting positions at the end of the day
USE_EOD_EXIT = False
# Set the number of minutes prior to the close to exit positions
EOD_EXIT_MINUTES_BEFORE_CLOSE = 5
# Turn on exiting positions at the start of the day (if held overnight)
EXIT_ON_OPEN = True
# Set the number of minutes prior to the close to exit positions on expiry day
EXPIRY_DAY_EXIT_MINUTES_BEFORE_CLOSE = 5
#-------------------------------------------------------------------------------
# INDICATOR INPUTS
# Set the time to start allowing indicators to update (premarket session)
# TradingView premarket starts at 7am ET [DT.time(7,0)]
# QuantConnnect premarket data starts at 4am ET [DT.time(4,0)]
INDICATOR_UPDATE_START_TIME = DT.time(7,0)
# Set the time to stop allowing indicators to update (after hours session)
# TradingView after hours session ends at 6pm ET [DT.time(18,0)]
# QuantConnnect after hours session data ends at 8pm ET [DT.time(20,0)]
INDICATOR_UPDATE_STOP_TIME = DT.time(18,0)
# Define the Stochastic
STOCHASTIC_PERIOD = 14
STOCHASTIC_KPERIOD = 3
STOCHASTIC_DPERIOD = 3
# Define the oversold and overbought levels
OVERSOLD = 20
OVERBOUGHT = 80
# Salty Support/Resistance levels
# Uses daily bars
ATR_PERIOD = 14
# Define the support/resistance levels to use from previous daily close
SR_LEVELS = [
0.382, 0.5, 0.618, 0.786,
1.0, 1.236, 1.382, 1.5, 1.618, 1.786,
2.0, 2.236, 2.382, 2.5, 2.618, 2.786,
3.0
]
#-------------------------------------------------------------------------------
# OPTION INPUTS
# Set the target (minimum) number of days to the expiry
MIN_DAYS_TO_EXPIRY = 2 # can be algorithm parameter (1 as minimum)
# Set the number of strikes above and below from the underlying price
# for the options selected
# Negative value will be in the money instead of out of the money
OTM_STRIKES = -1 # can be algorithm parameter
# Turn on/off allowing long trades
LONGS_ALLOWED = True
# Turn on/off allowing short trades
SHORTS_ALLOWED = True
#-------------------------------------------------------------------------------
# POSITION SIZING INPUTS
# Set risk per trade
RISK_PER_TRADE = 0.02 # decimal percent, e.g. 0.02=2.0%
#-------------------------------------------------------------------------------
# ENTRY FILTERS
# Turn on/off filtering out entries after extreme moves
# Crosses with %K already beyond these levels show price has already
# made the initial profitable move.
# Require longs to only be valid if %K <= ENTRY_STOCHASTIC_K_LONG_MAX
# Require shorts to only be valid if %K >= ENTRY_STOCHASTIC_K_SHORT_MIN
ENTRY_NOT_AFTER_EXTREME_MOVE_FILTER = True
ENTRY_STOCHASTIC_K_LONG_MAX = 50
ENTRY_STOCHASTIC_K_SHORT_MIN = 50
# Only enter if outside of the oversold/overbought zones?
# For long entry, this requires %K to be outside of oversold zone
# For short entry, this requires %K to be outside of the overbought zone
ENTRY_OUTSIDE_EXTREME_ZONE_FILTER = False
ENTRY_EXTREME_FILTER_OVERBOUGHT = 80
ENTRY_EXTREME_FILTER_OVERSOLD = 20
# Do not exit position at end of day if triggered within set minutes of close
LATE_ENTRY_HOLD_OVERNIGHT = False
LATE_ENTRY_MINUTES_BEFORE_CLOSE = 90
#-------------------------------------------------------------------------------
# ENTRY SIGNALS
# Turn on/off entries on crossovers after oversold (long) and crossunders after
# overbought (short)
ENTRY_CROSSES_AFTER_EXTREME_ZONES = True
# Turn on/off entering on bar following crosses if previous bar cross entry
# was not allowed because %K was still in extreme zone
# ENTRY_AFTER_CROSS_IN_EXTREME_ZONE = True
# Turn on/off entry at open
# enter long if in premarket %K drops to oversold zone, then crosses over %D
# enter short if in premarket %K enters overbought zone, then crosses under %D
ENTRY_AT_OPEN = False
# Turn on/off entering immediately at extreme levels
ENTRY_AT_EXTREMES = False
# Enter long (calls) at extreme oversold level
ENTRY_LONG_OVERSOLD = 8
# Enter short (puts) at extreme overbought level
ENTRY_SHORT_OVERBOUGHT = 92
#-------------------------------------------------------------------------------
# EXIT INPUTS
# Exit long trade if Stochastic %K is >= overbought level
# Exit short trade if Stochastic %D is <= oversold level
EXIT_AT_STOCHASTIC_EXTREME = False
EXIT_STOCHASTIC_OVERBOUGHT = 88
EXIT_STOCHASTIC_OVERSOLD = 12
# Exit based on SR levels (for profit)
TAKE_PROFIT_AT_SR = False
# Set the minimum distance required to allow take profit exit
MIN_TAKE_PROFIT_AMOUNT = 0.5 # can be algorithm parameter
# Set the percentage difference allowed to trigger at levels
SR_TOLERANCE_PERCENT = 0.01 # as decimal percentage, e.g. 0.01=1.0%
# NOTE: percents below are all decimal percents
# e.g. 0.50=50.0%, 1.0=100.0%
# Set stop loss exit
STOP_LOSS_PERCENT = 0.5 # can be algorithm parameter
# Set take profit
USE_TAKE_PROFIT = True
TAKE_PROFIT_PERCENT = 2.0 # can be algorithm parameter
# Set trailing stop
USE_TRAILING_STOP = False
TRAILING_STOP_PROFIT_TRIGGER = 1.0 # can be algorithm parameter
TRAILING_STOP_PERCENT = 0.10 # can be algorithm parameter
#-------------------------------------------------------------------------------
# LOGGING DETAILS
# What logs to print?
PRINT_SIGNALS = True # print signals info
PRINT_ORDERS = True # print new orders
################################################################################
############################ END OF ALL USER INPUTS ############################
################################################################################
# VALIDATE USER INPUTS - DO NOT CHANGE BELOW!!!
#-------------------------------------------------------------------------------
# Verify start date
try:
START_DT = DT.datetime.strptime(START_DATE, '%m-%d-%Y')
except:
raise ValueError(
f"Invalid START_DATE format ({START_DATE}). Must be in MM-DD-YYYY "
"format."
)
# Verify end date
try:
if END_DATE:
END_DT = DT.datetime.strptime(END_DATE, '%m-%d-%Y')
except:
raise ValueError(
f"Invalid END_DATE format ({END_DATE}). Must be in MM-DD-YYYY "
"format or set to None to run to date."
)
#-------------------------------------------------------------------------------
# Verify DATA_RESOLUTION input
DATA_RESOLUTION = DATA_RESOLUTION.upper()
if DATA_RESOLUTION not in ['SECOND', 'MINUTE', 'HOUR']:
raise ValueError(
f"Invalid DATA_RESOLUTION ({DATA_RESOLUTION}). Must be 'SECOND', "
f"'MINUTE' or 'HOUR'."
)
#-------------------------------------------------------------------------------
# Set the data normalizaton mode for price data
# either 'ADJUSTED' (backtesting), or 'RAW' (necessary for live trading)
# Raw is always required for options!
DATA_MODE = 'RAW'
#-------------------------------------------------------------------------------
# Define benchmark equity
# Also used for scheduling functions, so make sure it has same trading hours
# as instruments traded.
BENCHMARK = TICKER
#-------------------------------------------------------------------------------
# Calculate the number of days in the backtest
PLOT_LIMIT = 4000
if not END_DATE:
today = DT.datetime.today()
BT_DAYS = (today-START_DT).days
else:
BT_DAYS = (END_DT-START_DT).days
# Convert calendar days to estimated market days
# Round up to the nearest integer
# This uses // for integer division which rounds down
# Take division for negative number to round up, then negate the negative
BT_DAYS = -(-BT_DAYS*252//365)
# Calculate the frequency of days that we can create a new plot
# Use the same approach as above to round up to the nearest integer
PLOT_EVERY_DAYS = -(-BT_DAYS//PLOT_LIMIT)# Standard library imports
import datetime as DT
# from datetime import timedelta
# from datetime import date
# from dateutil.parser import parse
# import decimal
import math
# import numpy as np
# import pandas as pd
import pytz
from System.Drawing import Color
# QuantConnect specific imports
from AlgorithmImports import *
# import QuantConnect as qc
# Import from files
from custom_models import CustomExerciseModel
from notes_and_inputs import *
################################################################################
class SymbolData(object):
"""Class to store data for a specific symbol."""
def __init__(self, algo, symbol_object):
"""Initialize SymbolData object."""
# Save a reference to the QCAlgorithm class
self.algo = algo
# Save the string symbol and the .Symbol object
self.symbol = symbol_object.Value
self.symbol_object = symbol_object
# Get the symbol's exchange market info
self.get_exchange_info()
# Add strategy variables
self.add_strategy_variables()
# Add the bars and indicators required
self.add_bars()
self.add_indicators()
# Schedule functions
self.schedule_functions()
#-------------------------------------------------------------------------------
def get_exchange_info(self):
"""Get the security's exchange info."""
# Get the SecurityExchangeHours Class object for the symbol
self.exchange_hours = self.algo.Securities[self.symbol].Exchange.Hours
# Create a datetime I know the market was open for the full day
dt = DT.datetime(2021, 1, 4)
# Get the next open datetime from the SecurityExchangeHours Class
mkt_open_dt = self.exchange_hours.GetNextMarketOpen(dt, False)
# Save the typical market open and close times
self.mkt_open = mkt_open_dt.time()
mkt_close_dt = self.exchange_hours.GetNextMarketClose(dt, False)
self.mkt_close = mkt_close_dt.time()
# Get the exchange timezone
self.mkt_tz = pytz.timezone(str(self.exchange_hours.TimeZone))
# Create pytz timezone objects for the exchange tz and local tz
exchange_tz = self.mkt_tz
local_tz = pytz.timezone(TIMEZONE)
# Get the difference in the timezones
# REF: http://pytz.sourceforge.net/#tzinfo-api
# for pytz timezone.utcoffset() method
# 3600 seconds/hour
exchange_utc_offset_hrs = int(exchange_tz.utcoffset(dt).seconds/3600)
local_utc_offset_hrs = int(local_tz.utcoffset(dt).seconds/3600)
self.offset_hrs = exchange_utc_offset_hrs-local_utc_offset_hrs
# NOTE: offset hours are very helpful if you want to schedule functions
# around market open/close times
# Get the market close time for the local time zone
self.mkt_close_local_tz = \
(mkt_close_dt-DT.timedelta(hours=self.offset_hrs)).time()
#-------------------------------------------------------------------------------
def add_bars(self):
"""Add bars required."""
# Create the desired custom bar consolidator for the symbol
consolidator = TradeBarConsolidator(timedelta(minutes=self.algo.BAR_MINUTES))
# Create an event handler to be called on each new consolidated bar
consolidator.DataConsolidated += self.on_data_consolidated
# Link the consolidator with our symbol and add it to the algo manager
self.algo.SubscriptionManager.AddConsolidator(self.symbol, consolidator)
# Save the consolidator
self.consolidator = consolidator
# Create the daily bar consolidator (used for S/R levels)
daily_consolidator = TradeBarConsolidator(self.daily_calendar)
# Create an event handler to be called on each new consolidated bar
daily_consolidator.DataConsolidated += self.on_daily_consolidated
# Link daily_consolidator with our symbol and add it to the algo manager
self.algo.SubscriptionManager.AddConsolidator(
self.symbol, daily_consolidator
)
# Save daily_consolidator link so we can remove it when necessary
self.daily_consolidator = daily_consolidator
#-------------------------------------------------------------------------------
def daily_calendar(self, dt):
"""
Set up daily consolidator calendar info for the US equity market.
This should return a start datetime object that is timezone unaware
with a valid date/time for the desired securities' exchange's time zone.
"""
# Need to handle case where algo initializes and this function is called
# for the first time.
if not self.calendar_initialized:
# Since this doesn't matter, we'll pass dt as start and one day
# as the timedelta until end_dt
start_dt = dt
end_dt = start_dt + DT.timedelta(1)
self.calendar_initialized = True
return CalendarInfo(start_dt, end_dt-start_dt)
# Create a datetime.datetime object to represent the market open for the
# **EXCHANGE** timezone
start = dt.replace(
hour=self.mkt_open.hour,
minute=self.mkt_open.minute,
second=0,
microsecond=0
)
# Get today's end time from the SecurityExchangeHours Class object
end = self.exchange_hours.GetNextMarketClose(start, False)
# Catch when start is after the passed dt
# QC now throws an error in this case
if start > dt:
# To handle the QC error, pass period for no data
# Set the end to be the next desired start
end = start
# And set start to dt to avoid QC throwing error
start = dt
# This will result in the next dt being the desired start time
# Return the start datetime and the consolidation period
return CalendarInfo(start, end-start)
#-------------------------------------------------------------------------------
def add_indicators(self):
"""Add indicators required."""
# Create an empty list to hold all indicators
# Will add (indicator, update_method) tuples
# where update_method is either 'high', 'low', 'close' or 'bar'
self.indicators = []
# Create the Stochastic
self.stochastic = Stochastic(
STOCHASTIC_PERIOD, STOCHASTIC_KPERIOD, STOCHASTIC_DPERIOD
)
self.indicators.append((self.stochastic, 'bar'))
# Keep track of the previous stochastic %K value
self.previous_stochastic_k = None
# Keep track of %K crossing %D
self.k_gt_d = RollingWindow[bool](2)
self.k_lt_d = RollingWindow[bool](2)
# Create a chart to plot the Stochastic
self.chart_name = f'{self.symbol} Stochastics Plot'
chart = Chart(self.chart_name)
# Add series to the chart for the cmf
chart.AddSeries(
Series('%K', SeriesType.Line, '$', Color.Blue, ScatterMarkerSymbol.Circle)
)
chart.AddSeries(
Series('%D', SeriesType.Line, '$', Color.Orange, ScatterMarkerSymbol.Circle)
)
# Create a list to hold the daily indicators
self.daily_indicators = []
# Create the Average True Range
self.atr = AverageTrueRange(ATR_PERIOD)
self.daily_indicators.append((self.atr, 'bar'))
# Initialize the S/R levels
self.support = {}
self.resistance = {}
for percentage in SR_LEVELS:
self.support[percentage] = None
self.resistance[percentage] = None
# Get the min number of bars required
indicator_bars = [
int(3.0*(STOCHASTIC_PERIOD+STOCHASTIC_KPERIOD+STOCHASTIC_DPERIOD))
]
self.min_bars = max(indicator_bars)
# Keep a rolling window of bars
self.bar_window = RollingWindow[TradeBar](self.min_bars)
# Custom rolling windows
self.ma_fast_gt_slow = RollingWindow[bool](2)
self.ma_fast_lt_slow = RollingWindow[bool](2)
# Get the min number of daily bars required
daily_indicator_bars = [int(3.0*ATR_PERIOD)]
self.min_daily_bars = max(daily_indicator_bars)
# Keep a rolling window of daily bars
self.daily_bar_window = RollingWindow[TradeBar](self.min_daily_bars)
# Warm up the indicators with historical data
self.warmup_indicators()
#-------------------------------------------------------------------------------
def add_strategy_variables(self):
"""Add strategy specific variables."""
self.option_expiry = None
self.option_data_subcriptions = []
self.reset_trade_variables()
self.trading = False
self.end_of_day_exit_ignore = False
self.previous_bar_crossover = False
self.previous_bar_crossunder = False
# Keep track of calendar initialized
self.calendar_initialized = False
# Keep track of premarket variables
self.reset_premarket_variables()
#-------------------------------------------------------------------------------
def reset_premarket_variables(self):
"""Reset variables used for premarket tracking."""
self.premarket_overbought = False
self.premarket_oversold = False
self.premarket_crossover = False
self.premarket_crossunder = False
#-------------------------------------------------------------------------------
def reset_trade_variables(self):
"""Reset trade specific variables."""
self.stop_loss_price = None
self.stop_loss_order = None
self.trailing_stop_activation_price = None
self.trade_best_price = None
self.trailing_stop_activated = False
self.take_profit_order = None
self.call = None
self.put = None
self.position = None
self.price_at_entry = None
#-------------------------------------------------------------------------------
def schedule_functions(self):
"""Schedule functions required for the class."""
# Start trading
self.algo.Schedule.On(
self.algo.DateRules.EveryDay(self.symbol_object),
self.algo.TimeRules.AfterMarketOpen(
self.symbol_object,
self.algo.START_TRADING_MINUTES_AFTER_OPEN
),
self.start_trading
)
# Stop trading
self.algo.Schedule.On(
self.algo.DateRules.EveryDay(self.symbol_object),
self.algo.TimeRules.BeforeMarketClose(
self.symbol_object,
STOP_TRADING_MINUTES_BEFORE_CLOSE
),
self.stop_trading
)
# End of day exit
if USE_EOD_EXIT:
self.algo.Schedule.On(
self.algo.DateRules.EveryDay(self.symbol_object),
self.algo.TimeRules.BeforeMarketClose(
self.symbol_object,
EOD_EXIT_MINUTES_BEFORE_CLOSE
),
self.end_of_day_exit
)
# On expiry day exit
self.algo.Schedule.On(
self.algo.DateRules.EveryDay(self.symbol_object),
self.algo.TimeRules.BeforeMarketClose(
self.symbol_object,
EXPIRY_DAY_EXIT_MINUTES_BEFORE_CLOSE
),
self.expiry_day_end_of_day_exit
)
# On market open exit
if EXIT_ON_OPEN:
self.algo.Schedule.On(
self.algo.DateRules.EveryDay(self.symbol_object),
self.algo.TimeRules.AfterMarketOpen(self.symbol_object, -10),
self.go_flat
)
# After market close
self.algo.Schedule.On(
self.algo.DateRules.EveryDay(self.symbol_object),
self.algo.TimeRules.BeforeMarketClose(self.symbol_object, -30),
self.after_market_close
)
#-------------------------------------------------------------------------------
def start_trading(self):
"""Time to start trading."""
self.trading = True
#-------------------------------------------------------------------------------
def stop_trading(self):
"""Time to stop trading."""
if self.trading:
self.trading = False
# Reset premarket variables
self.reset_premarket_variables()
#-------------------------------------------------------------------------------
def end_of_day_exit(self):
"""Exit open trades at the end of the day and stop trading."""
self.stop_trading()
if not self.end_of_day_exit_ignore:
self.go_flat()
#-------------------------------------------------------------------------------
def expiry_day_end_of_day_exit(self):
"""
Exit open trades at the end of the day if expiry day and stop trading.
"""
# Catch if the current expiration is today
if self.option_expiry == self.algo.Time.date():
# Trigger the end of day exit logic
self.end_of_day_exit()
#-------------------------------------------------------------------------------
def after_market_close(self):
"""Function called 30minutes after the market close."""
# Catch if the current expiration is today
# if self.option_expiry == self.algo.Time.date():
# Set it back to None
self.option_expiry = None
#-------------------------------------------------------------------------------
def on_data_consolidated(self, sender, bar):
"""Event handler for desired custom bars."""
# Manually update all of the indicators
# Ignore if bar start time is before the start time
if bar.Time.time() < INDICATOR_UPDATE_START_TIME:
return
# Ignore if bar end time is after stop time
elif bar.EndTime.time() > INDICATOR_UPDATE_STOP_TIME:
return
self.update_indicators(bar)
# Keep track of potential options to trade
if not self.algo.IsWarmingUp and not self.warming_up:
self.track_options()
# Check for trading signals once the algo is no longer warming up,
# indicators are ready, and trading is allowed
if not self.algo.IsWarmingUp and not self.warming_up \
and self.indicators_ready and self.trading:
self.check_for_signals(new_bar=True)
# Update self.previous_bar_crossover and self.previous_bar_crossunder
self.previous_bar_crossover = self.stochastic_crossover
self.previous_bar_crossunder = self.stochastic_crossunder
#-------------------------------------------------------------------------------
def on_daily_consolidated(self, sender, bar):
"""Event handler for daily bars."""
# Manually update all of the daily indicators
self.update_daily_indicators(bar)
#-------------------------------------------------------------------------------
def track_options(self):
"""Track options to consider trading."""
# Get a list of the available options
options = self.algo.OptionChainProvider.GetOptionContractList(
self.symbol, self.algo.Time
)
# Get the new expiration date
if self.option_expiry is None:
# Get the possible expirations that meet MIN_DAYS_TO_EXPIRY
min_dt = self.algo.Time.date() + DT.timedelta(days=self.algo.MIN_DAYS_TO_EXPIRY)
expiries = [
x.ID.Date.date() for x in options if x.ID.Date.date() >= min_dt
]
# Get unique expiries as list
expiries = list(set(expiries))
# Sort list earliest to latest
expiries.sort()
# Use the first valid expiry
self.option_expiry = expiries[0]
# Check if we have the desired expiration
if self.option_expiry:
expiry = self.option_expiry
# Get a list of strikes for only the desired expiry
call_strikes = [
x.ID.StrikePrice for x in options \
if x.ID.Date.date() == expiry and x.ID.OptionRight == 0
]
put_strikes = [
x.ID.StrikePrice for x in options \
if x.ID.Date.date() == expiry and x.ID.OptionRight == 1
]
# Get a list of matching strikes - valid for calls and puts both
# This is an intersection
call_strikes = set(call_strikes)
put_strikes = set(put_strikes)
strikes = list(call_strikes.intersection(put_strikes))
# Return if no strikes
if len(strikes) == 0:
return
# Otherwise sort the list
strikes.sort()
# Get the underlying price and differences to it for all strikes
price = self.price
strike_differences = [(x, price-x) for x in strikes]
# Sort the lists by smallest difference from underlying price
strike_differences.sort(key=lambda x:x[1])
# Get the ATM strike
atm_strike_difference = min([abs(x[1]) for x in strike_differences])
try:
atm_strike_index = [x[1] for x in strike_differences].index(
atm_strike_difference
)
except:
try:
atm_strike_index = [x[1] for x in strike_differences].index(
-atm_strike_difference
)
except:
self.MyLog(f"Error getting the ATM strike!")
return
atm_strike = strike_differences[atm_strike_index][0]
atm_strike_index = strikes.index(atm_strike)
# Get the possible OTM call and put strikes
otm_strike_int = max(2, abs(self.algo.OTM_STRIKES)) # use this as a minimum
# We'll go otm_strike_int on either side of the current target
# just in case the underlying moves significantly before we
# enter the trade.
diff = int(otm_strike_int*2)
possible_call_strikes = strikes[
atm_strike_index-diff:atm_strike_index+diff+1
]
possible_call_strikes.sort()
possible_put_strikes = strikes[
atm_strike_index-diff:atm_strike_index+diff+1
]
possible_put_strikes.sort()
# Create empty lists for possible calls and puts
self.possible_calls = []
self.possible_puts = []
# Subscribe to the possible options to buy
for strike in possible_call_strikes:
# Get the target call and subscribe to it's data
calls = [
x for x in options \
if x.ID.Date.date() == expiry and x.ID.OptionRight == 0 \
and x.ID.StrikePrice == strike
]
if len(calls) > 0:
call = calls[0]
# Add option contract, if it's not currently subscribed to
if call not in self.option_data_subcriptions:
self.algo.AddOptionContract(
call, self.algo.option_resolution
)
self.option_data_subcriptions.append(call)
# below doesn't seem to be working...
security = self.algo.Securities[call]
security.SetOptionAssignmentModel(
NullOptionAssignmentModel()
)
security.SetOptionExerciseModel(CustomExerciseModel())
# option = self.AddOption("SPY")
self.possible_calls.append(call)
for strike in possible_put_strikes:
# Get the target put and subscribe to it's data
puts = [
x for x in options \
if x.ID.Date.date() == expiry and x.ID.OptionRight == 1 \
and x.ID.StrikePrice == strike
]
if len(puts) > 0:
put = puts[0]
# Add option contract, if it's not currently subscribed to
if put not in self.option_data_subcriptions:
self.algo.AddOptionContract(
put, self.algo.option_resolution
)
self.option_data_subcriptions.append(put)
# below doesn't seem to be working...
security = self.algo.Securities[put]
security.SetOptionAssignmentModel(
NullOptionAssignmentModel()
)
security.SetOptionExerciseModel(CustomExerciseModel())
self.possible_puts.append(put)
#-------------------------------------------------------------------------------
@property
def indicators_ready(self):
"""Check if all of the indicators used are ready (warmed up)."""
# Loop through all indicators
for indicator, update_method in self.indicators:
# Return False if the indicator is not ready
if not indicator.IsReady:
return False
# for indicator, update_method in self.higher_tf_indicators:
# # Return False if the indicator is not ready
# if not indicator.IsReady:
# return False
for indicator, update_method in self.daily_indicators:
# Return False if the indicator is not ready
if not indicator.IsReady:
return False
# Check any RollingWindows
if not self.bar_window.IsReady:
return False
# Check any custom RollingWindows
if not self.k_gt_d.IsReady:
return False
if not self.k_lt_d.IsReady:
return False
# Otherwise all indicators are ready, so return True
return True
#-------------------------------------------------------------------------------
def update_indicators(self, bar):
"""Manually update all of the symbol's indicators."""
# Keep track of self.previous_stochastic_k
if self.stochastic.IsReady:
self.previous_stochastic_k = self.stochastic.StochK.Current.Value
# Plot if not warming up
if not self.warming_up:
k = self.stochastic.StochK.Current.Value
d = self.stochastic.StochD.Current.Value
self.algo.Plot(self.chart_name, '%K', k)
self.algo.Plot(self.chart_name, '%D', d)
# If the indicator only needs a single value (like close), pass the time
# and the value to the Update() method.
# e.g. indicator.Update(bar.EndTime, bar.Close)
# If the indicator needs multiple parts of a bar like the open, high,
# low, and close, only pass the bar to the Update() method.
# e.g. indicator.Update(bar)
# Loop through the indicators
for indicator, update_method in self.indicators:
if update_method == 'close':
indicator.Update(bar.EndTime, bar.Close)
elif update_method == 'high':
indicator.Update(bar.EndTime, bar.High)
elif update_method == 'low':
indicator.Update(bar.EndTime, bar.Low)
elif update_method == 'bar':
indicator.Update(bar)
# Handle any RollingWindows below
# Update the bar window
self.bar_window.Add(bar)
# Handle custom RollingWindows
if self.stochastic.IsReady:
k = self.stochastic.StochK.Current.Value
d = self.stochastic.StochD.Current.Value
self.k_gt_d.Add(k > d)
self.k_lt_d.Add(k < d)
# Handle entry at open variables
if not self.warming_up and ENTRY_AT_OPEN \
and bar.EndTime.time() <= DT.time(9,30):
# Bar is still premarket
# Get the %K and %D values
k = self.stochastic.StochK.Current.Value
d = self.stochastic.StochD.Current.Value
# Check for crosses
if self.stochastic_crossover:
self.premarket_crossover = True
elif self.stochastic_crossunder:
self.premarket_crossunder = True
# Check for k overbought or oversold
if k >= OVERBOUGHT:
self.premarket_overbought = True
self.premarket_oversold = False
# Also reset crossunder
self.premarket_crossunder = False
elif k <= OVERSOLD:
self.premarket_overbought = False
self.premarket_oversold = True
# Also reset crossover
self.premarket_crossover = False
#-------------------------------------------------------------------------------
def update_daily_indicators(self, bar):
"""Manually update all of the symbol's daily indicators."""
# Loop through the indicators
for indicator, update_method in self.daily_indicators:
if update_method == 'close':
indicator.Update(bar.EndTime, bar.Close)
elif update_method == 'high':
indicator.Update(bar.EndTime, bar.High)
elif update_method == 'low':
indicator.Update(bar.EndTime, bar.Low)
elif update_method == 'bar':
indicator.Update(bar)
# Handle any RollingWindows below
# Update the bar window
self.daily_bar_window.Add(bar)
# Check if the ATR is ready
if self.atr.IsReady:
atr = self.atr.Current.Value
# Calculate the support/resistance levels
close = bar.Close
for percentage in SR_LEVELS:
self.support[percentage] = close - atr*percentage
self.resistance[percentage] = close + atr*percentage
#-------------------------------------------------------------------------------
def adjust_indicators(self, adjustment, split=False):
"""Adjust all indicators for splits or dividends."""
# Need to update to handle split -> volume handled differently
# Get a list of the current bars
bars = list(self.bar_window)
# Current order is newest to oldest (default for rolling window)
# Reverse the list to be oldest to newest
bars.reverse()
# Reset all indicators
self.reset_indicators()
# Loop through the bars from oldest to newest
for bar in bars:
# Adjust the bar by the adjustment factor
bar.Open *= adjustment
bar.High *= adjustment
bar.Low *= adjustment
bar.Close *= adjustment
# Use the bar to update the indicators
# This also adds the bar to the rolling window
self.update_indicators(bar)
# Get a list of the current daily bars
daily_bars = list(self.daily_bar_window)
# Current order is newest to oldest (default for rolling window)
# Reverse the list to be oldest to newest
daily_bars.reverse()
# Reset all indicators
self.reset_daily_indicators()
# Loop through the bars from oldest to newest
for bar in daily_bars:
# Adjust the bar by the adjustment factor
bar.Open *= adjustment
bar.High *= adjustment
bar.Low *= adjustment
bar.Close *= adjustment
# Use the bar to update the indicators
# This also adds the bar to the rolling window
self.update_daily_indicators(bar)
#-------------------------------------------------------------------------------
def reset_indicators(self):
"""Manually reset all of the indicators."""
# Loop through all indicators and reset them
for indicator, update_method in self.indicators:
indicator.Reset()
# Reset the bar window
self.bar_window.Reset()
#-------------------------------------------------------------------------------
def reset_daily_indicators(self):
"""Manually reset all of the daily indicators."""
# Loop through all indicators and reset them
for indicator, update_method in self.daily_indicators:
indicator.Reset()
# Reset the bar window
self.daily_bar_window.Reset()
#-------------------------------------------------------------------------------
def warmup_indicators(self):
"""Warm up indicators using historical data."""
self.warming_up = True
# Get a list of historical minute trade bars
# num_bars = int(2.0*self.min_bars*HIGHER_BAR_MINUTES)
num_bars = int(2.0*self.min_bars*self.algo.BAR_MINUTES)
bars = self.algo.History[TradeBar](
self.symbol_object,
num_bars,
Resolution.Minute
)
# Loop through the bars and update the consolidator
for bar in bars:
self.consolidator.Update(bar)
# self.higher_consolidator.Update(bar)
# Get a list of historical daily trade bars
daily_bars = self.algo.History[TradeBar](
self.symbol_object,
self.min_daily_bars,
Resolution.Daily
)
# Loop through the bars and update the consolidator
for bar in daily_bars:
# Instead of passing to consolidator, pass directly to event handler
self.on_daily_consolidated(None, bar)
self.warming_up = False
# Throw error if indicators are not ready when backtesting
if not self.indicators_ready and not self.algo.LiveMode:
raise
#-------------------------------------------------------------------------------
def check_trailing_stop(self):
"""Check if we need to updated the trailing stop."""
# Get the current option price
option_price = self.algo.Securities[self.option].Price
# Wait until we initialize the trade best price
if self.trade_best_price is None:
self.trade_best_price = option_price
# Check for new best trade price
if option_price > self.trade_best_price:
self.trade_best_price = option_price
# Check if the trailing stop has not been activated
if not self.trailing_stop_activated:
# Check if it's now triggered
if option_price >= self.trailing_stop_activation_price:
self.trailing_stop_activated = True
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.option} trailing stop activated. Price "
f"({option_price}) >= activation price "
f"({self.trailing_stop_activation_price})"
)
# Skip if the trailing stop is not activated
if not self.trailing_stop_activated:
return
# Otherwise check for new trailing stop price
trailing_stop_price = round(
self.trade_best_price*(1-self.algo.TRAILING_STOP_PERCENT),2
)
# Update the stop loss order on a higher trailing stop price
if trailing_stop_price > self.stop_loss_price:
# Update the order
self.update_stop_order(trailing_stop_price)
#-------------------------------------------------------------------------------
def check_for_signals(self, new_bar=False):
"""Check for new trading signals."""
# Debug
# if self.algo.Time.month == 2 and self.algo.Time.day >= 3:
# print('debug')
# Get the current price. Return if not valid.
price = self.price
if price is None:
return
# Check for trailing stop and an active position
if USE_TRAILING_STOP and self.position != None:
self.check_trailing_stop()
# Check for long position
if self.position == 'long':
# Check for long exit signal
if self.long_exit_signal(price):
self.go_flat()
# Check for short position
elif self.position == 'short':
# Check for short exit signal
if self.short_exit_signal(price):
self.go_flat()
# Otherwise flat
# not using else here to catch after exit triggered
# but only on a new bar!
if self.position is None and new_bar:
# Check for long entry signal
if self.long_entry_signal(price):
# Save price at entry
self.price_at_entry = price
self.go_long()
# Check for short entry signal
elif self.short_entry_signal(price):
# Save price at entry
self.price_at_entry = price
self.go_short()
#-------------------------------------------------------------------------------
def long_entry_signal(self, price):
"""Check if there is a valid long entry signal."""
# First apply filters to disqualify entry
# Require longs allowed
if not LONGS_ALLOWED:
return False
# Require the current k to be <= ENTRY_STOCHASTIC_K_LONG_MAX
k = self.stochastic.StochK.Current.Value
d = self.stochastic.StochD.Current.Value
if ENTRY_NOT_AFTER_EXTREME_MOVE_FILTER:
if k > ENTRY_STOCHASTIC_K_LONG_MAX:
return False
# Require the current k to be > oversold zone
if ENTRY_OUTSIDE_EXTREME_ZONE_FILTER:
if k <= ENTRY_EXTREME_FILTER_OVERSOLD:
return False
# Working here...
# # Next look for specific entry criteria
# if ENTRY_AFTER_CROSS_IN_EXTREME_ZONE:
# # Check if previous bar was a crossover
# # and Stochastic %K still > %D
# if self.previous_bar_crossover and (k>d):
# #
if ENTRY_CROSSES_AFTER_EXTREME_ZONES \
and self.algo.Time.time() >= DT.time(9,30):
# Require previous Stochastic %K below OVERSOLD
# and Stochastic %K crossing over Stochastic %D
if (self.previous_stochastic_k < OVERSOLD) \
and self.stochastic_crossover:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} LONG ENTRY SIGNAL: Stochastic %K "
f"({k:.2f}) crossing over %D ({d:.2f}) and previous %K="
f"{self.previous_stochastic_k:.2f}, price={price}"
)
return True
if ENTRY_AT_OPEN and self.algo.Time.time() < DT.time(9,31):
# Vaild if premarket oversold followed by crossover
# Also make sure that K is still over D
if self.premarket_oversold and self.premarket_crossover and (k>d):
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} LONG ENTRY SIGNAL: Stochastic %K "
f"oversold in premarket then crossed over %D. K "
f"({k:.2f}) still > %D ({d:.2f}), price={price}"
)
return True
if ENTRY_AT_EXTREMES:
# Enter if k drops <= ENTRY_LONG_OVERSOLD
if k <= ENTRY_LONG_OVERSOLD:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} LONG ENTRY SIGNAL: Stochastic %K "
f"extremely oversold ({k:.2f}) <= {ENTRY_LONG_OVERSOLD},"
f" price={price}"
)
return True
# If not triggered above, then entry not valid
return False
#-------------------------------------------------------------------------------
def short_entry_signal(self, price):
"""Check if there is a valid short entry signal."""
# # Debug
# if self.algo.Time.month == 2 and self.algo.Time.day >= 9:
# print('debug')
# First apply filters to disqualify entry
# Require shorts allowed
if not SHORTS_ALLOWED:
return False
# Require the current k to be >= ENTRY_STOCHASTIC_K_SHORT_MIN
k = self.stochastic.StochK.Current.Value
d = self.stochastic.StochD.Current.Value
if ENTRY_NOT_AFTER_EXTREME_MOVE_FILTER:
if k < ENTRY_STOCHASTIC_K_SHORT_MIN:
return False
# Require the current k to be < overbought zone
if ENTRY_OUTSIDE_EXTREME_ZONE_FILTER:
if k >= ENTRY_EXTREME_FILTER_OVERBOUGHT:
return False
# Next look for specific entry criteria
if ENTRY_CROSSES_AFTER_EXTREME_ZONES \
and self.algo.Time.time() >= DT.time(9,30):
# Require previous Stochastic %K above OVERBOUGHT
# and Stochastic %K crossing over Stochastic %D
if (self.previous_stochastic_k > OVERBOUGHT) \
and self.stochastic_crossunder:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} SHORT ENTRY SIGNAL: Stochastic %K "
f"({k:.2f}) crossing under %D ({d:.2f}) and previous "
f"%K={self.previous_stochastic_k:.2f}, price={price}"
)
return True
if ENTRY_AT_OPEN and self.algo.Time.time() < DT.time(9,31):
# Vaild if premarket overbought followed by crossunder
# Also make sure that K is still below D
if self.premarket_overbought and self.premarket_crossunder and (k<d):
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} SHORT ENTRY SIGNAL: Stochastic %K "
f"overbought in premarket then crossed under %D. K "
f"({k:.2f}) still < %D ({d:.2f}), price={price}"
)
return True
if ENTRY_AT_EXTREMES:
# Enter if k rises >= ENTRY_SHORT_OVERBOUGHT
if k >= ENTRY_SHORT_OVERBOUGHT:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} LONG ENTRY SIGNAL: Stochastic %K "
f"extremely overbought ({k:.2f}) >= "
f"{ENTRY_SHORT_OVERBOUGHT}, price={price}"
)
return True
# If not triggered above, then entry not valid
return False
#-------------------------------------------------------------------------------
def long_exit_signal(self, price):
"""Check if there is a valid long exit signal."""
# Valid if the Stochastic %K is >= EXIT_STOCHASTIC_OVERBOUGHT
k = self.stochastic.StochK.Current.Value
if k >= EXIT_STOCHASTIC_OVERBOUGHT and EXIT_AT_STOCHASTIC_EXTREME:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} LONG EXIT SIGNAL: Stochastic %K ({k:.2f}) "
f">= {EXIT_STOCHASTIC_OVERBOUGHT}"
)
return True
# Valid on profitable move to support/resistance level
if TAKE_PROFIT_AT_SR:
# Loop through SR levels
for percentage in SR_LEVELS:
# Ignore if the profit distance is too small
support = self.support[percentage]*(1-SR_TOLERANCE_PERCENT)
diff = support-self.price_at_entry
if diff > self.algo.MIN_TAKE_PROFIT_AMOUNT:
# Check if we've triggered at this level
if price >= support:
# Log message when desired
if PRINT_SIGNALS:
pct = round(percentage*100.0,3)
self.algo.MyLog(
f"{self.symbol} LONG EXIT SIGNAL: close "
f"({price}) >= S[{pct}] ({support})"
)
return True
resistance = self.resistance[percentage]*(1-SR_TOLERANCE_PERCENT)
diff = resistance-self.price_at_entry
if diff > self.algo.MIN_TAKE_PROFIT_AMOUNT:
# Check if we've triggered at this level
if price >= resistance:
# Log message when desired
if PRINT_SIGNALS:
pct = round(percentage*100.0,3)
self.algo.MyLog(
f"{self.symbol} LONG EXIT SIGNAL: close "
f"({price}) >= R[{pct}] ({resistance})"
)
return True
# Otherwise False
return False
#-------------------------------------------------------------------------------
def short_exit_signal(self, price):
"""Check if there is a valid short exit signal."""
# Valid if the Stochastic %K is <= EXIT_STOCHASTIC_OVERSOLD
k = self.stochastic.StochK.Current.Value
if k <= EXIT_STOCHASTIC_OVERSOLD and EXIT_AT_STOCHASTIC_EXTREME:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.symbol} SHORT EXIT SIGNAL: Stochastic %K ({k:.2f}) "
f"<= {EXIT_STOCHASTIC_OVERSOLD}"
)
return True
# Valid on profitable move to support/resistance level
if TAKE_PROFIT_AT_SR:
# Loop through SR levels
for percentage in SR_LEVELS:
# Ignore if the profit distance is too small
support = self.support[percentage]*(1+SR_TOLERANCE_PERCENT)
diff = self.price_at_entry-support
if diff > self.algo.MIN_TAKE_PROFIT_AMOUNT:
# Check if we've triggered at this level
if price <= support:
# Log message when desired
if PRINT_SIGNALS:
pct = round(percentage*100.0,3)
self.algo.MyLog(
f"{self.symbol} SHORT EXIT SIGNAL: close "
f"({price}) <= S[{pct}] ({support})"
)
return True
resistance = self.resistance[percentage]*(1+SR_TOLERANCE_PERCENT)
diff = self.price_at_entry-resistance
if diff > self.algo.MIN_TAKE_PROFIT_AMOUNT:
# Check if we've triggered at this level
if price <= resistance:
# Log message when desired
if PRINT_SIGNALS:
pct = round(percentage*100.0,3)
self.algo.MyLog(
f"{self.symbol} SHORT EXIT SIGNAL: close "
f"({price}) <= R[{pct}] ({resistance})"
)
return True
# Otherwise False
return False
#-------------------------------------------------------------------------------
def select_call(self):
"""Select the call option to trade."""
# Get the list of strikes
strikes = [x.ID.StrikePrice for x in self.possible_calls]
strikes.sort()
# Get the underlying price and differences to it for all strikes
price = self.price
strike_differences = [(x, abs(price-x)) for x in strikes]
# Sort the list by smallest difference from underlying price
strike_differences.sort(key=lambda x:x[1])
# Get the ATM strike
atm_strike_difference = min([abs(x[1]) for x in strike_differences])
atm_strike_index = [x[1] for x in strike_differences].index(
atm_strike_difference
)
atm_strike = strike_differences[atm_strike_index][0]
atm_strike_index = strikes.index(atm_strike)
# Get the target OTM strike for the trade
if len(strikes) > atm_strike_index+self.algo.OTM_STRIKES:
# Catch if we don't have the strike (index < 0)
if (atm_strike_index+self.algo.OTM_STRIKES) >= 0:
target_strike = strikes[atm_strike_index+self.algo.OTM_STRIKES]
else:
# Best we can do now is use the lowest possible strike
target_strike = strikes[0]
else:
target_strike = strikes[-1]
strike_differences = [
(x, abs(target_strike-x)) for x in strikes
]
strike_differences.sort(key=lambda x:x[1])
# Get the target OTM call option
for strike, difference in strike_differences:
if strike in strikes:
target_calls = [
x for x in self.possible_calls \
if x.ID.StrikePrice == strike
]
if len(target_calls) > 0:
target_call = target_calls[0]
return target_call
#-------------------------------------------------------------------------------
def select_put(self):
"""Select the put option to trade."""
# Get the list of strikes
strikes = [x.ID.StrikePrice for x in self.possible_puts]
strikes.sort()
# Get the underlying price and differences to it for all strikes
price = self.price
strike_differences = [(x, abs(price-x)) for x in strikes]
# Sort the list by smallest difference from underlying price
strike_differences.sort(key=lambda x:x[1])
# Get the ATM strike
atm_strike_difference = min([abs(x[1]) for x in strike_differences])
atm_strike_index = [x[1] for x in strike_differences].index(
atm_strike_difference
)
atm_strike = strike_differences[atm_strike_index][0]
atm_strike_index = strikes.index(atm_strike)
# Get the target OTM strike for the trade
try:
# Catch if we don't have the strike (index < 0)
if (atm_strike_index-self.algo.OTM_STRIKES) >= 0:
target_strike = strikes[atm_strike_index-self.algo.OTM_STRIKES]
else:
# Best we can do now is use the highest possible strike
target_strike = strikes[-1]
except:
target_strike = strikes[0]
strike_differences = [
(x, abs(target_strike-x)) for x in strikes
]
strike_differences.sort(key=lambda x:x[1])
# Get the target OTM put option
for strike, difference in strike_differences:
if strike in strikes:
target_puts = [
x for x in self.possible_puts \
if x.ID.StrikePrice == strike
]
if len(target_puts) > 0:
target_put = target_puts[0]
return target_put
#-------------------------------------------------------------------------------
def go_long(self):
"""Take a long position."""
# self.algo.SetHoldings(self.symbol_object, 1.0)
option = self.select_call()
if option is None:
return
# Get the option bid/ask
# bid = self.algo.Securities[option].BidPrice
ask = self.algo.Securities[option].AskPrice
# Ignore if ask is 0
if ask == 0:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"Ignoring (long) buy {option} signal because cannot get "
f"price info."
)
return
# Get the target number of options to buy
target_risk = self.algo.RISK_PER_TRADE*self.algo.Portfolio.TotalPortfolioValue
risk_per_contract = 100.0*self.algo.STOP_LOSS_PERCENT*ask
target_qty = int(target_risk/risk_per_contract)
if target_qty != 0:
# Log message when desired
if PRINT_ORDERS:
self.algo.MyLog(f"Buying {target_qty} of {option}. Ask={ask:.2f}.")
order = self.algo.MarketOrder(option, target_qty, tag='call entry')
# Catch invalid order
if order.Status != OrderStatus.Invalid:
self.call = option
self.option = self.call
self.position = 'long'
# else:
# raise ValueError(f"Error trying to buy {option}")
self.late_entry_check()
#-------------------------------------------------------------------------------
def go_short(self):
"""Take a short position."""
# self.algo.SetHoldings(self.symbol_object, -1.0)
option = self.select_put()
if option is None:
return
# Get the option bid/ask
# bid = self.algo.Securities[option].BidPrice
ask = self.algo.Securities[option].AskPrice
# Ignore if ask is 0
if ask == 0:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"Ignoring (short) buy {option} signal because cannot get "
f"price info."
)
return
# Get the target number of options to buy
target_risk = self.algo.RISK_PER_TRADE*self.algo.Portfolio.TotalPortfolioValue
risk_per_contract = 100.0*self.algo.STOP_LOSS_PERCENT*ask
target_qty = int(target_risk/risk_per_contract)
if target_qty != 0:
# Log message when desired
if PRINT_ORDERS:
self.algo.MyLog(f"Buying {target_qty} of {option}. Ask={ask:.2f}.")
order = self.algo.MarketOrder(option, target_qty, tag='put entry')
# Catch invalid order
if order.Status != OrderStatus.Invalid:
self.put = option
self.option = self.put
self.position = 'short'
# else:
# raise ValueError(f"Error trying to buy {option}")
self.late_entry_check()
#-------------------------------------------------------------------------------
def late_entry_check(self):
"""Check for late entry to possibly ignore the end of day exit."""
# Check if a late in the day entry should be checked
# Ignore if day of expiry
self.end_of_day_exit_ignore = False
if LATE_ENTRY_HOLD_OVERNIGHT and \
self.option_expiry != self.algo.Time.date():
# Get the number of minutes until the close
close_dt = self.algo.Time.replace(
hour=self.mkt_close_local_tz.hour,
minute=self.mkt_close_local_tz.minute
)
minutes_to_close = (close_dt-self.algo.Time).seconds/60.0
if minutes_to_close <= LATE_ENTRY_MINUTES_BEFORE_CLOSE:
# Log message when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{minutes_to_close} minutes to close, so the new "
f"position will NOT be exited at the end of the day!"
)
self.end_of_day_exit_ignore = True
#-------------------------------------------------------------------------------
def go_flat(self):
"""Flatten the open position."""
# self.algo.SetHoldings(self.symbol_object, 0)
if self.call:
# Get the current qty
order_qty = -self.algo.Portfolio[self.call].Quantity
# Log message when desired
if PRINT_ORDERS:
self.algo.MyLog(f"Selling {-order_qty} of {self.call}.")
self.algo.MarketOrder(self.call, order_qty, tag='call exit')
self.call = None
if self.put:
# Get the current qty
order_qty = -self.algo.Portfolio[self.put].Quantity
# Log message when desired
if PRINT_ORDERS:
self.algo.MyLog(f"Selling {-order_qty} of {self.put}.")
self.algo.MarketOrder(self.put, order_qty, tag='put exit')
self.put = None
self.position = None
#-------------------------------------------------------------------------------
@property
def price(self):
"""Return the current price."""
try:
return self.algo.Portfolio[self.symbol_object].Price
except:
try:
return self.algo.Securities[self.symbol_object].Price
except:
return None
#-------------------------------------------------------------------------------
@property
def stochastic_crossover(self):
"""Return if there is a Stochastic crossover."""
# Check if the rolling window is ready
if self.k_gt_d.IsReady:
return self.k_gt_d[0] and not self.k_gt_d[1]
# Otherwise return False
return False
#-------------------------------------------------------------------------------
@property
def stochastic_crossunder(self):
"""Return if there is a Stochastic crossunder."""
# Check if the rolling window is ready
if self.k_lt_d.IsReady:
return self.k_lt_d[0] and not self.k_lt_d[1]
# Otherwise return False
return False
#-------------------------------------------------------------------------------
def cancel_open_exit_orders(self):
"""Cancel any open exit orders."""
# Cancel stop loss order, if one
if self.stop_loss_order:
response = self.stop_loss_order.Cancel()
if response.IsSuccess:
self.stop_loss_order = None
elif not self.algo.LiveMode:
raise ValueError(f"Error trying to cancel stop loss order!")
# Cancel take profit order, if one
if self.take_profit_order:
response = self.take_profit_order.Cancel()
if response.IsSuccess:
self.take_profit_order = None
elif not self.algo.LiveMode:
raise ValueError(f"Error trying to cancel take profit order!")
#-------------------------------------------------------------------------------
def update_stop_order(self, price):
"""Update the desired stop order."""
# Get the stop order ticket
ticket = self.stop_loss_order
if ticket is None:
return
# Update the price
ticket.UpdateStopPrice(price)
self.stop_loss_price = price
# Print details when desired
if PRINT_SIGNALS:
self.algo.MyLog(
f"{self.option} trailing stop price updated to {price}"
)
#-------------------------------------------------------------------------------
def on_order_event(self, order_event):
"""New order event."""
# Get the order details
order = self.algo.Transactions.GetOrderById(order_event.OrderId)
order_qty = int(order.Quantity)
avg_fill = order_event.FillPrice
tag = order.Tag
symbol = order.Symbol
# Log message when desired
if PRINT_ORDERS:
self.algo.MyLog(
f"{symbol} {tag} order filled: {order_qty} @ {avg_fill}"
)
# Check for entry order
if 'entry' in tag:
# Check if we use a stop loss order
# if USE_STOP_LOSS:
# Check if also using trailing stop and profit trigger is 0
if USE_TRAILING_STOP \
and self.algo.TRAILING_STOP_PROFIT_TRIGGER == 0:
# Use the closest stop
stop_percent = min(
self.algo.TRAILING_STOP_PERCENT,
self.algo.STOP_LOSS_PERCENT
)
else:
stop_percent = self.algo.STOP_LOSS_PERCENT
# Calulate the stop loss price
stop_loss = round(avg_fill*(1-stop_percent),2)
# Create the stop loss order
self.stop_loss_price = stop_loss
self.stop_loss_order = self.algo.StopMarketOrder(
symbol, -order_qty, stop_loss, tag='stop loss exit'
)
# Check if the trailing stop is used
if USE_TRAILING_STOP:
# Set trade best price to the filled entry price
self.trade_best_price = avg_fill
# Get the activation price
activation = round(
avg_fill*(1+self.algo.TRAILING_STOP_PROFIT_TRIGGER),2
)
self.trailing_stop_activation_price = activation
# Check if the profit trigger is 0
if self.algo.TRAILING_STOP_PROFIT_TRIGGER == 0:
# Set trailing stop activated to be true
self.trailing_stop_activated = True
# # See if the stop loss is not used
# if not USE_STOP_LOSS:
# # Calulate the stop loss price
# stop_percent = self.algo.TRAILING_STOP_PERCENT
# stop_loss = round(avg_fill*(1-stop_percent),2)
# # Create the stop loss order
# self.stop_loss_price = stop_loss
# self.stop_loss_order = self.algo.StopMarketOrder(
# symbol, -order_qty, stop_loss, tag='stop loss exit'
# )
# Check if we use a take profit order
if USE_TAKE_PROFIT:
# Calulate the take profit price
tp_price = round(avg_fill*(1+self.algo.TAKE_PROFIT_PERCENT),2)
# Create the take profit order
self.take_profit_price = tp_price
self.take_profit_order = self.algo.LimitOrder(
symbol, -order_qty, tp_price, tag='take profit exit'
)
# Check for exit order
elif 'exit' in tag:
# Check for stop loss order filled
if 'stop loss' in tag:
self.stop_loss_order = None
# Check for take profit order filled
elif 'take profit' in tag:
self.take_profit_order = None
# Cancel any open exit orders
self.cancel_open_exit_orders()
# Reset trade variables
self.reset_trade_variables()