Overall Statistics
Total Trades
950
Average Win
0.39%
Average Loss
-0.27%
Compounding Annual Return
280.107%
Drawdown
9.500%
Expectancy
0.274
Net Profit
41.020%
Sharpe Ratio
7.287
Probabilistic Sharpe Ratio
96.985%
Loss Rate
48%
Win Rate
52%
Profit-Loss Ratio
1.44
Alpha
2.108
Beta
-0.021
Annual Standard Deviation
0.291
Annual Variance
0.085
Information Ratio
4.383
Tracking Error
0.596
Treynor Ratio
-100.665
Total Fees
$2006.00
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 Piotroski(QCAlgorithm):
    
    def Initialize(self):
        ''' Backtesting Parameters '''
        self.SetStartDate(2020, 1, 1)  
        #self.SetEndDate(2020, 1, 1)    
        self.minutesAfterOpen = 10
        self.SetCash(50000)            

        ''' Universe Settings '''
        self.benchmark = Symbol.Create("SPY", SecurityType.Equity, Market.USA)
        self.UniverseSettings.Resolution = Resolution.Minute        
        self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
        self.topScoreSymbolsCoarse = 10000
        self.topScoreSymbolsFine   = 50

        ''' Schedule Settings '''
        self.AddEquity("SPY", Resolution.Minute)
        self.SetBenchmark("SPY")
        self.Schedule.On(self.DateRules.EveryDay("SPY"), self.TimeRules.AfterMarketOpen("SPY", 1), Action(self.Buy))
        self.Schedule.On(self.DateRules.EveryDay("SPY"), self.TimeRules.AfterMarketOpen("SPY", 60), Action(self.Liquidate))
        
        ''' Other Settings '''
        self.month   = -1
        self.symbols = []
        self.changes = {}
        
    def CoarseSelectionFunction(self, coarse):
        if self.month != self.Time.month:
            sortedCoarse = [x for x in coarse 
                            if x.HasFundamentalData 
                            and x.Price > 5]
            sortedDollarVolume =  sorted(sortedCoarse, key=lambda x: x.DollarVolume, reverse=True) 
            topCoarse = sortedDollarVolume[:self.topScoreSymbolsCoarse]
            return [x.Symbol for x in topCoarse]
        else: return self.symbols

    def FineSelectionFunction(self, fine):
        if self.month != self.Time.month:
            self.month = self.Time.month

            ''' Retrieve all stocks that have the valid variation ratios that we want '''
            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]
                                    
            ''' Using the FScore class, retrieve the stocks that have a score of X or higher '''    
            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() > 6
                                                                ]
                                                                
            self.qualityStocks = sortedByFScore
            
            ''' The Piotroski score ranks quality stocks, but we still need to determine value by filtering it more '''
            sortedByNormalizedPE = sorted(sortedByFScore, key=lambda x: x.EarningReports.BasicAverageShares.ThreeMonths *
                                                                        x.EarningReports.BasicEPS.TwelveMonths *
                                                                        x.ValuationRatios.NormalizedPERatio, reverse = True)
                                                                        
            self.Debug(str(len(self.qualityStocks)))
            topFine = sortedByNormalizedPE[:self.topScoreSymbolsFine]

            self.symbols = [i.Symbol for i in topFine]
            return self.symbols
        else: return self.symbols
        
    def Buy(self):
        self.changes = {}
        for symbol in self.symbols:
            history = self.History(symbol, 2, Resolution.Minute)
            if history.empty: continue
            self.Debug(history)
            first     = history.head(1)['close'].iloc[0]
            last      = history.tail(1)['open'].iloc[0]
            if last < first: self.changes[symbol] = (last - first)/first
        
        if len(self.changes) > 0:    
            sortedSymbols = sorted(self.changes.items(), key=lambda x: x[1], reverse=False)[:10]
            for symbol in sortedSymbols:
                self.Debug(str(self.Time) + " - BUY: " + str(symbol[0]) + " - PERCENT GAINED: " + str(symbol[1]))
                self.SetHoldings(symbol[0], .25)

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