| Overall Statistics |
|
Total Orders 3 Average Win 0.40% Average Loss 0% Compounding Annual Return 12.893% Drawdown 0% Expectancy 0 Start Equity 1000000 End Equity 1003994.97 Net Profit 0.399% Sharpe Ratio 2.135 Sortino Ratio 0 Probabilistic Sharpe Ratio 94.319% Loss Rate 0% Win Rate 100% Profit-Loss Ratio 0 Alpha 0.037 Beta -0.125 Annual Standard Deviation 0.019 Annual Variance 0 Information Ratio 0.953 Tracking Error 0.075 Treynor Ratio -0.326 Total Fees $2.63 Estimated Strategy Capacity $19000.00 Lowest Capacity Asset KROS XDI9347V4LET Portfolio Turnover 0.13% |
###############################################################################
# Standard library imports
# from dateutil.relativedelta import *
import datetime as DT
# from dateutil.parser import parse
# import decimal
# import numpy as np
# import pandas as pd
# import pickle
# import pytz
# from System.Drawing import Color
# 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.set_backtest_details()
# Add strategy variables
self.add_strategy_variables()
# Add instrument data to the algo
self.add_instrument_data()
# Schedule functions
self.schedule_functions()
# Warmup
# self.set_warm_up(DT.timedelta(days=5))
#------------------------------------------------------------------------------
def set_backtest_details(self):
"""Set the backtest details."""
self.set_start_date(START_DT.year, START_DT.month, START_DT.day)
if END_DATE:
self.set_end_date(END_DT.year, END_DT.month, END_DT.day)
self.set_cash(CASH)
self.set_time_zone(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.set_brokerage_model(
BrokerageName.INTERACTIVE_BROKERS_BROKERAGE,
AccountType.MARGIN
)
# Configure all universe securities
# This sets the data normalization mode to raw
self.set_security_initializer(self.custom_security_initializer)
# Adjust the cash buffer from the default 2.5% to custom setting
self.settings.free_portfolio_value_percentage = FREE_PORTFOLIO_VALUE_PCT
# Disable margin calls
self.portfolio.margin_call_model = MarginCallModel.NULL
# Use precise daily end times
self.settings.daily_precise_end_time = True
#------------------------------------------------------------------------------
def custom_security_initializer(self, security):
"""Configure settings for securities added to our universe."""
# Set data normalization mode
if DATA_MODE == 'ADJUSTED':
# Throw error if live
if self.live_mode:
raise ValueError(f"Must use 'RAW' DATA_MODE for live trading!")
security.set_data_normalization_mode(
DataNormalizationMode.ADJUSTED
)
elif DATA_MODE == 'RAW':
security.set_data_normalization_mode(DataNormalizationMode.RAW)
else:
raise ValueError(f"Invalid DATA_MODE: {DATA_MODE}")
# Check for specific security type
# if security.type == SecurityType.EQUITY:
# Set the margin model
security.margin_model = PatternDayTradingMarginModel()
# Overwrite the security buying power
# security.set_buying_power_model(BuyingPowerModel.NULL)
security.set_buying_power_model(SecurityMarginModel(4.0))
# Overwrite the fee model
# security.set_fee_model(ConstantFeeModel(0))
#------------------------------------------------------------------------------
def add_strategy_variables(self):
"""Create required strategy variables."""
# Read algo parameters
self.PH_PCT_CHG = self.GetParameter('PH_PCT_CHG', PH_PCT_CHG)
self.PC_PCT_CHG = self.GetParameter('PC_PCT_CHG', PC_PCT_CHG)
self.PM_PCT_CHG = self.GetParameter('PM_PCT_CHG', PM_PCT_CHG)
self.MIN_PREVIOUS_VOL = \
self.GetParameter('MIN_PREVIOUS_VOL', MIN_PREVIOUS_VOL)
self.MIN_VOL = self.GetParameter('MIN_VOL', MIN_VOL)
self.STOP_PCT = self.GetParameter('STOP_PCT', STOP_PCT)
self.TAKE_PROFIT_PCT = self.GetParameter(
'TAKE_PROFIT_PCT', TAKE_PROFIT_PCT
)
self.FIXED_DOLLAR_SIZE = self.GetParameter(
'FIXED_DOLLAR_SIZE', FIXED_DOLLAR_SIZE
)
self.MAX_TRADES = self.GetParameter('MAX_TRADES', MAX_TRADES)
# Always update the universe when initializing
self.update_universe = True
# Keep track of SymbolData class instances and Symbols
# These are symbols that have already been filtered to be in our
# desired universe.
self.symbol_data = {}
self.symbol_objects = {}
# Keep track of the last valid market day (for previous day reference)
self.previous_day = None
# Keep track if trading is allowed & symbols that pass previous day check
self.trading = False
self.prefiltered_symbols = []
self.positions = 0
#------------------------------------------------------------------------------
def add_instrument_data(self):
"""Add instrument data to the algo."""
# Set data resolution
if DATA_RESOLUTION == 'SECOND':
self.resolution = Resolution.SECOND
elif DATA_RESOLUTION == 'MINUTE':
self.resolution = Resolution.MINUTE
else:
raise ValueError(f"Invalid DATA_RESOLUTION: {DATA_RESOLUTION}")
# Set the universe data properties desired
self.universe_settings.resolution = self.resolution
self.universe_settings.extended_market_hours = True
self.universe_settings.minimum_time_in_universe = MIN_TIME_IN_UNIVERSE
# Use custom coarse and filter universe selection
self.add_universe(self.custom_universe_filter)
# Add the benchmark
self.bm = self.add_equity(BENCHMARK, self.resolution).symbol
self.set_benchmark(self.bm)
#------------------------------------------------------------------------------
def schedule_functions(self):
"""Scheduling the functions required by the algo."""
# Update self.update_universe variable True when desired
if UNIVERSE_FREQUENCY == 'DAILY':
date_rules = self.date_rules.every_day(self.bm)
elif UNIVERSE_FREQUENCY == 'WEEKLY':
# Want to schedule at end of the week, so actual update on
# start of the next week
date_rules = self.date_rules.week_end(self.bm)
else: # 'MONTHLY'
# Want to schedule at end of the month, so actual update on
# first day of the next month
date_rules = self.date_rules.month_end(self.bm)
# Timing is after the market closes
self.schedule.on(
date_rules,
self.time_rules.before_market_close(self.bm, -5),
self.on_update_universe
)
# Start trading at the pre-market open @ 4am ET
self.schedule.on(
self.date_rules.every_day(self.bm),
self.time_rules.at(4, 0),
self.start_trading
)
# Exit open trades the desired number of minutes before the close
self.schedule.on(
self.date_rules.every_day(self.bm),
self.time_rules.before_market_close(self.bm, EOD_EXIT_MINUTES),
self.end_of_day_exit
)
# Schedule benchmark end of day event 5 minutes after the close
# Used to plot the benchmark on the equity curve
self.schedule.on(
self.date_rules.every_day(self.bm),
self.time_rules.before_market_close(self.bm, -5),
self.benchmark_on_end_of_day
)
#-------------------------------------------------------------------------------
def on_update_universe(self):
"""Event called when rebalancing is desired."""
# Update the variable to trigger the universe to be updated
self.update_universe = True
#-------------------------------------------------------------------------------
def custom_universe_filter(self, fundamental):
"""
Perform custom filters on universe.
Called once per day.
Returns all stocks meeting the desired criteria.
"""
# # Slow - so only for debugging
# # if self.time.day == 23:
# # print('debug')
# # Loop through all fundamental objects
# companies = ['KROS','Keros Therapeutics']
# target_tickers = ['KROS']
# fundamental_dict = {x.symbol: x for x in fundamental}
# kros = fundamental_dict.get(
# Symbol.create("KROS", SecurityType.EQUITY, Market.USA)
# )
# for f in fundamental:
# # Catch a specific company name
# if f.company_reference.standard_name is not None:
# for c in companies:
# if c in f.company_reference.standard_name:
# print('debug')
# # Catch a specific ticker symbol
# if f.symbol.value in target_tickers:
# # Why is 'ALAR' company_reference all None?
# # REF: https://finance.yahoo.com/quote/ALAR/
# if f.symbol.value == 'ALAR':
# self.my_log(
# f"ALAR company references all None? "
# f"{f.company_reference.standard_name}"
# )
# # Why is 'SIGA' not even in fundamental?
# # REF: https://finance.yahoo.com/quote/SIGA/
# Check if the universe doesn't need to be updated
if not self.update_universe and not self.is_warming_up:
# Return unchanged universe
return Universe.UNCHANGED
if ONLY_TRADE_TARGET_TICKERS:
filtered = [
f for f in fundamental if f.symbol.value in TARGET_TICKERS
]
else:
# First filter based on properties that will never change
# Filter based on allowed exchange
if USE_EXCHANGE_FILTER:
filtered = [
f for f in fundamental \
if f.security_reference.exchange_id in ALLOWED_EXCHANGE
]
else:
filtered = [f for f in fundamental]
# Check if fundamental data is required
if REQUIRE_FUNDAMENTAL_DATA:
# Filter all securities with fundamental data
filtered = [f for f in filtered if f.has_fundamental_data]
# Filter stocks based on primary share class
if PRIMARY_SHARES:
filtered = [
f for f in filtered if f.security_reference.is_primary_share
]
# Now filter based on properties that constantly change
# Filter by price
filtered = [
f for f in filtered \
if f.price >= MIN_PRICE and f.price <= MAX_PRICE
]
# Filter by allowed market cap
filtered = [
f for f in filtered \
if f.market_cap >= MIN_MARKET_CAP \
and f.market_cap <= MAX_MARKET_CAP
]
# Return a unique list of symbol objects
self.symbols = [f.symbol for f in filtered]
# Print universe details when desired
if PRINT_UNIVERSE or self.live_mode:
self.my_log(f"Universe filter returned {len(self.symbols)} stocks")
# tickers = [f.Value for f in self.symbols]
# tickers.sort()
# self.my_log(
# f"Universe filter returned {len(self.symbols)} stocks: {tickers}"
# )
# Set update universe variable back to False and return universe symbols
if not self.is_warming_up:
self.update_universe = False
# tickers = [x.value for x in self.symbols]
# if 'KROS' in tickers:
# self.my_log('KROS in universe')
# if 'SIGA' in tickers:
# self.my_log('SIGA in universe')
# if 'LQDA' in tickers:
# self.my_log('LQDA in universe')
# if 'ALAR' in tickers:
# self.my_log('ALAR in universe')
# if 'UBXG' in tickers:
# self.my_log('UBXG in universe')
return self.symbols
#-------------------------------------------------------------------------------
def my_log(self, message):
"""Add algo time to log if live trading. Otherwise just log message."""
if self.live_mode:
self.log(f'{self.time}: {message}')
else:
self.log(message)
#-------------------------------------------------------------------------------
def on_securities_changed(self, changes):
"""Built-in event handler for securities added and removed."""
# Loop through added securities
for security in changes.added_securities:
symbol = security.symbol
# Create a new SymbolData object for the security
self.symbol_data[symbol] = SymbolData(self, symbol)
# Save a link to the symbol object
self.symbol_objects[symbol.Value] = symbol
# Loop through removed securities
for security in changes.removed_securities:
symbol = security.symbol
# Get the SymbolData class instance
sd = self.symbol_data.get(symbol)
if sd:
# Dispose of the symbol's data handlers and pop off dictionary
sd.dispose()
data = self.symbol_data.pop(symbol, None)
# Remove symbol_object from dictionary
self.symbol_objects.pop(symbol.Value, None)
#-------------------------------------------------------------------------------
def on_splits(self, splits):
"""Built-in event handler for split events."""
# Loop through the splits
for symbol, split in splits.items():
# Verify this is not a warning
if split.Type == 1:
# Get the split factor
split = split.SplitFactor
# If this is for the benchmark, update the benchmark price
if symbol.value == BENCHMARK:
try:
# Adjust the first benchmark price for the split
self.bm_first_price *= split
except:
# Benchmark first price not set, so skip
self.my_log(
"Benchmark's first price not set. Trying to "
"adjust it for a split."
)
# Catch the appropriate symbol_data instance
sd = self.symbol_data.get(symbol)
if sd:
# Log message when desired
if PRINT_SPLITS or self.live_mode:
self.my_log(
f"New {symbol.value} split: split factor={split}. "
f"Updating {symbol.value}'s indicators."
)
# Adjust the previous bars by the split adjustment factor
sd.adjust_indicators(split, is_split=True)
#-------------------------------------------------------------------------------
def on_dividends(self, dividends):
"""Built-in event handler for dividend events."""
# Loop through the dividends
for symbol, dividend in dividends.items():
# Get the dividend distribution amount
dividend = dividend.Distribution
# Get last 2 daily prices
hist = self.history([symbol], 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
# If this is for the benchmark, then we update the benchmark price
if symbol.value == BENCHMARK:
try:
# Adjust the first benchmark price based on the af
self.bm_first_price *= af
except:
# Benchmark first price not set, so skip
self.my_log(
"Benchmark's first price not set. Trying to adjust it "
"for a dividend payment."
)
# Catch the appropriate symbol_data instance
sd = self.symbol_data.get(symbol)
if sd:
# Log message when desired
if PRINT_DIVIDENDS or self.live_mode:
self.my_log(
f"New {symbol.value} dividend={dividend}. Close={price}, "
f"previous close={previous_close}, so dividend "
f"adjustment factor={af}. Updating {symbol.value}'s "
"indicators."
)
# Adjust the previous bars by the dividend adjustment factor
sd.adjust_indicators(af)
#-------------------------------------------------------------------------------
def on_order_event(self, order_event):
"""Built-in event handler for orders."""
# Catch invalid order
if order_event.Status == OrderStatus.INVALID:
order = self.transactions.get_order_by_id(order_event.order_id)
msg = order_event.get_Message()
self.my_log(f"on_order_event() invalid order ({order}): {msg}")
# Check if filled
elif order_event.Status == OrderStatus.FILLED:
# Get the order's symbol
order = self.transactions.get_order_by_id(order_event.order_id)
symbol = order.symbol
# Get the SymbolData class instance
sd = self.symbol_data.get(symbol)
if sd:
# Pass to the SymbolData's handler
sd.on_order_event(order_event)
else:
msg = f"on_order_event() order without SymbolData class " + \
f"instance: ({order})"
self.my_log(msg)
if not self.live_mode:
raise ValueError(msg)
#------------------------------------------------------------------------------
def start_trading(self):
"""At beginning of trading day, reset trading allowed variable."""
# Return if warming up
if self.is_warming_up:
return
# Reset variables
self.trading = True
self.positions = 0
# Get a list of all symbols that pass the previous day filter
self.prefiltered_symbols = [
x for x, sd in self.symbol_data.items() \
if sd.previous_day_entry_filter
]
#------------------------------------------------------------------------------
def end_of_day_exit(self):
"""At end of trading day, so exit any open positions."""
positions = [
x for x in self.portfolio.Keys if self.portfolio[x].invested
]
for symbol in positions:
self.liquidate(symbol, tag='end of day exit')
self.trading = False
#------------------------------------------------------------------------------
def benchmark_on_end_of_day(self):
"""Event handler for end of trading day for the benchmark."""
self.plot_benchmark_on_equity_curve()
# Update prevous day reference
self.previous_day = self.time.date()
#------------------------------------------------------------------------------
def on_end_of_algorithm(self):
"""Built-in event handler for end of the backtest."""
# Plot the benchmark buy and hold value on the equity curve chart
self.plot_benchmark_on_equity_curve(force_plot=True)
#------------------------------------------------------------------------------
def plot_benchmark_on_equity_curve(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.my_log(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.total_holdings_value \
/ self.portfolio.total_portfolio_value
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.my_log(
f"Benchmark's first price = {self.bm_first_price}"
)
self.my_log(f"Benchmark's final price = {price}")
self.my_log(f"Benchmark buy & hold value = {bm_value}")from AlgorithmImports import *
"""
Custom Shorting Day Trading Strategy
Version 1.0.6
Platform: QuantConnect
By: Aaron Eller
r.aaron.eller@gmail.com
Revision Notes:
1.0.0 (08/28/2024) - Initial.
1.0.1 (08/29/2024) - Corrected the first_premarket_price.
1.0.3 (09/02/2024) - Changed to use daily regular session for the
previous day references.
1.0.4 (09/04/2024) - Added MAX_PRICE_FOR_ENTRY input and logic.
1.0.5 (09/05/2024) - Added USE_EXCHANGE_FILTER input and logic.
1.0.6 (12/19/2024) - Added TAKE_PROFIT_PCT input and logic.
- Changed to start trading as soon as the pre-market
session opens. Checks for possible entry signals
constantly rather than only at the market open.
- Added ONLY_TRADE_TARGET_TICKERS input and logic.
#** -> can be "Parameters" for optimization
References:
-QC (Lean) Class List
https://lean-api-docs.netlify.app/annotated.html
"""
# Standard library imports
import datetime as DT
###############################################################################
# Backtest inputs
# NOTE1: QC index data composition starts in August 2009
# NOTE2: if not using index data, can start in 1998
START_DATE = "12-02-2024" # must be in "MM-DD-YYYY" format
END_DATE = "12-13-2024" # must be in "MM-DD-YYYY" format or None
CASH = 1_000_000 # starting portfolio value
#------------------------------------------------------------------------------
# DATA INPUTS
# Set the data resolution required
# Must be 'SECOND' or 'MINUTE'
DATA_RESOLUTION = 'MINUTE'
# Set the data normalization mode - either 'RAW' or 'ADJUSTED'
# For live trading, must use 'RAW'
DATA_MODE = 'RAW'
# Set the Benchmark
BENCHMARK = 'SPY'
# Only trade target tickers?
ONLY_TRADE_TARGET_TICKERS = True
# When True, universe will only include these stocks
TARGET_TICKERS = ['KROS']
#------------------------------------------------------------------------------
# CUSTOM UNIVERSE INPUTS
# How often to update the universe?
# Options: 'DAILY', 'WEEKLY', 'MONTHLY'
UNIVERSE_FREQUENCY = 'DAILY'
# Set the minimum number of days to leave a stock in the universe
# This helps with making the universe output more stable
MIN_TIME_IN_UNIVERSE = 21 # approximately 1 month
REQUIRE_FUNDAMENTAL_DATA = False # if True, stocks only / no etfs e.g.
MIN_PRICE = 1.50 # set to 0 to disable
MAX_PRICE = 70.0 # set to 1e6 to disable
MAX_PRICE_FOR_ENTRY = 100.0 # set to 1e6 to disable
# Market cap filtering / e6=million/e9=billion/e12=trillion
MIN_MARKET_CAP = 500e6 #30e6 # 0 to disable
MAX_MARKET_CAP = 4e9 #100e12 # extremely high value like 100e12 to disable
# Turn on/off specific exchanges allowed
USE_EXCHANGE_FILTER = False
# When used, turn on/off specific exchanges
ARCX = False # Archipelago Electronic Communications Network
ASE = False # American Stock Exchange
BATS = False # Better Alternative Trading System
NAS = True # Nasdaq Stock Exchange
NYS = True # New York Stock Exchange
# Only allow a stock's primary shares?
# Example is Alphabet (Google):
# A shares = GOOGL (voting rights) -> considered the 'primary share' class.
# C shares = GOOG (no voting rights)
# REF:
'''https://www.investopedia.com/ask/answers/052615/whats-difference-between-
googles-goog-and-googl-stock-tickers.asp'''
PRIMARY_SHARES = False
#------------------------------------------------------------------------------
# ENTRY INPUTS
# Checks for signals at the market open
# All percents below are decimal percents. E.g. 0.05=5.0%
# Set the minimum pct change required from the previous day high
PH_PCT_CHG = 0.2 #**
# Set the minimum pct change required from the previous day close
PC_PCT_CHG = 0.2 #**
# Set the minimum pct change required from the first premarket price
PM_PCT_CHG = 0.2 #**
# Set the previous day's minimum allowed volume
MIN_PREVIOUS_VOL = 100_000 #**
# Set the current day's minimum trading volume
MIN_VOL = 10_000 #**
#------------------------------------------------------------------------------
# EXIT INPUTS
# Set the percentage stop to use (as a decimal percent, e.g. 0.20=20.0%)
STOP_PCT = 0.20 #**
# Set the target to exit for profit (as a decimal percent, e.g. 0.40=40.0%)
TAKE_PROFIT_PCT = 0.40 #**
# Set the number of minutes prior to the market close to exit
EOD_EXIT_MINUTES = 1
#------------------------------------------------------------------------------
# POSITION SIZING INPUTS
# Set the desired dollar amount to invest per position
FIXED_DOLLAR_SIZE = 10_000 #**
# Set the max number of trades allowed per day
MAX_TRADES = 10 #**
#------------------------------------------------------------------------------
# LOGGING INPUTS
PRINT_UNIVERSE = False # print universe info
PRINT_DIVIDENDS = False # turn on/off logs for new dividends
PRINT_SPLITS = False # turn on/off logs for new splits
PRINT_SIGNALS = True # print new signal information
PRINT_ORDERS = True # print new orders
################################################################################
############################ END OF ALL USER INPUTS ############################
################################################################################
# VALIDATE USER INPUTS - DO NOT CHANGE BELOW!!!
#-------------------------------------------------------------------------------
# POSITION SIZING INPUTS
# Set the percentage of the portfolio to free up to avoid buying power issues
FREE_PORTFOLIO_VALUE_PCT = 0.025 # decimal percent, e.g. 0.025=2.5% (default)
# Set the time zone -> do not change! Algo will use hard coded times!
TIMEZONE = "US/Eastern" # e.g. "US/Eastern", "US/Central", "US/Pacific"
#-------------------------------------------------------------------------------
# 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 universe update frequency
UNIVERSE_FREQUENCY = UNIVERSE_FREQUENCY.upper()
if UNIVERSE_FREQUENCY not in ['DAILY', 'WEEKLY', 'MONTHLY']: #'QUARTERLY'
raise ValueError(
f"Invalid UNIVERSE_FREQUENCY ({UNIVERSE_FREQUENCY}). "
f"Must be ['DAILY', 'WEEKLY', 'MONTHLY', 'QUARTERLY']."
)
#-------------------------------------------------------------------------------
# Verify the DATA_MODE input
DATA_MODE = DATA_MODE.upper()
if DATA_MODE not in ['RAW', 'ADJUSTED']:
raise ValueError(
f"Invalid DATA_MODE ({DATA_MODE}). Must be: 'RAW' or 'ADJUSTED'"
)
#-------------------------------------------------------------------------------
# Verify DATA_RESOLUTION input
DATA_RESOLUTION = DATA_RESOLUTION.upper()
if DATA_RESOLUTION not in ['SECOND', 'MINUTE']:
raise ValueError(
f"Invalid DATA_RESOLUTION ({DATA_RESOLUTION}). "
"Must be: 'SECOND' or 'MINUTE'"
)
#-------------------------------------------------------------------------------
# Get list of the allowed exchanges
ALLOWED_EXCHANGE = []
if ARCX:
ALLOWED_EXCHANGE.append('ARCX')
if ASE:
ALLOWED_EXCHANGE.append('ASE')
if BATS:
ALLOWED_EXCHANGE.append('BATS')
if NAS:
ALLOWED_EXCHANGE.append('NAS')
if NYS:
ALLOWED_EXCHANGE.append('NYS')
# Throw error if no exchange is allowed
if len(ALLOWED_EXCHANGE) == 0:
raise ValueError("At least one exchange must be set True.")
#-------------------------------------------------------------------------------
# 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 dateutil.relativedelta import relativedelta
# import numpy as np
# import pandas as pd
import pytz
# import statistics
# QuantConnect specific imports
# import QuantConnect as qc
from AlgorithmImports import *
# Import from files
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 .Symbol object
self.symbol_object = symbol_object
self.ticker = symbol_object.Value
self.symbol = str(self.symbol_object.id)
# Get the symbol's exchange market info
self.get_exchange_info()
# Add strategy variables
self.set_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 securities exchange info."""
# Get the SecurityExchangeHours Class object for the symbol
security = self.algo.securities[self.symbol_object]
self.exchange_hours = security.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.get_next_market_open(dt, False)
# Save the typical (regualar session) market open and close times
self.mkt_open = mkt_open_dt.time()
mkt_close_dt = self.exchange_hours.get_next_market_close(dt, False)
self.mkt_close = mkt_close_dt.time()
# Get the exchange timezone
self.mkt_tz = pytz.timezone(str(self.exchange_hours.time_zone))
# 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()
self.mkt_open_local_tz = \
(mkt_open_dt-DT.timedelta(hours=self.offset_hrs)).time()
# Get the min price
symbol_properties = security.symbol_properties
# Get and save the contract specs
self.min_tick = symbol_properties.minimum_price_variation
#-------------------------------------------------------------------------------
def set_strategy_variables(self):
"""Set strategy specific variables."""
self.warming_up = False
self.reset_trade_variables()
# Get the min tick size
self.tick_size = self.algo.securities[
self.symbol_object
].symbol_properties.minimum_price_variation
self.entry_limit_order = None
#-------------------------------------------------------------------------------
def reset_trade_variables(self):
"""Reset trade specific variables."""
self.stop_price = None
self.stop_loss_order = None
self.best_price = None
self.take_profit_order = None
#-------------------------------------------------------------------------------
def add_bars(self):
"""Add bars required."""
# Create a daily bar consolidator
self.calendar_initialized = False
consolidator = TradeBarConsolidator(self.daily_calendar)
# Create an event handler to be called on each new consolidated bar
consolidator.data_consolidated += self.on_data_consolidated
# Link the consolidator with our symbol and add it to the algo manager
self.algo.subscription_manager.add_consolidator(
self.symbol_object, consolidator
)
# Save consolidator link so we can remove it when necessary
self.consolidator = consolidator
# Create an intraday minute bar consolidator
consolidator2 = TradeBarConsolidator(DT.timedelta(minutes=1))
# Create an event handler to be called on each new consolidated bar
consolidator2.data_consolidated += self.on_intraday_data_consolidated
# Link the consolidator with our symbol and add it to the algo manager
self.algo.subscription_manager.add_consolidator(
self.symbol_object, consolidator2
)
# Save consolidator link so we can remove it when necessary
self.consolidator2 = consolidator2
#-------------------------------------------------------------------------------
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.get_next_market_close(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 dispose(self):
"""Stop the data consolidators."""
# Remove the consolidators from the algo manager
self.consolidator.data_consolidated -= self.on_data_consolidated
self.algo.subscription_manager.remove_consolidator(
self.symbol_object, self.consolidator
)
self.consolidator2.data_consolidated -= self.on_intraday_data_consolidated
self.algo.subscription_manager.remove_consolidator(
self.symbol_object, self.consolidator2
)
#-------------------------------------------------------------------------------
def add_indicators(self):
"""Add indicators and other required variables."""
# Create an empty list to hold all indicators
# Will add (indicator, update_method) tuples
# where update_method is either 'high', 'low', 'close', 'volume', or
# 'bar'
self.indicators = []
min_intraday_bars = [2]
self.todays_volume = 0 # used for the cumulative volume
# Keep track of bars for the daily indicators
self.min_bars = 5 # only need last 1 daily bar
self.bar_window = RollingWindow[TradeBar](self.min_bars)
# Keep track of bars for the intraday indicators
# We need 1min bars from 4am to 930am -> 330 minutes
self.min_bars_intraday = 330
self.bar_window_intraday = RollingWindow[TradeBar](self.min_bars_intraday)
# Warm up the indicators with historical data
self.warmup_indicators()
self.reset_intraday_indicators()
#-------------------------------------------------------------------------------
def schedule_functions(self):
"""Schedule functions required by the algo."""
# Reset the intraday indicators at midnight every day
self.algo.schedule.on(
self.algo.date_rules.every_day(self.symbol_object),
self.algo.time_rules.at(0,0),
self.reset_intraday_indicators
)
#-------------------------------------------------------------------------------
def reset_indicators(self):
"""Reset indicators required."""
# Loop through list of indicators
for indicator, _ in self.indicators:
indicator.reset()
# Handle custom indicators - if any
# Reset the rolling windows
self.bar_window.reset()
#-------------------------------------------------------------------------------
def reset_intraday_indicators(self):
"""Reset indicators required."""
# Handle custom indicators - if any
# Reset the rolling windows
self.bar_window_intraday.reset()
self.todays_volume = 0
self.traded_today = False
#-------------------------------------------------------------------------------
def adjust_indicators(self, adjustment_factor, is_split=False):
"""Adjust all indicators for splits or dividends."""
self.warming_up = True
# Get a list of the current bars
bars = list(self.bar_window)
# bars_intraday = list(self.bar_window_intraday)
# Current order is newest to oldest (default for rolling window)
# Reverse the list to be oldest to newest
bars.reverse()
# bars_intraday.reverse()
# Reset all indicators
self.reset_indicators()
self.reset_intraday_indicators()
# Loop through the daily bars from oldest to newest
for bar in bars:
# Adjust the bar by the adjustment factor
bar.Open *= adjustment_factor
bar.High *= adjustment_factor
bar.Low *= adjustment_factor
bar.Close *= adjustment_factor
# Update volume on split adjustment
if is_split:
vol_adjustment = 1.0/adjustment_factor
bar.Volume *= vol_adjustment
# Use the bar to update the indicators
# This also adds the bar to the rolling window
self.update_indicators(bar)
self.warming_up = False
#-------------------------------------------------------------------------------
def warmup_indicators(self):
"""Warm up indicators using historical data."""
# Update warmup variable
self.warming_up = True
# Get historical daily trade bars for the symbol
min_days = int(10.0*self.min_bars)
bars = self.algo.history[TradeBar](
self.symbol_object,
DT.timedelta(days=min_days),
Resolution.DAILY
)
# Loop through the bars and update the indicators
for bar in bars:
# Pass directly to the event handler (no consolidating necessary)
self.on_data_consolidated(None, bar)
# We don't need to warm up intraday indicators
# Update warmup variable back to False
self.warming_up = False
#-------------------------------------------------------------------------------
def on_data_consolidated(self, sender, bar):
"""Event handler for desired daily bar."""
# Skip if not regular session
if not self.warming_up:
if bar.time.time() != self.mkt_open_local_tz:
return
# Manually update all of the indicators
self.update_indicators(bar)
#-------------------------------------------------------------------------------
def update_indicators(self, bar):
"""Manually update all of the symbol's indicators."""
# Loop through the indicators
for indicator, update_method in self.indicators:
if update_method == 'close':
indicator.update(bar.end_time, bar.close)
elif update_method == 'high':
indicator.update(bar.end_time, bar.high)
elif update_method == 'low':
indicator.update(bar.end_time, bar.low)
elif update_method == 'bar':
indicator.update(bar)
elif update_method == 'volume':
indicator.update(bar.end_time, bar.volume)
# Add bar to the rolling window
self.bar_window.add(bar)
#-------------------------------------------------------------------------------
def on_intraday_data_consolidated(self, sender, bar):
"""Event handler for desired intraday bar."""
# Manually update the intraday indicators
self.update_intraday_indicators(bar)
# Check for signal if trading allowed
if self.algo.trading \
and self.symbol_object in self.algo.prefiltered_symbols \
and self.algo.positions < self.algo.MAX_TRADES:
# Go short on an entry signal
if self.short_entry_signal(bar):
self.go_short(bar)
#-------------------------------------------------------------------------------
def update_intraday_indicators(self, bar):
"""Manually update all of the symbol's intraday indicators."""
# Update cummulative volume for today
self.todays_volume += bar.volume
# Add bar to the rolling window
self.bar_window_intraday.add(bar)
#-------------------------------------------------------------------------------
# @property
# def indicators_ready(self):
# """Return whether the indicators are warmed up or not."""
# # Loop through list of indicators
# for indicator, _ in self.indicators:
# if not indicator.is_ready:
# return False
# if not self.bar_window.is_ready:
# return False
# if not self.bar_window_intraday.is_ready:
# return False
# # Otherwise True
# return True
#-------------------------------------------------------------------------------
@property
def current_qty(self):
"""Return the current traded symbol quantity held in the portfolio."""
return self.algo.portfolio[self.symbol_object].quantity
#-------------------------------------------------------------------------------
@property
def price(self):
"""Return the current traded symbol quantity held in the portfolio."""
return self.algo.securities[self.symbol_object].price
#-------------------------------------------------------------------------------
@property
def previous_day_entry_filter(self):
"""
Return whether the stock meets all of the entry requirements from the
previous day.
"""
# Check if the symbol is in the universe
if self.symbol_object not in self.algo.symbols:
return False
# Check yesterday's volume
elif self.previous_volume < self.algo.MIN_PREVIOUS_VOL:
return False
# Check yesterday's date
elif self.previous_date != self.algo.previous_day \
and self.algo.previous_day is not None:
return False
return True
#-------------------------------------------------------------------------------
def short_entry_signal(self, bar):
"""
Check for a short entry signal.
"""
# # Debugging
# if self.ticker == 'KROS':
# self.algo.my_log(f"KROS check entry_filter()")
# elif self.ticker == 'LQDA':
# self.algo.my_log(f"LQDA check entry_filter()")
# elif self.ticker == 'ALAR':
# self.algo.my_log(f"ALAR check entry_filter()")
# elif self.ticker == 'UBXG':
# self.algo.my_log(f"UBXG check entry_filter()")
# # Check if the symbol is in the universe
# if self.symbol_object not in self.algo.symbols:
# return False
# Not valid if we already have a position
if self.current_qty < 0:
return False
elif self.traded_today:
return False
# Check for min/max price
price = bar.close
if price < MIN_PRICE:
return False
elif price > MAX_PRICE_FOR_ENTRY:
return False
# # Check yesterday's volume
# if self.previous_volume < self.algo.MIN_PREVIOUS_VOL:
# return False
# Check today's volume
if self.todays_volume < self.algo.MIN_VOL:
return False
# Check if price is min percentage from previous high
pct_high = self.pct_chg_from_previous_high(price)
if pct_high > -self.algo.PH_PCT_CHG:
return False
# Check if price is min percentage from previous close
pct_close = self.pct_chg_from_previous_close()
if pct_close > -self.algo.PC_PCT_CHG:
return False
# Check if price is min percentage from first premarket price
first_price = self.first_premarket_price
if first_price == 0:
return False
pct_first = (price-first_price)/first_price
if pct_first > -self.algo.PM_PCT_CHG:
return False
# # Check yesterday's date
# if self.previous_date != self.algo.previous_day \
# and self.algo.previous_day is not None:
# self.algo.my_log(
# f"{self.ticker} Previously a signal, but now filtered out. The "
# f"previous date is {self.previous_date} vs. "
# f"{self.algo.previous_day}"
# )
# return False
# Otherwise valid
# Print info
if PRINT_SIGNALS or self.algo.live_mode:
self.algo.my_log(
f"{self.ticker} SHORT ENTRY SIGNAL: price={price}, "
f"PH={self.previous_high} ({round(100*pct_high,2)}%), "
f"PC={self.previous_close} ({round(100*pct_close,2)}%), "
f"PV={self.previous_volume}, "
f"PM={self.first_premarket_price} ({round(100*pct_first,2)}%), "
f"today's vol={self.todays_volume}"
)
return True
#-------------------------------------------------------------------------------
@property
def previous_date(self):
"""Return the previous day's date."""
try:
return self.bar_window[0].end_time.date()
except:
return 0
#-------------------------------------------------------------------------------
@property
def previous_volume(self):
"""Return the previous day's volume."""
try:
return self.bar_window[0].volume
except:
return 0
#-------------------------------------------------------------------------------
@property
def previous_high(self):
"""Return the previous day's high."""
try:
return self.bar_window[0].high
except:
return 0
#-------------------------------------------------------------------------------
@property
def previous_close(self):
"""Return the previous day's close."""
try:
return self.bar_window[0].close
except:
return 0
#-------------------------------------------------------------------------------
@property
def first_premarket_price(self):
"""Return the first premarket price for the day."""
try:
# First bar is the last in the rolling window
count = self.bar_window_intraday.count
return self.bar_window_intraday[count-1].open
except:
return 0
#-------------------------------------------------------------------------------
def pct_chg_from_previous_high(self, price):
"""Return the current percent change from the previous high."""
try:
previous_high = self.bar_window[0].high
pct_chg = (price-previous_high)/previous_high
return pct_chg
except:
return 0
#-------------------------------------------------------------------------------
def pct_chg_from_previous_close(self):
"""Return the current percent change from the previous close."""
try:
price = self.price
previous_close = self.bar_window[0].close
pct_chg = (price-previous_close)/previous_close
return pct_chg
except:
return 0
#-------------------------------------------------------------------------------
def update_stop_order(self, price, tag):
"""Update the desired stop order."""
# Get the stop order ticket
ticket = self.stop_loss_order
if ticket is None:
return
# Update the price
# price = self.round_price(price)
ticket.UpdateStopPrice(price, tag=tag)
self.stop_price = price
# Print details when desired
if PRINT_ORDERS or self.algo.live_mode:
self.algo.my_log(
f"{self.ticker} trailing stop price updated to {price}"
)
#-------------------------------------------------------------------------------
def round_price(self, price):
"""Round the given price to the nearest tick value."""
# Get the priced rounded to the nearest tick value
return round(price, 2)
# price = round(price/self.min_tick)*self.min_tick
# # Only return the desired number of decimals
# try:
# num_left_decimal = str(price).index('.')
# except:
# # min tick uses exponential format
# # Create a decimal with max of 12 decimals
# tmp = round(decimal.Decimal(price),12).normalize()
# num_left_decimal = '{:f}'.format(tmp).index('.')
# length = num_left_decimal + self.min_tick_decimals + 1
# # return float(str(price)[:length])
# return float('{:f}'.format(price)[:length])
#-------------------------------------------------------------------------------
def valid_order(self, order):
"""Return True / False if the order placed is valid."""
# Check order status
if order.Status == OrderStatus.INVALID:
return False
else:
return True
#-------------------------------------------------------------------------------
def update_limit_order(self, ticket, price):
"""Update the limit order's price."""
update_settings = UpdateOrderFields()
update_settings.limit_price = price
response = ticket.update(update_settings)
if not response.is_success:
self.algo.debug(
f"{self.symbol} limit order update request not successful!"
)
#-------------------------------------------------------------------------------
def go_short(self, bar):
"""Go short the desired amount."""
# Get the desired order qty
order_qty = -int(self.algo.FIXED_DOLLAR_SIZE/bar.close)
# Use a limit order if not after the market open
new_entry = False
if self.algo.time.time() < DT.time(9,30):
# Sell at the bid price
limit_price = round(
self.algo.securities[self.symbol_object].bid_price, 2
)
# Check if we already have an open limit order
if self.entry_limit_order:
# Update the price
self.update_limit_order(self.entry_limit_order, limit_price)
else:
# Place a new order
order = self.algo.limit_order(
self.symbol_object, order_qty, limit_price,
tag='short entry limit'
)
self.entry_limit_order = order
# Check order status
if not self.valid_order(order):
self.algo.my_log(
f"Invalid {self.ticker} short entry order!"
)
else:
new_entry = True
else:
# Cancel entry limit order - if one
if self.entry_limit_order:
self.entry_limit_order.cancel()
# Place a market order
order = self.algo.market_order(
self.symbol_object, order_qty, tag='short entry'
)
# Check order status
if not self.valid_order(order):
self.algo.my_log(
f"Invalid {self.ticker} short entry order!"
)
else:
new_entry = True
# Increment the algo's positions
if new_entry:
self.algo.positions += 1
self.traded_today = True
#-------------------------------------------------------------------------------
def go_flat(self):
"""Go short the desired amount."""
# Cancel any open exit orders
self.cancel_exit_orders()
# Get the desired order qty
order_qty = -self.current_qty
# Place the exit order
order = self.algo.market_order(
self.symbol_object, order_qty, tag='exit'
)
# Check order status
if not self.valid_order(order):
if not self.algo.live_mode:
raise ValueError(f"Invalid {self.ticker} exit order!")
#-------------------------------------------------------------------------------
def cancel_exit_orders(self):
"""Cancel any open exit orders."""
if self.stop_loss_order:
self.stop_loss_order.cancel()
if self.take_profit_order:
self.take_profit_order.cancel()
#-------------------------------------------------------------------------------
def on_order_event(self, order_event):
"""Built-in event handler for orders."""
# Get the order's info
order = self.algo.transactions.get_order_by_id(order_event.order_id)
symbol = order.symbol
tag = order.tag
qty = int(order.quantity)
avg_fill = order_event.fill_price
# Log message when desired
if PRINT_ORDERS or self.algo.live_mode:
self.algo.my_log(
f"{symbol.Value} {tag} order for {qty} shares filled @ "
f"{avg_fill:.2f}"
)
# Catch entry order
if 'entry' in tag: # don't do tag == 'entry bla bla'
# Set entry order back to None
if 'limit':
self.entry_limit_order = None
# Calculate the stop loss AND take profit prices
if qty > 0: # long
# not valid for strategy!
if self.algo.live_mode:
self.go_flat()
else:
raise ValueError(f"Invalid long {symbol.value} position!")
else: # short
stop_price = round((1+self.algo.STOP_PCT)*avg_fill,2)
tp_price = round((1-self.algo.TAKE_PROFIT_PCT)*avg_fill,2)
self.stop_price = stop_price
self.best_price = avg_fill
# Place and save the stop loss order
order_qty = -qty
order = self.algo.stop_market_order(
symbol, order_qty, stop_price, tag='stop loss exit'
)
if not self.valid_order(order) and not self.algo.live_mode:
raise
self.stop_loss_order = order
# Place and save the take profit order
order2 = self.algo.limit_order(
symbol, order_qty, tp_price, tag='take profit exit'
)
if not self.valid_order(order2) and not self.algo.live_mode:
raise
self.take_profit_order = order2
# Catch exit order
elif 'exit' in tag:
# Catch stop order
if 'stop loss' in tag:
self.stop_loss_order = None
# Catch take profit
elif 'take profit' in tag:
self.take_profit_order = None
# Reset trade if qty 0
if self.current_qty == 0:
# Cancel any open exit order
self.cancel_exit_orders()
self.reset_trade_variables()