From 3c37704a5b97406377173a72b3214f444116d0f4 Mon Sep 17 00:00:00 2001 From: Joe Jevnik Date: Mon, 15 Sep 2014 11:02:04 -0400 Subject: [PATCH] ENH: Adds a new api method `schedule_function`. schedule_function takes a date rule, a time rule, and a function and will call the function, passing context and data only when the two rules fire. This allows for code that is conditional to the datetime of the algo. This is implemented internally with `Event` objects which are pairings of `EventRule`s and callbacks. handle_data becomes a special event with a rule that always fires. This makes the logic for handling events more complete and compact. --- tests/test_algorithm.py | 42 ++- tests/utils/test_argcheck.py | 268 +++++++++++++++++ tests/utils/test_events.py | 357 ++++++++++++++++++++++ tests/utils/test_factory.py | 6 +- zipline/algorithm.py | 48 +++ zipline/api.py | 3 +- zipline/finance/trading.py | 17 +- zipline/gens/tradesimulation.py | 6 +- zipline/utils/argcheck.py | 332 +++++++++++++++++++++ zipline/utils/events.py | 513 ++++++++++++++++++++++++++++++++ 10 files changed, 1585 insertions(+), 7 deletions(-) create mode 100644 tests/utils/test_argcheck.py create mode 100644 tests/utils/test_events.py create mode 100644 zipline/utils/argcheck.py create mode 100644 zipline/utils/events.py diff --git a/tests/test_algorithm.py b/tests/test_algorithm.py index 94c2b0ac..224417d5 100644 --- a/tests/test_algorithm.py +++ b/tests/test_algorithm.py @@ -12,7 +12,7 @@ # 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 datetime from datetime import timedelta from mock import MagicMock from six.moves import range @@ -77,6 +77,7 @@ 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 +from zipline.utils.events import DateRuleFactory, TimeRuleFactory from zipline.algorithm import TradingAlgorithm @@ -174,6 +175,44 @@ class TestMiscellaneousAPI(TestCase): sim_params=self.sim_params) algo.run(self.source) + def test_schedule_function(self): + date_rules = DateRuleFactory + time_rules = TimeRuleFactory + + def incrementer(algo, data): + algo.func_called += 1 + self.assertEqual( + algo.get_datetime().time(), + datetime.time(hour=14, minute=31), + ) + + def initialize(algo): + algo.func_called = 0 + algo.days = 1 + algo.date = None + algo.schedule_function( + func=incrementer, + date_rule=date_rules.every_day(), + time_rule=time_rules.market_open(), + ) + + def handle_data(algo, data): + if not algo.date: + algo.date = algo.get_datetime().date() + + if algo.date < algo.get_datetime().date(): + algo.days += 1 + algo.date = algo.get_datetime().date() + + algo = TradingAlgorithm( + initialize=initialize, + handle_data=handle_data, + sim_params=self.sim_params, + ) + algo.run(self.source) + + self.assertEqual(algo.func_called, algo.days) + class TestTransformAlgorithm(TestCase): def setUp(self): @@ -840,7 +879,6 @@ class TestTradingControls(TestCase): self.check_algo_succeeds(algo, handle_data, order_count=20) def test_long_only(self): - # Sell immediately -> fail immediately. def handle_data(algo, data): algo.order(self.sid, -1) diff --git a/tests/utils/test_argcheck.py b/tests/utils/test_argcheck.py new file mode 100644 index 00000000..c357518c --- /dev/null +++ b/tests/utils/test_argcheck.py @@ -0,0 +1,268 @@ +# +# 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 unittest import TestCase + +from zipline.utils.argcheck import ( + verify_callable_argspec, + Argument, + NoStarargs, + UnexpectedStarargs, + NoKwargs, + UnexpectedKwargs, + NotCallable, + NotEnoughArguments, + TooManyArguments, + MismatchedArguments, +) + + +class TestArgCheck(TestCase): + def test_not_callable(self): + """ + Check the results of a non-callable object. + """ + not_callable = 'a' + + with self.assertRaises(NotCallable): + verify_callable_argspec(not_callable) + + def test_no_starargs(self): + """ + Tests when a function does not have *args and it was expected. + """ + def f(a): + pass + + with self.assertRaises(NoStarargs): + verify_callable_argspec(f, expect_starargs=True) + + def test_starargs(self): + """ + Tests when a function has *args and it was expected. + """ + def f(*args): + pass + + verify_callable_argspec(f, expect_starargs=True) + + def test_unexcpected_starargs(self): + """ + Tests a function that unexpectedly accepts *args. + """ + def f(*args): + pass + + with self.assertRaises(UnexpectedStarargs): + verify_callable_argspec(f, expect_starargs=False) + + def test_ignore_starargs(self): + """ + Tests checking a function ignoring the presence of *args. + """ + def f(*args): + pass + + def g(): + pass + + verify_callable_argspec(f, expect_starargs=Argument.ignore) + verify_callable_argspec(g, expect_starargs=Argument.ignore) + + def test_no_kwargs(self): + """ + Tests when a function does not have **kwargs and it was expected. + """ + def f(): + pass + + with self.assertRaises(NoKwargs): + verify_callable_argspec(f, expect_kwargs=True) + + def test_kwargs(self): + """ + Tests when a function has **kwargs and it was expected. + """ + def f(**kwargs): + pass + + verify_callable_argspec(f, expect_kwargs=True) + + def test_unexpected_kwargs(self): + """ + Tests a function that unexpectedly accepts **kwargs. + """ + def f(**kwargs): + pass + + with self.assertRaises(UnexpectedKwargs): + verify_callable_argspec(f, expect_kwargs=False) + + def test_ignore_kwargs(self): + """ + Tests checking a function ignoring the presence of **kwargs. + """ + def f(**kwargs): + pass + + def g(): + pass + + verify_callable_argspec(f, expect_kwargs=Argument.ignore) + verify_callable_argspec(g, expect_kwargs=Argument.ignore) + + def test_arg_subset(self): + """ + Tests when the args are a subset of the expectations. + """ + def f(a, b): + pass + + with self.assertRaises(NotEnoughArguments): + verify_callable_argspec( + f, [Argument('a'), Argument('b'), Argument('c')] + ) + + def test_arg_superset(self): + def f(a, b, c): + pass + + with self.assertRaises(TooManyArguments): + verify_callable_argspec(f, [Argument('a'), Argument('b')]) + + def test_no_default(self): + """ + Tests when an argument expects a default and it is not present. + """ + def f(a): + pass + + with self.assertRaises(MismatchedArguments): + verify_callable_argspec(f, [Argument('a', 1)]) + + def test_default(self): + """ + Tests when an argument expects a default and it is present. + """ + def f(a=1): + pass + + verify_callable_argspec(f, [Argument('a', 1)]) + + def test_ignore_default(self): + """ + Tests that ignoring defaults works as intended. + """ + def f(a=1): + pass + + verify_callable_argspec(f, [Argument('a')]) + + def test_mismatched_args(self): + def f(a, b): + pass + + with self.assertRaises(MismatchedArguments): + verify_callable_argspec(f, [Argument('c'), Argument('d')]) + + def test_ignore_args(self): + """ + Tests the ignore argument list feature. + """ + def f(a): + pass + + def g(): + pass + + h = 'not_callable' + + verify_callable_argspec(f) + verify_callable_argspec(g) + with self.assertRaises(NotCallable): + verify_callable_argspec(h) + + def test_out_of_order(self): + """ + Tests the case where arguments are not in the correct order. + """ + def f(a, b): + pass + + with self.assertRaises(MismatchedArguments): + verify_callable_argspec(f, [Argument('b'), Argument('a')]) + + def test_wrong_default(self): + """ + Tests the case where a default is expected, but the default provided + does not match the one expected. + """ + def f(a=1): + pass + + with self.assertRaises(MismatchedArguments): + verify_callable_argspec(f, [Argument('a', 2)]) + + def test_any_default(self): + """ + Tests the any_default option. + """ + def f(a=1): + pass + + def g(a=2): + pass + + def h(a): + pass + + expected_args = [Argument('a', Argument.any_default)] + verify_callable_argspec(f, expected_args) + verify_callable_argspec(g, expected_args) + with self.assertRaises(MismatchedArguments): + verify_callable_argspec(h, expected_args) + + def test_ignore_name(self): + """ + Tests ignoring a param name. + """ + def f(a): + pass + + def g(b): + pass + + def h(c=1): + pass + + expected_args = [Argument(Argument.ignore, Argument.no_default)] + verify_callable_argspec(f, expected_args) + verify_callable_argspec(f, expected_args) + with self.assertRaises(MismatchedArguments): + verify_callable_argspec(h, expected_args) + + def test_bound_method(self): + class C(object): + def f(self, a, b): + pass + + method = C().f + + verify_callable_argspec(method, [Argument('a'), Argument('b')]) + with self.assertRaises(NotEnoughArguments): + # Assert that we don't count self. + verify_callable_argspec( + method, + [Argument('self'), Argument('a'), Argument('b')], + ) diff --git a/tests/utils/test_events.py b/tests/utils/test_events.py new file mode 100644 index 00000000..92c5017b --- /dev/null +++ b/tests/utils/test_events.py @@ -0,0 +1,357 @@ +# +# 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 datetime +import random +from itertools import islice, dropwhile +from six.moves import range, map +from nose_parameterized import parameterized +from unittest import TestCase + +import numpy as np + +from zipline.finance.trading import TradingEnvironment +import zipline.utils.events +from zipline.utils.events import ( + EventRule, + StatelessRule, + Always, + Never, + AfterOpen, + ComposedRule, + BeforeClose, + NotHalfDay, + NthTradingDayOfWeek, + NDaysBeforeLastTradingDayOfWeek, + NthTradingDayOfMonth, + NDaysBeforeLastTradingDayOfMonth, + StatefulRule, + OncePerDay, + _build_offset, + _build_date, + _build_time, + EventManager, + Event, +) + + +# A day known to be a half day. +HALF_DAY = datetime.date(year=2014, month=7, day=3) + +# A day known to be a full day. +FULL_DAY = datetime.date(year=2014, month=9, day=24) + + +def param_range(*args): + return ([n] for n in range(*args)) + + +class TestUtils(TestCase): + @parameterized.expand([ + ('_build_date', _build_date), + ('_build_time', _build_time), + ]) + def test_build_none(self, name, f): + with self.assertRaises(ValueError): + f(None, {}) + + def test_build_offset_both(self): + with self.assertRaises(ValueError): + _build_offset(datetime.timedelta(minutes=1), {'minutes': 1}) + + def test_build_offset_kwargs(self): + kwargs = {'minutes': 1} + self.assertEqual( + _build_offset(None, kwargs), + datetime.timedelta(**kwargs), + ) + + def test_build_offset_td(self): + td = datetime.timedelta(minutes=1) + self.assertEqual( + _build_offset(td, {}), + td, + ) + + def test_build_date_both(self): + with self.assertRaises(ValueError): + _build_date( + datetime.date(year=2014, month=9, day=25), { + 'year': 2014, + 'month': 9, + 'day': 25, + }, + ) + + def test_build_date_kwargs(self): + kwargs = {'year': 2014, 'month': 9, 'day': 25} + self.assertEqual( + _build_date(None, kwargs), + datetime.date(**kwargs), + ) + + def test_build_date_date(self): + date = datetime.date(year=2014, month=9, day=25) + self.assertEqual( + _build_date(date, {}), + date, + ) + + def test_build_time_both(self): + with self.assertRaises(ValueError): + _build_time( + datetime.time(hour=1, minute=5), { + 'hour': 1, + 'minute': 5, + }, + ) + + def test_build_time_kwargs(self): + kwargs = {'hour': 1, 'minute': 5} + self.assertEqual( + _build_time(None, kwargs), + datetime.time(**kwargs), + ) + + +class TestEventManager(TestCase): + def setUp(self): + self.em = EventManager() + self.event1 = Event(Always(), lambda context, data: None) + self.event2 = Event(Always(), lambda context, data: None) + + def test_add_event(self): + self.em.add_event(self.event1) + self.assertEqual(len(self.em._events), 1) + + def test_add_event_prepend(self): + self.em.add_event(self.event1) + self.em.add_event(self.event2, prepend=True) + self.assertEqual([self.event2, self.event1], self.em._events) + + def test_add_event_append(self): + self.em.add_event(self.event1) + self.em.add_event(self.event2) + self.assertEqual([self.event1, self.event2], self.em._events) + + def test_checks_should_trigger(self): + class CountingRule(Always): + count = 0 + + def should_trigger(self, dt): + CountingRule.count += 1 + return True + + for r in [CountingRule] * 5: + self.em.add_event( + Event(r(), lambda context, data: None, check_args=False) + ) + + self.em.handle_data(None, None, datetime.datetime.now()) + + self.assertEqual(CountingRule.count, 5) + + +class TestEventRule(TestCase): + def test_is_abstract(self): + with self.assertRaises(TypeError): + EventRule() + + def test_not_implemented(self): + with self.assertRaises(NotImplementedError): + super(Always, Always()).should_trigger('a') + + +class RuleTestCase(TestCase): + @classmethod + def setUpClass(cls): + cls.env = TradingEnvironment.instance() + cls.class_ = None # Mark that this is the base class. + + def setUp(self): + # Select a random sample of 5 trading days + self.trading_days = self._get_random_days(5) + + def _get_random_days(self, n): + """ + Returns a random selection n trading days. + """ + index = random.sample(range(len(self.env.trading_days)), n) + test_dts = (self.env.trading_days[i] for i in index) + return (self.env.market_minutes_for_day(dt) for dt in test_dts) + + @property + def minutes(self): + for d in self.trading_days: + for m in d: + yield m.to_datetime() + + def test_completeness(self): + """ + Tests that all rules are being tested. + """ + if not self.class_: + return # This is the base class testing, it is always complete. + + dem = { + k for k, v in vars(zipline.utils.events).iteritems() + if isinstance(v, type) + and issubclass(v, self.class_) + and v is not self.class_ + } + ds = { + k[5:] for k in dir(self) + if k.startswith('test') and k[5:] in dem + } + self.assertTrue( + dem <= ds, + msg='This suite is missing tests for the following classes:\n' + + '\n'.join(map(repr, dem - ds)), + ) + + +class TestStatelessRules(RuleTestCase): + @classmethod + def setUpClass(cls): + super(TestStatelessRules, cls).setUpClass() + + cls.class_ = StatelessRule + + cls.sept_days = cls.env.days_in_range( + np.datetime64(datetime.date(year=2014, month=9, day=1)), + np.datetime64(datetime.date(year=2014, month=9, day=30)), + ) + + cls.sept_week = cls.env.minutes_for_days_in_range( + datetime.date(year=2014, month=9, day=21), + datetime.date(year=2014, month=9, day=26), + ) + + def test_Always(self): + should_trigger = Always().should_trigger + self.assertTrue(all(map(should_trigger, self.minutes))) + + def test_Never(self): + should_trigger = Never().should_trigger + self.assertFalse(any(map(should_trigger, self.minutes))) + + def test_AfterOpen(self): + should_trigger = AfterOpen(minutes=5, hours=1).should_trigger + for d in self.trading_days: + for m in islice(d, 65): + self.assertFalse(should_trigger(m)) + for m in islice(d, 65, None): + self.assertTrue(should_trigger(m)) + + def test_BeforeClose(self): + should_trigger = BeforeClose(hours=1, minutes=5).should_trigger + for d in self.trading_days: + for m in d[0:-65]: + self.assertFalse(should_trigger(m)) + for m in d[-65:]: + self.assertTrue(should_trigger(m)) + + def test_NotHalfDay(self): + should_trigger = NotHalfDay().should_trigger + self.assertTrue(should_trigger(FULL_DAY)) + self.assertFalse(should_trigger(HALF_DAY)) + + @parameterized.expand(param_range(5)) + def test_NthTradingDayOfWeek(self, n): + should_trigger = NthTradingDayOfWeek(n).should_trigger + prev_day = None + n_tdays = 0 + for m in dropwhile(lambda n: not should_trigger(n), self.sept_week): + if should_trigger(m): + self.assertEqual(n_tdays, n) + else: + self.assertNotEqual(n_tdays, n) + + if not prev_day or prev_day < m.date(): + n_tdays += 1 + prev_day = m.date() + + @parameterized.expand(param_range(5)) + def test_NDaysBeforeLastTradingDayOfWeek(self, n): + should_trigger = NDaysBeforeLastTradingDayOfWeek(n).should_trigger + for m in self.sept_week: + if should_trigger(m): + n_tdays = 0 + date = m.to_datetime().date() + next_date = self.env.next_trading_day(date) + while next_date.day > date.day: + date = next_date + next_date = self.env.next_trading_day(date) + n_tdays += 1 + + self.assertEqual(n_tdays, n) + + @parameterized.expand(param_range(30)) + def test_NthTradingDayOfMonth(self, n): + should_trigger = NthTradingDayOfMonth(n).should_trigger + for n_tdays, d in enumerate(self.sept_days): + for m in self.env.market_minutes_for_day(d): + if should_trigger(m): + self.assertEqual(n_tdays, n) + else: + self.assertNotEqual(n_tdays, n) + + @parameterized.expand(param_range(30)) + def test_NDaysBeforeLastTradingDayOfMonth(self, n): + should_trigger = NDaysBeforeLastTradingDayOfMonth(n).should_trigger + for n_days_before, d in enumerate(reversed(self.sept_days)): + for m in self.env.market_minutes_for_day(d): + if should_trigger(m): + self.assertEqual(n_days_before, n) + else: + self.assertNotEqual(n_days_before, n) + + def test_ComposedRule(self): + rule1 = Always() + rule2 = Never() + + composed = rule1 & rule2 + self.assertIsInstance(composed, ComposedRule) + self.assertIs(composed.first, rule1) + self.assertIs(composed.second, rule2) + self.assertFalse(any(map(composed.should_trigger, self.minutes))) + + +class TestStatefulRules(RuleTestCase): + @classmethod + def setUpClass(cls): + super(TestStatefulRules, cls).setUpClass() + + cls.class_ = StatefulRule + + def test_OncePerDay(self): + class RuleCounter(StatefulRule): + """ + A rule that counts the number of times another rule triggers + but forwards the results out. + """ + count = 0 + + def should_trigger(self, dt): + st = self.rule.should_trigger(dt) + if st: + self.count += 1 + return st + + rule = RuleCounter(OncePerDay()) + for m in self.minutes: + rule.should_trigger(m) + + # We are only using 5 trading days. + self.assertEqual(rule.count, 5) diff --git a/tests/utils/test_factory.py b/tests/utils/test_factory.py index cac242dd..828c31a9 100644 --- a/tests/utils/test_factory.py +++ b/tests/utils/test_factory.py @@ -14,12 +14,14 @@ # limitations under the License. from unittest import TestCase -from zipline.utils.factory import (load_from_yahoo, - load_bars_from_yahoo) + import pandas as pd import pytz import numpy as np +from zipline.utils.factory import (load_from_yahoo, + load_bars_from_yahoo) + class TestFactory(TestCase): def test_load_from_yahoo(self): diff --git a/zipline/algorithm.py b/zipline/algorithm.py index 125af57f..57de4c91 100644 --- a/zipline/algorithm.py +++ b/zipline/algorithm.py @@ -65,6 +65,13 @@ from zipline.gens.tradesimulation import AlgorithmSimulator from zipline.sources import DataFrameSource, DataPanelSource from zipline.transforms.utils import StatefulTransform from zipline.utils.api_support import ZiplineAPI, api_method +import zipline.utils.events +from zipline.utils.events import ( + EventManager, + make_eventrule, + DateRuleFactory, + TimeRuleFactory, +) from zipline.utils.factory import create_simulation_parameters import zipline.protocol @@ -181,6 +188,8 @@ class TradingAlgorithm(object): self._before_trading_start = None self._analyze = None + self.event_manager = EventManager() + if self.algoscript is not None: exec_(self.algoscript, self.namespace) self._initialize = self.namespace.get('initialize') @@ -203,6 +212,17 @@ class TradingAlgorithm(object): self._before_trading_start = kwargs.pop('before_trading_start', None) + self.event_manager.add_event( + zipline.utils.events.Event( + zipline.utils.events.Always(), + # We pass handle_data.__func__ to get the unbound method. + # We will explicitly pass the algorithm to bind it again. + self.handle_data.__func__, + check_args=False, + ), + prepend=True, + ) + # If method not defined, NOOP if self._initialize is None: self._initialize = lambda x: None @@ -492,6 +512,34 @@ class TradingAlgorithm(object): def get_environment(self): return self._environment + def add_event(self, rule=None, callback=None, check_args=True): + """ + Adds an event to the algorithm's EventManager. + """ + self.event_manager.add_event( + zipline.utils.events.Event(rule, callback, check_args=check_args), + ) + + @api_method + def schedule_function(self, + func, + date_rule=None, + time_rule=None, + half_days=True, + check_args=False): + """ + Schedules a function to be called with some timed rules. + """ + # Defaults to every day 30 minutes before close. + date_rule = date_rule or DateRuleFactory.every_day() + time_rule = time_rule or TimeRuleFactory.market_close(minutes=30) + + self.add_event( + make_eventrule(date_rule, time_rule, half_days), + func, + check_args=check_args, + ) + @api_method def record(self, *args, **kwargs): """ diff --git a/zipline/api.py b/zipline/api.py index 97436932..017efa76 100644 --- a/zipline/api.py +++ b/zipline/api.py @@ -19,7 +19,7 @@ import zipline from .finance import (commission, slippage) -from .utils import math_utils +from .utils import math_utils, events from zipline.finance.slippage import ( FixedSlippage, @@ -33,6 +33,7 @@ batch_transform = zipline.transforms.BatchTransform __all__ = [ 'slippage', 'commission', + 'events', 'math_utils', 'batch_transform', 'FixedSlippage', diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 61c92371..fb31d6e6 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -19,6 +19,7 @@ import datetime 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 @@ -179,13 +180,27 @@ class TradingEnvironment(object): dt = self.normalize_date(test_date) delta = datetime.timedelta(days=-1) - while self.first_trading_day < test_date: + while self.first_trading_day < dt: dt += delta if dt in self.trading_days: return dt return None + def add_trading_days(self, n, date): + if n > 0: + return reduce( + lambda a, b: self.next_trading_day(a), + range(n), + date, + ) + else: + return reduce( + lambda a, b: self.previous_trading_day(a), + range(abs(n)), + date, + ) + def days_in_range(self, start, end): mask = ((self.trading_days >= start) & (self.trading_days <= end)) diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 172c39c3..36230d20 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -250,7 +250,11 @@ class AlgorithmSimulator(object): Call the user's handle_data, returning any orders placed by the algo during the call. """ - self.algo.handle_data(self.current_data) + self.algo.event_manager.handle_data( + self.algo, + self.current_data, + self.simulation_dt, + ) orders = self.algo.blotter.new_orders self.algo.blotter.new_orders = [] return orders diff --git a/zipline/utils/argcheck.py b/zipline/utils/argcheck.py new file mode 100644 index 00000000..b0320d46 --- /dev/null +++ b/zipline/utils/argcheck.py @@ -0,0 +1,332 @@ +# +# 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 namedtuple +import inspect +from itertools import chain +from six.moves import map, zip_longest + +from zipline.errors import ZiplineError + + +Argspec = namedtuple('Argspec', ['args', 'starargs', 'kwargs']) + + +def singleton(cls): + instances = {} + + def getinstance(): + if cls not in instances: + instances[cls] = cls() + return instances[cls] + + return getinstance + + +@singleton +class Ignore(object): + def __str__(self): + return 'Argument.ignore' + __repr__ = __str__ + + +@singleton +class NoDefault(object): + def __str__(self): + return 'Argument.no_default' + __repr__ = __str__ + + +@singleton +class AnyDefault(object): + def __str__(self): + return 'Argument.any_default' + __repr__ = __str__ + + +class Argument(namedtuple('Argument', ['name', 'default'])): + """ + An argument to a function. + Argument.no_default is a value representing no default to the argument. + Argument.ignore is a value that says you should ignore the default value. + """ + no_default = NoDefault() + any_default = AnyDefault() + ignore = Ignore() + + def __new__(cls, name=ignore, default=ignore): + return super(Argument, cls).__new__(cls, name, default) + + def __str__(self): + if self.has_no_default(self) or self.ignore_default(self): + return str(self.name) + else: + return '='.join([str(self.name), str(self.default)]) + + def __repr__(self): + return 'Argument(%s, %s)' % (repr(self.name), repr(self.default)) + + def _defaults_match(self, arg): + return any(map(Argument.ignore_default, [self, arg])) \ + or (self.default is Argument.any_default + and arg.default is not Argument.no_default) \ + or (arg.default is Argument.any_default + and self.default is not Argument.no_default) \ + or self.default == arg.default + + def _names_match(self, arg): + return self.name == arg.name \ + or self.name is Argument.ignore \ + or arg.name is Argument.ignore + + def matches(self, arg): + return self._names_match(arg) and self._defaults_match(arg) + __eq__ = matches + + @staticmethod + def parse_argspec(callable_): + """ + Takes a callable and returns a tuple with the list of Argument objects, + the name of *args, and the name of **kwargs. + If *args or **kwargs is not present, it will be None. + This returns a namedtuple called Argspec that has three fields named: + args, starargs, and kwargs. + """ + args, varargs, keywords, defaults = inspect.getargspec(callable_) + defaults = list(defaults or []) + + if getattr(callable_, '__self__', None) is not None: + # This is a bound method, drop the self param. + args = args[1:] + + first_default = len(args) - len(defaults) + return Argspec( + [Argument(arg, Argument.no_default + if n < first_default else defaults[n - first_default]) + for n, arg in enumerate(args)], + varargs, + keywords, + ) + + @staticmethod + def has_no_default(arg): + return arg.default is Argument.no_default + + @staticmethod + def ignore_default(arg): + return arg.default is Argument.ignore + + +def _expect_extra(expected, present, exc_unexpected, exc_missing, exc_args): + """ + Checks for the presence of an extra to the argument list. Raises expections + if this is unexpected or if it is missing and expected. + """ + if present: + if not expected: + raise exc_unexpected(*exc_args) + elif expected and expected is not Argument.ignore: + raise exc_missing(*exc_args) + + +def verify_callable_argspec(callable_, + expected_args=Argument.ignore, + expect_starargs=Argument.ignore, + expect_kwargs=Argument.ignore): + """ + Checks the callable_ to make sure that it satisfies the given + expectations. + expected_args should be an iterable of Arguments in the order you expect to + receive them. + expect_starargs means that the function should or should not take a *args + param. expect_kwargs says the callable should or should not take **kwargs + param. + If expected_args, expect_starargs, or expect_kwargs is Argument.ignore, + then the checks related to that argument will not occur. + + Example usage: + + callable_check( + f, + [Argument('a'), Argument('b', 1)], + expect_starargs=True, + expect_kwargs=Argument.ignore + ) + """ + if not callable(callable_): + raise NotCallable(callable_) + + expected_arg_list = list( + expected_args if expected_args is not Argument.ignore else [] + ) + + args, starargs, kwargs = Argument.parse_argspec(callable_) + + exc_args = callable_, args, starargs, kwargs + + # Check the *args. + _expect_extra( + expect_starargs, + starargs, + UnexpectedStarargs, + NoStarargs, + exc_args, + ) + # Check the **kwargs. + _expect_extra( + expect_kwargs, + kwargs, + UnexpectedKwargs, + NoKwargs, + exc_args, + ) + + if expected_args is Argument.ignore: + # Ignore the argument list checks. + return + + if len(args) < len(expected_arg_list): + # One or more argument that we expected was not present. + raise NotEnoughArguments( + callable_, + args, + starargs, + kwargs, + [arg for arg in expected_arg_list if arg not in args], + ) + elif len(args) > len(expected_arg_list): + raise TooManyArguments( + callable_, args, starargs, kwargs + ) + + # Empty argument that will not match with any actual arguments. + missing_arg = Argument(object(), object()) + + for expected, provided in zip_longest(expected_arg_list, + args, + fillvalue=missing_arg): + if not expected.matches(provided): + raise MismatchedArguments( + callable_, args, starargs, kwargs + ) + + +class BadCallable(TypeError, AssertionError, ZiplineError): + """ + The given callable is not structured in the expected way. + """ + _lambda_name = (lambda: None).__name__ + + def __init__(self, callable_, args, starargs, kwargs): + self.callable_ = callable_ + self.args = args + self.starargs = starargs + self.kwargsname = kwargs + + self.kwargs = {} + + def format_callable(self): + if self.callable_.__name__ == self._lambda_name: + fmt = '%s %s' + name = 'lambda' + else: + fmt = '%s(%s)' + name = self.callable_.__name__ + + return fmt % ( + name, + ', '.join( + chain( + (str(arg) for arg in self.args), + ('*' + sa for sa in (self.starargs,) if sa is not None), + ('**' + ka for ka in (self.kwargsname,) if ka is not None), + ) + ) + ) + + @property + def msg(self): + return str(self) + + +class NoStarargs(BadCallable): + def __str__(self): + return '%s does not allow for *args' % self.format_callable() + + +class UnexpectedStarargs(BadCallable): + def __str__(self): + return '%s should not allow for *args' % self.format_callable() + + +class NoKwargs(BadCallable): + def __str__(self): + return '%s does not allow for **kwargs' % self.format_callable() + + +class UnexpectedKwargs(BadCallable): + def __str__(self): + return '%s should not allow for **kwargs' % self.format_callable() + + +class NotCallable(BadCallable): + """ + The provided 'callable' is not actually a callable. + """ + def __init__(self, callable_): + self.callable_ = callable_ + + def __str__(self): + return '%s is not callable' % self.format_callable() + + def format_callable(self): + try: + return self.callable_.__name__ + except AttributeError: + return str(self.callable_) + + +class NotEnoughArguments(BadCallable): + """ + The callback does not accept enough arguments. + """ + def __init__(self, callable_, args, starargs, kwargs, missing_args): + super(NotEnoughArguments, self).__init__( + callable_, args, starargs, kwargs + ) + self.missing_args = missing_args + + def __str__(self): + missing_args = list(map(str, self.missing_args)) + return '%s is missing argument%s: %s' % ( + self.format_callable(), + 's' if len(missing_args) > 1 else '', + ', '.join(missing_args), + ) + + +class TooManyArguments(BadCallable): + """ + The callback cannot be called by passing the expected number of arguments. + """ + def __str__(self): + return '%s accepts too many arguments' % self.format_callable() + + +class MismatchedArguments(BadCallable): + """ + The argument lists are of the same lengths, but not in the correct order. + """ + def __str__(self): + return '%s accepts mismatched parameters' % self.format_callable() diff --git a/zipline/utils/events.py b/zipline/utils/events.py new file mode 100644 index 00000000..562301f0 --- /dev/null +++ b/zipline/utils/events.py @@ -0,0 +1,513 @@ +# +# 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 abc import ABCMeta, abstractmethod +from collections import namedtuple +import six + +import datetime +import pandas as pd +import pytz + +from zipline.finance.trading import TradingEnvironment +from zipline.utils.argcheck import verify_callable_argspec, Argument + + +__all__ = [ + 'EventManager', + 'Event', + 'EventRule', + 'StatelessRule', + 'ComposedRule', + 'Always', + 'Never', + 'AfterOpen', + 'BeforeClose', + 'NotHalfDay', + 'NthTradingDayOfWeek', + 'NDaysBeforeLastTradingDayOfWeek', + 'NthTradingDayOfMonth', + 'NDaysBeforeLastTradingDayOfMonth', + 'StatefulRule', + 'OncePerDay', + + # Factory API + 'DateRuleFactory', + 'TimeRuleFactory', + 'make_eventrule', +] + + +def naive_to_utc(ts): + """ + Converts a UTC tz-naive timestamp to a tz-aware timestamp. + """ + # Drop the nanoseconds field. warn=False suppresses the warning + # that we are losing the nanoseconds; however, this is intended. + return pd.Timestamp(ts.to_pydatetime(warn=False), tz='UTC') + + +def ensure_utc(time, tz='UTC'): + """ + Normalize a time. If the time is tz-naive, assume it is UTC. + """ + if not time.tzinfo: + time = time.replace(tzinfo=pytz.timezone(tz)) + return time.replace(tzinfo=pytz.utc) + + +def _build_offset(offset, kwargs): + """ + Builds the offset argument for event rules. + """ + if offset is None: + if not kwargs: + return datetime.timedelta() # An empty offset (+0). + else: + return datetime.timedelta(**kwargs) + elif kwargs: + raise ValueError('Cannot pass kwargs and an offset') + else: + return offset + + +def _build_date(date, kwargs): + """ + Builds the date argument for event rules. + """ + if date is None: + if not kwargs: + raise ValueError('Must pass a date or kwargs') + else: + return datetime.date(**kwargs) + + elif kwargs: + raise ValueError('Cannot pass kwargs and a date') + else: + return date + + +def _build_time(time, kwargs): + """ + Builds the time argument for event rules. + """ + tz = kwargs.pop('tz', 'UTC') + if time: + if kwargs: + raise ValueError('Cannot pass kwargs and a time') + else: + return ensure_utc(time, tz) + elif not kwargs: + raise ValueError('Must pass a time or kwargs') + else: + return datetime.time(**kwargs) + + +class EventManager(object): + """ + Manages a list of Event objects. + This manages the logic for checking the rules and dispatching to the + handle_data function of the Events. + """ + def __init__(self): + self._events = [] + + def add_event(self, event, prepend=False): + """ + Adds an event to the manager. + """ + if prepend: + self._events.insert(0, event) + else: + self._events.append(event) + + def handle_data(self, context, data, dt): + for event in self._events: + event.handle_data(context, data, dt) + + +class Event(namedtuple('Event', ['rule', 'callback'])): + """ + An event is a pairing of an EventRule and a callable that will be invoked + with the current algorithm context, data, and datetime only when the rule + is triggered. + """ + def __new__(cls, rule=None, callback=None, check_args=True): + callback = callback or (lambda *args, **kwargs: None) + if check_args: + # Check the callback provided. + verify_callable_argspec( + callback, + [Argument('context' if check_args else Argument.ignore), + Argument('data' if check_args else Argument.ignore)] + ) + + # Make sure that the rule's should_trigger is valid. This will + # catch potential errors much more quickly and give a more helpful + # error. + verify_callable_argspec( + getattr(rule, 'should_trigger'), + [Argument('dt')] + ) + + return super(cls, cls).__new__(cls, rule=rule, callback=callback) + + def handle_data(self, context, data, dt): + """ + Calls the callable only when the rule is triggered. + """ + if self.rule.should_trigger(dt): + self.callback(context, data) + + +class EventRule(six.with_metaclass(ABCMeta)): + """ + An event rule checks a datetime and sees if it should trigger. + """ + env = TradingEnvironment.instance() + + @abstractmethod + def should_trigger(self, dt): + """ + Checks if the rule should trigger with it's current state. + This method should be pure and NOT mutate any state on the object. + """ + raise NotImplementedError('should_trigger') + + +class StatelessRule(EventRule): + """ + A stateless rule has no state. + This is reentrant and will always give the same result for the + same datetime. + Because these are pure, they can be composed to create new rules. + """ + def and_(self, rule): + """ + Logical and of two rules, triggers only when both rules trigger. + This follows the short circuiting rules for normal and. + """ + return ComposedRule(self, rule, ComposedRule.lazy_and) + __and__ = and_ + + +class ComposedRule(StatelessRule): + """ + A rule that composes the results of two rules with some composing function. + The composing function should be a binary function that accepts the results + first(dt) and second(dt) as positional arguments. + For example, operator.and_. + If lazy=True, then the lazy composer is used instead. The lazy composer + expects a function that takes the two should_trigger functions and the + datetime. This is useful of you don't always want to call should_trigger + for one of the rules. For example, this is used to implement the & and | + operators so that they will have the same short circuit logic that is + expected. + """ + def __init__(self, first, second, composer): + if not (isinstance(first, StatelessRule) + and isinstance(second, StatelessRule)): + raise ValueError('Only two StatelessRules can be composed') + + self.first = first + self.second = second + self.composer = composer + + def should_trigger(self, dt): + """ + Composes the two rules with a lazy composer. + """ + return self.composer( + self.first.should_trigger, + self.second.should_trigger, + dt, + ) + + @staticmethod + def lazy_and(first_should_trigger, second_should_trigger, dt): + """ + Lazily ands the two rules. This will NOT call the should_trigger of the + second rule if the first one returns False. + """ + return first_should_trigger(dt) and second_should_trigger(dt) + + +class Always(StatelessRule): + """ + A rule that always triggers. + """ + @staticmethod + def always_trigger(dt): + """ + A should_trigger implementation that will always trigger. + """ + return True + should_trigger = always_trigger + + +class Never(StatelessRule): + """ + A rule that never triggers. + """ + @staticmethod + def never_trigger(dt): + """ + A should_trigger implementation that will never trigger. + """ + return False + should_trigger = never_trigger + + +class AfterOpen(StatelessRule): + """ + A rule that triggers for some offset after the market opens. + Example that triggers triggers after 30 minutes of the market opening: + + >>> AfterOpen(minutes=30) + """ + def __init__(self, offset=None, **kwargs): + self.offset = _build_offset(offset, kwargs) + + def should_trigger(self, dt): + return self.env.get_open_and_close(dt)[0] + self.offset <= dt + + +class BeforeClose(StatelessRule): + """ + A rule that triggers for some offset time before the market closes. + Example that triggers for the last 30 minutes every day: + + >>> BeforeClose(minutes=30) + """ + def __init__(self, offset=None, **kwargs): + self.offset = _build_offset(offset, kwargs) + + def should_trigger(self, dt): + return self.env.get_open_and_close(dt)[1] - self.offset < dt + + +class NotHalfDay(StatelessRule): + """ + A rule that only triggers when it is not a half day. + """ + def should_trigger(self, dt): + return dt not in self.env.early_closes + + +class NthTradingDayOfWeek(StatelessRule): + """ + A rule that triggers on the nth trading day of the week. + This is zero-indexed, n=0 is the first trading day of the week. + """ + def __init__(self, n=0): + if n not in range(5): + raise ValueError('n must be in [0,5)') + self.td_delta = n + + def should_trigger(self, dt): + return self.env.add_trading_days( + self.td_delta, + self.get_first_trading_day_of_week(dt), + ) == dt.date() + + def get_first_trading_day_of_week(self, dt): + prev = dt + dt = self.env.previous_trading_day(dt) + # Backtrack until we hit a week border, then jump to the next trading + # day. + while dt.day < prev.day: + prev = dt + dt = self.env.previous_trading_day(dt) + return prev.date() + + +class NDaysBeforeLastTradingDayOfWeek(StatelessRule): + """ + A rule that triggers n days before the last trading day of the week. + """ + def __init__(self, n): + if n not in range(5): + raise ValueError('n must be in [0,5)') + self.td_delta = -n + self.date = None + + def should_trigger(self, dt): + return self.env.add_trading_days( + self.td_delta, + self.get_last_trading_day_of_week(dt), + ) == dt.date() + + def get_last_trading_day_of_week(self, dt): + prev = dt + dt = self.env.next_trading_day(dt) + # Traverse forward until we hit a week border, then jump back to the + # previous trading day. + while dt.day > prev.day: + prev = dt + dt = self.env.next_trading_day(dt) + return prev.date() + + +class NthTradingDayOfMonth(StatelessRule): + """ + A rule that triggers on the nth trading day of the month. + This is zero-indexed, n=0 is the first trading day of the month. + """ + def __init__(self, n=0): + if n not in range(31): + raise ValueError('n must be in [0,31)') + self.td_delta = n + self.month = None + self.day = None + + def should_trigger(self, dt): + return self.get_nth_trading_day_of_month(dt) == dt.date() + + def get_nth_trading_day_of_month(self, dt): + if self.month == dt.month: + # We already computed the day for this month. + return self.day + + if not self.td_delta: + self.day = self.get_first_trading_day_of_month(dt) + else: + self.day = self.env.add_trading_days( + self.td_delta, + self.get_first_trading_day_of_month(dt), + ).date() + + return self.day + + def get_first_trading_day_of_month(self, dt): + self.month = dt.month + + dt = dt.replace(day=1) + self.first_day = (dt if self.env.is_trading_day(dt) + else self.env.next_trading_day(dt)).date() + return self.first_day + + +class NDaysBeforeLastTradingDayOfMonth(StatelessRule): + """ + A rule that triggers n days before the last trading day of the month. + """ + def __init__(self, n=0): + if n not in range(31): + raise ValueError('n must be in [0,31)') + self.td_delta = -n + self.month = None + self.day = None + + def should_trigger(self, dt): + return self.get_nth_to_last_trading_day_of_month(dt) == dt.date() + + def get_nth_to_last_trading_day_of_month(self, dt): + if self.month == dt.month: + # We already computed the last day for this month. + return self.day + + if not self.td_delta: + self.day = self.get_last_trading_day_of_month(dt) + else: + self.day = self.env.add_trading_days( + self.td_delta, + self.get_last_trading_day_of_month(dt), + ).date() + + return self.day + + def get_last_trading_day_of_month(self, dt): + self.month = dt.month + + self.last_day = self.env.previous_trading_day( + dt.replace(month=(dt.month % 12) + 1, day=1) + ).date() + return self.last_day + + +# Stateful rules + + +class StatefulRule(EventRule): + """ + A stateful rule has state. + This rule will give different results for the same datetimes depending + on the internal state that this holds. + StatefulRules wrap other rules as state transformers. + """ + def __init__(self, rule=None): + self.rule = rule or Always() + + def new_should_trigger(self, callable_): + """ + Replace the should trigger implementation for the current rule. + """ + self.should_trigger = callable_ + + +class OncePerDay(StatefulRule): + def __init__(self, rule=None): + self.date = None + self.triggered = False + super(OncePerDay, self).__init__(rule) + + def should_trigger(self, dt): + dt_date = dt.date() + if self.date is None or self.date != dt_date: + # initialize or reset for new date + self.triggered = False + self.date = dt_date + + if not self.triggered and self.rule.should_trigger(dt): + self.triggered = True + return True + + +# Factory API + +class DateRuleFactory(object): + every_day = Always + + @staticmethod + def month_start(offset=0): + return NthTradingDayOfMonth(n=offset) + + @staticmethod + def month_end(offset=0): + return NDaysBeforeLastTradingDayOfMonth(n=offset) + + @staticmethod + def week_start(offset=0): + return NthTradingDayOfWeek(n=offset) + + @staticmethod + def week_end(offset=0): + return NDaysBeforeLastTradingDayOfWeek(n=offset) + + +class TimeRuleFactory(object): + market_open = AfterOpen + market_close = BeforeClose + + +def make_eventrule(date_rule, time_rule, half_days=True): + """ + Constructs an event rule from the factory api. + """ + if half_days: + inner_rule = date_rule & time_rule + else: + inner_rule = date_rule & time_rule & NotHalfDay() + + return OncePerDay(rule=inner_rule)