ENH: Clock now fires a BEFORE_TRADING_START_BAR event.

`AlgorithmSimulator` listens to that event to call the algorithm's
`before_trading_start` method.

tests/finance/test_cancel_policy.py

@@ -17,7 +17,7 @@ from unittest import TestCase
 from zipline.finance.cancel_policy import NeverCancel, EODCancel
 from zipline.gens.sim_engine import (
     BAR,
-    DAY_END
+    SESSION_END
 )
 
 
@@ -25,10 +25,10 @@ class CancelPolicyTestCase(TestCase):
 
     def test_eod_cancel(self):
         cancel_policy = EODCancel()
-        self.assertTrue(cancel_policy.should_cancel(DAY_END))
+        self.assertTrue(cancel_policy.should_cancel(SESSION_END))
         self.assertFalse(cancel_policy.should_cancel(BAR))
 
     def test_never_cancel(self):
         cancel_policy = NeverCancel()
-        self.assertFalse(cancel_policy.should_cancel(DAY_END))
+        self.assertFalse(cancel_policy.should_cancel(SESSION_END))
         self.assertFalse(cancel_policy.should_cancel(BAR))

tests/test_blotter.py

@@ -25,7 +25,7 @@ from zipline.finance.execution import (
     StopOrder,
 )
 
-from zipline.gens.sim_engine import DAY_END, BAR
+from zipline.gens.sim_engine import SESSION_END, BAR
 from zipline.finance.cancel_policy import EODCancel, NeverCancel
 from zipline.finance.slippage import (
     DEFAULT_VOLUME_SLIPPAGE_BAR_LIMIT,
@@ -143,7 +143,7 @@ class BlotterTestCase(WithLogger,
         self.assertEqual(blotter.new_orders[0].status, ORDER_STATUS.OPEN)
         self.assertEqual(blotter.new_orders[1].status, ORDER_STATUS.OPEN)
 
-        blotter.execute_cancel_policy(DAY_END)
+        blotter.execute_cancel_policy(SESSION_END)
         for order_id in order_ids:
             order = blotter.orders[order_id]
             self.assertEqual(order.status, ORDER_STATUS.CANCELLED)
@@ -161,7 +161,7 @@ class BlotterTestCase(WithLogger,
         blotter.execute_cancel_policy(BAR)
         self.assertEqual(blotter.new_orders[0].status, ORDER_STATUS.OPEN)
 
-        blotter.execute_cancel_policy(DAY_END)
+        blotter.execute_cancel_policy(SESSION_END)
         self.assertEqual(blotter.new_orders[0].status, ORDER_STATUS.OPEN)
 
     def test_order_rejection(self):

tests/test_clock.py

@@ -0,0 +1,183 @@
+from datetime import time
+from unittest import TestCase
+import pandas as pd
+import numpy as np
+from zipline.gens.sim_engine import (
+    MinuteSimulationClock,
+    SESSION_START,
+    BEFORE_TRADING_START_BAR,
+    BAR,
+    MINUTE_END,
+    SESSION_END
+)
+
+from zipline.utils.calendars import get_calendar
+from zipline.utils.calendars.trading_calendar import days_at_time
+
+
+class TestClock(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.nyse_calendar = get_calendar("NYSE")
+
+        # july 15 is friday, so there are 3 sessions in this range (15, 18, 19)
+        cls.sessions = cls.nyse_calendar.sessions_in_range(
+            pd.Timestamp("2016-07-15"),
+            pd.Timestamp("2016-07-19")
+        )
+
+        trading_o_and_c = cls.nyse_calendar.schedule.ix[cls.sessions]
+        cls.opens = trading_o_and_c['market_open'].values.astype(np.int64)
+        cls.closes = trading_o_and_c['market_close'].values.astype(np.int64)
+
+    def test_bts_before_session(self):
+        clock = MinuteSimulationClock(
+            self.sessions,
+            self.opens,
+            self.closes,
+            days_at_time(self.sessions, time(6, 17), "US/Eastern"),
+            False
+        )
+
+        all_events = list(clock)
+
+        def _check_session_bts_first(session_label, events, bts_dt):
+            minutes = self.nyse_calendar.minutes_for_session(session_label)
+
+            self.assertEqual(393, len(events))
+
+            self.assertEqual(events[0], (session_label, SESSION_START))
+            self.assertEqual(events[1], (bts_dt, BEFORE_TRADING_START_BAR))
+            for i in range(2, 392):
+                self.assertEqual(events[i], (minutes[i - 2], BAR))
+            self.assertEqual(events[392], (minutes[-1], SESSION_END))
+
+        _check_session_bts_first(
+            self.sessions[0],
+            all_events[0:393],
+            pd.Timestamp("2016-07-15 6:17", tz='US/Eastern')
+        )
+
+        _check_session_bts_first(
+            self.sessions[1],
+            all_events[393:786],
+            pd.Timestamp("2016-07-18 6:17", tz='US/Eastern')
+        )
+
+        _check_session_bts_first(
+            self.sessions[2],
+            all_events[786:],
+            pd.Timestamp("2016-07-19 6:17", tz='US/Eastern')
+        )
+
+    def test_bts_during_session(self):
+        self.verify_bts_during_session(
+            time(11, 45), [
+                pd.Timestamp("2016-07-15 11:45", tz='US/Eastern'),
+                pd.Timestamp("2016-07-18 11:45", tz='US/Eastern'),
+                pd.Timestamp("2016-07-19 11:45", tz='US/Eastern')
+            ],
+            135
+        )
+
+    def test_bts_on_first_minute(self):
+        self.verify_bts_during_session(
+            time(9, 30), [
+                pd.Timestamp("2016-07-15 9:30", tz='US/Eastern'),
+                pd.Timestamp("2016-07-18 9:30", tz='US/Eastern'),
+                pd.Timestamp("2016-07-19 9:30", tz='US/Eastern')
+            ],
+            1
+        )
+
+    def test_bts_on_last_minute(self):
+        self.verify_bts_during_session(
+            time(16, 00), [
+                pd.Timestamp("2016-07-15 16:00", tz='US/Eastern'),
+                pd.Timestamp("2016-07-18 16:00", tz='US/Eastern'),
+                pd.Timestamp("2016-07-19 16:00", tz='US/Eastern')
+            ],
+            390
+        )
+
+    def verify_bts_during_session(self, bts_time, bts_session_times, bts_idx):
+        def _check_session_bts_during(session_label, events, bts_dt):
+            minutes = self.nyse_calendar.minutes_for_session(session_label)
+
+            self.assertEqual(393, len(events))
+
+            self.assertEqual(events[0], (session_label, SESSION_START))
+
+            for i in range(1, bts_idx):
+                self.assertEqual(events[i], (minutes[i - 1], BAR))
+
+            self.assertEqual(
+                events[bts_idx],
+                (bts_dt, BEFORE_TRADING_START_BAR)
+            )
+
+            for i in range(bts_idx + 1, 391):
+                self.assertEqual(events[i], (minutes[i - 2], BAR))
+
+            self.assertEqual(events[392], (minutes[-1], SESSION_END))
+
+        clock = MinuteSimulationClock(
+            self.sessions,
+            self.opens,
+            self.closes,
+            days_at_time(self.sessions, bts_time, "US/Eastern"),
+            False
+        )
+
+        all_events = list(clock)
+
+        _check_session_bts_during(
+            self.sessions[0],
+            all_events[0:393],
+            bts_session_times[0]
+        )
+
+        _check_session_bts_during(
+            self.sessions[1],
+            all_events[393:786],
+            bts_session_times[1]
+        )
+
+        _check_session_bts_during(
+            self.sessions[2],
+            all_events[786:],
+            bts_session_times[2]
+        )
+
+
+    def test_bts_after_session(self):
+        clock = MinuteSimulationClock(
+            self.sessions,
+            self.opens,
+            self.closes,
+            days_at_time(self.sessions, time(19, 05), "US/Eastern"),
+            False
+        )
+
+        all_events = list(clock)
+
+        # since 19:05 Eastern is after the NYSE is closed, we don't emit
+        # BEFORE_TRADING_START.  therefore, each day has SESSION_START,
+        # 390 BARs, and then SESSION_END
+
+        def _check_session_bts_after(session_label, events):
+            minutes = self.nyse_calendar.minutes_for_session(session_label)
+
+            self.assertEqual(392, len(events))
+            self.assertEqual(events[0], (session_label, SESSION_START))
+
+            for i in range(1, 391):
+                self.assertEqual(events[i], (minutes[i - 1], BAR))
+
+            self.assertEqual(events[-1], (minutes[389], SESSION_END))
+
+        for i in range(0, 2):
+            _check_session_bts_after(
+                self.sessions[i],
+                all_events[(i * 392): ((i + 1) * 392)]
+            )

tests/test_tradesimulation.py

@@ -12,6 +12,8 @@
 # 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 datetime import time
+
 import pandas as pd
 from mock import patch
 
@@ -23,6 +25,7 @@ from zipline.sources.benchmark_source import BenchmarkSource
 from zipline.test_algorithms import NoopAlgorithm
 from zipline.utils import factory
 from zipline.testing.core import FakeDataPortal
+from zipline.utils.calendars.trading_calendar import days_at_time
 
 
 class BeforeTradingAlgorithm(TradingAlgorithm):
@@ -75,10 +78,18 @@ class TestTradeSimulation(TestCase):
             algo = BeforeTradingAlgorithm(sim_params=params)
             algo.run(FakeDataPortal())
 
-            self.assertEqual(len(algo.perf_tracker.sim_params.sessions),
-                             num_days)
+            self.assertEqual(
+                len(algo.perf_tracker.sim_params.sessions),
+                num_days
+            )
 
-            self.assertTrue(params.sessions.equals(
-                pd.DatetimeIndex(algo.before_trading_at)),
-                "Expected %s but was %s."
-                % (params.sessions, algo.before_trading_at))
+            bts_minutes = days_at_time(
+                params.sessions, time(8, 45), "US/Eastern"
+            )
+
+            self.assertTrue(
+                bts_minutes.equals(
+                    pd.DatetimeIndex(algo.before_trading_at)
+                ),
+                "Expected %s but was %s." % (params.sessions,
+                                             algo.before_trading_at))

zipline/algorithm.py

@@ -15,7 +15,7 @@
 from copy import copy
 import operator as op
 import warnings
-from datetime import tzinfo
+from datetime import tzinfo, time
 import logbook
 import pytz
 import pandas as pd
@@ -94,9 +94,9 @@ from zipline.utils.api_support import (
     require_not_initialized,
     ZiplineAPI,
     disallowed_in_before_trading_start)
-
 from zipline.utils.input_validation import ensure_upper_case, error_keywords, \
     expect_types, optional, coerce_string
+from zipline.utils.calendars.trading_calendar import days_at_time
 from zipline.utils.cache import CachedObject, Expired
 from zipline.utils.calendars import get_calendar
 
@@ -497,28 +497,33 @@ class TradingAlgorithm(object):
         trading_o_and_c = self.trading_calendar.schedule.ix[
             self.sim_params.sessions]
         market_closes = trading_o_and_c['market_close'].values.astype(np.int64)
+        minutely_emission = False
 
         if self.sim_params.data_frequency == 'minute':
             market_opens = trading_o_and_c['market_open'].values.astype(
-                np.int64)
+                np.int64
+            )
 
             minutely_emission = self.sim_params.emission_rate == "minute"
-
-            return MinuteSimulationClock(
-                self.sim_params.sessions,
-                market_opens,
-                market_closes,
-                minutely_emission
-            )
         else:
             # in daily mode, we want to have one bar per session, timestamped
             # as the last minute of the session.
-            return MinuteSimulationClock(
-                self.sim_params.sessions,
-                market_closes,
-                market_closes,
-                False
-            )
+            market_opens = market_closes
+
+        # FIXME generalize these values
+        before_trading_start_minutes = days_at_time(
+            self.sim_params.sessions,
+            time(8, 45),
+            "US/Eastern"
+        )
+
+        return MinuteSimulationClock(
+            self.sim_params.sessions,
+            market_opens,
+            market_closes,
+            before_trading_start_minutes,
+            minute_emission=minutely_emission,
+        )
 
     def _create_benchmark_source(self):
         return BenchmarkSource(
@@ -1545,6 +1550,7 @@ class TradingAlgorithm(object):
                 self.datetime, self._in_before_trading_start, self.data_portal)
             self._account = \
                 self.perf_tracker.get_account(self.performance_needs_update)
+
             self.account_needs_update = False
             self.performance_needs_update = False
         return self._account

zipline/finance/cancel_policy.py

@@ -17,7 +17,7 @@ import abc
 from abc import abstractmethod
 from six import with_metaclass
 
-from zipline.gens.sim_engine import DAY_END
+from zipline.gens.sim_engine import SESSION_END
 
 
 class CancelPolicy(with_metaclass(abc.ABCMeta)):
@@ -58,7 +58,7 @@ class EODCancel(CancelPolicy):
         self.warn_on_cancel = warn_on_cancel
 
     def should_cancel(self, event):
-        return event == DAY_END
+        return event == SESSION_END
 
 
 class NeverCancel(CancelPolicy):

zipline/gens/sim_engine.pyx

@@ -24,26 +24,31 @@ NANOS_IN_MINUTE = _nanos_in_minute
 
 cpdef enum:
     BAR = 0
-    DAY_START = 1
-    DAY_END = 2
+    SESSION_START = 1
+    SESSION_END = 2
     MINUTE_END = 3
+    BEFORE_TRADING_START_BAR = 4
 
 cdef class MinuteSimulationClock:
-    cdef object trading_days
+    cdef object sessions
     cdef bool minute_emission
     cdef np.int64_t[:] market_opens, market_closes
-    cdef public dict minutes_by_day, minutes_to_day
+    cdef object before_trading_start_minutes
+    cdef dict minutes_by_session, minutes_to_session
 
     def __init__(self,
-                 trading_days,
+                 sessions,
                  market_opens,
                  market_closes,
+                 before_trading_start_minutes,
                  minute_emission=False):
         self.minute_emission = minute_emission
         self.market_opens = market_opens
         self.market_closes = market_closes
-        self.trading_days = trading_days
-        self.minutes_by_day = self.calc_minutes_by_day()
+        self.sessions = sessions
+        self.minutes_by_session = self.calc_minutes_by_session()
+
+        self.before_trading_start_minutes = before_trading_start_minutes
 
     @cython.boundscheck(False)
     @cython.wraparound(False)
@@ -59,28 +64,50 @@ cdef class MinuteSimulationClock:
 
     @cython.boundscheck(False)
     @cython.wraparound(False)
-    cdef dict calc_minutes_by_day(self):
-        cdef dict minutes_by_day
-        cdef int day_idx
-        cdef object day
+    cdef dict calc_minutes_by_session(self):
+        cdef dict minutes_by_session
+        cdef int session_idx
+        cdef object session
 
-        minutes_by_day = {}
-        for day_idx, day in enumerate(self.trading_days):
-            minutes_by_day[day] = pd.to_datetime(
-                self.market_minutes(day_idx), utc=True, box=True)
-        return minutes_by_day
+        minutes_by_session = {}
+        for session_idx, session in enumerate(self.sessions):
+            minutes_by_session[session] = pd.to_datetime(
+                self.market_minutes(session_idx), utc=True, box=True)
+        return minutes_by_session
 
     def __iter__(self):
         minute_emission = self.minute_emission
 
-        for day in self.trading_days:
-            yield day, DAY_START
+        for idx, session in enumerate(self.sessions):
+            yield session, SESSION_START
 
-            minutes = self.minutes_by_day[day]
+            bts_minute = self.before_trading_start_minutes[idx]
+            regular_minutes = self.minutes_by_session[session]
 
-            for minute in minutes:
-                yield minute, BAR
-                if minute_emission:
-                    yield minute, MINUTE_END
+            # we have to search anew every session, because there is no
+            # guarantee that any two session start on the same minute
+            bts_idx = regular_minutes.searchsorted(bts_minute)
 
-            yield minutes[-1], DAY_END
+            if bts_idx == len(regular_minutes):
+                # before_trading_start is after the last close, so don't emit
+                # it
+                for minute in regular_minutes:
+                    yield minute, BAR
+                    if minute_emission:
+                        yield minute, MINUTE_END
+            else:
+                # emit all the minutes before bts_minute
+                for minute in regular_minutes[0:bts_idx]:
+                    yield minute, BAR
+                    if minute_emission:
+                        yield minute, MINUTE_END
+
+                yield bts_minute, BEFORE_TRADING_START_BAR
+
+                # emit all the minutes after bts_minute
+                for minute in regular_minutes[bts_idx:]:
+                    yield minute, BAR
+                    if minute_emission:
+                        yield minute, MINUTE_END
+
+            yield regular_minutes[-1], SESSION_END

zipline/gens/tradesimulation.py

@@ -21,9 +21,10 @@ from six import viewkeys
 
 from zipline.gens.sim_engine import (
     BAR,
-    DAY_START,
-    DAY_END,
-    MINUTE_END
+    SESSION_START,
+    SESSION_END,
+    MINUTE_END,
+    BEFORE_TRADING_START_BAR
 )
 
 log = Logger('Trade Simulation')
@@ -181,9 +182,6 @@ class AlgorithmSimulator(object):
                     algo.blotter.process_splits(splits)
                     perf_tracker.position_tracker.handle_splits(splits)
 
-            # call before trading start
-            algo.before_trading_start(current_data)
-
         def handle_benchmark(date, benchmark_source=self.benchmark_source):
             algo.perf_tracker.all_benchmark_returns[date] = \
                 benchmark_source.get_value(date)
@@ -202,7 +200,7 @@ class AlgorithmSimulator(object):
 
             if algo.data_frequency == 'minute':
                 def execute_order_cancellation_policy():
-                    algo.blotter.execute_cancel_policy(DAY_END)
+                    algo.blotter.execute_cancel_policy(SESSION_END)
 
                 def calculate_minute_capital_changes(dt):
                     # process any capital changes that came between the last
@@ -220,16 +218,20 @@ class AlgorithmSimulator(object):
                 if action == BAR:
                     for capital_change_packet in every_bar(dt):
                         yield capital_change_packet
-                elif action == DAY_START:
+                elif action == SESSION_START:
                     for capital_change_packet in once_a_day(dt):
                         yield capital_change_packet
-                elif action == DAY_END:
-                    # End of the day.
+                elif action == SESSION_END:
+                    # End of the session.
                     if emission_rate == 'daily':
                         handle_benchmark(normalize_date(dt))
                     execute_order_cancellation_policy()
 
                     yield self._get_daily_message(dt, algo, algo.perf_tracker)
+                elif action == BEFORE_TRADING_START_BAR:
+                    # call before trading start
+                    algo.on_dt_changed(dt)
+                    algo.before_trading_start(self.current_data)
                 elif action == MINUTE_END:
                     handle_benchmark(dt)
                     minute_msg = \

zipline/utils/calendars/trading_calendar.py

@@ -757,6 +757,9 @@ def days_at_time(days, t, tz, day_offset=0):
     day_offset : int
         The number of days we want to offset @days by
     """
+    if len(days) == 0:
+        return days
+
     # Offset days without tz to avoid timezone issues.
     days = DatetimeIndex(days).tz_localize(None)
     days_offset = days + DateOffset(days=day_offset)