wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-26 13:18:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

ENH: Replaces the simple transforms with history calls. Switches

Browse Source 
transforms to quantopian syntax.

Adds the sid attribute to the siddata so it is aware of which security
it represents.
This commit is contained in:
Joe Jevnik
2014-11-17 13:16:30 -05:00
parent 7889f93522
commit e3d8b1034e
17 changed files with 427 additions and 904 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_algorithm.py
-14
View File
@@ -81,7 +81,6 @@ from zipline.sources import (SpecificEquityTrades,
DataPanelSource,
RandomWalkSource)
from zipline.transforms import MovingAverage
from zipline.finance.execution import LimitOrder
from zipline.finance.trading import SimulationParameters
from zipline.utils.api_support import set_algo_instance
@@ -335,19 +334,6 @@ class TestTransformAlgorithm(TestCase):
np.testing.assert_array_equal(res1, res2)
def test_transform_registered(self):
algo = TestRegisterTransformAlgorithm(
sim_params=self.sim_params,
sids=[133]
)
algo.run(self.source)
assert 'mavg' in algo.registered_transforms
assert algo.registered_transforms['mavg']['args'] == (['price'],)
assert algo.registered_transforms['mavg']['kwargs'] == \
{'window_length': 2, 'market_aware': True}
assert algo.registered_transforms['mavg']['class'] is MovingAverage
def test_data_frequency_setting(self):
self.sim_params.data_frequency = 'daily'
algo = TestRegisterTransformAlgorithm(
tests/test_transforms.py
+168 -296
View File
@@ -1,5 +1,5 @@
#
# Copyright 2013 Quantopian, Inc.
# Copyright 2014 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -12,331 +12,203 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytz
import numpy as np
from datetime import timedelta, datetime
from itertools import chain
from datetime import timedelta
from functools import wraps
from itertools import product
from nose_parameterized import parameterized
import operator
import random
from six import itervalues
from six.moves import map
from unittest import TestCase
from nose_parameterized import parameterized
from six.moves import range
import numpy as np
from numpy.testing import assert_allclose
from zipline.utils.test_utils import setup_logger
from zipline.protocol import (
DATASOURCE_TYPE,
Event,
)
from zipline.sources import SpecificEquityTrades
from zipline.transforms.utils import StatefulTransform, EventWindow
from zipline.transforms import MovingVWAP
from zipline.transforms import MovingAverage
from zipline.transforms import MovingStandardDev
from zipline.transforms import Returns
from zipline.algorithm import TradingAlgorithm
import zipline.utils.factory as factory
from zipline.api import add_transform, get_datetime
def to_dt(msg):
return Event({'dt': msg})
def handle_data_wrapper(f):
@wraps(f)
def wrapper(context, data):
dt = get_datetime()
if dt.date() != context.current_date:
context.warmup -= 1
context.mins_for_days.append(1)
context.current_date = dt.date()
else:
context.mins_for_days[-1] += 1
for n in (1, 2, 3):
if n in data:
if data[n].dt == dt:
context.vol_bars[n].append(data[n].volume)
else:
context.vol_bars[n].append(0)
context.price_bars[n].append(data[n].price)
else:
context.price_bars[n].append(np.nan)
context.vol_bars[n].append(0)
context.last_close_prices[n] = context.price_bars[n][-2]
if context.warmup < 0:
return f(context, data)
return wrapper
class NoopEventWindow(EventWindow):
"""
A no-op EventWindow subclass for testing the base EventWindow logic.
Keeps lists of all added and dropped events.
"""
def __init__(self, market_aware, days, delta):
EventWindow.__init__(self, market_aware, days, delta)
def initialize_with(test_case, tfm_name, days):
def initalize(context):
context.test_case = test_case
context.days = days
context.mins_for_days = []
context.price_bars = (None, [np.nan], [np.nan], [np.nan])
context.vol_bars = (None, [np.nan], [np.nan], [np.nan])
if context.days:
context.warmup = days + 1
else:
context.warmup = 2
self.added = []
self.removed = []
self._fields = []
context.current_date = None
@property
def fields(self):
return self._fields
context.last_close_prices = [np.nan, np.nan, np.nan, np.nan]
add_transform(tfm_name, days)
def handle_add(self, event):
self.added.append(event)
def handle_remove(self, event):
self.removed.append(event)
return initalize
class TestEventWindow(TestCase):
def setUp(self):
self.sim_params = factory.create_simulation_parameters()
def windows_with_frequencies(*args):
args = args or (None,)
return product(('daily', 'minute'), args)
setup_logger(self)
self.monday = datetime(2012, 7, 9, 16, tzinfo=pytz.utc)
self.eleven_normal_days = [self.monday + i * timedelta(days=1)
for i in range(11)]
def with_algo(f):
name = f.__name__
if not name.startswith('test_'):
raise ValueError('This must decorate a test case')
# Modify the end of the period slightly to exercise the
# incomplete day logic.
self.eleven_normal_days[-1] -= timedelta(minutes=1)
self.eleven_normal_days.append(self.monday +
timedelta(days=11, seconds=1))
tfm_name = name[len('test_'):]
# Second set of dates to test holiday handling.
self.jul4_monday = datetime(2012, 7, 2, 16, tzinfo=pytz.utc)
self.week_of_jul4 = [self.jul4_monday + i * timedelta(days=1)
for i in range(5)]
@wraps(f)
def wrapper(self, data_frequency, days=None):
sim_params, source = self.sim_and_source[data_frequency]
def test_market_aware_window_normal_week(self):
window = NoopEventWindow(
market_aware=True,
delta=None,
days=3
algo = TradingAlgorithm(
initialize=initialize_with(self, tfm_name, days),
handle_data=handle_data_wrapper(f),
sim_params=sim_params,
)
events = [to_dt(date) for date in self.eleven_normal_days]
lengths = []
# Run the events.
for event in events:
window.update(event)
# Record the length of the window after each event.
lengths.append(len(window.ticks))
algo.run(source)
# The window stretches out during the weekend because we wait
# to drop events until the weekend ends. The last window is
# briefly longer because it doesn't complete a full day. The
# window then shrinks once the day completes
self.assertEquals(lengths, [1, 2, 3, 3, 3, 4, 5, 5, 5, 3, 4, 3])
self.assertEquals(window.added, events)
self.assertEquals(window.removed, events[:-3])
return wrapper
def test_market_aware_window_holiday(self):
window = NoopEventWindow(
market_aware=True,
delta=None,
days=2
class TransformTestCase(TestCase):
"""
Tests the simple transforms by running them through a zipline.
"""
@classmethod
def setUpClass(cls):
random.seed(0)
cls.sids = (1, 2, 3)
minute_sim_ps = factory.create_simulation_parameters(
num_days=3,
sids=cls.sids,
data_frequency='minute',
emission_rate='minute',
)
events = [to_dt(date) for date in self.week_of_jul4]
lengths = []
# Run the events.
for event in events:
window.update(event)
# Record the length of the window after each event.
lengths.append(len(window.ticks))
self.assertEquals(lengths, [1, 2, 3, 3, 2])
self.assertEquals(window.added, events)
self.assertEquals(window.removed, events[:-2])
daily_sim_ps = factory.create_simulation_parameters(
num_days=30,
sids=cls.sids,
data_frequency='daily',
emission_rate='daily',
)
cls.sim_and_source = {
'minute': (minute_sim_ps, factory.create_minutely_trade_source(
cls.sids,
trade_count=45,
sim_params=minute_sim_ps,
)),
'daily': (daily_sim_ps, factory.create_trade_source(
cls.sids,
trade_count=90,
trade_time_increment=timedelta(days=1),
sim_params=daily_sim_ps,
)),
}
def tearDown(self):
setup_logger(self)
class TestFinanceTransforms(TestCase):
def setUp(self):
self.sim_params = factory.create_simulation_parameters()
setup_logger(self)
trade_history = factory.create_trade_history(
133,
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
self.sim_params
)
self.source = trade_history
def intersperse_custom_events(self, events):
"""
Take a stream of events and return the same stream with a minimal event
of type CUSTOM following each trade event. Used to test graceful
handling of CUSTOM events that are missing required transform fields.
Each test consumes a source, we need to rewind it.
"""
return list(
chain.from_iterable(
(
event,
Event(
initial_values={
'dt': event.dt,
'sid': event.sid,
'source_id': "fake_custom_source",
'type': DATASOURCE_TYPE.CUSTOM
}
)
)
for event in events
for _, source in itervalues(self.sim_and_source):
source.rewind()
@parameterized.expand(windows_with_frequencies(1, 2, 3, 4))
@with_algo
def test_mavg(context, data):
"""
Tests the mavg transform by manually keeping track of the prices
in a naiive way and asserting that our mean is the same.
"""
mins = sum(context.mins_for_days[-context.days:])
for sid in data:
assert_allclose(
data[sid].mavg(context.days),
np.mean(context.price_bars[sid][-mins:]),
)
)
def tearDown(self):
self.log_handler.pop_application()
@parameterized.expand(windows_with_frequencies(2, 3, 4))
@with_algo
def test_stddev(context, data):
"""
Tests the stddev transform by manually keeping track of the prices
in a naiive way and asserting that our stddev is the same.
This accounts for the corrected ddof.
"""
mins = sum(context.mins_for_days[-context.days:])
@parameterized.expand([
('with_custom', True),
('without_custom', False),
])
def test_vwap(self, name, add_custom_events):
vwap = MovingVWAP(
market_aware=True,
window_length=2
)
for sid in data:
assert_allclose(
data[sid].stddev(context.days),
np.std(context.price_bars[sid][-mins:], ddof=1),
)
if add_custom_events:
self.source = self.intersperse_custom_events(self.source)
@parameterized.expand(windows_with_frequencies(2, 3, 4))
@with_algo
def test_vwap(context, data):
"""
Tests the vwap transform by manually keeping track of the prices
and volumes in a naiive way and asserting that our hand-rolled vwap is
the same
"""
mins = sum(context.mins_for_days[-context.days:])
for sid in data:
prices = context.price_bars[sid][-mins:]
vols = context.vol_bars[sid][-mins:]
manual_vwap = sum(
map(operator.mul, np.nan_to_num(np.array(prices)), vols),
) / sum(vols)
transformed = list(vwap.transform(self.source))
assert_allclose(
data[sid].vwap(context.days),
manual_vwap,
)
# Output values. Unprocessed custom events will not have a field
# corresponding to the transform hash.
tnfm_vals = [message[vwap.get_hash()] for message in transformed
if message.type != DATASOURCE_TYPE.CUSTOM]
# "Hand calculated" values.
expected = [
(10.0 * 100) / 100.0,
((10.0 * 100) + (10.0 * 100)) / (200.0),
# We should drop the first event here.
((10.0 * 100) + (11.0 * 100)) / (200.0),
# We should drop the second event here.
((11.0 * 100) + (11.0 * 300)) / (400.0)
]
@parameterized.expand(windows_with_frequencies())
@with_algo
def test_returns(context, data):
for sid in data:
last_close = context.last_close_prices[sid]
returns = (data[sid].price - last_close) / last_close
# Output should match the expected.
self.assertEquals(tnfm_vals, expected)
@parameterized.expand([
('with_custom', True),
('without_custom', False),
])
def test_returns(self, name, add_custom_events):
# Daily returns.
returns = Returns(1)
if add_custom_events:
self.source = self.intersperse_custom_events(self.source)
transformed = list(returns.transform(self.source))
tnfm_vals = [message[returns.get_hash()] for message in transformed
if message.type != DATASOURCE_TYPE.CUSTOM]
# No returns for the first event because we don't have a
# previous close.
expected = [0.0, 0.0, 0.1, 0.0]
self.assertEquals(tnfm_vals, expected)
# Two-day returns. An extra kink here is that the
# factory will automatically skip a weekend for the
# last event. Results shouldn't notice this blip.
trade_history = factory.create_trade_history(
133,
[10.0, 15.0, 13.0, 12.0, 13.0],
[100, 100, 100, 300, 100],
timedelta(days=1),
self.sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
returns = StatefulTransform(Returns, 2)
transformed = list(returns.transform(self.source))
tnfm_vals = [message[returns.get_hash()] for message in transformed]
expected = [
0.0,
0.0,
(13.0 - 10.0) / 10.0,
(12.0 - 15.0) / 15.0,
(13.0 - 13.0) / 13.0
]
self.assertEquals(tnfm_vals, expected)
@parameterized.expand([
('with_custom', True),
('without_custom', False),
])
def test_moving_average(self, name, add_custom_events):
mavg = MovingAverage(
market_aware=True,
fields=['price', 'volume'],
window_length=2
)
if add_custom_events:
self.source = self.intersperse_custom_events(self.source)
transformed = list(mavg.transform(self.source))
# Output values.
tnfm_prices = [message[mavg.get_hash()].price
for message in transformed
if message.type != DATASOURCE_TYPE.CUSTOM]
tnfm_volumes = [message[mavg.get_hash()].volume
for message in transformed
if message.type != DATASOURCE_TYPE.CUSTOM]
# "Hand-calculated" values
expected_prices = [
((10.0) / 1.0),
((10.0 + 10.0) / 2.0),
# First event should get dropped here.
((10.0 + 11.0) / 2.0),
# Second event should get dropped here.
((11.0 + 11.0) / 2.0)
]
expected_volumes = [
((100.0) / 1.0),
((100.0 + 100.0) / 2.0),
# First event should get dropped here.
((100.0 + 100.0) / 2.0),
# Second event should get dropped here.
((100.0 + 300.0) / 2.0)
]
self.assertEquals(tnfm_prices, expected_prices)
self.assertEquals(tnfm_volumes, expected_volumes)
@parameterized.expand([
('with_custom', True),
('without_custom', False),
])
def test_moving_stddev(self, name, add_custom_events):
trade_history = factory.create_trade_history(
133,
[10.0, 15.0, 13.0, 12.0],
[100, 100, 100, 100],
timedelta(days=1),
self.sim_params
)
stddev = MovingStandardDev(
market_aware=True,
window_length=3,
)
self.source = SpecificEquityTrades(event_list=trade_history)
if add_custom_events:
self.source = self.intersperse_custom_events(self.source)
transformed = list(stddev.transform(self.source))
vals = [message[stddev.get_hash()] for message in transformed
if message.type != DATASOURCE_TYPE.CUSTOM]
expected = [
None,
np.std([10.0, 15.0], ddof=1),
np.std([10.0, 15.0, 13.0], ddof=1),
np.std([15.0, 13.0, 12.0], ddof=1),
]
# np has odd rounding behavior, cf.
# http://docs.scipy.org/doc/np/reference/generated/np.std.html
for v1, v2 in zip(vals, expected):
if v1 is None:
self.assertIsNone(v2)
continue
self.assertEquals(round(v1, 5), round(v2, 5))
assert_allclose(
data[sid].returns(),
returns,
)
zipline/algorithm.py
+35 -14
View File
@@ -505,23 +505,44 @@ class TradingAlgorithm(object):
return daily_stats
def add_transform(self, transform_class, tag, *args, **kwargs):
"""Add a single-sid, sequential transform to the model.
@api_method
def add_transform(self, transform, days=None):
"""
Ensures that the history container will have enough size to service
a simple transform.
:Arguments:
transform_class : class
Which transform to use. E.g. mavg.
tag : str
How to name the transform. Can later be access via:
data[sid].tag()
Extra args and kwargs will be forwarded to the transform
instantiation.
transform : string
The transform to add. must be an element of:
{'mavg', 'stddev', 'vwap', 'returns'}.
days : int <default=None>
The maximum amount of days you will want for this transform.
This is not needed for 'returns'.
"""
self.registered_transforms[tag] = {'class': transform_class,
'args': args,
'kwargs': kwargs}
if transform not in {'mavg', 'stddev', 'vwap', 'returns'}:
raise ValueError('Invalid transform')
if transform == 'returns':
if days is not None:
raise ValueError('returns does use days')
self.add_history(2, '1d', 'price')
return
elif days is None:
raise ValueError('no number of days specified')
if self.sim_params.data_frequency == 'daily':
mult = 1
freq = '1d'
else:
mult = 390
freq = '1m'
bars = mult * days
self.add_history(bars, freq, 'price')
if transform == 'vwap':
self.add_history(bars, freq, 'volume')
@api_method
def get_environment(self):
zipline/examples/olmar.py
+73 -77
View File
@@ -5,7 +5,6 @@ from datetime import datetime
import pytz
from zipline.algorithm import TradingAlgorithm
from zipline.transforms import MovingAverage
from zipline.utils.factory import load_from_yahoo
from zipline.finance import commission
@@ -19,99 +18,97 @@ zipline_logging.push_application()
STOCKS = ['AMD', 'CERN', 'COST', 'DELL', 'GPS', 'INTC', 'MMM']
class OLMAR(TradingAlgorithm):
"""
On-Line Portfolio Moving Average Reversion
# On-Line Portfolio Moving Average Reversion
More info can be found in the corresponding paper:
http://icml.cc/2012/papers/168.pdf
"""
def initialize(self, eps=1, window_length=5):
self.stocks = STOCKS
self.m = len(self.stocks)
self.price = {}
self.b_t = np.ones(self.m) / self.m
self.last_desired_port = np.ones(self.m) / self.m
self.eps = eps
self.init = True
self.days = 0
self.window_length = window_length
self.add_transform(MovingAverage, 'mavg', ['price'],
window_length=window_length)
# More info can be found in the corresponding paper:
# http://icml.cc/2012/papers/168.pdf
def initialize(algo, eps=1, window_length=5):
algo.stocks = STOCKS
algo.m = len(algo.stocks)
algo.price = {}
algo.b_t = np.ones(algo.m) / algo.m
algo.last_desired_port = np.ones(algo.m) / algo.m
algo.eps = eps
algo.init = True
algo.days = 0
algo.window_length = window_length
algo.add_transform('mavg', 5)
self.set_commission(commission.PerShare(cost=0))
algo.set_commission(commission.PerShare(cost=0))
def handle_data(self, data):
self.days += 1
if self.days < self.window_length:
return
if self.init:
self.rebalance_portfolio(data, self.b_t)
self.init = False
return
def handle_data(algo, data):
algo.days += 1
if algo.days < algo.window_length:
return
m = self.m
if algo.init:
rebalance_portfolio(algo, data, algo.b_t)
algo.init = False
return
x_tilde = np.zeros(m)
b = np.zeros(m)
m = algo.m
# find relative moving average price for each security
for i, stock in enumerate(self.stocks):
price = data[stock].price
# Relative mean deviation
x_tilde[i] = data[stock]['mavg']['price'] / price
x_tilde = np.zeros(m)
b = np.zeros(m)
###########################
# Inside of OLMAR (algo 2)
x_bar = x_tilde.mean()
# find relative moving average price for each security
for i, stock in enumerate(algo.stocks):
price = data[stock].price
# Relative mean deviation
x_tilde[i] = data[stock].mavg(algo.window_length) / price
# market relative deviation
mark_rel_dev = x_tilde - x_bar
###########################
# Inside of OLMAR (algo 2)
x_bar = x_tilde.mean()
# Expected return with current portfolio
exp_return = np.dot(self.b_t, x_tilde)
weight = self.eps - exp_return
variability = (np.linalg.norm(mark_rel_dev)) ** 2
# market relative deviation
mark_rel_dev = x_tilde - x_bar
# test for divide-by-zero case
if variability == 0.0:
step_size = 0
else:
step_size = max(0, weight / variability)
# Expected return with current portfolio
exp_return = np.dot(algo.b_t, x_tilde)
weight = algo.eps - exp_return
variability = (np.linalg.norm(mark_rel_dev)) ** 2
b = self.b_t + step_size * mark_rel_dev
b_norm = simplex_projection(b)
np.testing.assert_almost_equal(b_norm.sum(), 1)
# test for divide-by-zero case
if variability == 0.0:
step_size = 0
else:
step_size = max(0, weight / variability)
self.rebalance_portfolio(data, b_norm)
b = algo.b_t + step_size * mark_rel_dev
b_norm = simplex_projection(b)
np.testing.assert_almost_equal(b_norm.sum(), 1)
# update portfolio
self.b_t = b_norm
rebalance_portfolio(algo, data, b_norm)
def rebalance_portfolio(self, data, desired_port):
# rebalance portfolio
desired_amount = np.zeros_like(desired_port)
current_amount = np.zeros_like(desired_port)
prices = np.zeros_like(desired_port)
# update portfolio
algo.b_t = b_norm
if self.init:
positions_value = self.portfolio.starting_cash
else:
positions_value = self.portfolio.positions_value + \
self.portfolio.cash
for i, stock in enumerate(self.stocks):
current_amount[i] = self.portfolio.positions[stock].amount
prices[i] = data[stock].price
def rebalance_portfolio(algo, data, desired_port):
# rebalance portfolio
desired_amount = np.zeros_like(desired_port)
current_amount = np.zeros_like(desired_port)
prices = np.zeros_like(desired_port)
desired_amount = np.round(desired_port * positions_value / prices)
if algo.init:
positions_value = algo.portfolio.starting_cash
else:
positions_value = algo.portfolio.positions_value + \
algo.portfolio.cash
self.last_desired_port = desired_port
diff_amount = desired_amount - current_amount
for i, stock in enumerate(algo.stocks):
current_amount[i] = algo.portfolio.positions[stock].amount
prices[i] = data[stock].price
for i, stock in enumerate(self.stocks):
self.order(stock, diff_amount[i])
desired_amount = np.round(desired_port * positions_value / prices)
algo.last_desired_port = desired_port
diff_amount = desired_amount - current_amount
for i, stock in enumerate(algo.stocks):
algo.order(stock, diff_amount[i])
def simplex_projection(v, b=1):
@@ -155,10 +152,9 @@ if __name__ == '__main__':
import pylab as pl
start = datetime(2004, 1, 1, 0, 0, 0, 0, pytz.utc)
end = datetime(2008, 1, 1, 0, 0, 0, 0, pytz.utc)
data = load_from_yahoo(stocks=STOCKS, indexes={}, start=start,
end=end)
data = load_from_yahoo(stocks=STOCKS, indexes={}, start=start, end=end)
data = data.dropna()
olmar = OLMAR()
olmar = TradingAlgorithm(handle_data=handle_data, initialize=initialize)
results = olmar.run(data)
results.portfolio_value.plot()
pl.show()
zipline/finance/trading.py
-1
View File
@@ -20,7 +20,6 @@ from functools import wraps
import pandas as pd
import numpy as np
from six.moves import reduce
from zipline.data.loader import load_market_data
from zipline.utils import tradingcalendar
zipline/gens/tradesimulation.py
+2 -1
View File
@@ -304,5 +304,6 @@ class AlgorithmSimulator(object):
try:
sid_data = self.current_data[event.sid]
except KeyError:
sid_data = self.current_data[event.sid] = SIDData()
sid_data = self.current_data[event.sid] = SIDData(event.sid)
sid_data.__dict__.update(event.__dict__)
zipline/protocol.py
+114 -3
View File
@@ -18,6 +18,9 @@ import pandas as pd
from . utils.protocol_utils import Enum
from zipline.finance.trading import with_environment
from zipline.utils.algo_instance import get_algo_instance
# Datasource type should completely determine the other fields of a
# message with its type.
DATASOURCE_TYPE = Enum(
@@ -196,10 +199,29 @@ class Positions(dict):
class SIDData(object):
# Cache some data on the class so that this is shared for all instances of
# siddata.
_history_cache_dt = None
_history_cache = {}
_returns_cache_dt = None
_returns_cache = None
_minute_bar_cache_dt = None
_minute_bar_cache = {}
def __init__(self, sid, initial_values=None):
self._sid = sid
self._freqstr = None
# To check if we have data, we use the __len__ which depends on the
# __dict__. Because we are foward defining the attributes needed, we
# need to account for their entrys in the __dict__.
# We will add 1 because we need to account for the _initial_len entry
# itself.
self._initial_len = len(self.__dict__) + 1
def __init__(self, initial_values=None):
if initial_values:
self.__dict__ = initial_values
self.__dict__.update(initial_values)
@property
def datetime(self):
@@ -226,7 +248,7 @@ class SIDData(object):
self.__dict__[name] = value
def __len__(self):
return len(self.__dict__)
return len(self.__dict__) - self._initial_len
def __contains__(self, name):
return name in self.__dict__
@@ -234,6 +256,95 @@ class SIDData(object):
def __repr__(self):
return "SIDData({0})".format(self.__dict__)
def _get_buffer(self, bars, field='price'):
cls = self.__class__
algo = get_algo_instance()
now = algo.datetime
if now != cls._history_cache_dt:
cls._history_cache_dt = now
cls._history_cache = {}
if field not in self._history_cache \
or bars > len(cls._history_cache[field].index):
hst = algo.history(
bars, self._freqstr, field, ffill=True,
)
# Assert that the column holds ints, not security objects.
if not isinstance(self._sid, str):
hst.columns = hst.columns.astype(int)
self._history_cache[field] = hst
return cls._history_cache[field][self._sid][-bars:]
def _get_bars(self, days):
"""
Gets the number of bars needed for the current number of days.
Figures this out based on the algo datafrequency and caches the result.
"""
def daily_get_bars(days):
return days
@with_environment()
def minute_get_bars(days, env=None):
cls = self.__class__
now = get_algo_instance().datetime
if now != cls._minute_bar_cache_dt:
cls._minute_bar_cache_dt = now
cls._minute_bar_cache = {}
if days not in cls._minute_bar_cache:
# Cache this calculation to happen once per bar, even if we
# use another transform with the same number of days.
prev = env.previous_trading_day(now)
ds = env.days_in_range(
env.add_trading_days(-days + 2, prev),
prev,
)
ms = sum(210 if d in env.early_closes else 390 for d in ds)
ms += \
(now - env.get_open_and_close(now)[0]).total_seconds() / 60
cls._minute_bar_cache[days] = ms + 1 # Account for this minute
return cls._minute_bar_cache[days]
if get_algo_instance().sim_params.data_frequency == 'daily':
self._freqstr = '1d'
self._get_bars = daily_get_bars
else:
self._freqstr = '1m'
self._get_bars = minute_get_bars
# Not actually recursive.
return self._get_bars(days)
def mavg(self, days):
return self._get_buffer(self._get_bars(days)).mean()
def stddev(self, days):
return self._get_buffer(self._get_bars(days)).std(ddof=1)
def vwap(self, days):
bars = self._get_bars(days)
prices = self._get_buffer(bars)
vols = self._get_buffer(bars, field='volume')
return (prices * vols).sum() / vols.sum()
def returns(self):
algo = get_algo_instance()
now = algo.datetime
if now != self._returns_cache_dt:
self._returns_cache_dt = now
self._returns_cache = algo.history(2, '1d', 'price', ffill=True)
hst = self._returns_cache[self._sid]
return (hst.iloc[-1] - hst.iloc[0]) / hst.iloc[0]
class BarData(object):
"""
zipline/sources/test_source.py
+1 -1
View File
@@ -31,7 +31,7 @@ from zipline.protocol import (
DATASOURCE_TYPE
)
from zipline.gens.utils import hash_args
from zipline.utils.tradingcalendar import trading_days
from zipline.finance.trading import with_environment
def create_trade(sid, price, amount, datetime, source_id="test_factory"):
zipline/test_algorithms.py
-5
View File
@@ -443,15 +443,10 @@ class SetLongOnlyAlgorithm(TradingAlgorithm):
from zipline.transforms import BatchTransform, batch_transform
from zipline.transforms import MovingAverage
class TestRegisterTransformAlgorithm(TradingAlgorithm):
def initialize(self, *args, **kwargs):
self.add_transform(MovingAverage, 'mavg', ['price'],
market_aware=True,
window_length=2)
self.set_slippage(FixedSlippage())
def handle_data(self, data):
zipline/transforms/__init__.py
-8
View File
@@ -13,17 +13,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from . mavg import MovingAverage
from . stddev import MovingStandardDev
from . vwap import MovingVWAP
from . returns import Returns
from . batch_transform import BatchTransform, batch_transform
__all__ = [
'MovingAverage',
'MovingStandardDev',
'MovingVWAP',
'Returns',
'BatchTransform',
'batch_transform'
]
zipline/transforms/batch_transform.py
+8 -1
View File
@@ -447,7 +447,14 @@ class BatchTransform(object):
# with CUSTOM data events, there may be different fields
# per sid. So the allowable keys are the union of all events.
union = set.union(*sid_keys)
unwanted_fields = set(['portfolio', 'sid', 'dt', 'type', 'source_id'])
unwanted_fields = {
'portfolio',
'sid',
'dt',
'type',
'source_id',
'_initial_len',
}
return union - unwanted_fields
def _get_field_names(self, event):
zipline/transforms/mavg.py
-152
View File
@@ -1,152 +0,0 @@
#
# Copyright 2013 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from collections import defaultdict
from six import string_types, with_metaclass
from zipline.transforms.utils import EventWindow, TransformMeta
class MovingAverage(with_metaclass(TransformMeta)):
"""
Class that maintains a dictionary from sids to
MovingAverageEventWindows. For each sid, we maintain moving
averages over any number of distinct fields (For example, we can
maintain a sid's average volume as well as its average price.)
"""
def __init__(self, fields='price',
market_aware=True, window_length=None, delta=None):
if isinstance(fields, string_types):
fields = [fields]
self.fields = fields
self.market_aware = market_aware
self.delta = delta
self.window_length = window_length
# Market-aware mode only works with full-day windows.
if self.market_aware:
assert self.window_length and not self.delta,\
"Market-aware mode only works with full-day windows."
# Non-market-aware mode requires a timedelta.
else:
assert self.delta and not self.window_length, \
"Non-market-aware mode requires a timedelta."
# No way to pass arguments to the defaultdict factory, so we
# need to define a method to generate the correct EventWindows.
self.sid_windows = defaultdict(self.create_window)
def create_window(self):
"""
Factory method for self.sid_windows.
"""
return MovingAverageEventWindow(
self.fields,
self.market_aware,
self.window_length,
self.delta
)
def update(self, event):
"""
Update the event window for this event's sid. Return a dict
from tracked fields to moving averages.
"""
# This will create a new EventWindow if this is the first
# message for this sid.
window = self.sid_windows[event.sid]
window.update(event)
return window.get_averages()
class Averages(object):
"""
Container for averages.
"""
def __getitem__(self, name):
"""
Allow dictionary lookup.
"""
return self.__dict__[name]
class MovingAverageEventWindow(EventWindow):
"""
Iteratively calculates moving averages for a particular sid over a
given time window. We can maintain averages for arbitrarily many
fields on a single sid. (For example, we might track average
price as well as average volume for a single sid.) The expected
functionality of this class is to be instantiated inside a
MovingAverage transform.
"""
def __init__(self, fields, market_aware, days, delta):
# Call the superclass constructor to set up base EventWindow
# infrastructure.
EventWindow.__init__(self, market_aware, days, delta)
# We maintain a dictionary of totals for each of our tracked
# fields.
self._fields = fields
self.totals = defaultdict(float)
@property
def fields(self):
return self._fields
# Subclass customization for adding new events.
def handle_add(self, event):
# Sanity check on the event.
# Increment our running totals with data from the event.
for field in self.fields:
self.totals[field] += event[field]
# Subclass customization for removing expired events.
def handle_remove(self, event):
# Decrement our running totals with data from the event.
for field in self.fields:
self.totals[field] -= event[field]
def average(self, field):
"""
Calculate the average value of our ticks over a single field.
"""
# Sanity check.
assert field in self.fields
# Averages are None by convention if we have no ticks.
if len(self.ticks) == 0:
return 0.0
# Calculate and return the average. len(self.ticks) is O(1).
else:
return self.totals[field] / len(self.ticks)
def get_averages(self):
"""
Return a dict of all our tracked averages.
"""
out = Averages()
for field in self.fields:
out.__dict__[field] = self.average(field)
return out
zipline/transforms/returns.py
-102
View File
@@ -1,102 +0,0 @@
#
# Copyright 2013 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from zipline.errors import WrongDataForTransform
from zipline.transforms.utils import TransformMeta
from collections import defaultdict, deque
from six import with_metaclass
class Returns(with_metaclass(TransformMeta)):
"""
Class that maintains a dictionary from sids to the sid's
closing price N trading days ago.
"""
def __init__(self, window_length):
self.window_length = window_length
self.mapping = defaultdict(self._create)
def update(self, event):
"""
Update and return the calculated returns for this event's sid.
"""
tracker = self.mapping[event.sid]
tracker.update(event)
return tracker.returns
def _create(self):
return ReturnsFromPriorClose(
self.window_length
)
class ReturnsFromPriorClose(object):
"""
Records the last N closing events for a given security as well as the
last event for the security. When we get an event for a new day, we
treat the last event seen as the close for the previous day.
"""
def __init__(self, window_length):
self.closes = deque()
self.last_event = None
self.returns = 0.0
self.window_length = window_length
def update(self, event):
self.check_required_fields(event)
if self.last_event:
# Day has changed since the last event we saw. Treat
# the last event as the closing price for its day and
# clear out the oldest close if it has expired.
if self.last_event.dt.date() != event.dt.date():
self.closes.append(self.last_event)
# We keep an event for the end of each trading day, so
# if the number of stored events is greater than the
# number of days we want to track, the oldest close
# is expired and should be discarded.
while len(self.closes) > self.window_length:
# Pop the oldest event.
self.closes.popleft()
# We only generate a return value once we've seen enough days
# to give a sensible value. Would be nice if we could query
# db for closes prior to our initial event, but that would
# require giving this transform database creds, which we want
# to avoid.
if len(self.closes) == self.window_length:
last_close = self.closes[0].price
change = event.price - last_close
self.returns = change / last_close
# the current event is now the last_event
self.last_event = event
def check_required_fields(self, event):
"""
We only allow events with a price field to be run through
the returns transform.
"""
if not hasattr(event, 'price'):
raise WrongDataForTransform(
transform="ReturnsEventWindow",
fields=['price'])
zipline/transforms/stddev.py
-117
View File
@@ -1,117 +0,0 @@
#
# Copyright 2014 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from collections import defaultdict
from math import sqrt
from six import with_metaclass
from zipline.transforms.utils import EventWindow, TransformMeta
import zipline.utils.math_utils as zp_math
class MovingStandardDev(with_metaclass(TransformMeta)):
"""
Class that maintains a dictionary from sids to
MovingStandardDevWindows. For each sid, we maintain a the
standard deviation of all events falling within the specified
window.
"""
def __init__(self, market_aware=True, window_length=None, delta=None):
self.market_aware = market_aware
self.delta = delta
self.window_length = window_length
# Market-aware mode only works with full-day windows.
if self.market_aware:
assert self.window_length and not self.delta,\
"Market-aware mode only works with full-day windows."
# Non-market-aware mode requires a timedelta.
else:
assert self.delta and not self.window_length, \
"Non-market-aware mode requires a timedelta."
# No way to pass arguments to the defaultdict factory, so we
# need to define a method to generate the correct EventWindows.
self.sid_windows = defaultdict(self.create_window)
def create_window(self):
"""
Factory method for self.sid_windows.
"""
return MovingStandardDevWindow(
self.market_aware,
self.window_length,
self.delta
)
def update(self, event):
"""
Update the event window for this event's sid. Return a dict
from tracked fields to moving averages.
"""
# This will create a new EventWindow if this is the first
# message for this sid.
window = self.sid_windows[event.sid]
window.update(event)
return window.get_stddev()
class MovingStandardDevWindow(EventWindow):
"""
Iteratively calculates standard deviation for a particular sid
over a given time window. The expected functionality of this
class is to be instantiated inside a MovingStandardDev.
"""
def __init__(self, market_aware=True, window_length=None, delta=None):
# Call the superclass constructor to set up base EventWindow
# infrastructure.
EventWindow.__init__(self, market_aware, window_length, delta)
self.sum = 0.0
self.sum_sqr = 0.0
@property
def fields(self):
return ['price']
def handle_add(self, event):
self.sum += event.price
self.sum_sqr += event.price ** 2
def handle_remove(self, event):
self.sum -= event.price
self.sum_sqr -= event.price ** 2
def get_stddev(self):
# Sample standard deviation is undefined for a single event or
# no events.
if len(self) <= 1:
return None
else:
average = self.sum / len(self)
s_squared = (self.sum_sqr - self.sum * average) \
/ (len(self) - 1)
if zp_math.tolerant_equals(0, s_squared):
return 0.0
stddev = sqrt(s_squared)
return stddev
zipline/transforms/vwap.py
-101
View File
@@ -1,101 +0,0 @@
#
# Copyright 2014 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from collections import defaultdict
from six import with_metaclass
from zipline.transforms.utils import EventWindow, TransformMeta
class MovingVWAP(with_metaclass(TransformMeta)):
"""
Class that maintains a dictionary from sids to VWAPEventWindows.
"""
def __init__(self, market_aware=True, delta=None, window_length=None):
self.market_aware = market_aware
self.delta = delta
self.window_length = window_length
# Market-aware mode only works with full-day windows.
if self.market_aware:
assert self.window_length and not self.delta,\
"Market-aware mode only works with full-day windows."
# Non-market-aware mode requires a timedelta.
else:
assert self.delta and not self.window_length, \
"Non-market-aware mode requires a timedelta."
# No way to pass arguments to the defaultdict factory, so we
# need to define a method to generate the correct EventWindows.
self.sid_windows = defaultdict(self.create_window)
def create_window(self):
"""Factory method for self.sid_windows."""
return VWAPEventWindow(
self.market_aware,
window_length=self.window_length,
delta=self.delta
)
def update(self, event):
"""
Update the event window for this event's sid. Returns the
current vwap for the sid.
"""
# This will create a new EventWindow if this is the first
# message for this sid.
window = self.sid_windows[event.sid]
window.update(event)
return window.get_vwap()
class VWAPEventWindow(EventWindow):
"""
Iteratively maintains a vwap for a single sid over a given
timedelta.
"""
def __init__(self, market_aware=True, window_length=None, delta=None):
EventWindow.__init__(self, market_aware, window_length, delta)
self._fields = ('price', 'volume')
self.flux = 0.0
self.totalvolume = 0.0
@property
def fields(self):
return self._fields
# Subclass customization for adding new events.
def handle_add(self, event):
self.flux += event.volume * event.price
self.totalvolume += event.volume
# Subclass customization for removing expired events.
def handle_remove(self, event):
self.flux -= event.volume * event.price
self.totalvolume -= event.volume
def get_vwap(self):
"""
Return the calculated vwap for this sid.
"""
# By convention, vwap is None if we have no events.
if len(self.ticks) == 0:
return None
else:
return (self.flux / self.totalvolume)
zipline/utils/algo_instance.py
+24
View File
@@ -0,0 +1,24 @@
#
# Copyright 2014 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import threading
context = threading.local()
def get_algo_instance():
return getattr(context, 'algorithm', None)
def set_algo_instance(algo):
context.algorithm = algo
zipline/utils/api_support.py
+2 -11
View File
@@ -14,18 +14,9 @@
# limitations under the License.
from functools import wraps
import zipline.api
import threading
context = threading.local()
def get_algo_instance():
return getattr(context, 'algorithm', None)
def set_algo_instance(algo):
context.algorithm = algo
from zipline.utils.algo_instance import get_algo_instance, set_algo_instance
class ZiplineAPI(object):