| Overall Statistics |
|
Total Trades 69 Average Win 0.05% Average Loss -0.06% Compounding Annual Return -1.201% Drawdown 0.800% Expectancy -0.138 Net Profit -0.304% Sharpe Ratio -0.765 Probabilistic Sharpe Ratio 14.423% Loss Rate 53% Win Rate 47% Profit-Loss Ratio 0.83 Alpha -0.008 Beta 0.01 Annual Standard Deviation 0.011 Annual Variance 0 Information Ratio -0.143 Tracking Error 0.092 Treynor Ratio -0.818 Total Fees $69.00 Estimated Strategy Capacity $1400000.00 Lowest Capacity Asset SGEN S2TCB9V1OIG5 |
"""
Moving Average Cross Universe Strategy
Version 1.0.0
Platform: QuantConnect
By: Aaron Eller
www.excelintrading.com
aaron@excelintrading.com
Revision Notes:
1.0.0 (01/17/2020) - Initial. Started from "Universe Strategy_v103". Also
copied SymbolData logic from
"Moving Average Crossover_v107".
References:
-QC (Lean) Class List
https://lean-api-docs.netlify.app/annotated.html
-OrderTicket properties
https://lean-api-docs.netlify.app/classQuantConnect_1_1Orders_1_1OrderTicket.html
-QC Universe
https://www.quantconnect.com/docs/algorithm-reference/universes
-QC Universe Settings
https://www.quantconnect.com/docs/algorithm-reference/universes#Universes-Universe-Settings
-QC Universe Fundamentals
https://www.quantconnect.com/docs/data-library/fundamentals
-Speeding up QC Universe
https://www.quantconnect.com/forum/discussion/7875/speeding-up-universe-selection/p1
"""
# Standard library imports
import datetime as DT
###############################################################################
# Backtest inputs
START_DATE = "07-01-2021" # must be in "MM-DD-YYYY" format
END_DATE = "09-30-2021" #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", or "MINUTE"
DATA_RESOLUTION = 'MINUTE' # NEW
# How often to update the universe?
# Options: 'daily', 'weekly', or 'monthly'
UNIVERSE_FREQUENCY = 'monthly' # NEW
#-------------------------------------------------------------------------------
# COARSE UNIVERSE SELECTION INPUTS
REQUIRE_FUNDAMENTAL_DATA = True # if True, stocks only / no etfs e.g.
MIN_PRICE = 10.0 # set to 0 to disable
MAX_PRICE = 1e6 # set to 1e6 to disable
MIN_DAILY_VOLUME = 0 # set to 0 to disable
MIN_DAILY_DOLLAR_VOLUME = 10e6 # dollar volume = last price times volume
#-------------------------------------------------------------------------------
# FINE UNIVERSE SELECTION INPUTS
# Market cap filtering / e6=million/e9=billion/e12=trillion
MIN_MARKET_CAP = 10e9
MAX_MARKET_CAP = 10e12
# Turn on/off specific exchanges allowed
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?
PRIMARY_SHARES = True
# Turn on/off specific sectors allowed
BASIC_MATERIALS = True
CONSUMER_CYCLICAL = True
FINANCIAL_SERVICES = True
REAL_ESTATE = True
CONSUMER_DEFENSIVE = True
HEALTHCARE = True
UTILITIES = True
COMMUNICATION_SERVICES = True
ENERGY = True
INDUSTRIALS = True
TECHNOLOGY = True
# Set Morningstar Industry Groups not allowed
# Use Industry Group Code from:
# https://www.quantconnect.com/docs/data-library/fundamentals#Fundamentals-Asset-Classification
# Scroll down to Industry Groups section for codes
GROUPS_NOT_ALLOWED = [
# 10320, # MorningstarIndustryGroupCode.Banks
# 10322, # MorningstarIndustryGroupCode.CreditServices
# 10323, # MorningstarIndustryGroupCode.Insurance
# 10324, # MorningstarIndustryGroupCode.InsuranceLife
# 10325, # MorningstarIndustryGroupCode.InsurancePropertyAndCasualty
# 10326, # MorningstarIndustryGroupCode.InsurancePropertyAndCasualty
]
# Set Morningstar Industries not allowed
# Use Industry Codes from:
# https://www.quantconnect.com/docs/data-library/fundamentals#Fundamentals-Asset-Classification
# Scroll down to Industries section for codes
INDUSTRIES_NOT_ALLOWED = [
# 10218038, # MorningstarIndustryCode.Gambling
# 10320043, # MorningstarIndustryCode.BanksGlobal
# 10320044, # MorningstarIndustryCode.BanksRegionalAfrica
# 10320045, # MorningstarIndustryCode.BanksRegionalAsia
# 10320046, # MorningstarIndustryCode.BanksRegionalAustralia
# 10320047, # MorningstarIndustryCode.BanksRegionalCanada
# 10320048, # MorningstarIndustryCode.BanksRegionalEurope
# 10320049, # MorningstarIndustryCode.BanksRegionalLatinAmerica
# 10320050, # MorningstarIndustryCode.BanksRegionalUS
# 10320051, # MorningstarIndustryCode.SavingsAndCooperativeBanks
# 10320052, # MorningstarIndustryCode.SpecialtyFinance
# 10321053, # MorningstarIndustryCode.CapitalMarkets
# 10322056, # MorningstarIndustryCode.CreditServices
# 10323057, # MorningstarIndustryCode.InsuranceDiversified
# 10324058, # MorningstarIndustryCode.InsuranceLife
# 10325059, # MorningstarIndustryCode.InsurancePropertyAndCasualty
# 10326060, # MorningstarIndustryCode.InsuranceReinsurance
# 10326061, # MorningstarIndustryCode.InsuranceSpecialty
# 20636085, # MorningstarIndustryCode.DrugManufacturersMajor
# 20636086, # MorningstarIndustryCode.DrugManufacturersSpecialtyAndGeneric
# 20637087, # MorningstarIndustryCode.HealthCarePlans
# 20640091, # MorningstarIndustryCode.DiagnosticsAndResearch
]
# Set the minimum number of days to leave a stock in a universe
# This helps with making the universe output more stable
MIN_TIME_IN_UNIVERSE = 65
# Set the minimum number of days with historical data
MIN_TRADING_DAYS = 200
#-------------------------------------------------------------------------------
# ENTRY SIGNAL INPUTS
# Set the moving average periods
EMA_FAST_PERIOD = 50
EMA_SLOW_PERIOD = 200
# How many minutes prior to the open to check for new signals?
# Ideally this is called AFTER the universe filters run!
SIGNAL_CHECK_MINUTES = 30
#-------------------------------------------------------------------------------
# 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 max number of positions allowed
MAX_POSITIONS = 20
# Calculate max % of portfolio per position
MAX_PCT_PER_POSITION = 1.0/MAX_POSITIONS
#-------------------------------------------------------------------------------
# EXIT SIGNAL INPUTS
# Turn on/off stop loss
STOP_LOSS = True
# Set stop loss percentage as a decimal percent, e.g. 0.02=2.0%
SL_PCT = 0.02
# Turn on/off trailing stop
TRAILING_STOP = True
# Starts and trails based on SL_PCT
#-------------------------------------------------------------------------------
# Turn on/off end of day exit
EOD_EXIT = False
# When end of exit exit is desired, how many minutes prior to the market close
# should positions be liquidated?
EOD_EXIT_MINUTES = 15
#-------------------------------------------------------------------------------
# Turn on/off exiting on fast EMA crossing under the slow EMA
EMA_CROSSUNDER_EXIT = True # full exit when triggered
#-------------------------------------------------------------------------------
# Set the RSI period
RSI_PERIOD = 14
# Turn on/off RSI exit #1
RSI_EXIT_1 = False
RSI_EXIT_1_VALUE = 60 # long exit when RSI crosses below this value
RSI_EXIT_1_PCT = 0.50 # decimal percent, 0.50=50.0%
# Turn on/off RSI exit #2
RSI_EXIT_2 = False
RSI_EXIT_2_VALUE = 50 # long exit when RSI crosses below this value
RSI_EXIT_2_PCT = 0.50 # decimal percent, 0.50=50.0%
#-------------------------------------------------------------------------------
# Turn on/off days helds exit #1
DAYS_HELD_EXIT_1 = False
DAYS_HELD_EXIT_1_VALUE = 2
DAYS_HELD_EXIT_1_PCT = 0.25 # decimal percent, e.g. 0.25=25.0%
# Turn on/off days helds exit #2
DAYS_HELD_EXIT_2 = False
DAYS_HELD_EXIT_2_VALUE = 4
DAYS_HELD_EXIT_2_PCT = 0.25 # decimal percent, e.g. 0.25=25.0%
# Turn on/off days helds exit #3
DAYS_HELD_EXIT_3 = False
DAYS_HELD_EXIT_3_VALUE = 6
DAYS_HELD_EXIT_3_PCT = 0.25 # decimal percent, e.g. 0.25=25.0%
# Turn on/off days helds exit #4
DAYS_HELD_EXIT_4 = False
DAYS_HELD_EXIT_4_VALUE = 8
DAYS_HELD_EXIT_4_PCT = 0.25 # decimal percent, e.g. 0.25=25.0%
#-------------------------------------------------------------------------------
# Turn on/off profit target 1
PROFIT_TARGET_1 = True
# Set profit target percentage as a decimal percent, e.g. 0.05=5.0%
PT1_PCT = 0.05
# Set profit target order percentage as a decimal percent, e.g. 0.50=50.0%
PT1_ORDER_PCT = 0.50
# Turn on/off profit target 2
PROFIT_TARGET_2 = True
# Set profit target percentage as a decimal percent, e.g. 0.07=7.0%
PT2_PCT = 0.07
# Set profit target order percentage as a decimal percent, e.g. 0.25=25.0%
PT2_ORDER_PCT = 0.25
# Turn on/off profit target 3
PROFIT_TARGET_3 = True
# Set profit target percentage as a decimal percent, e.g. 0.09=9.0%
PT3_PCT = 0.09
# Set profit target order percentage as a decimal percent, e.g. 0.25=25.0%
PT3_ORDER_PCT = 0.25
#-------------------------------------------------------------------------------
# How long to keep rolling window of indicators?
INDICATOR_WINDOW_LENGTH = 5 # change to 30
#-------------------------------------------------------------------------------
# BENCHMARK DETAILS
# Turn on/off using the custom benchmark plot on the strategy equity chart
PLOT_BENCHMARK_ON_STRATEGY_EQUITY_CHART = True
# Define benchmark equity
# Currently set to not be able to trade the benchmark!
# Also used for scheduling functions, so make sure it has same trading hours
# as instruments traded.
BENCHMARK = "SPY"
#-------------------------------------------------------------------------------
# LOGGING DETAILS
# What logs to print?
PRINT_COARSE = True # print summary of coarse universe selection
PRINT_FINE = True # print summary of fine universe selection
PRINT_ENTRIES = True # print summary of daily entry signals
PRINT_EXITS = True # print exit signals triggered
PRINT_ORDERS = True # print new orders
################################################################################
############################ END OF ALL USER INPUTS ############################
################################################################################
# VALIDATE USER INPUTS - DO NOT CHANGE BELOW!!!
#-------------------------------------------------------------------------------
# Verify start date
try:
START_DT = datetime.strptime(START_DATE, '%m-%d-%Y')
except:
raise ValueError("Invalid START_DATE format ({}). Must be in MM-DD-YYYY "
"format.".format(START_DATE))
# Verify end date
try:
if END_DATE:
END_DT = datetime.strptime(END_DATE, '%m-%d-%Y')
except:
raise ValueError("Invalid END_DATE format ({}). Must be in MM-DD-YYYY "
"format or set to None to run to date.".format(END_DATE))
#-------------------------------------------------------------------------------
# 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.")
#-------------------------------------------------------------------------------
# Get list of the sectors NOT allowed
SECTORS_NOT_ALLOWED = []
if not BASIC_MATERIALS:
SECTORS_NOT_ALLOWED.append(101)
if not CONSUMER_CYCLICAL:
SECTORS_NOT_ALLOWED.append(102)
if not FINANCIAL_SERVICES:
SECTORS_NOT_ALLOWED.append(103)
if not REAL_ESTATE:
SECTORS_NOT_ALLOWED.append(104)
if not CONSUMER_DEFENSIVE:
SECTORS_NOT_ALLOWED.append(205)
if not HEALTHCARE:
SECTORS_NOT_ALLOWED.append(206)
if not UTILITIES:
SECTORS_NOT_ALLOWED.append(207)
if not COMMUNICATION_SERVICES:
SECTORS_NOT_ALLOWED.append(308)
if not ENERGY:
SECTORS_NOT_ALLOWED.append(309)
if not INDUSTRIALS:
SECTORS_NOT_ALLOWED.append(310)
if not TECHNOLOGY:
SECTORS_NOT_ALLOWED.append(311)
#-------------------------------------------------------------------------------
# Verify DATA_RESOLUTION input
DATA_RESOLUTION = DATA_RESOLUTION.upper()
resolutions = ['SECOND', 'MINUTE']
if DATA_RESOLUTION not in resolutions:
raise ValueError(f"Invalid DATA_RESOLUTION ({DATA_RESOLUTION}). "
f"Must be: {resolutions}")
#-------------------------------------------------------------------------------
# Verify universe update frequency
if UNIVERSE_FREQUENCY not in ['daily', 'weekly', 'monthly']:
raise ValueError(f"UNIVERSE_FREQUENCY ({UNIVERSE_FREQUENCY}) must be "
f"'daily', 'weekly', or 'monthly'.")
#-------------------------------------------------------------------------------
# Verify profit target orders
# Make sure order percentages <= 1.0
pt_order_total = 0
if PROFIT_TARGET_1:
pt_order_total += PT1_ORDER_PCT
if PROFIT_TARGET_2:
pt_order_total += PT2_ORDER_PCT
if PROFIT_TARGET_3:
pt_order_total += PT3_ORDER_PCT
if pt_order_total > 1:
raise ValueError(f"Invalid PT_ORDER_PCTS. Total ({pt_order_total}) > 1")
#-------------------------------------------------------------------------------
# Verify RSI exit percent totals doesn't exceed 1
if RSI_EXIT_1_PCT+RSI_EXIT_2_PCT > 1:
raise ValueError("Invalid RSI exit percents. Total cannot exceed 1.")
#-------------------------------------------------------------------------------
# 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)
#-------------------------------------------------------------------------------
# Calculate the period to warm up the data
BARS_PER_DAY = 1
# Get the minimum number of bars required to fill all indicators
MIN_BARS = max([EMA_FAST_PERIOD, EMA_SLOW_PERIOD, RSI_PERIOD]) \
+ INDICATOR_WINDOW_LENGTH
# Calculate the number of market warmup days required
MARKET_WARMUP_DAYS = -(-MIN_BARS//BARS_PER_DAY)
# Add a 10% buffer
MARKET_WARMUP_DAYS = int(1.1*MARKET_WARMUP_DAYS)
# Convert the number of market days to be actual calendar days
# Assume 252 market days per calendar year (or 365 calendar days)
CALENDAR_WARMUP_DAYS = int(MARKET_WARMUP_DAYS*(365/252))# 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
# QuantConnect specific imports
# import QuantConnect as qc
# 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 and the symbol's string
self.symbol_object = symbol_object
self.symbol = symbol_object.ID.Symbol # NEW
# Get the symbol's exchange market info
self.get_exchange_info()
# Add strategy specific variables
self.add_strategy_variables()
# Add the bars and indicators required
self.add_bars_indicators()
#-------------------------------------------------------------------------------
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_object].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_strategy_variables(self):
"""Add other required variables for the strategy."""
# Initialize order variables
self.reset_order_variables()
#-------------------------------------------------------------------------------
def reset_order_variables(self):
"""Reset order variables for the strategy."""
self.cost_basis = None
self.trade_best_price = None
self.sl_order = None
self.sl_price = None
self.pt_order1 = None
self.pt_order2 = None
self.pt_order3 = None
self.days_held = 0
#-------------------------------------------------------------------------------
def add_bars_indicators(self):
"""Add bars, indicators, and other required variables."""
# Create the desired daily bar consolidator for the symbol
consolidator = TradeBarConsolidator(self.daily_US_equity_calendar)
# 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_object, consolidator)
# Save consolidator link so we can remove it when necessary
self.consolidator = consolidator
# Create indicators to be based on the desired consolidated bars
self.ema_fast = ExponentialMovingAverage(EMA_FAST_PERIOD)
self.ema_slow = ExponentialMovingAverage(EMA_SLOW_PERIOD)
self.rsi = RelativeStrengthIndex(RSI_PERIOD)
# Create rolling windows of whether the fast EMA is > or < slow EMA
# format: RollingWindow[object type](length)
self.fast_ema_gt_slow_ema = RollingWindow[bool](2)
self.fast_ema_lt_slow_ema = RollingWindow[bool](2)
# Create a rolling window of the last closing prices
# used to make sure there is enough data to start trading
self.window_closes = RollingWindow[float](MIN_TRADING_DAYS)
# Create rolling windows for desired data
self.window_ema_fast = RollingWindow[float](INDICATOR_WINDOW_LENGTH)
self.window_ema_slow = RollingWindow[float](INDICATOR_WINDOW_LENGTH)
self.window_rsi = RollingWindow[float](INDICATOR_WINDOW_LENGTH)
self.window_bar = RollingWindow[TradeBar](INDICATOR_WINDOW_LENGTH)
# Keep a list of all indicators - for indicators_ready property
self.indicators = [self.ema_fast, self.ema_slow, self.rsi,
self.fast_ema_gt_slow_ema, self.fast_ema_lt_slow_ema,
self.window_closes, self.window_ema_fast, self.window_ema_slow,
self.window_rsi, self.window_bar]
# Get min bars required to initialize the indicators
self.min_bars = MARKET_WARMUP_DAYS
# Warm up the indicators with historical data
self.warmup_indicators()
#-------------------------------------------------------------------------------
def daily_US_equity_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.
Useful Refs:
datetime.replace() method:
https://docs.python.org/3/library/datetime.html#datetime.datetime.replace
"""
# 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)
# Need to handle case where algo initializes and this function is called
# for the first time. When that occurs, the current datetime when you
# requested a backtest is passed to this function as dt. This is the
# only time that dt has a timezone attached to it.
if dt.tzinfo:
# We need to make sure that the start datetime that we return from
# this function is BEFORE dt, otherwise it will throw this error:
# FuncPeriodSpecification: Please use a function that computes a
# date/time in the past
# (e.g.: Time.StartOfWeek and Time.StartOfMonth)
if start > dt:
# Make the start datetime go back by one day
start -= timedelta(1)
# Get today's end time from the SecurityExchangeHours Class object
# exchange_class = self.algo.Securities[self.symbol].Exchange
# exchange_hrs_class = self.algo.Securities[self.symbol].Exchange.Hours
# which is saved as self.exchange_hours
end = self.exchange_hours.GetNextMarketClose(start, False)
# Return the start datetime and the consolidation period
return CalendarInfo(start, end-start)
#-------------------------------------------------------------------------------
def warmup_indicators(self):
"""Warm up indicators using historical data."""
# Get historical data in pandas dataframe
df = self.algo.History(
[self.symbol_object],
self.min_bars,
Resolution.Daily
)
# Drops level 0 (Symbols), so only time as index
df.reset_index(level=0, inplace=True)
# Get only the desired columns
columns = ['open', 'high', 'low', 'close', 'volume']
try:
df = df[columns]
# Drop all rows with nans
df.dropna(inplace=True)
except:
self.algo.Debug(f"{self.symbol} ERROR warming up indicators")
return
# Loop through rows of df and update indicators
for index, row in df.iterrows():
# Create TradeBar
bar = TradeBar(index, self.symbol, row['open'], row['high'],
row['low'], row['close'], row['volume'])
self.update_indicators(bar)
# Log message that the indicators have been warmed up
# self.algo.Debug(f"{self.symbol} indicators warmed up with history")
#-------------------------------------------------------------------------------
def on_data_consolidated(self, sender, bar):
"""Event handler for desired custom bars."""
# Manually update all of the indicators
self.update_indicators(bar)
#-------------------------------------------------------------------------------
@property
# REF: https://www.programiz.com/python-programming/property
def indicators_ready(self):
"""Check if all of the indicators used are ready (warmed up)."""
# Return False if any indicator is not ready
for indicator in self.indicators:
if not indicator.IsReady:
return False
# Otherwise all indicators are ready, so return True
return True
#-------------------------------------------------------------------------------
@property
def ema_cross_over(self):
"""Return if there is an EMA cross-over."""
# Check if the rolling window is ready
if self.fast_ema_gt_slow_ema.IsReady:
# Need latest value True (fast > slow ema)
# and previous one False (fast <= slow ema)
return self.fast_ema_gt_slow_ema[0] \
and not self.fast_ema_gt_slow_ema[1]
# Otherwise return False
return False
#-------------------------------------------------------------------------------
@property
def ema_cross_under(self):
"""Return if there is an EMA cross-under."""
# Check if the rolling window is ready
if self.fast_ema_lt_slow_ema.IsReady:
# Need latest value True (fast < slow ema)
# and previous one False (fast >= slow ema)
return self.fast_ema_lt_slow_ema[0] \
and not self.fast_ema_lt_slow_ema[1]
# Otherwise return False
return False
#-------------------------------------------------------------------------------
@property
def fast_slow_pct_difference(self):
"""Return the percent difference between the fast and slow EMAs."""
# Check if both EMAs are ready
if self.ema_fast.IsReady and self.ema_slow.IsReady:
return (self.ema_fast.Current.Value-self.ema_slow.Current.Value) \
/self.ema_slow.Current.Value
# Otherwise return 0
return 0
#-------------------------------------------------------------------------------
@property
def current_qty(self):
"""Return the current quantity held in the portfolio."""
return self.algo.Portfolio[self.symbol_object].Quantity
#-------------------------------------------------------------------------------
def long_entry_signal(self):
"""Check if there is a valid long entry signal."""
# Trigger on an EMA cross-over
if self.ema_cross_over:
# Log message when desired
if PRINT_ENTRIES:
self.algo.Log(f"{self.symbol} LONG ENTRY SIGNAL: EMA CROSSOVER")
return True
#-------------------------------------------------------------------------------
def long_exit_signal_checks(self):
"""Check if there are any valid long exit signals."""
# Trigger on an EMA cross-under
if EMA_CROSSUNDER_EXIT and self.ema_cross_under:
# Log message when desired
if PRINT_EXITS:
self.algo.Log(f"{self.symbol} LONG EXIT SIGNAL: EMA CROSSUNDER")
self.algo.Liquidate(self.symbol_object)
return
# Check for RSI exits
exit_pct = 0
if RSI_EXIT_1:
# Check for RSI below the RSI_EXIT_1_VALUE
if self.rsi.Current.Value < RSI_EXIT_1_VALUE:
exit_pct += RSI_EXIT_1_PCT
# Log message when desired
if PRINT_EXITS:
self.algo.Log(f"{self.symbol} LONG EXIT SIGNAL: RSI "
f"({self.rsi.Current.Value}) < {RSI_EXIT_1_VALUE}. Now "
f"closing {exit_pct*100.0}% of the position.")
if RSI_EXIT_2:
# Check for RSI below the RSI_EXIT_2_VALUE
if self.rsi.Current.Value < RSI_EXIT_2_VALUE:
exit_pct += RSI_EXIT_2_PCT
# Log message when desired
if PRINT_EXITS:
self.algo.Log(f"{self.symbol} LONG EXIT SIGNAL: RSI "
f"({self.rsi.Current.Value}) < {RSI_EXIT_2_VALUE}. Now "
f"closing {exit_pct*100.0}% of the position.")
# Check for days held exits
if DAYS_HELD_EXIT_1:
if self.days_held == DAYS_HELD_EXIT_1_VALUE:
exit_pct += DAYS_HELD_EXIT_1_PCT
# Make sure we don't trade more than the full position (100%)
exit_pct = min(1, exit_pct)
# Log message when desired
if PRINT_EXITS:
self.algo.Log(f"{self.symbol} LONG EXIT SIGNAL: Days "
f"held ({self.days_held}) = {DAYS_HELD_EXIT_1_VALUE}. "
f"Now closing {exit_pct*100.0}% of the position.")
if DAYS_HELD_EXIT_2:
if self.days_held == DAYS_HELD_EXIT_2_VALUE:
exit_pct += DAYS_HELD_EXIT_2_PCT
# Make sure we don't trade more than the full position (100%)
exit_pct = min(1, exit_pct)
# Log message when desired
if PRINT_EXITS:
self.algo.Log(f"{self.symbol} LONG EXIT SIGNAL: Days "
f"held ({self.days_held}) = {DAYS_HELD_EXIT_2_VALUE}. "
f"Now closing {exit_pct*100.0}% of the position.")
if DAYS_HELD_EXIT_3:
if self.days_held == DAYS_HELD_EXIT_3_VALUE:
exit_pct += DAYS_HELD_EXIT_3_PCT
# Make sure we don't trade more than the full position (100%)
exit_pct = min(1, exit_pct)
# Log message when desired
if PRINT_EXITS:
self.algo.Log(f"{self.symbol} LONG EXIT SIGNAL: Days "
f"held ({self.days_held}) = {DAYS_HELD_EXIT_3_VALUE}. "
f"Now closing {exit_pct*100.0}% of the position.")
# Check if any of the position should be closed
if exit_pct > 0:
# Get actual quantity to exit
exit_qty = -self.current_qty*exit_pct
# Place market order to exit
self.algo.MarketOrder(self.symbol_object, exit_qty)
#-------------------------------------------------------------------------------
def update_indicators(self, bar):
"""Manually update all of the symbol's indicators."""
# Update the EMAs and RSI
self.ema_fast.Update(bar.EndTime, bar.Close)
self.ema_slow.Update(bar.EndTime, bar.Close)
self.rsi.Update(bar.EndTime, bar.Close)
# Update rolling windows when ready
if self.ema_fast.IsReady:
self.window_ema_fast.Add(self.ema_fast.Current.Value)
if self.ema_slow.IsReady:
self.window_ema_slow.Add(self.ema_slow.Current.Value)
if self.ema_fast.IsReady and self.ema_slow.IsReady:
self.fast_ema_gt_slow_ema.Add(
self.ema_fast.Current.Value>self.ema_slow.Current.Value)
self.fast_ema_lt_slow_ema.Add(
self.ema_fast.Current.Value<self.ema_slow.Current.Value)
if self.rsi.IsReady:
self.window_rsi.Add(self.rsi.Current.Value)
self.window_closes.Add(bar.Close)
self.window_bar.Add(bar)
# Update the trades best price if trailing stop is used
if TRAILING_STOP:
self.update_trade_best_price(bar)
# Increment days held counter if there is a positioin
if self.current_qty != 0:
self.days_held += 1
#-------------------------------------------------------------------------------
def update_trade_best_price(self, bar):
"""Update the trade's best price if there is an open position."""
# Check if there is an open position
if self.current_qty > 0: # long
# Update the trade best price when appropriate
if not self.trade_best_price:
# Should be set, so raise error to debug
raise
elif bar.High > self.trade_best_price:
self.trade_best_price = bar.High
# Get the current stop loss price
sl_price = self.get_stop_price()
# Check for increase in stop loss price
if sl_price > self.sl_price:
# Update the stop loss order's price
self.update_order_stop_price(self.sl_order, sl_price)
self.sl_price = sl_price
#-------------------------------------------------------------------------------
def get_stop_price(self):
"""Get the current stop loss price."""
# Calculate the stop loss price
if self.current_qty > 0: # long
return round(self.trade_best_price*(1-SL_PCT),2)
elif self.current_qty < 0: # short
return round(self.trade_best_price*(1+SL_PCT),2)
#-------------------------------------------------------------------------------
def cancel_exit_orders(self):
"""Cancel any open exit orders."""
# Cancel open profit target order #1, if one
if self.pt_order1:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Cancelling {self.symbol} open profit target #1 "
f"order.")
try:
self.pt_order1.Cancel()
except:
self.algo.Log(f"Error trying to cancel {self.symbol} profit "
f"target #1 order.")
# Cancel open profit target order #2, if one
if self.pt_order2:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Cancelling {self.symbol} open profit target #2 "
f"order.")
try:
self.pt_order2.Cancel()
except:
self.algo.Log(f"Error trying to cancel {self.symbol} profit "
f"target #2 order.")
# Cancel open profit target order #3, if one
if self.pt_order3:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Cancelling {self.symbol} open profit target #3 "
f"order.")
try:
self.pt_order3.Cancel()
except:
self.algo.Log(f"Error trying to cancel {self.symbol} profit "
f"target #3 order.")
# Cancel open stop order, if one
if self.sl_order:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Cancelling {self.symbol} open stop order.")
try:
self.sl_order.Cancel()
except:
self.algo.Log(f"Error trying to cancel {self.symbol} stop "
f"loss order.")
# Reset order variables
self.reset_order_variables()
#-------------------------------------------------------------------------------
def get_pt_exit_quantities(self, initial=False):
"""Get the profit target exit quantities for orders #1, #2, #3."""
# Get the exit order qty
exit_qty = -self.current_qty
# Initialize each to 0
pt1_exit_qty = 0
pt2_exit_qty = 0
pt3_exit_qty = 0
# Get PT1, PT2, PT3 exit qtys
pt_exit_qty = 0
if (initial and PROFIT_TARGET_1) or self.pt_order1:
pt1_exit_qty = int(exit_qty*PT1_ORDER_PCT)
pt_exit_qty += pt1_exit_qty
if (initial and PROFIT_TARGET_1) or self.pt_order2:
pt2_exit_qty = int(exit_qty*PT2_ORDER_PCT)
pt_exit_qty += pt2_exit_qty
if (initial and PROFIT_TARGET_1) or self.pt_order3:
pt3_exit_qty = int(exit_qty*PT3_ORDER_PCT)
pt_exit_qty += pt3_exit_qty
# Make sure pt_exit_qty equals exit_qty
if pt_exit_qty != exit_qty:
# Get the difference
diff = exit_qty-pt_exit_qty
# Add the difference to the last pt order
if (initial and PROFIT_TARGET_1) or self.pt_order3:
pt3_exit_qty += diff
elif (initial and PROFIT_TARGET_1) or self.pt_order2:
pt2_exit_qty += diff
elif (initial and PROFIT_TARGET_1) or self.pt_order1:
pt2_exit_qty += diff
# Return the quantities
return pt1_exit_qty, pt2_exit_qty, pt3_exit_qty
#-------------------------------------------------------------------------------
def update_exit_orders(self):
"""Update any open exit orders."""
# Get the exit order qty
exit_qty = -self.current_qty
# Get the desired profit target exit order quantities
pt1_exit_qty, pt2_exit_qty, pt3_exit_qty = self.get_pt_exit_quantities()
# Update open profit target order #1, if one
if self.pt_order1:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Updating {self.symbol} open profit target #1 "
f"order qty to {pt1_exit_qty}.")
# Get the profit taking order ticket, then update it
ticket = self.pt_order1
self.update_order_qty(ticket, pt1_exit_qty)
# Update open profit target order #2, if one
if self.pt_order2:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Updating {self.symbol} open profit target #2 "
f"order qty to {pt2_exit_qty}.")
# Get the profit taking order ticket, then update it
ticket = self.pt_order2
self.update_order_qty(ticket, pt2_exit_qty)
# Update open profit target order #3, if one
if self.pt_order3:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Updating {self.symbol} open profit target #3 "
f"order qty to {pt3_exit_qty}.")
# Get the profit taking order ticket, then update it
ticket = self.pt_order3
self.update_order_qty(ticket, pt3_exit_qty)
# Update open stop loss order, if one
if self.sl_order:
# Log message whe desired
if PRINT_ORDERS:
self.algo.Log(f"Updating {self.symbol} open stop loss "
f"order qty to {exit_qty}.")
# Get the profit taking order ticket, then update it
ticket = self.sl_order
self.update_order_qty(ticket, exit_qty)
#-------------------------------------------------------------------------------
def update_order_qty(self, ticket, qty):
"""Update the desired order ticket's qty."""
ticket.UpdateQuantity(qty, tag=f'updating qty to {qty}')
#-------------------------------------------------------------------------------
def update_order_stop_price(self, ticket, price):
"""Update the desired order ticket's stop price."""
ticket.UpdateStopPrice(price, tag=f'updating stop price to {price}')
#-------------------------------------------------------------------------------
def update_order_limit_price(self, ticket, price):
"""Update the desired order ticket's limit price."""
ticket.UpdateLimitPrice(price, tag=f'updating limit price to {price}')
#-------------------------------------------------------------------------------
def place_sl_order(self):
"""Place the desired stop loss order."""
# Get the current stop loss price
self.sl_price = self.get_stop_price()
# Place and save the stop loss order
self.sl_order = self.algo.StopMarketOrder(
self.symbol_object, -self.current_qty, self.sl_price)
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} stop order placed at {self.sl_price}")
#-------------------------------------------------------------------------------
def place_pt_orders(self):
"""Place the desired profit target orders."""
# Get the desired profit target exit order quantities
pt1_exit_qty, pt2_exit_qty, pt3_exit_qty = \
self.get_pt_exit_quantities(initial=True)
# Check for placing a profit target order #1
if PROFIT_TARGET_1:
# Calculate the profit target price
if self.current_qty > 0: # long
pt_price = round(self.cost_basis*(1+PT1_PCT),2)
elif self.current_qty < 0: # short
pt_price = round(self.cost_basis*(1-PT1_PCT),2)
# Place and save the stop loss order
self.pt_order1 = self.algo.LimitOrder(
self.symbol_object, pt1_exit_qty, pt_price)
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} profit target #1 order: "
f"{pt1_exit_qty} at {pt_price}")
# Check for placing a profit target order #2
if PROFIT_TARGET_2:
# Calculate the profit target price
if self.current_qty > 0: # long
pt_price = round(self.cost_basis*(1+PT2_PCT),2)
elif self.current_qty < 0: # short
pt_price = round(self.cost_basis*(1-PT2_PCT),2)
# Place and save the stop loss order
self.pt_order2 = self.algo.LimitOrder(
self.symbol_object, pt2_exit_qty, pt_price)
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} profit target #2 order: "
f"{pt2_exit_qty} at {pt_price}")
# Check for placing a profit target order #3
if PROFIT_TARGET_3:
# Calculate the profit target price
if self.current_qty > 0: # long
pt_price = round(self.cost_basis*(1+PT3_PCT),2)
elif self.current_qty < 0: # short
pt_price = round(self.cost_basis*(1-PT3_PCT),2)
# Place and save the stop loss order
self.pt_order3 = self.algo.LimitOrder(
self.symbol_object, pt2_exit_qty, pt_price)
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} profit target #3 order: "
f"{pt2_exit_qty} at {pt_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
# Get current qty of symbol
qty = self.current_qty
# Check for entry order
if order_qty == qty:
# Entry order filled
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} entry order filled: {order_qty}"
f" at {avg_fill}")
# Save the cost basis and trade best price
self.cost_basis = avg_fill
self.trade_best_price = avg_fill
# Place a stop loss order when desired
if STOP_LOSS or TRAILING_STOP:
self.place_sl_order()
# Place profit target orders when desired
self.place_pt_orders()
# Done with event, so return
return
# Check for stop order
if self.sl_order:
# Check for matching order ids
if order_event.OrderId == self.sl_order.OrderId:
# Stop order filled
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} stop order filled: "
f"{order_qty} at {avg_fill}")
# Cancel open exit orders
self.cancel_exit_orders()
# Done with event, so return
return
# Check for profit target order #1
if self.pt_order1:
# Check for matching order ids
if order_event.OrderId == self.pt_order1.OrderId:
# Profit target order filled
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} profit target order #1 "
f"filled: {order_qty} at {avg_fill}")
# Check if the position is still open
if qty != 0:
# Update open exit orders
self.update_exit_orders()
else:
# Cancel open exit orders
self.cancel_exit_orders()
# Set order to None
self.pt_order1 = None
# Done with event, so return
return
# Check for profit target order #2
if self.pt_order2:
# Check for matching order ids
if order_event.OrderId == self.pt_order2.OrderId:
# Profit target order filled
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} profit target order #2 "
f"filled: {order_qty} at {avg_fill}")
# Check if the position is still open
if qty != 0:
# Update open exit orders
self.update_exit_orders()
else:
# Cancel open exit orders
self.cancel_exit_orders()
# Set order to None
self.pt_order2 = None
# Done with event, so return
return
# Check for profit target order #3
if self.pt_order3:
# Check for matching order ids
if order_event.OrderId == self.pt_order3.OrderId:
# Profit target order filled
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} profit target order #3 "
f"filled: {order_qty} at {avg_fill}")
# Check if the position is still open
if qty != 0:
# Update open exit orders
self.update_exit_orders()
else:
# Cancel open exit orders
self.cancel_exit_orders()
# Set order to None
self.pt_order3 = None
# Done with event, so return
return
# Check for full exit order
if qty == 0:
# Exit order filled
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} exit order filled: "
f"{order_qty} at {avg_fill}")
# Cancel open exit orders
self.cancel_exit_orders()
# Done with event, so return
return
# Check for pyramid entry order (qty and order_qty have the same signs)
if qty*order_qty > 0:
# This strategy doesn't have pyramid entries, so raise error
raise
# Otherwise a partial exit order
else:
# Partial exit order filled
# Log message when desired
if PRINT_ORDERS:
self.algo.Log(f"{self.symbol} partial exit order filled: "
f"{order_qty} at {avg_fill}")
# Update open exit orders
self.update_exit_orders()
# Done with event, so return
return###############################################################################
# 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
# QuantConnect specific imports
# import QuantConnect as qc
# Import from files
from notes_and_inputs import *
from symbol_data import *
###############################################################################
class CustomTradingStrategy(QCAlgorithm):
def Initialize(self):
"""Initialize algorithm."""
# Set backtest details
self.SetBacktestDetails()
# Add instrument data to the algo
self.AddInstrumentData()
# Schedule functions
self.ScheduleFunctions()
# Warm up the indicators prior to the start date
# self.SetWarmUp()
# This doesn't work with universe filters
# Instead we'll use History() to warm up indicators when SymbolData
# class objects get created
#-------------------------------------------------------------------------------
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.Cash) # CASH ACCOUNT
# Configure all universe securities
# This sets the data normalization mode
# You can also set custom fee, slippage, fill, and buying power models
self.SetSecurityInitializer(self.CustomSecurityInitializer)
# Adjust the cash buffer from the default 2.5% to custom setting
self.Settings.FreePortfolioValuePercentage = FREE_PORTFOLIO_VALUE_PCT
#-------------------------------------------------------------------------------
def AddInstrumentData(self):
"""Add instrument data to the algo."""
# Set data resolution based on input
if DATA_RESOLUTION == 'SECOND':
resolution = Resolution.Second
elif DATA_RESOLUTION == 'MINUTE':
resolution = Resolution.Minute
# Define the desired universe
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# Set universe data properties desired
self.UniverseSettings.Resolution = resolution
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = \
DataNormalizationMode.Adjusted
self.UniverseSettings.MinimumTimeInUniverse = MIN_TIME_IN_UNIVERSE
# Add data for the benchmark and set benchmark
# Always use minute data for the benchmark
self.bm = self.AddEquity(BENCHMARK, Resolution.Minute).Symbol
self.SetBenchmark(BENCHMARK)
# Create a dictionary to hold SymbolData class objects
self.symbol_data = {}
# Create a variable to tell the algo when to update the universe
self.update_universe = True # update at the beginning of the backtest
#-------------------------------------------------------------------------------
def ScheduleFunctions(self):
"""Scheduling the functions required by the algo."""
# For live trading, universe selection occurs approximately 04:00-07:00
# EST on Tue-Sat.
# For backtesting, universe selection occurs at 00:00 EST
# We need to update the self.update_universe variable
# before both of these scenarios are triggered
# Desired order of events:
# 1. Update the self.update_universe variable True
# end of week/month, 5 min after market close
# 2. Coarse/Fine universe filters run and update universe
# run everyday at either 00:00 or 04:00 EST
# Update self.update_universe variable True when desired
if UNIVERSE_FREQUENCY == 'daily':
date_rules = self.DateRules.EveryDay(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.DateRules.WeekEnd(self.bm)
elif UNIVERSE_FREQUENCY == 'monthly':
# Want to schedule at end of the month, so actual update on
# first day of the next month
date_rules = self.DateRules.MonthEnd(self.bm)
# Timing is after the market closes
self.Schedule.On(
date_rules,
self.TimeRules.BeforeMarketClose(self.bm, -5),
self.UpdateUniverse
)
# Calling -5 minutes "BeforeMarketClose" schedules the function 5
# minutes "after market close"
# Calling -5 minutes "AfterMarketOpen" schedules the function 5
# minutes "before market open"
# Now the coarse/fine universe filters will run automatically either at
# 00:00 EST for backtesting or
# 04:00 EST for live trading
# Check for new signals SIGNAL_CHECK_MINUTES before the market open
self.Schedule.On(
self.DateRules.EveryDay(self.bm),
self.TimeRules.AfterMarketOpen(self.bm, -SIGNAL_CHECK_MINUTES),
self.CheckForSignals
)
# Check if end of day exit is desired
if EOD_EXIT:
# Schedule function to liquidate the portfolio EOD_EXIT_MINUTES
# before the market close
self.Schedule.On(
self.DateRules.EveryDay(self.bm),
self.TimeRules.BeforeMarketClose(self.bm, EOD_EXIT_MINUTES),
self.LiquidatePortfolio
)
# Check if we want to plot the benchmark on the equity curve
if PLOT_BENCHMARK_ON_STRATEGY_EQUITY_CHART:
# Schedule benchmark end of day event 5 minutes after the close
self.Schedule.On(
self.DateRules.EveryDay(self.bm),
self.TimeRules.BeforeMarketClose(self.bm, -5),
self.BenchmarkOnEndOfDay
)
#-------------------------------------------------------------------------------
def UpdateUniverse(self):
"""Event called when rebalancing is desired."""
# Update variable to trigger the universe to be updated
self.update_universe = True
#-------------------------------------------------------------------------------
def CoarseSelectionFunction(self, coarse):
"""
Perform coarse filters on universe.
Called once per day.
Returns all stocks meeting the desired criteria.
Attributes available:
.AdjustedPrice
.DollarVolume
.HasFundamentalData
.Price -> always the raw price!
.Volume
"""
# # Testing - catch specific symbol
# for x in coarse:
# # if str(x.Symbol).split(" ")[0] in ['AAPL']:
# if x.Symbol.ID.Symbol in ['AAPL']:
# # Stop and debug below
# print(x)
# Check if the universe doesn't need to be updated
if not self.update_universe:
# Return unchanged universe
return Universe.Unchanged
# Otherwise update the universe based on the desired filters
# Filter all securities with appropriate price and volume
filtered_coarse = [x for x in coarse if \
x.Price >= MIN_PRICE and \
x.Price <= MAX_PRICE and \
x.Volume >= MIN_DAILY_VOLUME and \
x.DollarVolume >= MIN_DAILY_DOLLAR_VOLUME
]
# Check if fundamental data is required
if REQUIRE_FUNDAMENTAL_DATA:
# Filter all securities with fundamental data
filtered_coarse = \
[x for x in filtered_coarse if x.HasFundamentalData]
# Return the symbol objects
symbols = [x.Symbol for x in filtered_coarse]
# Print universe details when desired
if PRINT_COARSE:
self.Log(f"Coarse filter returned {len(symbols)} stocks.")
return symbols
#-------------------------------------------------------------------------------
def FineSelectionFunction(self, fine):
"""
Perform fine filters on universe.
Called once per day.
Returns all stocks meeting the desired criteria.
Attribues available:
.AssetClassification
.CompanyProfile
.CompanyReference
.EarningRatios
.EarningReports
.FinancialStatements
.MarketCap
.OperationRatios
.Price -> always the raw price!
.ValuationRatios
"""
# # Testing - catch specific symbol
# for x in coarse:
# # if str(x.Symbol).split(" ")[0] in ['AAPL']:
# if x.Symbol.ID.Symbol in ['AAPL']:
# # Stop and debug below
# print(x)
# Check if the universe doesn't need to be updated
if not self.update_universe:
# Return unchanged universe
return Universe.Unchanged
# Otherwise update the universe based on the desired filters
# Filter by allowed exchange and market cap
symbols = [x for x in fine if \
x.SecurityReference.ExchangeId in ALLOWED_EXCHANGE and \
x.MarketCap >= MIN_MARKET_CAP and \
x.MarketCap <= MAX_MARKET_CAP
]
# Filter stocks based on primary share class
if PRIMARY_SHARES:
symbols = [x for x in symbols if x.SecurityReference.IsPrimaryShare]
# Filter stocks based on disallowed sectors
if len(SECTORS_NOT_ALLOWED) > 0:
symbols = [x for x in symbols if \
x.AssetClassification.MorningstarSectorCode \
not in SECTORS_NOT_ALLOWED
]
# Filter stocks based on disallowed industry groups
if len(GROUPS_NOT_ALLOWED) > 0:
symbols = [x for x in symbols if \
x.AssetClassification.MorningstarIndustryGroupCode \
not in GROUPS_NOT_ALLOWED
]
# Filter stocks based on disallowed industries
if len(INDUSTRIES_NOT_ALLOWED) > 0:
symbols = [x for x in symbols if \
x.AssetClassification.MorningstarIndustryCode \
not in INDUSTRIES_NOT_ALLOWED
]
# Return the symbol objects
self.symbols = [x.Symbol for x in symbols]
# Print universe details when desired
if PRINT_FINE:
self.Log(f"Fine filter returned {len(self.symbols)} stocks.")
# Set update universe variable back to False
self.update_universe = False
return self.symbols
#-------------------------------------------------------------------------------
def OnSecuritiesChanged(self, changes):
"""Event handler for changes to our universe."""
# Loop through securities added to the universe
for security in changes.AddedSecurities:
# Get the security symbol string
symbol = security.Symbol
# Skip if BENCHMARK - we cannot trade this!
if symbol.ID.Symbol == BENCHMARK:
continue
# Create a new symbol_data object for the security
self.symbol_data[symbol] = SymbolData(self, symbol) # Using QC.Symbol for key
# Loop through securities removed from the universe
for security in changes.RemovedSecurities:
# Get the security symbol string
symbol = security.Symbol
# Liquidate removed securities
if security.Invested:
# Log message when desired
if PRINT_ORDERS:
self.Log(f"{symbol.ID.Symbol} removed from the universe, "
f"so closing open position.")
self.Liquidate(security.Symbol)
# Remove from symbol_data dictionary
if symbol in self.symbol_data:
# Remove desired bar consolidator for security
consolidator = self.symbol_data[symbol].consolidator
self.SubscriptionManager.RemoveConsolidator(
security.Symbol, consolidator)
# Remove symbol from symbol data
self.symbol_data.pop(symbol)
#-------------------------------------------------------------------------------
def LiquidatePortfolio(self):
"""Liquidate the entire portfolio. Also cancel any pending orders."""
# Log message when desired
if self.Portfolio.Invested and PRINT_ORDERS:
self.Log("Time for end of day exit. Liquidating the portfolio.")
self.Liquidate()
#-------------------------------------------------------------------------------
def CheckForSignals(self):
"""Event called when signal checks are desired."""
# self.Log(f"CheckForSignals: {self.Time}")
# Check for exit signals
self.CheckForExits()
# Check for entry signals
self.CheckForEntries()
#-------------------------------------------------------------------------------
def CheckForExits(self):
"""Check for exit signals."""
# Get list of current positions
positions = [symbol for symbol in self.Portfolio.Keys \
if self.Portfolio[symbol].Invested]
# Loop through positions
for sym in positions:
# Check for long position
if self.Portfolio[sym].Quantity > 0: # long
# Check for long exit signal
self.symbol_data[sym].long_exit_signal_checks()
#-------------------------------------------------------------------------------
def CheckForEntries(self):
"""Check for entry signals."""
# Get list of current positions
positions = [symbol for symbol in self.Portfolio.Keys \
if self.Portfolio[symbol].Invested]
# Get number of new entries allowed
new_entry_num = MAX_POSITIONS-len(positions)
# Return if no new positions are allowed
if new_entry_num == 0:
return
# Create a list of long symbol, pct_change tuples
long_tuples = []
long_tuples_symbols = []
# Loop through the SymbolData class objects
for sym in self.symbol_data.keys():
# Skip if already invested
if self.Securities[sym].Invested:
continue
# Check if indicators are ready
if self.symbol_data[sym].indicators_ready:
# Check for long entry signal
if self.symbol_data[sym].long_entry_signal():
# Add to the long list
long_tuples.append(
(sym,
self.symbol_data[sym].fast_slow_pct_difference)
)
long_tuples_symbols.append(sym)
# Check if the number of new entries exceeds limit
if len(long_tuples) > new_entry_num:
# Sort the entry_tuples list of tuples by largest pct_change
# pct_change is the second element of the tuple
# reverse=True for descending order (highest to lowest)
long_tuples = sorted(long_tuples, key=lambda x:x[1], reverse=True)
# Only keep the top new_entry_num
long_tuples = long_tuples[:new_entry_num]
# Get list of long symbol objects
long_symbols = []
for tup in long_tuples:
if tup[0] in long_tuples_symbols:
# Verify there is data available
if self.Securities[tup[0]].HasData:
long_symbols.append(tup[0])
else: # must ignore signal, order will not be handled
continue
# Print entry signal summary when desired
if PRINT_ENTRIES:
# Log message
if len(long_symbols) > 0:
long_symbol_strings = [x.ID.Symbol for x in long_symbols]
self.Log(f"{len(long_symbols)} LONG entry signal(s): "
f"{long_symbol_strings}")
# Place entry orders
for symbol_object in long_symbols:
self.SetHoldings(symbol_object, MAX_PCT_PER_POSITION)
#-------------------------------------------------------------------------------
def OnOrderEvent(self, orderEvent):
"""Built-in event handler for orders."""
# Skip if not filled
if orderEvent.Status != OrderStatus.Filled:
return
# Get the order's symbol
symbol = orderEvent.Symbol
# Call on_order_event for the symbol's SymbolData class
self.symbol_data[symbol].on_order_event(orderEvent)
#-------------------------------------------------------------------------------
# def OnData(self, data):
# """Event handler for new data pumped into the algo."""
# pass
#-------------------------------------------------------------------------------
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."""
# Check if we want to plot the benchmark
if PLOT_BENCHMARK_ON_STRATEGY_EQUITY_CHART:
self.PlotBenchmarkOnEquityCurve(True)
# self.Log("End of Backtest")
#-------------------------------------------------------------------------------
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.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.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.Log("Benchmark's first price = {}".format(
self.bm_first_price))
self.Log(f"Benchmark's final price = {price}")
self.Log(f"Benchmark buy & hold value = {bm_value}")
#-------------------------------------------------------------------------------
def CustomSecurityInitializer(self, security):
"""
Define models to be used for securities as they are added to the
algorithm's universe.
"""
# Define the data normalization mode
security.SetDataNormalizationMode(DataNormalizationMode.Adjusted)
# 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()