| Overall Statistics |
|
Total Trades 76 Average Win 94.70% Average Loss -4.49% Compounding Annual Return 18250.522% Drawdown 54.100% Expectancy 8.208 Net Profit 9015.099% Sharpe Ratio 35189.462 Probabilistic Sharpe Ratio 83.773% Loss Rate 58% Win Rate 42% Profit-Loss Ratio 21.10 Alpha 320158.695 Beta -1.171 Annual Standard Deviation 9.098 Annual Variance 82.776 Information Ratio 35115.425 Tracking Error 9.117 Treynor Ratio -273325.367 Total Fees $8880.55 |
from System.Collections.Generic import List
from QuantConnect.Data.UniverseSelection import *
import operator
from math import ceil,floor
from scipy import stats
import numpy as np
from datetime import timedelta
class Test(QCAlgorithm):
def Initialize(self):
''' Backtesting Parameters '''
self.SetStartDate(2020, 1, 1)
# self.SetEndDate(2014, 1, 1)
self.SetCash(30000)
''' Universe Settings '''
self.benchmark = Symbol.Create("SPY", SecurityType.Equity, Market.USA)
self.UniverseSettings.Resolution = Resolution.Minute
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
''' Minor Parameters '''
self.month = -1
self.leverage = 2
self.leniency = .95
''' Parameters with meaning '''
self.volumeMin = 1000000 # Ensures we are looking at liquid stocks
self.piotroskiMin = 5 # The higher the better (6, 7, 8, 9) leaves us with the top 40% best quality companies
self.minuteBeforeEOD = 60 # Ensure we have enough time to sell before the end of the day
self.priceMin = 0 # Ensures we do not get the stocks with extremely high risk (Penny stocks)
self.sectorHistory = 21 # Performance last iteration/month
self.stockHistory = 21 # Performance last iteration/month
''' Parameters without meaning '''
self.monthlyStocks = 6
''' Schedule Settings '''
self.AddEquity("SPY", Resolution.Minute)
self.SetBenchmark("SPY")
self.Schedule.On(self.DateRules.EveryDay("SPY"), self.TimeRules.AfterMarketOpen("SPY"), Action(self.Daily))
self.Schedule.On(self.DateRules.MonthEnd("SPY"), self.TimeRules.BeforeMarketClose("SPY", self.minuteBeforeEOD), self.Liquidate)
''' Data Structures '''
self.symbols, self.topSectors,self.sectorPerformances = [], [], {}
''' Sectors '''
self.sectorStocks = {"XLV": MorningstarSectorCode.Healthcare,
"XLK": MorningstarSectorCode.Technology,
"XLI": MorningstarSectorCode.Industrials,
"XLU": MorningstarSectorCode.Utilities,
"XLF": MorningstarSectorCode.FinancialServices,
"XLP": MorningstarSectorCode.ConsumerDefensive,
"XLY": MorningstarSectorCode.ConsumerCyclical,
"XLB": MorningstarSectorCode.BasicMaterials,
"XLE": MorningstarSectorCode.Energy,
"PSR": MorningstarSectorCode.RealEstate,
"IYZ": MorningstarSectorCode.CommunicationServices
}
def CoarseSelectionFunction(self, coarse):
if self.month != self.Time.month:
self.month = self.Time.month
sortedCoarse = [x for x in coarse
if x.HasFundamentalData
and x.Volume > self.volumeMin
and x.Price > self.priceMin]
self.topSectors = []
self.sectorPerformances = {}
for x in self.sectorStocks:
key = Symbol.Create(x, SecurityType.Equity, Market.USA)
history = self.History(key, self.sectorHistory, Resolution.Daily) # Parameter (1)
# self.Debug(history)
if history.empty: continue
first = history.head(1)['close'].iloc[0]
last = history.tail(1)['open'].iloc[0]
self.sectorPerformances[self.sectorStocks[x]] = (last-first)/last
self.topSectors = [x for x in self.sectorPerformances if self.sectorPerformances[x] > 0]
return [x.Symbol for x in sortedCoarse]
else: return Universe.Unchanged
def FineSelectionFunction(self, fine):
filteredFine = [x for x in fine if x.FinancialStatements.IncomeStatement.NetIncome.TwelveMonths
and x.FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.TwelveMonths
and x.OperationRatios.ROA.ThreeMonths
and x.OperationRatios.ROA.OneYear
and x.FinancialStatements.BalanceSheet.ShareIssued.ThreeMonths
and x.FinancialStatements.BalanceSheet.ShareIssued.TwelveMonths
and x.OperationRatios.GrossMargin.ThreeMonths
and x.OperationRatios.GrossMargin.OneYear
and x.OperationRatios.LongTermDebtEquityRatio.ThreeMonths
and x.OperationRatios.LongTermDebtEquityRatio.OneYear
and x.OperationRatios.CurrentRatio.ThreeMonths
and x.OperationRatios.CurrentRatio.OneYear
and x.OperationRatios.AssetsTurnover.ThreeMonths
and x.OperationRatios.AssetsTurnover.OneYear
and x.ValuationRatios.NormalizedPERatio
and x.EarningReports.BasicAverageShares.ThreeMonths
and x.EarningReports.BasicEPS.TwelveMonths
and x.ValuationRatios.PayoutRatio > 0]
sortedByFScore = [x for x in filteredFine if FScore(x.FinancialStatements.IncomeStatement.NetIncome.TwelveMonths,
x.FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.TwelveMonths,
x.OperationRatios.ROA.ThreeMonths,
x.OperationRatios.ROA.OneYear,
x.FinancialStatements.BalanceSheet.ShareIssued.ThreeMonths,
x.FinancialStatements.BalanceSheet.ShareIssued.TwelveMonths,
x.OperationRatios.GrossMargin.ThreeMonths,
x.OperationRatios.GrossMargin.OneYear,
x.OperationRatios.LongTermDebtEquityRatio.ThreeMonths,
x.OperationRatios.LongTermDebtEquityRatio.OneYear,
x.OperationRatios.CurrentRatio.ThreeMonths,
x.OperationRatios.CurrentRatio.OneYear,
x.OperationRatios.AssetsTurnover.ThreeMonths,
x.OperationRatios.AssetsTurnover.OneYear).ObjectiveScore() > self.piotroskiMin
]
sortedBySector = [x for x in sortedByFScore if x.AssetClassification.MorningstarSectorCode in self.topSectors]
self.symbols = [i.Symbol for i in sortedBySector]
return self.symbols
def Daily(self):
if self.month == self.Time.month and not self.Portfolio.Invested: self.Rebalance()
def Rebalance(self):
filterByPrice = [x for x in self.symbols if self.ActiveSecurities[x].Price > 0]
sortByPrice = sorted(filterByPrice, key=lambda x: (self.ActiveSecurities[x].Price), reverse = False)
buyingPower = (self.Portfolio.MarginRemaining / (self.monthlyStocks/self.leverage)) * self.leniency
if buyingPower > 0:
for symbol in sortByPrice:
if self.Portfolio.MarginRemaining >= buyingPower:
history = self.History(symbol, self.stockHistory, Resolution.Daily)
if history.empty: return
pOpen = history.head(1)['open'].iloc[0]
pClose = history.tail(1)['close'].iloc[0]
if pClose > pOpen:
orderSize = buyingPower / self.Securities[symbol].Price
self.MarketOrder(symbol, orderSize)
else: self.Log("Insufficient Buying Power: " + str(self.Portfolio.MarginRemaining))
class FScore(object):
def __init__(self,
netincome,
operating_cashflow,
roa_current,
roa_past,
issued_current,
issued_past,
grossm_current,
grossm_past,
longterm_current,
longterm_past,
curratio_current,
curratio_past,
assetturn_current,
assetturn_past):
self.netincome = netincome
self.operating_cashflow = operating_cashflow
self.roa_current = roa_current
self.roa_past = roa_past
self.issued_current = issued_current
self.issued_past = issued_past
self.grossm_current = grossm_current
self.grossm_past = grossm_past
self.longterm_current = longterm_current
self.longterm_past = longterm_past
self.curratio_current = curratio_current
self.curratio_past = curratio_past
self.assetturn_current = assetturn_current
self.assetturn_past = assetturn_past
def ObjectiveScore(self):
''' The Piotroski score is broken down into profitability; leverage, liquidity, and source of funds; and operating efficiency categories, as follows: '''
fscore = 0
''' Profitability Criteria '''
fscore += np.where(self.netincome > 0, 1, 0) # Positive Net Income (X Months?)
fscore += np.where(self.operating_cashflow > 0, 1, 0) # Positive Operating Cash Flow
fscore += np.where(self.roa_current > self.roa_past, 1, 0) # Positive Return on Assets
fscore += np.where(self.operating_cashflow > self.roa_current, 1, 0) # Cash flow from operations being greater than net income (quality of earnings)
''' Leverage, Liquidity, and Source of Dunds Criteria '''
fscore += np.where(self.longterm_current <= self.longterm_past, 1, 0) # Lower ratio of long term debt in the current period, compared to the previous year (decreased leverage)
fscore += np.where(self.curratio_current >= self.curratio_past, 1, 0) # Higher current ratio this year compared to the previous year (more liquidity)
fscore += np.where(self.issued_current <= self.issued_past, 1, 0) # No new shares were issued in the last year
''' Operating Efficiency Criteria '''
# A higher gross margin compared to the previous year
fscore += np.where(self.grossm_current >= self.grossm_past, 1, 0) # A higher gross margin compared to the previous year
fscore += np.where(self.assetturn_current >= self.assetturn_past, 1, 0) # A higher asset turnover ratio compared to the previous year (1 point)
return fscore