| Overall Statistics |
|
Total Trades 156 Average Win 0.57% Average Loss -0.32% Compounding Annual Return 47.529% Drawdown 4.300% Expectancy 1.178 Net Profit 49.427% Sharpe Ratio 3.138 Loss Rate 22% Win Rate 78% Profit-Loss Ratio 1.78 Alpha 0.354 Beta -0.172 Annual Standard Deviation 0.103 Annual Variance 0.011 Information Ratio 1.125 Tracking Error 0.124 Treynor Ratio -1.88 Total Fees $170.83 |
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 PiotroskiFscoreAlgorithm(QCAlgorithm):
'''In this algorithm we demonstrate how to define a universe as a combination of use the coarse fundamental data and fine fundamental data'''
def Initialize(self):
self.SetStartDate(2017,01,01) #Set Start Date
#self.SetEndDate(2017,05,02) #Set End Date
self.SetCash(100000) #Set Strategy Cash
self.flag1 = 1
self.flag2 = 0
self.flag3 = 0
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
self.AddEquity("SPY", Resolution.Minute)
self.SetBenchmark("SPY")
self.__numberOfSymbols = 100
self.__numberOfSymbolsFine = 30
self._changes = SecurityChanges.None
self.Schedule.On(self.DateRules.MonthStart("SPY"), self.TimeRules.AfterMarketOpen("SPY"), Action(self.Rebalancing))
self.splotName = 'Strategy Info'
sPlot = Chart(self.splotName)
sPlot.AddSeries(Series('Leverage', SeriesType.Line, 0))
self.AddChart(sPlot)
def CoarseSelectionFunction(self, coarse):
if self.flag1:
CoarseWithFundamental = [x for x in coarse if (x.HasFundamentalData) and (float(x.Price) > 5)]
sortedByDollarVolume = sorted(CoarseWithFundamental, key=lambda x: x.DollarVolume, reverse=True)
top = sortedByDollarVolume[:self.__numberOfSymbols]
return [i.Symbol for i in top]
else:
return []
def FineSelectionFunction(self, fine):
if self.flag1:
self.flag1 = 0
self.flag2 = 1
filtered_fine = [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]
sortedByfactor1 = [x for x in filtered_fine 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]
sortedByNormalizedPE = sorted(sortedByfactor1, key=lambda x: x.ValuationRatios.NormalizedPERatio, reverse = False)
topFine = sortedByNormalizedPE[:self.__numberOfSymbolsFine]
self.flag3 = self.flag3 + 1
self.topFine = [i.Symbol for i in topFine]
return [i.Symbol for i in topFine]
else:
return []
def OnData(self, data):
if self.flag3 > 0:
if self.flag2 == 1:
self.flag2 = 0
# if we have no changes, do nothing
if self._changes == SecurityChanges.None: return
# liquidate removed securities
for security in self._changes.RemovedSecurities:
if security.Invested:
self.Liquidate(security.Symbol)
for security in self.topFine:
self.SetHoldings(security, 1.0/float(len(self.topFine)))
self._changes = SecurityChanges.None;
# this event fires whenever we have changes to our universe
def OnSecuritiesChanged(self, changes):
self._changes = changes
def Rebalancing(self):
self.flag1 = 1
self.account_leverage = self.Portfolio.TotalAbsoluteHoldingsCost / self.Portfolio.TotalPortfolioValue
self.Plot(self.splotName,'Leverage', float(self.account_leverage))
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):
fscore = 0
fscore += np.where(self.netincome > 0, 1, 0)
fscore += np.where(self.operating_cashflow > 0, 1, 0)
fscore += np.where(self.roa_current > self.roa_past, 1, 0)
fscore += np.where(self.operating_cashflow > self.roa_current, 1, 0)
fscore += np.where(self.longterm_current <= self.longterm_past, 1, 0)
fscore += np.where(self.curratio_current >= self.curratio_past, 1, 0)
fscore += np.where(self.issued_current <= self.issued_past, 1, 0)
fscore += np.where(self.grossm_current >= self.grossm_past, 1, 0)
fscore += np.where(self.assetturn_current >= self.assetturn_past, 1, 0)
return fscore