| Overall Statistics |
|
Total Trades 147 Average Win 3.65% Average Loss -3.95% Compounding Annual Return -99.586% Drawdown 71.200% Expectancy -0.394 Net Profit -60.005% Sharpe Ratio -1.2 Probabilistic Sharpe Ratio 0.207% Loss Rate 69% Win Rate 31% Profit-Loss Ratio 0.92 Alpha 0 Beta 0 Annual Standard Deviation 0.819 Annual Variance 0.671 Information Ratio -1.2 Tracking Error 0.819 Treynor Ratio 0 Total Fees $3034.19 Estimated Strategy Capacity $240000.00 Lowest Capacity Asset RETO WPYH6M9J9085 |
''' TickerState
States a ticker can take while being processed against a brokerage.
'''
from enum import Enum, auto
class TickerState(Enum):
NEW = auto()
BUY = auto()
SHORT = auto()
LIQUIDATE = auto()
from datetime import datetime, timedelta
from QuantConnect.Data.UniverseSelection import *
from config import Config
from universe import UniverseSelection
from alpha import Alpha
from portfolio import PortfolioConstruction
from execute import OrderExecution
from stop import StopPrices
class OpeningRangeBreakout(QCAlgorithm):
''' OpeningRangeBreakout
'''
def Initialize(self):
# Configuration parameters
self.config = Config(self)
# Backtesting
self.SetStartDate(2021, 3, 1) # Set backtest Start Date
self.SetEndDate(2021, 5, 1) # Set backtest End Date
self.SetCash(35000) # Set backtest Strategy Cash
# Brokerage
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
# Construct our universe of equities to trade.
self.UniverseSettings.Resolution = Resolution.Minute
self.AddUniverseSelection(UniverseSelection(self.config))
# symbols = [Symbol.Create("SPY", SecurityType.Equity, Market.USA)]
# self.AddUniverseSelection(ManualUniverseSelectionModel(symbols))
# Alpha Model
# Generate trading signals.
self.stopPrices = StopPrices()
self.SetAlpha(Alpha(self.config, self.stopPrices))
# Portfolio Construction
# Determine the best portfolio based on the trading signals.
self.SetPortfolioConstruction(PortfolioConstruction(self.config))
# Risk Management Model
self.SetRiskManagement(NullRiskManagementModel())
# Execution Model
self.SetExecution(OrderExecution(self.config, self.stopPrices))
from datetime import timedelta
from ticker import TickerState
class Alpha(AlphaModel):
'''Alpha - Determine trading signals
Insights are evaluated every X minutes, e.g. 10,
but this method is called every minute.
The indicators are likewise updated every minute.
'''
def __init__(self, config, stopPrices):
self.config = config
self.stopPrices = stopPrices
self.symbolDataBySymbol = {}
def Update(self, algorithm, data):
''' Update - Scan symbol data for insights
Called every minute.
'''
def GenerateInsight(_sd, _direction, _state):
# Create and return an Insight.
# Set the SymbolData direction and state.
insight = Insight.Price(_sd.symbol, timedelta(minutes=self.config.consolidationPeriod), _direction)
_sd.direction = _direction
_sd.state = _state
return insight
def CalcMidpoint(_high, _low):
mid = abs((_high - _low) / 2)
base = _low if _low < _high else _high
return mid + base
# List of Insights to emit
insights = []
#
# Scan for insights among our universe of symbols.
#
removedSymbolData = []
for s, sd in self.symbolDataBySymbol.items():
if sd.IsReady:
# The opening bar status for this symbol is valid.
price = data[sd.symbol].Close if data.Bars.ContainsKey(sd.symbol) else None
if price is None:
# Tradebar is not ready.
continue
if algorithm.Portfolio[sd.symbol].Invested:
#
# Invested in the ticker.
#
if sd.state == TickerState.LIQUIDATE:
# End Of Day or ticker was removed.
insights.append(GenerateInsight(sd, InsightDirection.Flat, TickerState.LIQUIDATE))
# Clear opening bar.
sd.openingBar = None
if sd.remove:
# The ticker has been removed. Portfolio will handle closeout.
removedSymbolData.append(s)
# If invested, but TickerState is not LIQUIDATE, do nothing.
else:
#
# We are not invested in this ticker.
#
# Only trade during the specified opening period.
if (data.Bars[sd.symbol].Time.hour == self.config.exitTime.hour \
and data.Bars[sd.symbol].Time.minute > self.config.exitTime.minute) \
or (data.Bars[sd.symbol].Time.hour > self.config.exitTime.hour):
continue
# Check wheather we've already issued an insight, but the order has not yet executed.
# This should not happen with a default timeout of 5 seconds, but just to be complete,
# guard against it anyway.
# TODO Could we linger here if an order execution fails, since we ignore those?
# TODO sd.state == LIQUIDATE won't get us out of this state if .Invested is not set.
if sd.state == TickerState.BUY or sd.state == TickerState.SHORT:
continue
# Determine if there is an Opening Range Breakout.
if price > sd.openingBar.High:
# Opening Range Breakout HIGH.
# Calculate appropriate StopMarketOrder price.
self.stopPrices[sd.symbol] = CalcMidpoint(sd.openingBar.High, sd.openingBar.Low)
# BUY the ticker.
insights.append(GenerateInsight(sd, InsightDirection.Up, TickerState.BUY))
elif price < sd.openingBar.Low:
# Opening Range Breakout LOW.
# Calculate appropriate StopMarketOrder price.
self.stopPrices[sd.symbol] = CalcMidpoint(sd.openingBar.High, sd.openingBar.Low)
# SHORT the ticker.
insights.append(GenerateInsight(sd, InsightDirection.Down, TickerState.SHORT))
# Remove any deleted tickers.
for s in removedSymbolData:
del self.symbolDataBySymbol[s]
return insights
def OnSecuritiesChanged(self, algorithm, changes):
for added in changes.AddedSecurities:
symbolData = self.symbolDataBySymbol.get(added.Symbol)
if symbolData is None:
algorithm.Log("Adding ticker {}".format(added.Symbol.Value))
self.symbolDataBySymbol[added.Symbol] = SymbolData(self.config, algorithm, added.Symbol)
else:
algorithm.Log("Resetting ticker {}".format(added.Symbol.Value))
for removed in changes.RemovedSecurities:
algorithm.Log("Removing ticker {}".format(removed.Symbol.Value))
symbolData = self.symbolDataBySymbol.get(removed.Symbol)
if symbolData is not None:
if algorithm.Portfolio[removed.Symbol].Invested:
# We are invested in this ticker, so we must liquidate it.
symbolData.state = TickerState.LIQUIDATE
symbolData.remove = True
class SymbolData:
''' Indicator for one symbol of our universe.
Each indicator is updated automatically.
'''
def __init__(self, config, algorithm, symbol):
self.config = config
self.symbol = symbol
self.openingBar = None
self.direction = None
self.state = TickerState.NEW
self.remove = False
self.consolidator = algorithm.Consolidate(self.symbol, timedelta(minutes=self.config.consolidationPeriod), self.OnDataConsolidated)
@property
def IsReady(self):
''' The symbol data is ready when the
openingBar has been collected.
'''
return self.openingBar is not None
def OnDataConsolidated(self, bar):
if bar.Time.hour == self.config.entryTime.hour and bar.Time.minute == self.config.entryTime.minute:
self.openingBar = bar
elif bar.Time.hour == self.config.closeTime.hour and bar.Time.minute == self.config.closeTime.minute:
self.ClosePosition()
def ClosePosition(self):
self.state = TickerState.LIQUIDATE
# Instantiate QuantConnect parameters.
from datetime import *
from decimal import *
class Config():
def __init__(self, algorithm):
#
# Universe parameters
#
self.selectionDay = int(algorithm.GetParameter("selectionDay"))
self.minPrice = Decimal(algorithm.GetParameter("minPrice"))
self.minVolume = float(algorithm.GetParameter("minVolume"))
self.minDaysSinceIPO = int(algorithm.GetParameter("minDaysSinceIPO"))
self.maxFineUniverse = int(algorithm.GetParameter("maxFineUniverse"))
self.daysMemory = int(algorithm.GetParameter("daysMemory"))
#
# Alpha parameters
#
self.entryTime = datetime.strptime("09:30", "%H:%M")
self.exitTime = datetime.strptime("10:30", "%H:%M")
self.closeTime = datetime.strptime("13:30", "%H:%M")
self.consolidationPeriod = int(algorithm.GetParameter("consolidationPeriod"))
#
# Portfolio Construction parameters
#
self.initialMargin = float(algorithm.GetParameter("initialMargin"))
#
# Generic Simple Execution Model
#
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
from System import *
from QuantConnect import *
from QuantConnect.Orders import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Execution import *
from QuantConnect.Algorithm.Framework.Portfolio import *
class OrderExecution(ExecutionModel):
''' Immediately submits market orders to achieve the desired portfolio targets,
with stop loss and end-of-day liquidation.
'''
def __init__(self, config, stopPrices):
'''Initializes a new instance of the Generic Simple Execution class'''
self.config = config
self.stopPrices = stopPrices
def Execute(self, algorithm, targets):
''' Immediately submits orders for the specified portfolio targets.
Implicit Args:
stopPrices Stop prices for symbols being traded
Args:
algorithm: The algorithm instance
targets: The portfolio targets to be ordered
Positive quantity denotes an equity purchase
Negative quantity denotes an equity sale
Zero quantity denotes an equity liquidation
'''
for target in targets:
# Get quantity to be ordered.
quantity = target.Quantity
if quantity != 0:
# BUY or SHORT the equity.
algorithm.Log("Market order for {} shares of {}".format(quantity, target.Symbol.Value))
algorithm.MarketOrder(target.Symbol, quantity)
# Create StopMarketOrder with price calculated by the AlphaModel.
algorithm.Log("Stop Market order for {} shares of {} at ${:.2f}".format(quantity, target.Symbol.Value, self.stopPrices[target.Symbol]))
algorithm.StopMarketOrder(target.Symbol, quantity, self.stopPrices[target.Symbol])
_direction = OrderDirection.Buy if quantity > 0 else OrderDirection.Sell
algorithm.Log("Available cash ${:.2f}, remaining margin ${:.2f}".format(algorithm.Portfolio.Cash,
algorithm.Portfolio.GetMarginRemaining(target.Symbol, _direction)))
# Remove the Stop Market Order Price, which is matched to each BUY or SHORT.
# If we do not maintain queue discipline, this will eventually provoke a run time exception.
del self.stopPrices[target.Symbol]
else:
# Liquidating position. Will also cancel any StopMarket order.
algorithm.Log("Liquidating {}".format(target.Symbol.Value))
algorithm.Liquidate(target.Symbol)
algorithm.Log("Available cash now ${:.2f}".format(algorithm.Portfolio.Cash))
def OnSecuritiesChanged(self, algorithm, changes):
# Event handler for securities added and removed.
# changes.AddedSecurities:
# We do not need to do anything for AddedSecurities.
# changes.RemovedSecurities:
# Liquidate removed securities.
for removed in changes.RemovedSecurities:
algorithm.Log("Liquidating removed security {}".format(removed.Symbol.Value))
algorithm.Liquidate(removed.Symbol)
algorithm.Log("Available cash now ${:.2f}".format(algorithm.Portfolio.Cash))
#
# Portfolio Construction
#
from clr import AddReference
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm.Framework")
from QuantConnect import Resolution, Extensions
from QuantConnect.Algorithm.Framework.Alphas import *
from QuantConnect.Algorithm.Framework.Portfolio import *
from itertools import groupby
from datetime import datetime, timedelta
from math import floor
from ticker import TickerState
class PortfolioConstruction(PortfolioConstructionModel):
''' PortfolioConstruction - Construct a portfolio based
on Insights generated by the AlphaModel.
Liquidation means a ticker has been delisted, or is otherwise
unavailable.
'''
def __init__(self, config):
self.config = config
self.symbolDataBySymbol = {}
self.accumulatedProfit = 0
def CreateTargets(self, algorithm, insights):
''' CreateTargets for Portfolio Construction
Scan insights looking for tickets we can purchase,
hold, or sell. It is assumed that symbols can only
appear once in any list of insights.
'''
targets = [] # Portfolio targets to be returned for execution
deletedSymbols = [] # We will delete these symbols.
# Account for having possibly already expended some number of slices of our cash.
cashSlicesLeft = self.config.maxFineUniverse - len(self.symbolDataBySymbol)
for insight in insights:
sd = self.symbolDataBySymbol.get(insight.Symbol)
if sd is None:
#
# We do not own this ticker.
#
if cashSlicesLeft == 0:
# No cash / margin slices left; cannot BUY or SHORT this ticker.
continue
if insight.Direction == InsightDirection.Up:
# Buy this ticker.
# Calculate the number of shares to buy.
margin = algorithm.Portfolio.GetMarginRemaining(insight.Symbol, OrderDirection.Buy)
sharePrice = algorithm.Securities[insight.Symbol].Close
shares = floor((margin / cashSlicesLeft) / sharePrice)
# Create target for appropriate number of shares.
algorithm.Log("Buying long ticker {} {} shares at ${:.2f}".format(insight.Symbol.Value, shares, sharePrice))
targets.append(PortfolioTarget(insight.Symbol, shares))
# Record pending ticker purchase.
self.symbolDataBySymbol[insight.Symbol] = SymbolData(self.config,
insight.Symbol, sharePrice, shares, TickerState.BUY)
elif insight.Direction == InsightDirection.Down:
# Short the ticker.
# Calculate the number of shares to sell.
margin = algorithm.Portfolio.GetMarginRemaining(insight.Symbol, OrderDirection.Sell) * self.config.initialMargin
sharePrice = algorithm.Securities[insight.Symbol].Close
shares = floor((margin / cashSlicesLeft) / sharePrice)
# Create target for appropriate number of shares.
algorithm.Log("Short selling ticker {} {} shares at ${:.2f}".format(insight.Symbol.Value, -shares, sharePrice))
targets.append(PortfolioTarget(insight.Symbol, -shares))
# Record pending ticker purchase.
self.symbolDataBySymbol[insight.Symbol] = SymbolData(self.config,
insight.Symbol, sharePrice, -shares, TickerState.SHORT)
else:
#
# We own this ticker.
#
if insight.Direction == InsightDirection.Flat \
or sd.state == TickerState.LIQUIDATE:
# Liquidate the ticker.
# We expect shares to be zero for this insight,
# but in fact we don't care; this might be a delisted stock.
algorithm.Log("Liquidating ticker {} {} shares at ${:.2f}".format(sd.symbol.Value, sd.shares, sd.price))
# Liquidate the position.
targets.append(PortfolioTarget(insight.Symbol, 0))
sd.state = TickerState.LIQUIDATE
deletedSymbols.append(sd)
else:
algorithm.Log("ERROR: Unexpected insight direction or ticker state, should only be Flat or LIQUIDATE, \ninsight direction {}, ticker state {}".format(insight.Direction, sd.state))
deletedSymbols.append(sd)
# Remove the deleted symbols.
for sd in deletedSymbols:
self.symbolDataBySymbol.pop(sd.symbol, None)
# Return the portfolio targets.
return targets
def OnSecuritiesChanged(self, algorithm, changes):
# Event handler for securities added and removed.
# changes.AddedSecurities:
# We do not need to do anything for AddedSecurities.
# changes.RemovedSecurities:
for removed in changes.RemovedSecurities:
sd = self.symbolDataBySymbol.get(removed.Symbol)
if sd is not None:
sd.state = TickerState.LIQUIDATE
class SymbolData:
''' Symbol data for each ticker held in our portfolio.
'''
def __init__(self, config, symbol, price, shares, state):
self.config = config
self.symbol = symbol
self.price = price
self.shares = shares
self.state = state
class StopPrices(dict):
""" StopPrices
Dictionary holding Stop Market prices for equities being traded.
"""
def __init__(self, **kwargs):
# noinspection PyTypeChecker
dict.__init__(self, kwargs)
self.__dict__=self
from datetime import datetime, timedelta
from QuantConnect.Data.UniverseSelection import *
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
class UniverseSelection(FundamentalUniverseSelectionModel):
'''Select a new universe every week on the specified day.'''
def __init__(self, config, filterFineData=True, universeSettings=None):
super().__init__(filterFineData, universeSettings)
self.config = config
self.symbolDataBySymbol = {}
def SelectCoarse(self, algorithm, coarse):
# Only run on the selected day at midnight,
# otherwise, the universe has not changed.
if self.config.selectionDay != algorithm.Time.weekday():
return Universe.Unchanged
algorithm.Debug("SelectCoarse: start " + str(algorithm.Time.strftime("%Y-%m-%d %H:%M:%S")))
# Select equities with prices less than or equal to our standard bet
# divided by the minimum number of shares. Return that result sorted
# by lowest price. Price is the closing price on the previous trading day.
# Equity must have been listed at least minIPOTime days before universe selection time.
# Equity market capitalization must be at least minMarketCap.
# Equity must be on a USA market.
sortedByPrice = sorted([x for x in coarse if x.HasFundamentalData
and x.Symbol.SecurityType == SecurityType.Equity
and x.Symbol.ID.Market == Market.USA
and x.AdjustedPrice >= self.config.minPrice
and x.AdjustedPrice <= (algorithm.Portfolio.Cash / self.config.maxFineUniverse)
and x.DollarVolume >= self.config.minVolume],
key=lambda x: x.Price)
universe = [x.Symbol for x in sortedByPrice]
algorithm.Debug("SelectCoarse: universe length: " + str(len(universe)))
if len(universe) == 0:
return Universe.Unchanged
return universe
def SelectFine(self, algorithm, fine):
algorithm.Debug("SelectFine: start at {}".format(algorithm.Time.strftime("%Y-%m-%d %H:%M:%S")))
# Check all equities
for f in fine:
# Equity IPO must have happened some time ago.
if int((algorithm.Time - f.SecurityReference.IPODate).days) < self.config.minDaysSinceIPO:
continue
# Add the equity if we don't already know about it.
if f.Symbol not in self.symbolDataBySymbol:
self.symbolDataBySymbol[f.Symbol] = SymbolData(self.config, f.Symbol, f.Price)
# Update the equity with new history, and calculate the theoritical profit.
self.symbolDataBySymbol[f.Symbol].Update(algorithm)
# Sort to find the most profitable equities.
sortedByProfit = sorted(self.symbolDataBySymbol.keys(), key=lambda s: self.symbolDataBySymbol[s].profit, reverse=True)
# Trim the universe to length maxFineUniverse.
universe = sortedByProfit[:self.config.maxFineUniverse] if len(sortedByProfit) > self.config.maxFineUniverse else sortedByProfit
algorithm.Debug("SelectFine: universe length {}: ".format(len(universe)))
# for s in universe:
# algorithm.Debug("Symbol: {} total profit: {:.2f}".format(s.Value, self.symbolDataBySymbol[s].profit))
if len(universe) == 0:
return Universe.Unchanged
return universe
class SymbolData:
def __init__(self, config, symbol, price):
self.config = config
self.symbol = symbol
self.price = price
self.history = None
self.profit = 0
def Update(self, algorithm):
# algorithm.Debug("SymbolData.Update: Symbol: {} price: {:.2f}".format(self.symbol, self.price))
# Create a new period of look back history.
self.history = SymbolHistory(self.config, self.symbol, self.price)
self.history.Update(algorithm)
self.profit = self.history.profit
class SymbolHistory:
def __init__(self, config, symbol, price):
self.config = config
self.symbol = symbol
self.price = price
self.date = None
self.history = None
self.profit = 0
def Update(self, algorithm):
self.date = algorithm.Time
# algorithm.Debug("SymbolHistory.Update: Symbol: {} Price: {:.2f} date: {}".format(self.symbol, self.price, self.date))
timeAgo = self.date - timedelta(days=self.config.daysMemory)
self.history = algorithm.History(self.symbol, timeAgo, self.date, Resolution.Daily)
timeProfit = 0
for index, row in self.history.iterrows():
# Calculate estimated potential profit for each day's trading.
dayProfit = ((row["high"] - row["low"]) * (algorithm.Portfolio.Cash / self.config.maxFineUniverse)) / row["close"]
# Accumulate into profit over the entire time period.
timeProfit += dayProfit
self.profit = timeProfit
# algorithm.Debug("SymbolHistory.Update: timeProfit: {:.2f}".format(self.profit))