Merge branch 'new_world_order' of github.com:quantopian/zipline into new_world_order

tests/test_sorting.py

@@ -249,9 +249,11 @@ def compare_by_dt_source_id(x,y):
         return -1
     elif x.source_id > y.source_id:
         return 1
-
     else:
         return 0
 
 #Alias for ease of use                         
 comp = compare_by_dt_source_id
+
+def to_dt(msg):
+    return ndict({'dt': msg})

tests/test_transforms.py

@@ -5,9 +5,12 @@ from unittest2 import TestCase
 from zipline.utils.test_utils import setup_logger, teardown_logger
 
 import zipline.utils.factory as factory
-from zipline.finance.vwap import DailyVWAP, VWAPTransform
+
+from zipline.gens.tradegens import SpecificEquityTrades
+from zipline.gens.transform import StatefulTransform
+from zipline.gens.vwap import VWAP
+from zipline.gens.mavg import MovingAverage
 from zipline.finance.returns import ReturnsFromPriorClose
-from zipline.finance.movingaverage import MovingAverage
 from zipline.lines import SimulatedTrading
 from zipline.core.devsimulator import AddressAllocator
 
@@ -25,7 +28,7 @@ class ZiplineWithTransformsTestCase(TestCase):
             'sid'       : 133,
             'devel'     : True
         }
-        setup_logger(self, '/var/log/qexec/qexed.log')
+        setup_logger(self, '/var/log/qexec/qexec.log')
 
     def tearDown(self):
         teardown_logger(self)
@@ -48,25 +51,34 @@ class FinanceTransformsTestCase(TestCase):
         self.trading_environment = factory.create_trading_environment()
         setup_logger(self, '/var/log/qexec/qexec.log')
 
-    def tearDown(self):
-        self.log_handler.pop_application()
-
-    def test_vwap(self):
-
         trade_history = factory.create_trade_history(
             133,
-            [10.0, 10.0, 10.0, 11.0],
+            [10.0, 10.0, 11.0, 11.0],
             [100, 100, 100, 300],
             timedelta(days=1),
             self.trading_environment
         )
+        self.source = SpecificEquityTrades(event_list=trade_history)
 
-        vwap = DailyVWAP(days=2)
-        for trade in trade_history:
-            vwap.update(trade)
+    def tearDown(self):
+        self.log_handler.pop_application()
 
-        self.assertEqual(vwap.vwap, 10.75)
+    def test_vwap(self):
+        vwap = StatefulTransform(VWAP, timedelta(days = 2))
+        transformed = list(vwap.transform(self.source))
 
+        # Output values
+        tnfm_vals = [message.tnfm_value for message in transformed]
+        # "Hand calculated" values.
+        expected = [(10.0 * 100) / 100.0,
+                    ((10.0 * 100) + (10.0 * 100)) / (200.0),
+                    ((10.0 * 100) + (10.0 * 100) + (11.0 * 100)) / (300.0),
+                    # First event should get droppped here.
+                    ((10.0 * 100) + (11.0 * 100) + (11.0 * 300)) / (500.0)] 
+
+        # Output should match the expected.
+        assert tnfm_vals == expected
+        
 
     def test_returns(self):
         trade_history = factory.create_trade_history(
@@ -86,17 +98,29 @@ class FinanceTransformsTestCase(TestCase):
 
 
     def test_moving_average(self):
-        trade_history = factory.create_trade_history(
-            133,
-            [10.0, 10.0, 10.0, 11.0],
-            [100, 100, 100, 300],
-            timedelta(days=1),
-            self.trading_environment
-        )
-
-        ma = MovingAverage(days=2)
-        for trade in trade_history:
-            ma.update(trade)
-
-
-        self.assertEqual(ma.average, 10.5)
+        
+        mavg = StatefulTransform(
+            MovingAverage, 
+            timedelta(days = 2), 
+            ['price', 'volume']
+        ) 
+        
+        transformed = list(mavg.transform(self.source))
+        # Output values.
+        tnfm_prices = [message.tnfm_value.price for message in transformed]
+        tnfm_volumes = [message.tnfm_value.volume for message in transformed]
+        # "Hand-calculated" values
+        expected_prices = [((10.0) / 1.0),
+                           ((10.0 + 10.0) / 2.0),
+                           ((10.0 + 10.0 + 11.0) / 3.0),
+                           # First event should get dropped here.
+                           ((10.0 + 11.0 + 11.0) / 3.0)]
+        expected_volumes = [((100.0) / 1.0),
+                           ((100.0 + 100.0) / 2.0),
+                           ((100.0 + 100.0 + 100.0) / 3.0),
+                           # First event should get dropped here.
+                           ((100.0 + 100.0 + 300.0) / 3.0)]
+        
+        assert tnfm_prices == expected_prices
+        assert tnfm_volumes == expected_volumes
+        

zipline/finance/returns.py

@@ -1,26 +1,24 @@
 from collections import defaultdict
 from zipline.transforms.base import BaseTransform
 
-class ReturnsTransform(BaseTransform):
-
-    def init(self, name):
-        self.state = {}
-        self.state['name'] = name
-        self.by_sid = defaultdict(self._create)
-
-    @property
-    def get_id(self):
-        return self.state['name']
-
-
-    def transform(self, event):
-        cur = self.by_sid[event.sid]
-        cur.update(event)
-        self.state['value'] = cur.returns
-        return self.state
+class Returns(object):
+    """
+    Class that maintains a dictionary from sids to the event
+    representing the most recent closing price.
+    """
+    def __init__(self, days == 1):
+        self.days = days
+        self.mapping = defaultdict(self._create)
+        
+    def update(self, event):
+        """
+        Update and return the calculated returns for this event's sid.
+        """
+        sid_returns = self.mapping[event.sid].update(event)
+        return sid_returns
 
     def _create(self):
-        return ReturnsFromPriorClose()
+        return ReturnsFromPriorClose(days)
 
 class ReturnsFromPriorClose(object):
     """

zipline/gens/mavg.py

@@ -1,4 +1,3 @@
-
 from numbers import Number
 from datetime import datetime, timedelta
 from collections import defaultdict
@@ -8,15 +7,15 @@ from zipline.gens.transform import EventWindow
 
 class MovingAverage(object):
     """
-    Class that maintains a dictionary from sids to EventWindows
-    Upon receipt of each message we update the
-    corresponding window and return the calculated average.
+    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, delta, fields):
         self.delta = delta
         self.fields = fields
-
         # 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)
@@ -27,13 +26,9 @@ class MovingAverage(object):
 
     def update(self, event):
         """
-        Update the event window for this event's sid.  Return an ndict from
-        tracked fields to averages.
+        Update the event window for this event's sid.  Return an ndict
+        from tracked fields to moving averages.
         """
-        assert isinstance(event, ndict),"Bad event in MovingAverage: %s" % event
-        assert event.has_key('sid'), "No sid in MovingAverage: %s" % event
-        assert event.has_key('dt'), "No dt in MovingAverage: %s" % event
-
         # This will create a new EventWindow if this is the first
         # message for this sid.
         window = self.sid_windows[event.sid]
@@ -42,22 +37,34 @@ class MovingAverage(object):
 
 class MovingAverageEventWindow(EventWindow):
     """
-    Calculates a moving average over all specified fields.
+    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.
     """
-    # Subclass initializer.  The superclass also requires a timedelta
-    # argument, so instantiation should look like:
-    # mavg = MovingAverageEventWindow(timedelta(minutes=1), ['price']) 
-    def init(self, fields):
+
+    def __init__(self, delta, fields):
+
+        # Call the superclass constructor to set up base EventWindow
+        # infrastructure.
+        EventWindow.__init__(self, delta)
+
+        # We maintain a dictionary of totals for each of our tracked
+        # fields.
         self.fields = fields
         self.totals = defaultdict(float)
 
+    # Subclass customization for adding new events.
     def handle_add(self, event):
         # Sanity check on the event.
-        self.assert_all_fields(event)
+        self.assert_required_fields(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:
@@ -65,12 +72,12 @@ class MovingAverageEventWindow(EventWindow):
 
     def average(self, field):
         """
-        Calculate the average value of our ticks over a given field.
+        Calculate the average value of our ticks over a single field.
         """
         # Sanity check.
         assert field in self.fields
 
-        # Averages are 0 by convention if we have no ticks.
+        # Averages are None by convention if we have no ticks.
         if len(self.ticks) == 0:
             return 0.0
         
@@ -82,15 +89,14 @@ class MovingAverageEventWindow(EventWindow):
         """
         Return an ndict of all our tracked averages.
         """
-        out = ndict()
-        
+        out = ndict()        
         for field in self.fields:
             out[field] = self.average(field)
         return out
 
-    def assert_all_fields(self, event):
+    def assert_required_fields(self, event):
         """
-        We only track events with all the fields we care about.
+        We only allow events with all of our tracked fields.
         """
         for field in self.fields:
             assert event.has_key(field), \

zipline/gens/tradegens.py

@@ -3,12 +3,14 @@ Tools to generate trade events without a backing store. Useful for testing
 and zipline development
 """
 import random
+import pytz
+
 from itertools import chain, cycle, ifilter, izip
 from datetime import datetime, timedelta
 
 from zipline.gens.utils import hash_args, create_trade
 
-def date_gen(start = datetime(2006, 6, 6, 12),
+def date_gen(start = datetime(2006, 6, 6, 12, tzinfo=pytz.utc),
              delta = timedelta(minutes = 1),
              count = 100):
     """
@@ -24,9 +26,9 @@ def mock_prices(count, rand = False):
     """
 
     if rand:
-        return (random.uniform(0.0, 10.0) for i in xrange(count))
+        return (random.uniform(1.0, 10.0) for i in xrange(count))
     else:
-        return (float(i % 11) for i in xrange(1,count+1))
+        return (float(i % 10) + 1.0 for i in xrange(count))
 
 def mock_volumes(count, rand = False):
     """
@@ -70,7 +72,7 @@ class SpecificEquityTrades(object):
         # Unpack config dictionary with default values.
         self.count = kwargs.get('count', 500)
         self.sids = kwargs.get('sids', [1, 2])
-        self.start = kwargs.get('start', datetime(2012, 6, 6, 0))
+        self.start = kwargs.get('start', datetime(2008, 6, 6, 15, tzinfo = pytz.utc))
         self.delta = kwargs.get('delta', timedelta(minutes = 1))
 
         # Default to None for event_list and filter.

zipline/gens/transform.py

@@ -10,6 +10,7 @@ from numbers import Number
 from abc import ABCMeta, abstractmethod
 
 from zipline import ndict
+from zipline.utils.tradingcalendar import trading_days_between
 from zipline.gens.utils import assert_sort_unframe_protocol, \
     assert_transform_protocol, hash_args
 
@@ -156,15 +157,10 @@ class EventWindow:
     # Mark this as an abstract base class.
     __metaclass__ = ABCMeta
 
-    def __init__(self, delta, *args, **kwargs):
+    def __init__(self, delta):
         self.ticks  = deque()
         self.delta  = delta
-        self.init(*args, **kwargs)
         
-    @abstractmethod
-    def init(self):
-        raise NotImplementedError()
-
     @abstractmethod
     def handle_add(self, event):
         raise NotImplementedError()
@@ -193,7 +189,9 @@ class EventWindow:
             # Subclasses should override handle_remove to define
             # behavior for removing ticks.
             self.handle_remove(popped)
-
+            
+    # All event windows expect to receive events with datetime fields
+    # that arrive in sorted order.
     def assert_well_formed(self, event):
         assert isinstance(event, ndict), "Bad event in EventWindow:%s" % event
         assert event.has_key('dt'), "Missing dt in EventWindow:%s" % event

zipline/gens/vwap.py

@@ -0,0 +1,70 @@
+from numbers import Number
+from datetime import datetime, timedelta
+from collections import defaultdict
+
+from zipline import ndict
+from zipline.gens.transform import EventWindow
+
+class VWAP(object):
+    """
+    Class that maintains a dictionary from sids to VWAPEventWindows.
+    """
+    def __init__(self, delta):
+        self.delta = delta
+
+        # 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.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, delta):
+        EventWindow.__init__(self, delta)
+        self.flux = 0.0
+        self.totalvolume = 0.0
+
+    # Subclass customization for adding new events.
+    def handle_add(self, event):
+        # Sanity check on the event.
+        self.assert_required_fields(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)
+
+    # We need numerical price and volume to calculate a vwap.
+    def assert_required_fields(self, event):
+        assert isinstance(event.price, Number)
+        assert isinstance(event.volume, Number)

zipline/utils/date_utils.py

@@ -95,7 +95,7 @@ HOLIDAYS = {
     'july_4th'     : datetime(2008 , 7  , 4 ),
     'labor_day'    : datetime(2008 , 9  , 1 ),
     'tgiving'      : datetime(2008 , 11 , 27),
-    'christmas'    : datetime(2008 , 5  , 25),
+    'christmas'    : datetime(2008 , 12  , 25),
 }
 
 # Create a rule to recur every weekday starting today

zipline/utils/tradingcalendar.py

@@ -0,0 +1,333 @@
+import pytz
+
+from datetime import datetime, timedelta
+from dateutil import rrule
+from zipline.utils.date_utils import utcnow
+
+def market_opens(start, end, inclusive=False):
+    """
+    Returns all market opens between the start date and the end date.
+    Must use utc-stamped datetimes.
+    """
+    return opens.between(start, end, inc=inclusive)
+
+def market_closes(start, end, inclusive=False):
+    """
+    Returns all market closes between the start date and the end date.
+    Must use utc-stamped datetimes.
+    """
+    return closes.between(start, end, inc=inclusive)
+
+def trading_days_between(start, end):
+    """
+    Calculate the number of "complete" trading days between two
+    events.  We define this as the number of market opens that
+    occurred between start and end, with the caveat that we subtract 1
+    from this total if end falls on the same day as the last market
+    open and end occurs earlier in its own day than start.  This
+    reflects the fact that we haven't completed a full day
+    corresponding to the last market open.
+
+    Examples:
+    
+    1.)
+    start = Tuesday, Aug 7, 2012, 1:00 pm 
+    end = Wednesday, Aug 8, 2012, 1:30 pm
+
+    There is one market open between these dates, on the morning of
+    Wednesday the 8th.  This falls on the same calendar day as end,
+    but end is later in the day than start, so we count this as a full
+    day.  The correct output is 1.
+
+    2.)
+    start = Tuesday, Aug 7, 2012, 1:30 pm 
+    end = Wednesday, Aug 8, 2012, 1:00 pm
+    
+    There is one market open between these dayes, on the morning of
+    Wednesday the 8th.  This falls on the same calendar day as end,
+    and end is earlier in the day than start, so we do not count this
+    day as completed.  The correct output is 0.
+
+    3.)
+    start = Tuesday, Aug 7, 2012, 1:00 pm 
+    end = Saturday, Aug 11, 2012, 1:30 pm
+    
+    There are 3 market opens between these dates, occurring on
+    Wednesday, Thursday, and Friday.  The last open is not on
+    the same day as end, so we simply return 3
+
+    4.)
+    start = Tuesday, Aug 7, 2012, 1:30 pm 
+    end = Monday, Aug, 13, 2012, 1:00 pm
+    
+    There are 4 market opens between these dates, occurring on
+    Wednesday, Thursday, Friday, and the following Monday. The 
+    last open occurs on the same calendar day as end, and end
+    is earlier in the day than start, so we do not count the
+    last market day as completed.  The correct output is 3 days.
+    """
+    # Calculate the number of opens between the events.
+    opens = (market_opens(start, end))
+    days_between = len(opens)
+    if days_between == 0:
+        return days_between
+    
+    # If end falls on the same day as an open, subtract 1 from the
+    # total if end is earlier in its respective day than start.
+    last_open = opens[-1]
+    if last_open.date() == end.date() and earlier_in_day(end, start):
+        days_between -=1
+    
+    return days_between
+
+def earlier_in_day(d1, d2):
+    """
+    Return true if d1 falls earlier in its own day than d2.
+    """
+    d1 = d1.replace(year = d2.year, day = d2.day)
+    return d1 < d2
+
+WEEKDAYS = [rrule.MO, rrule.TU, rrule.WE, rrule.TH, rrule.FR]
+
+# Recurrence rule that generates all market opens since Jan 1, 1970.
+# This does not exclude holidays.
+market_opens_with_holidays = rrule.rrule(
+    rrule.DAILY,
+    byweekday=WEEKDAYS,
+    byhour = 14,
+    byminute = 30,
+    cache = True,
+    dtstart=datetime(1970, 1, 1, tzinfo = pytz.utc),
+)
+
+# Recurrence rule that generates all market closes since Jan 1, 1970.
+# This does not exclude holidays.
+market_closes_with_holidays = rrule.rrule(
+    rrule.DAILY,
+    byweekday=WEEKDAYS,
+    byhour = 21,
+    byminute = 0,
+    cache = True,
+    dtstart=datetime(1970, 1, 1, tzinfo = pytz.utc),
+)
+
+# Recurrence rules for excluding the market open/close on new years.
+new_years_opens = rrule.rrule(
+    rrule.MONTHLY,
+    byyearday = 1,
+    byhour = 14,
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+new_years_closes = rrule.rrule(
+    rrule.MONTHLY,
+    byyearday = 1,
+    byhour = 21,
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+
+# Recurrence rules for excluding MLK day. It is always the third
+# monday in January.
+mlk_opens = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 1,
+    byweekday = (rrule.MO(3)),
+    byhour = 14, 
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+mlk_closes = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 1,
+    byweekday = (rrule.MO(+3)),
+    byhour = 21, 
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+) 
+
+# Recurrence rules for generating the market open/close for
+# presidents' day.  Presidents' day always occurs on the third monday
+# of February.
+presidents_day_opens = rrule.rrule(
+    rrule.MONTHLY, 
+    bymonth = 2,
+    byweekday = (rrule.MO(3)),
+    byhour = 14,
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+presidents_day_closes = rrule.rrule(
+    rrule.MONTHLY, 
+    bymonth = 2,
+    byweekday = (rrule.MO(3)),
+    byhour = 21,
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+
+# Recurrence rules for generating the market open/close for good
+# friday.  Good friday always falls 2 days before easter, which
+# thankfully is a built-in refernce in this module.
+good_friday_opens = rrule.rrule(
+    rrule.DAILY,
+    byeaster = -2,
+    byhour = 14,
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+good_friday_closes = rrule.rrule(
+    rrule.DAILY,
+    byeaster = -2,
+    byhour = 21,
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+) 
+
+# Recurrence rules for generating the market open/close for memorial
+# day. Memorial day always occurs on the last monday of May.
+memorial_day_opens = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 5,
+    byweekday = (rrule.MO(-1)),
+    byhour = 14, 
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+memorial_day_closes = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 5,
+    byweekday = (rrule.MO(-1)),
+    byhour = 21, 
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+
+# Recurrence rules for generating the market open/close for July 4th.
+july_4th_opens = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 6,
+    bymonthday = 4,
+    byhour = 14,
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+july_4th_closes = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 6,
+    bymonthday = 4,
+    byhour = 21,
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+
+# Recurrence rule for generating the market open/close for labor day.
+# Labor day is always the first monday of September.
+labor_day_opens = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 9,
+    byweekday = (rrule.MO(1)),
+    byhour = 14, 
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+labor_day_closes = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 9,
+    byweekday = (rrule.MO(1)),
+    byhour = 21, 
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+
+# Recurrence rule for generating the market open/close for
+# thanksgiving.  Thanksgiving always falls on the fourth thursday in
+# November. (Who decides how these holidays work!?!)
+thanksgiving_opens = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 11,
+    byweekday = (rrule.TH(-1)),
+    byhour = 14,
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+thanksgiving_closes = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 11,
+    byweekday = (rrule.TH(-1)),
+    byhour = 21,
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+
+# Recurrence relation for generating the market open/close for
+# christmas.  Christmas always occurs on december 25th.
+
+christmas_opens = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 12,
+    bymonthday = 25,
+    byhour = 14,
+    byminute = 30,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+christmas_closes = rrule.rrule(
+    rrule.MONTHLY,
+    bymonth = 12,
+    bymonthday = 25,
+    byhour = 21,
+    byminute = 0,
+    cache = True,
+    dtstart = datetime(1970, 1,1,tzinfo = pytz.utc)
+)
+# All NYSE observed holidays.
+holiday_opens = [
+    new_years_opens,
+    mlk_opens,
+    presidents_day_opens,
+    good_friday_opens,
+    memorial_day_opens,
+    july_4th_opens,
+    labor_day_opens,
+    thanksgiving_opens,
+    christmas_opens
+]
+holiday_closes = [
+    new_years_closes,
+    mlk_closes,
+    presidents_day_closes,
+    good_friday_closes,
+    memorial_day_closes,
+    july_4th_closes,
+    labor_day_closes,
+    thanksgiving_closes,
+    christmas_closes
+]
+
+# Valid market opens are given by all market opens minus holidays.
+opens = rrule.rruleset()
+opens.rrule(market_opens_with_holidays)
+for holiday_rule in holiday_opens:
+    opens.exrule(holiday_rule)
+
+closes = rrule.rruleset()
+closes.rrule(market_closes_with_holidays)
+for holiday_rule in holiday_closes:
+    closes.exrule(holiday_rule)