wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-06-29 21:06:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

MAINT: Tweaks/cleanups in technical.py.

Browse Source 
- Use `expect_bounded` to check inputs.
- Add tests for expected failures from `MACDSignal`.
- Use `float64` instead of `float` in a few places.  This prevents
  diverging behavior on 32-bit systems.
- Docstring edits.
This commit is contained in:
Scott Sanderson
2016-11-28 13:00:24 -05:00
parent 48327266db
commit 9c05e5edfe
2 changed files with 85 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/pipeline/test_technical.py
+45 -7
View File
@@ -438,6 +438,35 @@ class MovingAverageConvergenceDivergenceTestCase(ZiplineTestCase):
slow_period + signal_period - 1,
)
def test_bad_inputs(self):
template = (
"MACDSignal() expected a value greater than or equal to 1"
" for argument %r, but got 0 instead."
)
with self.assertRaises(ValueError) as e:
MovingAverageConvergenceDivergenceSignal(fast_period=0)
self.assertEqual(template % 'fast_period', str(e.exception))
with self.assertRaises(ValueError) as e:
MovingAverageConvergenceDivergenceSignal(slow_period=0)
self.assertEqual(template % 'slow_period', str(e.exception))
with self.assertRaises(ValueError) as e:
MovingAverageConvergenceDivergenceSignal(signal_period=0)
self.assertEqual(template % 'signal_period', str(e.exception))
with self.assertRaises(ValueError) as e:
MovingAverageConvergenceDivergenceSignal(
fast_period=5,
slow_period=4,
)
expected = (
"'slow_period' must be greater than 'fast_period', but got\n"
"slow_period=4, fast_period=5"
)
self.assertEqual(expected, str(e.exception))
@parameter_space(
seed=range(2),
fast_period=[3, 5],
@@ -478,14 +507,23 @@ class MovingAverageConvergenceDivergenceTestCase(ZiplineTestCase):
close_df = pd.DataFrame(close)
fast_ewma = self.expected_ewma(
close_df,
fast_period)
fast_period,
)
slow_ewma = self.expected_ewma(
close_df,
slow_period)
expected_signal = self.expected_ewma(
fast_ewma-slow_ewma,
slow_period,
)
signal_ewma = self.expected_ewma(
fast_ewma - slow_ewma,
signal_period
).values[-1]
)
# Everything but the last row should be NaN.
self.assertTrue(signal_ewma.iloc[:-1].isnull().all().all())
# We're testing a single compute call, which we expect to be equivalent
# to the last row of the frame we calculated with pandas.
expected_signal = signal_ewma.values[-1]
np.testing.assert_almost_equal(
out,
@@ -505,7 +543,7 @@ class AnnualizedVolatilityTestCase(ZiplineTestCase):
nassets = 3
ann_vol = AnnualizedVolatility()
today = pd.Timestamp('2016', tz='utc')
assets = np.arange(nassets, dtype=np.float)
assets = np.arange(nassets, dtype=np.float64)
returns = np.full((ann_vol.window_length, nassets),
0.004,
dtype=np.float64)
@@ -527,7 +565,7 @@ class AnnualizedVolatilityTestCase(ZiplineTestCase):
nassets = 3
ann_vol = AnnualizedVolatility()
today = pd.Timestamp('2016', tz='utc')
assets = np.arange(nassets, dtype=np.float)
assets = np.arange(nassets, dtype=np.float64)
returns = np.random.normal(loc=0.001,
scale=0.01,
size=(ann_vol.window_length, nassets))
zipline/pipeline/factors/technical.py
+40 -28
View File
@@ -26,8 +26,7 @@ from numexpr import evaluate
from zipline.pipeline.data import USEquityPricing
from zipline.pipeline.mixins import SingleInputMixin
from zipline.utils.numpy_utils import ignore_nanwarnings
from zipline.utils.input_validation import expect_types
from zipline.utils.input_validation import expect_bounded, expect_types
from zipline.utils.math_utils import (
nanargmax,
nanargmin,
@@ -37,7 +36,11 @@ from zipline.utils.math_utils import (
nansum,
nanmin,
)
from zipline.utils.numpy_utils import rolling_window
from zipline.utils.numpy_utils import (
float64_dtype,
ignore_nanwarnings,
rolling_window,
)
from .factor import CustomFactor
@@ -401,13 +404,13 @@ class LinearWeightedMovingAverage(CustomFactor, SingleInputMixin):
ctx = ignore_nanwarnings()
def compute(self, today, assets, out, data):
num_days = data.shape[0]
ndays = data.shape[0]
# Initialize weights array
weights = arange(1, num_days + 1, dtype=float).reshape(num_days, 1)
weights = arange(1, ndays + 1, dtype=float64_dtype).reshape(ndays, 1)
# Compute normalizer
normalizer = (num_days * (num_days + 1)) / 2
normalizer = (ndays * (ndays + 1)) / 2
# Weight the data
weighted_data = data * weights
@@ -706,8 +709,6 @@ class MovingAverageConvergenceDivergenceSignal(CustomFactor):
trend in a stock's price.
**Default Inputs:** :data:`zipline.pipeline.data.USEquityPricing.close`
**Default Window Length:** Window length is automatically calculated as the
sum of slow_period and signal_period.
Parameters
----------
@@ -718,15 +719,24 @@ class MovingAverageConvergenceDivergenceSignal(CustomFactor):
signal_period' : int > 0, < fast_period
The window length for the signal line. Default is 9.
Returns
-------
The EWMA of the difference between "fast" EWMA and "slow" EWMA line using
`signal_period` as span.
Notes
-----
Unlike most Factors, MovingAverageConvergenceDivergence does not accept a
``window_length`` parameter. ``window_length`` is inferred from
``slow_period`` and ``signal_period``.
"""
inputs = [USEquityPricing.close]
inputs = (USEquityPricing.close,)
# We don't use the default form of `params` here because we want to
# dynamically calculate `window_length` from the period lengths in our
# __new__.
params = ('fast_period', 'slow_period', 'signal_period')
@expect_bounded(
__funcname='MACDSignal',
fast_period=(1, None), # These must all be >= 1.
slow_period=(1, None),
signal_period=(1, None),
)
def __new__(cls,
fast_period=12,
slow_period=26,
@@ -734,12 +744,13 @@ class MovingAverageConvergenceDivergenceSignal(CustomFactor):
*args,
**kwargs):
if signal_period <= 0:
raise ValueError("'signal_period' must be larger than 0.")
if slow_period <= fast_period or fast_period <= signal_period:
if slow_period <= fast_period:
raise ValueError(
"'slow_period' must be larger than 'fast_period'."
"'fast_period' must be larger than 'signal_period'."
"'slow_period' must be greater than 'fast_period', but got\n"
"slow_period={slow}, fast_period={fast}".format(
slow=slow_period,
fast=fast_period,
)
)
return super(MovingAverageConvergenceDivergenceSignal, cls).__new__(
@@ -753,10 +764,11 @@ class MovingAverageConvergenceDivergenceSignal(CustomFactor):
def _ewma(self, data, length):
decay_rate = 1.0 - (2.0 / (1.0 + length))
return average(data,
axis=1,
weights=exponential_weights(length, decay_rate)
)
return average(
data,
axis=1,
weights=exponential_weights(length, decay_rate)
)
def compute(self, today, assets, out, close, fast_period, slow_period,
signal_period):
@@ -778,19 +790,19 @@ class AnnualizedVolatility(CustomFactor):
https://en.wikipedia.org/wiki/Volatility_(finance)
The degree of variation of a series over time as measured by the standard
deviation of returns.
deviation of daily returns.
**Default Inputs:**
:data:`zipline.pipeline.factors.Returns(window_length=2)`
Parameters
----------
annualization_factor :
The number of time units per year. Defaults to average number of NYSE
trading days per year, 252.
annualization_factor : float, optional
The number of time units per year. Defaults is 252, the number of NYSE
trading days in a normal year.
"""
inputs = [Returns(window_length=2)]
params = {'annualization_factor': 252}
params = {'annualization_factor': 252.0}
window_length = 252
def compute(self, today, assets, out, returns, annualization_factor):