Pipeline Example: Piotrosky Score¶
A stock's Piotrosky Score is a simple measure of its "financial strength". It was introduced by Joseph Piotrosky in the paper:
The score identifies nine criteria as indicators of financial health and awards one point for each criterion that a stock meets, for a total of up to 9 points.
Factor definitions were sourced from Mark Segal's helpful post, from Investopedia's Article on Piotroski Score, and from The Graham Investor's Article on Piotroski Score.
In [16]:
from quantopian.research import run_pipeline
from quantopian.pipeline import Pipeline, CustomFactor
from quantopian.pipeline.data import morningstar as m
from quantopian.pipeline.data.builtin import USEquityPricing
TRADING_DAYS_PER_YEAR = 252
OLDEST = -1
NEWEST = 0
# Most definitions sourced from:
# http://www.grahaminvestor.com/articles/quantitative-tools/the-piotroski-score/
In [2]:
class NetIncomePositive(CustomFactor):
"""
Factor returning 1.0 if net_income is currently positive.
Otherwise returns 0.
'Net Income: Bottom line.
Score 1 if last year net income is positive.'
"""
inputs = [m.income_statement.net_income]
window_length = 1
def compute(self, today, assets, out, net_income):
# NOTE:
# This is actually more complicated than it looks.
# (net_income > 0) will return a numpy array of dtype `bool`,
# but `out` is an array of dtype `float64`.
#
# Numpy allows casts from bool to float though, with the result
# being that True values are cast to 1.0, and False values are cast to 0.0
out[:] = (net_income[NEWEST] > 0)
In [3]:
class OperatingCashFlowPositive(CustomFactor):
"""
Factor returning 1.0 if operating cash flow is currently positive.
Otherwise returns 0.
'Operating Cash Flow: A better earnings gauge.
Score 1 if last year cash flow is positive.'
"""
inputs = [m.cash_flow_statement.operating_cash_flow]
window_length = 1
def compute(self, today, assets, out, ocf):
# See note in NetPositiveIncome.compute for an explanation of why
# this assignment is allowed.
out[:] = (ocf[NEWEST] > 0)
In [4]:
class ReturnOnAssetsIncreased(CustomFactor):
"""
Factor returning 1.0 if this year's ROA exceeds last year's ROA.
Otherwise returns 0.
'Return On Assets: Measures Profitability.
Score 1 if last year ROA exceeds prior-year ROA.'
"""
inputs = [m.operation_ratios.roa]
window_length = TRADING_DAYS_PER_YEAR
def compute(self, today, assets, out, roa):
# See note in NetPositiveIncome.compute for an explanation of why
# this assignment is allowed.
out[:] = (roa[NEWEST] > roa[OLDEST])
In [5]:
class EarningsQuality(CustomFactor):
"""
Factor returning 1.0 if operating cash flow exceeds net income.
Otherwise returns 0.
'Quality of Earnings: Warns of Accounting Tricks.
Score 1 if last year operating cash flow exceeds net income.'
"""
inputs = [m.cash_flow_statement.operating_cash_flow,
m.income_statement.net_income,
]
window_length = 1
def compute(self, today, assets, out, ocf, net_income):
# See note in NetPositiveIncome.compute for an explanation of why
# this assignment is allowed.
out[:] = (ocf[NEWEST] > net_income[NEWEST])
In [6]:
class LongTermDebtDecreased(CustomFactor):
"""
Factor returning 1.0 if Long Term Debt to Equity Ratio decreased year over year.
Otherwise returns 0.
'Long-Term Debt vs. Assets: Is Debt Decreasing?
Score 1 if the ratio of long-term debt to assets is down from the year-ago value.
(If LTD is zero but assets are increasing, score 1 anyway.)'
"""
inputs = [m.operation_ratios.long_term_debt_equity_ratio]
window_length = TRADING_DAYS_PER_YEAR
def compute(self, today, assets, out, debt_to_equity):
# See note in NetPositiveIncome.compute for an explanation of why
# this assignment is allowed.
out[:] = (debt_to_equity[NEWEST] < debt_to_equity[OLDEST])
In [7]:
class CurrentRatioIncreased(CustomFactor):
"""
Factor returning 1.0 if Current Ratio increased year over year.
'Current Ratio: Measures increasing working capital.
Score 1 if CR has increased from the prior year.'
"""
inputs = [m.operation_ratios.current_ratio]
window_length = TRADING_DAYS_PER_YEAR
def compute(self, today, assets, out, current_ratio):
# See note in NetPositiveIncome.compute for an explanation of why
# this assignment is allowed.
out[:] = (current_ratio[NEWEST] > current_ratio[OLDEST])
In [8]:
class SharesOutstandingNotIncreased(CustomFactor):
"""
Factor returning 1.0 if shares outstanding decreased or stayed constant.
'Shares Outstanding: A Measure of potential dilution.
Score 1 if the number of shares outstanding is no greater than the year-ago figure.'
"""
inputs = [m.valuation.shares_outstanding]
window_length = TRADING_DAYS_PER_YEAR
def compute(self, today, assets, out, shares_outstanding):
# See note in NetPositiveIncome.compute for an explanation of why
# this assignment is allowed.
out[:] = (shares_outstanding[NEWEST] <= shares_outstanding[OLDEST])
In [9]:
class GrossMarginIncreased(CustomFactor):
"""
Factor returning 1.0 if gross margin increased year over year.
'Gross Margin: A measure of improving competitive position.
Score 1 if full-year GM exceeds the prior-year GM.'
"""
inputs = [m.operation_ratios.gross_margin]
window_length = TRADING_DAYS_PER_YEAR
def compute(self, today, assets, out, gross_margin):
out[:] = (gross_margin[NEWEST] > gross_margin[OLDEST])
In [10]:
class AssetTurnover(CustomFactor):
"""
Factor returning 1.0 asset turnover ratio increased year over year.
'Asset Turnover: Measures productivity.
Score 1 if the percentage increase in sales exceeds the percentage increase in total assets.'
"""
inputs = [m.operation_ratios.assets_turnover]
window_length = TRADING_DAYS_PER_YEAR
def compute(self, today, assets, out, turnover):
out[:] = (turnover[NEWEST] > turnover[OLDEST])
In [11]:
p_score = (
NetIncomePositive() +
OperatingCashFlowPositive() +
ReturnOnAssetsIncreased() +
EarningsQuality() +
LongTermDebtDecreased() +
CurrentRatioIncreased() +
SharesOutstandingNotIncreased() +
GrossMarginIncreased() +
AssetTurnover()
)
In [17]:
pipe = Pipeline(columns={'p_score': p_score,
'close': USEquityPricing.close.latest},
screen=p_score > 5)
In [18]:
pipe.show_graph(format='png')
Out[18]:
In [19]:
# Run a 1-year pipeline computeing the Piotrosky score.
# This takes a few minutes.
results = run_pipeline(pipe, '2014', '2015')
In [20]:
results
Out[20]:
In [ ]: