| Overall Statistics |
|
Total Trades 2185 Average Win 0.35% Average Loss -0.22% Compounding Annual Return 62.795% Drawdown 9.600% Expectancy 0.319 Net Profit 77.318% Sharpe Ratio 2.284 Probabilistic Sharpe Ratio 86.551% Loss Rate 48% Win Rate 52% Profit-Loss Ratio 1.54 Alpha 0.512 Beta 0.135 Annual Standard Deviation 0.235 Annual Variance 0.055 Information Ratio 1.026 Tracking Error 0.339 Treynor Ratio 3.967 Total Fees $2273.22 Estimated Strategy Capacity $2300000.00 |
from sklearn.linear_model import LinearRegression
import numpy as np
import pandas as pd
class CalibratedHorizontalThrustAssembly(QCAlgorithm):
def Initialize(self):
self.SetTimeZone("America/New_York")
self.SetStartDate(2020, 1, 1)
# self.SetEndDate(2020,2,5)
self.SetCash(100000)
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
self.AddEquity("SPY", Resolution.Minute)
self.SetBenchmark("SPY")
# self.AddAlpha( EmaCrossAlphaModel() )
self.UniverseSettings.Resolution = Resolution.Minute
self.AddUniverse(self.CoarseSelectionFilter, self.SelectFine)
self.Schedule.On(self.DateRules.Every(DayOfWeek.Wednesday), self.TimeRules.At(12, 15), self.handle_trade)
self.betas = []
self.SetPortfolioConstruction( InsightWeightingPortfolioConstructionModel() )
self.SetExecution( ImmediateExecutionModel() )
self.SetExecution( StandardDeviationExecutionModel( deviations = 1, period = 60, resolution=Resolution.Minute ) )
# self.SetRiskManagement(TrailingStopRiskManagementModel())
def CoarseSelectionFilter(self, coarse):
sortedByDollarVolume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
filteredByPrice = [x.Symbol for x in sortedByDollarVolume if x.Price > 10 and x.HasFundamentalData == True]
return filteredByPrice[:10]
def beta(self, asset_return, market_return):
asset_return = np.array(asset_return, dtype=np.float32)
market_return = np.array(market_return, dtype=np.float32)
corr = np.cov(asset_return, market_return)[0][1]/(np.std(asset_return)*np.std(market_return))
return (np.cov(asset_return, market_return)[0][1]/np.var(market_return))*abs(1/corr)
def merge_DataFrame_to_timeindex(self, array_1, array_2):
array_1["time"] = pd.to_datetime(array_1.index, unit='s').to_frame()
array_2["time.two"] = pd.to_datetime(array_2.index, unit='s').to_frame()
data = pd.concat([array_1, array_2], axis=1).dropna()
array_1 = data.iloc[:, 0]
array_2 = data.iloc[:, 1]
return array_1, array_2
def SelectFine(self, fine):
beta_values = {}
for x in fine:
ticker = x.Symbol
market = self.History(["SPY"], 200, Resolution.Daily)
market = market.loc[("SPY"), "close"]
df_multi = self.History(ticker, 200, Resolution.Daily)
df_single = df_multi.loc[ticker, "close"]
df_single, market = self.merge_DataFrame_to_timeindex(df_single,market)
beta_values[x] = self.beta(df_single, market)
sorted_by_beta = sorted(beta_values, key = lambda x: beta_values[x])
sorted_by_beta = [i.Symbol for i in sorted_by_beta]
window = int(self.GetParameter("window"))
self.betas = sorted_by_beta
return sorted_by_beta
def OnSecuritiesChanged(self, changes):
self.changes = changes
for security in changes.AddedSecurities:
self.AddEquity(security.Symbol)
def handle_trade(self):
weight_insight = 0.20
insights_long = []
for security in self.betas[0:4]:
insights_long.append(Insight.Price(security, timedelta(5), InsightDirection.Down, None, None, None, weight_insight))
self.EmitInsights(Insight.Group(insights_long))
insights_short = []
for security in self.betas[-5:]:
insights_short.append(Insight.Price(security, timedelta(5), InsightDirection.Up, None, None, None, weight_insight))
self.EmitInsights(Insight.Group(insights_short))
def handle_trade_sell(self):
invested = []
for security in self.betas:
if self.Portfolio[security].Invested == False:
invested.append(security)
portfolio_invested = [x.Symbol.Value for x in self.Portfolio.Values if x.Invested]
deinvest_list = np.setdiff1d(invested, portfolio_invested)
insights_sell = []
if not len(deinvest_list) == 0:
for deinvest in deinvest_list:
self.Debug(deinvest)
self.Liquidate(str(deinvest))
#insights_sell.append(Insight.Price(str(deinvest), timedelta(3), InsightDirection.Flat))
#self.EmitInsights(Insight.Group(insights_sell))
def OnData(self, data):
passfrom sklearn.linear_model import LinearRegression
import numpy as np
import pandas as pd
class CalibratedHorizontalThrustAssembly(QCAlgorithm):
def Initialize(self):
self.SetTimeZone("America/New_York")
self.SetStartDate(2017, 1, 1)
# self.SetEndDate(2020,2,5)
self.SetCash(100000)
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
self.AddEquity("SPY", Resolution.Hour)
self.UniverseSettings.Resolution = Resolution.Hour
self.AddUniverse(self.CoarseSelectionFilter, self.SelectFine)
self.Schedule.On(self.DateRules.Every(DayOfWeek.Wednesday), self.TimeRules.At(10, 0), self.handle_trade)
self.betas = []
self.wait = True
# self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.02))
def CoarseSelectionFilter(self, coarse):
sortedByDollarVolume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
filteredByPrice = [x.Symbol for x in sortedByDollarVolume if x.Price > 10 and x.HasFundamentalData == True]
return filteredByPrice[:40]
def beta(self, asset_return, market_return):
asset_return = np.array(asset_return, dtype=np.float32)
market_return = np.array(market_return, dtype=np.float32)
return np.cov(asset_return, market_return)[0][1]/np.var(market_return)
def merge_DataFrame_to_timeindex(self, array_1, array_2):
array_1["time"] = pd.to_datetime(array_1.index, unit='s').to_frame()
array_2["time.two"] = pd.to_datetime(array_2.index, unit='s').to_frame()
data = pd.concat([array_1, array_2], axis=1).dropna()
array_1 = data.iloc[:, 0]
array_2 = data.iloc[:, 1]
return array_1, array_2
def SelectFine(self, fine):
beta_values = {}
for x in fine:
ticker = x.Symbol
market = self.History(["SPY"], 20, Resolution.Daily)
market = market.loc[("SPY"), "close"]
df_multi = self.History(ticker, 20, Resolution.Daily)
df_single = df_multi.loc[ticker, "close"]
df_single, market = self.merge_DataFrame_to_timeindex(df_single,market)
beta_values[x] = self.beta(df_single, market)
sorted_by_beta = sorted(beta_values, key = lambda x: beta_values[x])
sorted_by_beta = [i.Symbol for i in sorted_by_beta]
self.betas = sorted_by_beta[-20:]
return sorted_by_beta[-20:]
def OnSecuritiesChanged(self, changes):
self.changes = changes
for security in changes.AddedSecurities:
self.AddEquity(security.Symbol)
def handle_trade(self):
invested = []
for security in self.betas:
if self.Portfolio[security].Invested == False:
weight = 1 / 20
self.SetHoldings(security, weight)
invested.append(security)
portfolio_invested = [x.Symbol.Value for x in self.Portfolio.Values if x.Invested]
deinvest_list = np.setdiff1d(invested, portfolio_invested)
if self.wait == False:
if not len(deinvest_list) == 0:
for deinvest in deinvest_list:
self.Liquidate(str(deinvest))
self.wait = False
def OnData(self, data):
pass