| Overall Statistics |
|
Total Trades 38 Average Win 0.20% Average Loss -0.61% Compounding Annual Return -89.064% Drawdown 5.000% Expectancy -0.216 Net Profit -2.986% Sharpe Ratio -10.971 Probabilistic Sharpe Ratio 3.506% Loss Rate 41% Win Rate 59% Profit-Loss Ratio 0.33 Alpha -0.946 Beta -0.071 Annual Standard Deviation 0.082 Annual Variance 0.007 Information Ratio -1.095 Tracking Error 0.158 Treynor Ratio 12.538 Total Fees $297.23 Estimated Strategy Capacity $47000.00 Lowest Capacity Asset COSM XWEMZPZHZXB9 |
#region imports
from AlgorithmImports import *
#endregion
# https://quantpedia.com/Screener/Details/13
# The investment universe consists of the 100 biggest companies by market capitalization.
# The investor goes long on the 10 stocks with the lowest performance in the previous month
# and goes short on the 10 stocks with the greatest performance from the previous month.
# The portfolio is rebalanced weekly.
from QuantConnect.Data.UniverseSelection import *
import math
import numpy as np
import pandas as pd
import scipy as sp
SL = 0.04; TP = 0.02;
class ShortTermReversalAlgorithm(QCAlgorithm):
# Order ticket for our stop order, Datetime when stop order was last hit
def Initialize(self):
self.SetStartDate(2022, 4, 4) # Set Start Date
self.SetEndDate(2022, 4, 8) # Set End Date
self.SetCash(100000) # Set Strategy Cash
self.lookback = 20
self.UniverseSettings.Resolution = Resolution.Minute
self.num_screener = 20
self.num_trade = 10
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
self.symbolDataDict = {}
self.AddEquity("SPY", Resolution.Minute)
self.Schedule.On(self.DateRules.Every(DayOfWeek.Monday, DayOfWeek.Monday), self.TimeRules.AfterMarketOpen("SPY", 0), self.Rebalance)
self.weekly_rebalance = True
self.filtered_coarse = []
self.filtered_fine = []
self.long_stocks = []
self.price = {}
self.fillTime = {}
def CoarseSelectionFunction(self, coarse):
if self.weekly_rebalance:
# Drop stocks with no fundamental data. Keep small cap stocks.
filtered_set = [x for x in coarse if (x.HasFundamentalData) and (float(x.Price) > 1.5)]
self.filtered_coarse = [x.Symbol for x in filtered_set]
# Sort stocks by highest dollar volume
sorted_set = sorted(filtered_set, key = lambda x: x.DollarVolume, reverse=True)
self.filtered_fine = [x.Symbol for x in sorted_set]
return self.filtered_coarse
else:
return self.filtered_fine
def FineSelectionFunction(self, fine):
if self.weekly_rebalance:
filtered_set = [x for x in fine if x.EarningReports.BasicEPS.TwelveMonths > 0
and x.ValuationRatios.PERatio > 0
and x.EarningReports.BasicAverageShares.ThreeMonths > 0
and x.MarketCap < 2e9]
# Take remaining top 500 by highest dollar volume
self.filtered_fine = [x.Symbol for x in filtered_set[:500]]
return self.filtered_fine
else:
return self.filtered_fine
def OnData(self, data):
if self.Time.hour < 10 or self.Time.minute != 1: return
if self.Time.hour >= 16 or self.Time.minute != 1: return
for symbol, symbolData in self.symbolDataDict.items():
# update the indicator value for newly added securities
if symbol not in self.addedSymbols:
symbolData.ROC.Update(IndicatorDataPoint(symbol, self.Time, self.Securities[symbol].Close))
if self.weekly_rebalance and self.filtered_fine:
self.addedSymbols = []
# sorted the stocks by the monthly return (RateOfReturn)
readyROC = {key: value for key, value in self.symbolDataDict.items() if value.ROC.IsReady}
sorted_symbolData = sorted(readyROC, key=lambda x: readyROC[x].ROC.Current.Value)
short_stocks = sorted_symbolData[-self.num_trade:]
self.long_stocks = sorted_symbolData[:self.num_trade]
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for i in invested:
if i not in short_stocks+self.long_stocks:
self.Liquidate(i)
# for short in short_stocks:
# quantity = self.CalculateOrderQuantity(short, 0.5/self.num_trade)
# self.MarketOrder(short, -quantity)
# #self.stopMarketTicket[short] =
# self.StopMarketOrder(short, quantity, 1.02 * self.Securities[short].Close)
# self.StopMarketOrder(short, quantity, 0.96 * self.Securities[short].Close)
#
#self.highestPrice[short] = 0
for long in self.long_stocks:
self.SetHoldings(long, 1.0/self.num_trade)
self.price[long] = data[long].Price
self.fillTime[long] = datetime.min
self.weekly_rebalance = False
else:
for long in self.long_stocks:
if self.Portfolio[long].Invested:
if self.price[long] > 0:
curr_price = data[long].Price
if curr_price >= self.price[long]*(1 + TP):
self.Liquidate(long, "Take Profit")
self.fillTime[long] = self.Time
self.price[long] = 0
elif curr_price < self.price[long]*(1 - SL):
self.Liquidate(long, "Stop Loss")
self.fillTime[long] = self.Time
self.price[long] = 0
else:
if (self.Time - self.fillTime[long]).days > 1:
self.SetHoldings(long, 1.0/self.num_trade)
self.price[long] = data[long].Price
# HighestPrice
# for stock in long_stocks:
# if self.Securities[stock].Close > self.highestPrice[stock]:
# #2. Save the new high to highestSPYPrice; then update the stop price to 90% of highestSPYPrice
# self.highestPrice[stock] = self.Securities[stock].Close
# updateFields = UpdateOrderFields()
# updateFields.StopPrice = self.Securities[stock].Close * 0.98
# self.Debug(updateFields.StopPrice)
# self.stopMarketTicket[stock].Update(updateFields)
# #3. Print the new stop price with Debug()
# # self.Debug(updateFields.StopPrice)
# for stock in short_stocks:
# if self.Securities[stock].Close > self.highestPrice[stock]:
# #2. Save the new high to highestSPYPrice; then update the stop price to 90% of highestSPYPrice
# self.highestPrice[stock] = self.Securities[stock].Close
# updateFields = UpdateOrderFields()
# updateFields.StopPrice = self.Securities[stock].Close * 1.05
# self.Debug(updateFields.StopPrice)
# self.stopMarketTicket[stock].Update(updateFields)
#3. Print the new stop price with Debug()
# self.Debug(updateFields.StopPrice)
def Rebalance(self):
self.weekly_rebalance = True
def OnSecuritiesChanged(self, changes):
for removed in changes.RemovedSecurities:
symbolData = self.symbolDataDict.pop(removed.Symbol, None)
# warm up the indicator with history price for newly added securities
self.addedSymbols = [x.Symbol for x in changes.AddedSecurities if x.Symbol.Value != "SPY"]
history = self.History(self.addedSymbols, self.lookback+10, Resolution.Daily)
for symbol in self.addedSymbols:
if symbol not in self.symbolDataDict.keys():
symbolData = SymbolData(symbol, self.lookback)
self.symbolDataDict[symbol] = symbolData
if str(symbol) in history.index:
symbolData.WarmUpIndicator(history.loc[str(symbol)])
class SymbolData:
'''Contains data specific to a symbol required by this model'''
def __init__(self, symbol, lookback):
self.symbol = symbol
self.ROC = RateOfChange(lookback)
def WarmUpIndicator(self, history):
# warm up the RateOfChange indicator with the history request
for tuple in history.itertuples():
item = IndicatorDataPoint(self.symbol, tuple.Index, float(tuple.close))
self.ROC.Update(item)
#region imports
from AlgorithmImports import *
#endregion
# https://quantpedia.com/Screener/Details/13
# The investment universe consists of the 100 biggest companies by market capitalization.
# The investor goes long on the 10 stocks with the lowest performance in the previous month
# and goes short on the 10 stocks with the greatest performance from the previous month.
# The portfolio is rebalanced weekly.
from QuantConnect.Data.UniverseSelection import *
import math
import numpy as np
import pandas as pd
import scipy as sp
class ShortTermReversalAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2020, 1, 1) # Set Start Date
self.SetEndDate(2022, 10, 2) # Set End Date
self.SetCash(5000000) # Set Strategy Cash
self.lookback = 20
self.UniverseSettings.Resolution = Resolution.Hour
self.num_screener = 20
self.num_trade = 10
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
self.symbolDataDict = {}
self.AddEquity("SPY", Resolution.Hour)
self.Schedule.On(self.DateRules.Every(DayOfWeek.Monday, DayOfWeek.Monday), self.TimeRules.AfterMarketOpen("SPY", 0), self.Rebalance)
self.weekly_rebalance = True
self.filtered_coarse = []
self.filtered_fine = []
def CoarseSelectionFunction(self, coarse):
if self.weekly_rebalance:
# drop stocks which have no fundamental data or have too low prices
selected = [x for x in coarse if (x.HasFundamentalData) and (float(x.Price) > 5)]
# rank the stocks by dollar volume
filtered = sorted(selected, key=lambda x: x.DollarVolume, reverse=True)
self.filtered_coarse = [ x.Symbol for x in filtered[:1000]]
return self.filtered_coarse
else:
return self.filtered_fine
def FineSelectionFunction(self, fine):
if self.weekly_rebalance:
filtered_fine = [x for x in fine if x.EarningReports.BasicEPS.TwelveMonths > 0
and x.ValuationRatios.PERatio > 0
and x.EarningReports.BasicAverageShares.ThreeMonths > 0
and x.EarningReports.BasicAverageShares.ThreeMonths > 0]
# filter 100 stocks with the top market cap
top = sorted(filtered_fine, key = lambda x: x.EarningReports.BasicAverageShares.ThreeMonths * (x.EarningReports.BasicEPS.TwelveMonths*x.ValuationRatios.PERatio), reverse=True)[:100]
self.filtered_fine = [i.Symbol for i in top]
return self.filtered_fine
else:
return self.filtered_fine
def OnData(self, data):
for symbol, symbolData in self.symbolDataDict.items():
# update the indicator value for newly added securities
if symbol not in self.addedSymbols:
symbolData.ROC.Update(IndicatorDataPoint(symbol, self.Time, self.Securities[symbol].Close))
if self.weekly_rebalance and self.filtered_fine:
self.addedSymbols = []
# sorted the stocks by the monthly return (RateOfReturn)
readyROC = {key: value for key, value in self.symbolDataDict.items() if value.ROC.IsReady}
sorted_symbolData = sorted(readyROC, key=lambda x: readyROC[x].ROC.Current.Value)
short_stocks = sorted_symbolData[-self.num_trade:]
long_stocks = sorted_symbolData[:self.num_trade]
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for i in invested:
if i not in short_stocks+long_stocks:
self.Liquidate(i)
for short in short_stocks:
self.SetHoldings(short, -0.5/self.num_trade)
for long in long_stocks:
self.SetHoldings(long, 0.5/self.num_trade)
self.weekly_rebalance = False
def Rebalance(self):
self.weekly_rebalance = True
def OnSecuritiesChanged(self, changes):
for removed in changes.RemovedSecurities:
symbolData = self.symbolDataDict.pop(removed.Symbol, None)
# warm up the indicator with history price for newly added securities
self.addedSymbols = [x.Symbol for x in changes.AddedSecurities if x.Symbol.Value != "SPY"]
history = self.History(self.addedSymbols, self.lookback+10, Resolution.Daily)
for symbol in self.addedSymbols:
if symbol not in self.symbolDataDict.keys():
symbolData = SymbolData(symbol, self.lookback)
self.symbolDataDict[symbol] = symbolData
if str(symbol) in history.index:
symbolData.WarmUpIndicator(history.loc[str(symbol)])
class SymbolData:
'''Contains data specific to a symbol required by this model'''
def __init__(self, symbol, lookback):
self.symbol = symbol
self.ROC = RateOfChange(lookback)
def WarmUpIndicator(self, history):
# warm up the RateOfChange indicator with the history request
for tuple in history.itertuples():
item = IndicatorDataPoint(self.symbol, tuple.Index, float(tuple.close))
self.ROC.Update(item)