| Overall Statistics |
|
Total Trades 207 Average Win 0.20% Average Loss -0.10% Compounding Annual Return 25.439% Drawdown 5.600% Expectancy -0.507 Net Profit 13.931% Sharpe Ratio 1.616 Sortino Ratio 1.56 Probabilistic Sharpe Ratio 74.296% Loss Rate 83% Win Rate 17% Profit-Loss Ratio 1.96 Alpha 0 Beta 0 Annual Standard Deviation 0.092 Annual Variance 0.008 Information Ratio 1.901 Tracking Error 0.092 Treynor Ratio 0 Total Fees $247.60 Estimated Strategy Capacity $23000000.00 Lowest Capacity Asset NWSVV VHJF6S7EZRL1 Portfolio Turnover 0.76% |
#region imports
from AlgorithmImports import *
#endregion
class ValueAlphaModel(AlphaModel):
securities = []
month = -1
def __init__(self, portfolio_size):
self.portfolio_size = portfolio_size
def Update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]:
# Only rebalance when there is QuoteBar data
if data.QuoteBars.Count == 0:
return []
# Rebalance monthly
if self.month == data.Time.month:
return []
self.month = data.Time.month
# Create insights to long the securities with the lowest price-to-book ratio
insights = []
expiry = self.securities[0].Exchange.Hours.GetNextMarketOpen(Expiry.EndOfMonth(data.Time), extendedMarketHours=False)
tradable_securities = [security for security in self.securities if security.Symbol in data.QuoteBars and security.Price > 0 and security.Fundamentals]
for security in sorted(tradable_securities, key=lambda security: security.Fundamentals.ValuationRatios.PBRatio)[:self.portfolio_size]:
insights.append(Insight.Price(security.Symbol, expiry, InsightDirection.Up))
return insights
def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
for security in changes.RemovedSecurities:
if security in self.securities:
self.securities.remove(security)
self.securities.extend(changes.AddedSecurities)# region imports
from AlgorithmImports import *
#from universe import LowPBRatioUniverseSelectionModel
#from alpha import ValueAlphaModel
#from portfolio import EqualWeightingRebalanceOnInsightsPortfolioConstructionModel
# endregion
class BuffetBargainHunterAlgorithm(QCAlgorithm):
undesired_symbols_from_previous_deployment = []
checked_symbols_from_previous_deployment = False
symbols_and_windows = []
symbol_data_by_symbol = {}
symbols_buy_fill_prices = {}
stopPercentage = 0.05
profPercentage = 0.3
liquidationQueue = []
def Initialize(self):
self.SetStartDate(2019, 6, 1)
self.SetEndDate(2019, 12, 28)
self.SetCash(1_000_000)
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0
self.SetSecurityInitializer(BrokerageModelSecurityInitializer(self.BrokerageModel, FuncSecuritySeeder(self.GetLastKnownPrices)))
self.UniverseSettings.Asynchronous = True
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.FillForward = True
self.symbol_data_by_symbol = {}
self.AddUniverse(self.Universe.ETF("SPY", Market.USA, self.UniverseSettings, self.selectionMethod))
#self.AddAlpha(ValueAlphaModel(self.GetParameter("portfolio_size", 100)))
#self.SetPortfolioConstruction(EqualWeightingRebalanceOnInsightsPortfolioConstructionModel(self))
#self.AddRiskManagement(NullRiskManagementModel())
#self.AddEquity("ADBE", Resolution.Daily)
self.SetExecution(ImmediateExecutionModel())
self.SetWarmUp(timedelta(30))
def selectionMethod(self, fundamental: List[Fundamental]) -> List[Symbol]:
universe_symbols = []
lisfun = list(fundamental)
for f in fundamental:
f_symbol = f.Symbol.Value
f_index = lisfun.index(f)
f_datetime = str(f.EndTime.month)+"/"+str(f.EndTime.day)+"/"+str(f.EndTime.year)
universe_symbols.append(f.Symbol)
if f.Symbol not in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[f.Symbol] = SymbolData(f.Symbol)
tickerHistory = self.History(f.Symbol, 1, Resolution.Daily)
try:
lastClosePrice = tickerHistory.iloc[0][0]
except IndexError as ir:
self.Log("INDEX ERROR in selection area when looking at tickerPrice " + str(f.EndTime) + ": " + str(ir) + " " + f.Symbol.ToString())
continue
try:
possible_log_message = self.symbol_data_by_symbol[f.Symbol].Update(f.EndTime, tickerHistory.iloc[0][0])
if possible_log_message != "":
# pass
self.Log(possible_log_message)
except Exception as e:
self.Log("EXCEPTION in selection area on " + str(f.EndTime) + ": " + e.Message + " " + f.Symbol.Value)
# self.Log("Runtime Error: single positional indexer is out-of-bounds. Ticker:" + f.Symbol.ToString())
except RuntimeError as r:
self.Log("RUNTIME ERROR in selection area on " + str(f.EndTime) + ": " + r.Message + " " + f.Symbol.Value)
except IndexError as ir:
self.Log("INDEX ERROR in selection area on " + str(f.EndTime) + ": " + str(ir) + " " + f.Symbol.Value)
except:
self.Log("SOME ERROR in selection area that wasn't Runtime Error or Exception. On " + str(sys.exc_info()[0]) + " " + str(f.EndTime) + " - " + f.Symbol.Value)
# Remove indicators for assets that are no longer in the ETF
# symbols_to_remove = [symbol for symbol in self.symbol_data_by_symbol.keys() if symbol not in universe_symbols]
# for symbol in symbols_to_remove:
# self.symbol_data_by_symbol.pop(symbol)
# self.Log("Symbol removed from ETF. " + symbol.Value)
# Select a subset of the ETF constituents based on the indicator value
universe = [symbol for symbol, symbol_data in self.symbol_data_by_symbol.items() if symbol_data.lowerLowsStraight]
# Return the selected assets
return universe
def OnData(self, slice) -> None:
self.holdingsSymbolsList = []
if len(self.liquidationQueue) > 0:
liqOrders = list(self.liquidationQueue)
for orderId in liqOrders:
liqTransaction = self.Transactions.GetOrderById(orderId)
if liqTransaction.Status == OrderStatus.Filled:
securitySymbol = liqTransaction.Symbol
securityTickerWatch = liqTransaction.Symbol.Value
if securitySymbol in self.symbol_data_by_symbol:
sec_record = self.symbol_data_by_symbol[securitySymbol]
else:
self.Log(securitySymbol.Value + " not found in self.symbol_data_by_symbol when handling liquidation stuff.")
try:
sec_record.resetEntryPrice()
except:
self.Log("Couldn't reset entry price after lquidation. On " + str(slice.Time) + str(sys.exc_info()[0]) + " " + " - " + securitySymbol.Value)
continue
self.liquidationQueue.remove(orderId)
if len(self.symbols_buy_fill_prices) > 0:
keysList = list(self.symbols_buy_fill_prices.keys())
for key in keysList:
securitySelected = self.symbol_data_by_symbol[key]
securityTicker = securitySelected.symbol.Value
symbolEntries = self.symbols_buy_fill_prices
symbolBuyEntry = self.symbols_buy_fill_prices[key]
avgEntryPrice = self.symbols_buy_fill_prices[key].AverageFillPrice
if self.symbols_buy_fill_prices[key].QuantityFilled != 0.0:
self.symbol_data_by_symbol[key].addEntryPrice(self.symbols_buy_fill_prices[key].AverageFillPrice)
self.Log("")
del self.symbols_buy_fill_prices[key]
if not self.IsWarmingUp:
for security_holding in self.Portfolio.Values:
tickerHistory = self.History(security_holding.Symbol, 1, Resolution.Daily)
tickerSymbolData = None
securityTickerWatch = security_holding.Symbol.Value
if security_holding.Symbol in self.symbol_data_by_symbol:
tickerSymbolData = self.symbol_data_by_symbol[security_holding.Symbol]
else:
self.Log(security_holding.Symbol.Value + " not found in self.symbol_data_by_symbol.")
try:
stopLevel1 = 0.85
highLevelHold1 = 0.0
hilodiff = 0.0
sixMHigh = 0.0
sixMLow = 0.0
if tickerSymbolData != None:
stopLevel1 = tickerSymbolData.stopLevel
highLevelHold1 = tickerSymbolData.highCloseDuringHold
hilodiff = tickerSymbolData.highLowDifference*100
sixMHigh = tickerSymbolData.sixMonthHighAtEntry
sixMLow = tickerSymbolData.sixMonthLowAtEntry
try:
lastClosePrice = tickerHistory.iloc[0][0]
except IndexError as ir:
self.Log("INDEX ERROR in onData when looking at tickerHistory " + str(slice.Time) + ": " + str(ir) + " " + security_holding.Symbol.Value)
continue
if lastClosePrice < security_holding.AveragePrice * stopLevel1 and security_holding.AveragePrice != 0 and security_holding.Quantity > 0:
liq_order = self.Liquidate(security_holding.Symbol, "Sold cuz loss. | Price:|"+str(security_holding.Price)+"| AvgPrice:|"+str(security_holding.AveragePrice)+"| StopLevel:|" + str(stopLevel1) + "| HighDuringHold:|" + str(highLevelHold1) + "| DropSize:|" + str(hilodiff) + "| 6M High: |" + str(sixMHigh) + "| 6M Low: |"+str(sixMLow))
self.liquidationQueue.append(liq_order.Id)
# if isinstance(liq_order, list):
# # if len(liq_order) != 0:
# #if all(one_order.Status == OrderStatus.Filled for one_order in liq_order):
# tickerSymbolData.resetEntryPrice()
# else:
# if liq_order.Status == OrderStatus.Filled:
# tickerSymbolData.resetEntryPrice()
elif lastClosePrice > security_holding.AveragePrice * (1.6) and security_holding.AveragePrice != 0 and security_holding.Quantity > 0:
liq_order = self.Liquidate(security_holding.Symbol, "Sold cuz prof. | Price:|"+str(security_holding.Price)+"| AvgPrice:|"+str(security_holding.AveragePrice)+"| StopLevel:|" + str(stopLevel1) + " ||| DropSize:|" + str(hilodiff) + "| 6M High: |" + str(sixMHigh) + "| 6M Low: |"+str(sixMLow))
self.liquidationQueue.append(liq_order.Id)
# if isinstance(liq_order, list):
# tickerSymbolData.resetEntryPrice()
# else:
# if liq_order.Status == OrderStatus.Filled:
# tickerSymbolData.resetEntryPrice()
except Exception as exc:
self.Log("EXCEPTION in onData section on " + str(slice.Time) + ": " + exc.Message + " - " + security_holding.Symbol.Value)
except RuntimeError as rte:
self.Log("RUNTIME ERROR in onData section on " + str(slice.Time) + ": " + rte.Message + " - " + security_holding.Symbol.Value)
except AttributeError as aer:
errormsg_fu = str(aer)
self.Log("ATTRIBUTE ERROR in onData section on " + str(slice.Time) + ": " + str(aer) + " - " + security_holding.Symbol.Value)
continue
except IndexError as ier:
errormsg_fu2 = str(ier)
self.Log("INDEX ERROR in onData section on " + str(slice.Time) + ": " + str(ier) + " " + security_holding.Symbol.Value)
continue
except NameError as ner:
errormsg_fu3 = str(ner)
self.Log("NAME ERROR in onData section on " + str(slice.Time) + ": " + str(ner) + " " + security_holding.Symbol.Value)
continue
except:
self.Log("SOME ERROR in onData section that wasn't Runtime Error or Exception. On " + str(slice.Time) + str(sys.exc_info()[0]) + " " + " - " + security_holding.Symbol.Value)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
if not self.IsWarmingUp:
for security in changes.AddedSecurities:
try:
if security not in self.Portfolio.Values or self.Portfolio.Values[security].Quantity == 0:
tickerHistory = self.History(security.Symbol, 1, Resolution.Daily)
closingPrice = tickerHistory.iloc[0][0]
self.symbols_buy_fill_prices[security.Symbol] = self.MarketOrder(security.Symbol, int(8000/closingPrice), tag="")
except Exception as e:
self.Log("EXCEPTION in adding securities: " + e.Message + " - " + security.Symbol.Value)
except RuntimeError as r:
self.Log("RUNTIME ERROR in adding securities on: " + r.Message + " - " + security.Symbol.Value)
except:
self.Log("SOME ERROR iin adding securities that wasn't Runtime Error or Exception - " + str(sys.exc_info()[0]) + " " + security.Symbol.Value)
for security in changes.RemovedSecurities:
pass
class SymbolData(object):
def __init__(self, symbol):
self.symbol = symbol
self.priceWindow = RollingWindow[float](7)
self.sixMonthsWindow = RollingWindow[float](126)
self.lowerLowsStraight = False
self.current_price = 0.0
#self.monthCandleWindow = RollingWindow[TradeBar](6)
self.avgEntryPrice = 0.0
self.stopLevel = 0.85
self.highCloseDuringHold = 0.0
self.highLowDifference = 0.0
self.sixMonthHigh = 0.0
self.sixMonthLow = 0.0
self.sixMonthHighAtEntry = 0.0
self.sixMonthLowAtEntry = 0.0
def Update(self, time, price):
self.current_price = price
self.sixMonthsWindow.Add(price)
if not self.priceWindow.IsReady:
self.lowerLowsStraight = False
self.priceWindow.Add(price)
return ""
log_message = ""
self.priceWindow.Add(price)
sortedSixMonths = list(self.priceWindow)[::-1]
self.sixMonthHigh = sortedSixMonths[0]
self.sixMonthLow = sortedSixMonths[len(sortedSixMonths)-1]
self.priceList = list(self.priceWindow)[::-1]
self.orderedPriceList = list(self.priceWindow)[::-1]
self.orderedPriceList.sort(reverse=True)
if self.priceList == self.orderedPriceList:
self.lowerLowsStraight = True
self.highLowDifference = (self.orderedPriceList[0]-self.orderedPriceList[6])/self.orderedPriceList[6]
if self.avgEntryPrice != 0.0:
if price >= 1.45 * self.avgEntryPrice and self.stopLevel < 1.12:
self.stopLevel = 1.12
if price >= 1.3 * self.avgEntryPrice and self.stopLevel < 1.06:
self.stopLevel = 1.06
if price >= 1.15 * self.avgEntryPrice and self.stopLevel < 1.0:
self.stopLevel = 1.0
if price > self.highCloseDuringHold:
self.highCloseDuringHold = price
# if self.symbol.Value == "TSLA":
# if time.day == 24 and time.month == 12:
# log_message += "TSLA Price " + str(time.year) + ": "
# for i in range(7):
# dps1 = str(self.priceList[i])
# dps2 = str(self.orderedPriceList[i])
# log_message += " " + dps1 + "/" + dps2 + " "
# log_message += " Flag says: " + str(self.lowerLowsStraight)
return log_message
def addEntryPrice(self, price):
self.avgEntryPrice = price
self.sixMonthHighAtEntry = self.sixMonthHigh
self.sixMonthLowAtEntry = self.sixMonthLow
def resetEntryPrice(self):
self.avgEntryPrice = 0.0
self.stopLevel = 0.85
self.highCloseDuringHold = 0.0
self.highLowDifference = 0.0#region imports
from AlgorithmImports import *
#endregion
class EqualWeightingRebalanceOnInsightsPortfolioConstructionModel(EqualWeightingPortfolioConstructionModel):
def __init__(self, algorithm):
super().__init__()
self.algorithm = algorithm
self.new_insights = False
def IsRebalanceDue(self, insights: List[Insight], algorithmUtc: datetime) -> bool:
if not self.new_insights:
self.new_insights = len(insights) > 0
is_rebalance_due = self.new_insights and not self.algorithm.IsWarmingUp and self.algorithm.CurrentSlice.QuoteBars.Count > 0
if is_rebalance_due:
self.new_insights = False
return is_rebalance_due