wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-04 19:25:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

ENH: Add MACDSignal, test with random input

Browse Source
This commit is contained in:
Ana Ruelas
2016-11-23 11:43:29 -05:00
parent 435d5acd14
commit 3363237123
4 changed files with 84 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/pipeline/test_technical.py
+33 -35
View File
@@ -17,13 +17,12 @@ from zipline.pipeline.factors import (
LinearWeightedMovingAverage,
RateOfChangePercentage,
TrueRange,
MovingAverageConvergenceDivergence,
MovingAverageConvergenceDivergenceSignal,
AnnualizedVolatility,
)
from zipline.testing import parameter_space
from zipline.testing.fixtures import ZiplineTestCase
from zipline.testing.predicates import assert_equal
from .base import BasePipelineTestCase
@@ -409,11 +408,24 @@ class TestTrueRange(ZiplineTestCase):
class MovingAverageConvergenceDivergenceTestCase(ZiplineTestCase):
def expected_ewma(self, data_df, window):
# Comment copied from `test_engine.py`:
# XXX: This is a comically inefficient way to compute a windowed EWMA.
# Don't use it outside of testing. We're using rolling-apply of an
# ewma (which is itself a rolling-window function) because we only want
# to look at ``window_length`` rows at a time.
return data_df.rolling(window).apply(
lambda sub: pd.DataFrame(sub)
.ewm(span=window)
.mean()
.values[-1])
def test_MACD_window_length_generation(self):
signal_period = random_integers(1, 90)
fast_period = random_integers(signal_period+1, signal_period+100)
slow_period = random_integers(fast_period+1, fast_period+100)
ewma = MovingAverageConvergenceDivergence(
ewma = MovingAverageConvergenceDivergenceSignal(
fast_period=fast_period,
slow_period=slow_period,
signal_period=signal_period,
@@ -424,34 +436,22 @@ class MovingAverageConvergenceDivergenceTestCase(ZiplineTestCase):
)
def test_moving_average_convergence_divergence(self):
nassets = 3
fast_period = 3
slow_period = 8
signal_period = 2
macd = MovingAverageConvergenceDivergence(
macd = MovingAverageConvergenceDivergenceSignal(
fast_period=fast_period,
slow_period=slow_period,
signal_period=signal_period,
)
today = pd.Timestamp('2016', tz='utc')
nassets = macd.window_length
assets = pd.Index(np.arange(nassets))
days_col = np.arange(start=-.05,
stop=.01*nassets-.05,
step=.01)[:, np.newaxis]
close = np.logspace(start=.01, stop=.10, num=nassets) - 1 + days_col
out = np.empty(shape=(nassets,), dtype=np.float64)
close = np.random.rand(macd.window_length, nassets)
dtype = [
('macd', 'f8'),
('signal', 'f8'),
('hist', 'f8'),
]
out = np.recarray(
shape=(nassets,),
dtype=dtype,
buf=np.empty(shape=(nassets,), dtype=dtype),
)
macd.compute(
today,
assets,
@@ -462,25 +462,23 @@ class MovingAverageConvergenceDivergenceTestCase(ZiplineTestCase):
signal_period,
)
expected_macd = np.array([0.01691553] * nassets)
expected_signal = np.array([0.01691553] * nassets)
expected_hist = np.array([0] * nassets)
close_df = pd.DataFrame(close)
fast_ewma = self.expected_ewma(
close_df,
fast_period)
slow_ewma = self.expected_ewma(
close_df,
slow_period)
expected_signal = self.expected_ewma(
fast_ewma-slow_ewma,
signal_period
).values[-1]
np.testing.assert_almost_equal(
out.macd,
expected_macd,
decimal=8
)
np.testing.assert_almost_equal(
out.signal,
out,
expected_signal,
decimal=8
)
np.testing.assert_almost_equal(
out.hist,
expected_hist,
decimal=8
)
class AnnualizedVolatilityTestCase(ZiplineTestCase):
@@ -502,7 +500,7 @@ class AnnualizedVolatilityTestCase(ZiplineTestCase):
ann_vol.compute(today, assets, out, returns, 252)
expected_vol = np.array([0] * nassets)
expected_vol = np.zeros(nassets)
np.testing.assert_almost_equal(
out,
expected_vol,
@@ -523,7 +521,7 @@ class AnnualizedVolatilityTestCase(ZiplineTestCase):
out = np.empty(shape=(nassets,), dtype=np.float64)
ann_vol.compute(today, assets, out, returns, 252)
mean = returns.sum(axis=0) / returns.shape[0]
mean = np.mean(returns, axis=0)
annualized_variance = ((returns - mean) ** 2).sum(axis=0) / \
returns.shape[0] * 252
expected_vol = np.sqrt(annualized_variance)
zipline/pipeline/factors/__init__.py
+6 -6
View File
@@ -14,6 +14,7 @@ from .statistical import (
RollingSpearmanOfReturns,
)
from .technical import (
AnnualizedVolatility,
Aroon,
AverageDollarVolume,
BollingerBands,
@@ -24,7 +25,9 @@ from .technical import (
FastStochasticOscillator,
IchimokuKinkoHyo,
LinearWeightedMovingAverage,
MACDSignal,
MaxDrawdown,
MovingAverageConvergenceDivergenceSignal,
RateOfChangePercentage,
Returns,
RSI,
@@ -32,12 +35,10 @@ from .technical import (
TrueRange,
VWAP,
WeightedAverageValue,
MovingAverageConvergenceDivergence,
MACD,
AnnualizedVolatility,
)
__all__ = [
'AnnualizedVolatility',
'Aroon',
'AverageDollarVolume',
'BollingerBands',
@@ -53,7 +54,9 @@ __all__ = [
'IchimokuKinkoHyo',
'Latest',
'LinearWeightedMovingAverage',
'MACDSignal',
'MaxDrawdown',
'MovingAverageConvergenceDivergenceSignal',
'RateOfChangePercentage',
'RecarrayField',
'Returns',
@@ -65,7 +68,4 @@ __all__ = [
'TrueRange',
'VWAP',
'WeightedAverageValue',
'MovingAverageConvergenceDivergence',
'MACD',
'AnnualizedVolatility',
]
zipline/pipeline/factors/technical.py
+36 -49
View File
@@ -14,7 +14,6 @@ from numpy import (
dstack,
exp,
fmax,
full,
inf,
isnan,
log,
@@ -36,6 +35,7 @@ from zipline.utils.math_utils import (
nanstd,
nansum,
nanmin,
exponential_weights,
)
from zipline.utils.numpy_utils import rolling_window
from .factor import CustomFactor
@@ -192,14 +192,6 @@ class _ExponentialWeightedFactor(SingleInputMixin, CustomFactor):
"""
params = ('decay_rate',)
@staticmethod
def weights(length, decay_rate):
"""
Return weighting vector for an exponential moving statistic on `length`
rows with a decay rate of `decay_rate`.
"""
return full(length, decay_rate, float) ** arange(length + 1, 1, -1)
@classmethod
@expect_types(span=Number)
def from_span(cls, inputs, window_length, span, **kwargs):
@@ -369,7 +361,7 @@ class ExponentialWeightedMovingAverage(_ExponentialWeightedFactor):
out[:] = average(
data,
axis=0,
weights=self.weights(len(data), decay_rate),
weights=exponential_weights(len(data), decay_rate),
)
@@ -434,7 +426,7 @@ class ExponentialWeightedMovingStdDev(_ExponentialWeightedFactor):
"""
def compute(self, today, assets, out, data, decay_rate):
weights = self.weights(len(data), decay_rate)
weights = exponential_weights(len(data), decay_rate)
mean = average(data, axis=0, weights=weights)
variance = average((data - mean) ** 2, axis=0, weights=weights)
@@ -681,9 +673,9 @@ class TrueRange(CustomFactor):
)
class MovingAverageConvergenceDivergence(_ExponentialWeightedFactor):
class MovingAverageConvergenceDivergenceSignal(CustomFactor):
"""
Moving Average Convergence/Divergence (MACD)
Moving Average Convergence/Divergence (MACD) Signal line
https://en.wikipedia.org/wiki/MACD
A technical indicator originally developed by Gerald Appel in the late
@@ -697,22 +689,21 @@ class MovingAverageConvergenceDivergence(_ExponentialWeightedFactor):
Parameters
----------
fast_period : int >= 0, <= window_length
fast_period : int > 0
The window length for the "fast" EWMA. Default is 12.
slow_period : int >= 0, <= window_length
slow_period : int > 0, > fast_period
The window length for the "slow" EWMA. Default is 26.
signal_period' : int >= 0, <= slow_period
signal_period' : int > 0, < fast_period
The window length for the signal line. Default is 9.
Returns
-------
MACD: The difference between "fast" EWMA and "slow" EWMA.
signal: The EWMA of the MACD line using `signal_period` as span.
hist: Difference between MACD and signal. (Divergence series)
The EWMA of the difference between "fast" EWMA and "slow" EWMA line using
`signal_period` as span.
"""
inputs = [USEquityPricing.close]
params = ('fast_period', 'slow_period', 'signal_period')
outputs = ('MACD', 'signal', 'hist')
def __new__(cls,
fast_period=12,
@@ -720,7 +711,16 @@ class MovingAverageConvergenceDivergence(_ExponentialWeightedFactor):
signal_period=9,
*args,
**kwargs):
return super(MovingAverageConvergenceDivergence, cls).__new__(
if signal_period <= 0:
raise ValueError("'signal_period' must be larger than 0.")
if slow_period <= fast_period or fast_period <= signal_period:
raise ValueError(
"'slow_period' must be larger than 'fast_period'."
"'fast_period' must be larger than 'signal_period'."
)
return super(MovingAverageConvergenceDivergenceSignal, cls).__new__(
cls,
fast_period=fast_period,
slow_period=slow_period,
@@ -729,38 +729,25 @@ class MovingAverageConvergenceDivergence(_ExponentialWeightedFactor):
*args, **kwargs
)
def calculate_ewma(self, data, length):
def _ewma(self, data, length):
decay_rate = 1.0 - (2.0 / (1.0 + length))
return average(data,
axis=1,
weights=self.weights(length, decay_rate))
def calculate_macd(self, col):
slow_EWMA = self.calculate_ewma(
rolling_window(
col,
self.params['slow_period']
),
self.params['slow_period'])
fast_EWMA = self.calculate_ewma(
rolling_window(
col,
self.params['fast_period']
)[-self.params['signal_period']:],
self.params['fast_period'])
macd = fast_EWMA - slow_EWMA
signal_line = self.calculate_ewma(
macd.reshape(-1, self.params['signal_period']),
self.params['signal_period'])
hist = macd[-1] - signal_line
return macd[-1], signal_line[-1], hist[-1]
weights=exponential_weights(length, decay_rate)
)
def compute(self, today, assets, out, close, fast_period, slow_period,
signal_period):
macd, sig, hist = zip(*map(self.calculate_macd, close.T))
out.macd[:] = macd
out.signal[:] = sig
out.hist[:] = hist
slow_EWMA = self._ewma(
rolling_window(close, slow_period),
slow_period
)
fast_EWMA = self._ewma(
rolling_window(close, fast_period)[-signal_period:],
fast_period
)
macd = fast_EWMA - slow_EWMA
out[:] = self._ewma(macd.T, signal_period)
class AnnualizedVolatility(CustomFactor):
@@ -785,9 +772,9 @@ class AnnualizedVolatility(CustomFactor):
window_length = 252
def compute(self, today, assets, out, returns, annualization_factor):
out[:] = nanstd(returns, ddof=0, axis=0) * (annualization_factor ** .5)
out[:] = nanstd(returns, axis=0) * (annualization_factor ** .5)
# Convenience aliases.
EWMA = ExponentialWeightedMovingAverage
EWMSTD = ExponentialWeightedMovingStdDev
MACD = MovingAverageConvergenceDivergence
MACDSignal = MovingAverageConvergenceDivergenceSignal
zipline/utils/math_utils.py
+9 -1
View File
@@ -14,7 +14,7 @@
# limitations under the License.
import math
from numpy import isnan
from numpy import isnan, full, arange
def tolerant_equals(a, b, atol=10e-7, rtol=10e-7, equal_nan=False):
@@ -77,3 +77,11 @@ def round_if_near_integer(a, epsilon=1e-4):
return round(a)
else:
return a
def exponential_weights(length, decay_rate):
"""
Return weighting vector for an exponential moving statistic on `length`
rows with a decay rate of `decay_rate`.
"""
return full(length, decay_rate, float) ** arange(length + 1, 1, -1)