diff --git a/tests/test_perf_tracking.py b/tests/test_perf_tracking.py
index 91c71703..411c955a 100644
--- a/tests/test_perf_tracking.py
+++ b/tests/test_perf_tracking.py
@@ -16,15 +16,16 @@
 from __future__ import division
 
 import collections
+import datetime
 import logging
 import operator
 
 import unittest
 from nose_parameterized import parameterized
-import datetime
 import pytz
 import itertools
 
+import pandas as pd
 from six.moves import range, zip
 
 import zipline.utils.factory as factory
@@ -37,9 +38,8 @@ from zipline.finance.trading import SimulationParameters
 from zipline.finance.blotter import Order
 from zipline.finance.commission import PerShare, PerTrade, PerDollar
 from zipline.finance import trading
-from zipline.protocol import DATASOURCE_TYPE
 from zipline.utils.factory import create_random_simulation_parameters
-import zipline.protocol
+import zipline.protocol as zp
 from zipline.protocol import Event
 
 logger = logging.getLogger('Test Perf Tracking')
@@ -49,44 +49,101 @@ oneday = datetime.timedelta(days=1)
 tradingday = datetime.timedelta(hours=6, minutes=30)
 
 
-def create_txn(event, price, amount):
-    mock_order = Order(None, None, event.sid, id=None)
-    txn = create_transaction(event, mock_order, price, amount)
-    txn.source_id = 'MockTransactionSource'
-    return txn
+def create_txn(trade_event, price, amount):
+    """
+    Create a fake transaction to be filled and processed prior to the execution
+    of a given trade event.
+    """
+    mock_order = Order(trade_event.dt, trade_event.sid, amount, id=None)
+    return create_transaction(trade_event, mock_order, price, amount)
 
 
 def benchmark_events_in_range(sim_params):
     return [
         Event({'dt': dt,
                'returns': ret,
-               'type':
-               zipline.protocol.DATASOURCE_TYPE.BENCHMARK,
-               'source_id': 'benchmarks'})
+               'type': zp.DATASOURCE_TYPE.BENCHMARK,
+               # We explicitly rely on the behavior that benchmarks sort before
+               # any other events.
+               'source_id': '1Abenchmarks'})
         for dt, ret in trading.environment.benchmark_returns.iterkv()
         if dt.date() >= sim_params.period_start.date()
         and dt.date() <= sim_params.period_end.date()
     ]
 
 
-def calculate_results(host, events):
+def calculate_results(host,
+                      trade_events,
+                      dividend_events=None,
+                      splits=None,
+                      txns=None):
+    """
+    Run the given events through a stripped down version of the loop in
+    AlgorithmSimulator.transform.
+
+    IMPORTANT NOTE FOR TEST WRITERS/READERS:
+
+    This loop has some wonky logic for the order of event processing for
+    datasource types.  This exists mostly to accomodate legacy tests accomodate
+    existing tests that were making assumptions about how events would be
+    sorted.
+
+    In particular:
+
+        - Dividends passed for a given date are processed PRIOR to any events
+          for that date.
+        - Splits passed for a given date are process AFTER any events for that
+          date.
+
+    Tests that use this helper should not be considered useful guarantees of
+    the behavior of AlgorithmSimulator on a stream containing the same events
+    unless the subgroups have been explicitly re-sorted in this way.
+    """
+
+    txns = txns or []
+    splits = splits or []
 
     perf_tracker = perf.PerformanceTracker(host.sim_params)
+    if dividend_events is not None:
+        dividend_frame = pd.DataFrame(
+            [
+                event.to_series(index=zp.DIVIDEND_FIELDS)
+                for event in dividend_events
+            ],
+        )
+        perf_tracker.update_dividends(dividend_frame)
 
-    events = sorted(events, key=lambda ev: ev.dt)
-    all_events = date_sorted_sources(events, host.benchmark_events)
+    # Raw trades
+    trade_events = sorted(trade_events, key=lambda ev: (ev.dt, ev.source_id))
 
-    filtered_events = (filt_event for filt_event in all_events
-                       if filt_event.dt <= events[-1].dt)
-    grouped_events = itertools.groupby(filtered_events, lambda x: x.dt)
+    # Add a benchmark event for each date.
+    trades_plus_bm = date_sorted_sources(trade_events, host.benchmark_events)
+
+    # Filter out benchmark events that are later than the last trade date.
+    filtered_trades_plus_bm = (filt_event for filt_event in trades_plus_bm
+                               if filt_event.dt <= trade_events[-1].dt)
+
+    grouped_trades_plus_bm = itertools.groupby(filtered_trades_plus_bm,
+                                               lambda x: x.dt)
     results = []
 
     bm_updated = False
-    for date, group in grouped_events:
+    for date, group in grouped_trades_plus_bm:
+
+        for txn in filter(lambda txn: txn.dt == date, txns):
+            # Process txns for this date.
+            perf_tracker.process_event(txn)
+
         for event in group:
+
             perf_tracker.process_event(event)
-            if event.type == DATASOURCE_TYPE.BENCHMARK:
+            if event.type == zp.DATASOURCE_TYPE.BENCHMARK:
                 bm_updated = True
+
+        for split in filter(lambda split: split.dt == date, splits):
+            # Process splits for this date.
+            perf_tracker.process_event(split)
+
         if bm_updated:
             msg = perf_tracker.handle_market_close_daily()
             results.append(msg)
@@ -105,62 +162,67 @@ class TestSplitPerformance(unittest.TestCase):
         self.benchmark_events = benchmark_events_in_range(self.sim_params)
 
     def test_split_long_position(self):
-        with trading.TradingEnvironment() as env:
-            events = factory.create_trade_history(
+        events = factory.create_trade_history(
+            1,
+            [20, 20],
+            [100, 100],
+            oneday,
+            self.sim_params
+        )
+
+        # set up a long position in sid 1
+        # 100 shares at $20 apiece = $2000 position
+        txns = [create_txn(events[0], 20, 100)]
+
+        # set up a split with ratio 3 occurring at the start of the second
+        # day.
+        splits = [
+            factory.create_split(
                 1,
-                [20, 20],
-                [100, 100],
-                oneday,
-                self.sim_params
-            )
+                3,
+                events[1].dt,
+            ),
+        ]
 
-            # set up a long position in sid 1
-            # 100 shares at $20 apiece = $2000 position
-            events.insert(0, create_txn(events[0], 20, 100))
+        results = calculate_results(self, events, txns=txns, splits=splits)
 
-            # set up a split with ratio 3
-            events.append(factory.create_split(1, 3,
-                          env.next_trading_day(events[1].dt)))
+        # should have 33 shares (at $60 apiece) and $20 in cash
+        self.assertEqual(2, len(results))
 
-            results = calculate_results(self, events)
+        latest_positions = results[1]['daily_perf']['positions']
+        self.assertEqual(1, len(latest_positions))
 
-            # should have 33 shares (at $60 apiece) and $20 in cash
-            self.assertEqual(2, len(results))
+        # check the last position to make sure it's been updated
+        position = latest_positions[0]
 
-            latest_positions = results[1]['daily_perf']['positions']
-            self.assertEqual(1, len(latest_positions))
+        self.assertEqual(1, position['sid'])
+        self.assertEqual(33, position['amount'])
+        self.assertEqual(60, position['cost_basis'])
+        self.assertEqual(60, position['last_sale_price'])
 
-            # check the last position to make sure it's been updated
-            position = latest_positions[0]
+        # since we started with $10000, and we spent $2000 on the
+        # position, but then got $20 back, we should have $8020
+        # (or close to it) in cash.
 
-            self.assertEqual(1, position['sid'])
-            self.assertEqual(33, position['amount'])
-            self.assertEqual(60, position['cost_basis'])
-            self.assertEqual(60, position['last_sale_price'])
+        # we won't get exactly 8020 because sometimes a split is
+        # denoted as a ratio like 0.3333, and we lose some digits
+        # of precision.  thus, make sure we're pretty close.
+        daily_perf = results[1]['daily_perf']
 
-            # since we started with $10000, and we spent $2000 on the
-            # position, but then got $20 back, we should have $8020
-            # (or close to it) in cash.
+        self.assertTrue(
+            zp_math.tolerant_equals(8020,
+                                    daily_perf['ending_cash'], 1))
 
-            # we won't get exactly 8020 because sometimes a split is
-            # denoted as a ratio like 0.3333, and we lose some digits
-            # of precision.  thus, make sure we're pretty close.
-            daily_perf = results[1]['daily_perf']
-
-            self.assertTrue(
-                zp_math.tolerant_equals(8020,
-                                        daily_perf['ending_cash'], 1))
-
-            for i, result in enumerate(results):
-                for perf_kind in ('daily_perf', 'cumulative_perf'):
-                    perf_result = result[perf_kind]
-                    # prices aren't changing, so pnl and returns should be 0.0
-                    self.assertEqual(0.0, perf_result['pnl'],
-                                     "day %s %s pnl %s instead of 0.0" %
-                                     (i, perf_kind, perf_result['pnl']))
-                    self.assertEqual(0.0, perf_result['returns'],
-                                     "day %s %s returns %s instead of 0.0" %
-                                     (i, perf_kind, perf_result['returns']))
+        for i, result in enumerate(results):
+            for perf_kind in ('daily_perf', 'cumulative_perf'):
+                perf_result = result[perf_kind]
+                # prices aren't changing, so pnl and returns should be 0.0
+                self.assertEqual(0.0, perf_result['pnl'],
+                                 "day %s %s pnl %s instead of 0.0" %
+                                 (i, perf_kind, perf_result['pnl']))
+                self.assertEqual(0.0, perf_result['returns'],
+                                 "day %s %s returns %s instead of 0.0" %
+                                 (i, perf_kind, perf_result['returns']))
 
 
 class TestCommissionEvents(unittest.TestCase):
@@ -197,28 +259,29 @@ class TestCommissionEvents(unittest.TestCase):
             transactions = [create_txn(events[0], 20, i)
                             for i in [50, 100, 150]]
 
-            # Create commission models
+            # Create commission models and validate that produce expected
+            # commissions.
             models = [PerShare(cost=0.01, min_trade_cost=1.00),
                       PerTrade(cost=5.00),
                       PerDollar(cost=0.0015)]
+            expected_results = [3.50, 15.0, 9.0]
 
-            # Aggregate commission amounts
-            total_commission = 0
-            for model in models:
+            for model, expected in zip(models, expected_results):
+                total_commission = 0
                 for trade in transactions:
                     total_commission += model.calculate(trade)[1]
-            self.assertEqual(total_commission, 27.5)
+                self.assertEqual(total_commission, expected)
 
-            cash_adj_dt = self.sim_params.first_open \
-                + datetime.timedelta(hours=3)
-            cash_adjustment = factory.create_commission(1, 300.0,
-                                                        cash_adj_dt)
+            # Verify that commission events are handled correctly by
+            # PerformanceTracker.
+            cash_adj_dt = events[0].dt
+            cash_adjustment = factory.create_commission(1, 300.0, cash_adj_dt)
+            events.append(cash_adjustment)
 
             # Insert a purchase order.
-            events.insert(0, create_txn(events[0], 20, 1))
+            txns = [create_txn(events[0], 20, 1)]
+            results = calculate_results(self, events, txns=txns)
 
-            events.insert(1, cash_adjustment)
-            results = calculate_results(self, events)
             # Validate that we lost 320 dollars from our cash pool.
             self.assertEqual(results[-1]['cumulative_perf']['ending_cash'],
                              9680)
@@ -230,31 +293,31 @@ class TestCommissionEvents(unittest.TestCase):
         """
         Ensure no div-by-zero errors.
         """
-        with trading.TradingEnvironment():
-            events = factory.create_trade_history(
-                1,
-                [10, 10, 10, 10, 10],
-                [100, 100, 100, 100, 100],
-                oneday,
-                self.sim_params
-            )
+        events = factory.create_trade_history(
+            1,
+            [10, 10, 10, 10, 10],
+            [100, 100, 100, 100, 100],
+            oneday,
+            self.sim_params
+        )
 
-            cash_adj_dt = self.sim_params.first_open \
-                + datetime.timedelta(hours=3)
-            cash_adjustment = factory.create_commission(1, 300.0,
-                                                        cash_adj_dt)
+        # Buy and sell the same sid so that we have a zero position by the
+        # time of events[3].
+        txns = [
+            create_txn(events[0], 20, 1),
+            create_txn(events[1], 20, -1),
+        ]
 
-            # Insert a purchase order.
-            events.insert(0, create_txn(events[0], 20, 1))
+        # Add a cash adjustment at the time of event[3].
+        cash_adj_dt = events[3].dt
+        cash_adjustment = factory.create_commission(1, 300.0, cash_adj_dt)
 
-            # Sell that order.
-            events.insert(1, create_txn(events[1], 20, -1))
+        events.append(cash_adjustment)
 
-            events.insert(2, cash_adjustment)
-            results = calculate_results(self, events)
-            # Validate that we lost 300 dollars from our cash pool.
-            self.assertEqual(results[-1]['cumulative_perf']['ending_cash'],
-                             9700)
+        results = calculate_results(self, events, txns=txns)
+        # Validate that we lost 300 dollars from our cash pool.
+        self.assertEqual(results[-1]['cumulative_perf']['ending_cash'],
+                         9700)
 
     def test_commission_no_position(self):
         """
@@ -269,12 +332,11 @@ class TestCommissionEvents(unittest.TestCase):
                 self.sim_params
             )
 
-            cash_adj_dt = self.sim_params.first_open \
-                + datetime.timedelta(hours=3)
-            cash_adjustment = factory.create_commission(1, 300.0,
-                                                        cash_adj_dt)
+            # Add a cash adjustment at the time of event[3].
+            cash_adj_dt = events[3].dt
+            cash_adjustment = factory.create_commission(1, 300.0, cash_adj_dt)
+            events.append(cash_adjustment)
 
-            events.insert(0, cash_adjustment)
             results = calculate_results(self, events)
             # Validate that we lost 300 dollars from our cash pool.
             self.assertEqual(results[-1]['cumulative_perf']['ending_cash'],
@@ -312,24 +374,27 @@ class TestDividendPerformance(unittest.TestCase):
                 oneday,
                 self.sim_params
             )
-
             dividend = factory.create_dividend(
                 1,
                 10.00,
                 # declared date, when the algorithm finds out about
                 # the dividend
-                events[1].dt,
-                # ex_date, when the algorithm is credited with the
-                # dividend
+                events[0].dt,
+                # ex_date, the date before which the algorithm must hold stock
+                # to receive the dividend
                 events[1].dt,
                 # pay date, when the algorithm receives the dividend.
                 events[2].dt
             )
 
-            txn = create_txn(events[0], 10.0, 100)
-            events.insert(0, txn)
-            events.insert(1, dividend)
-            results = calculate_results(self, events)
+            # Simulate a transaction being filled prior to the ex_date.
+            txns = [create_txn(events[0], 10.0, 100)]
+            results = calculate_results(
+                self,
+                events,
+                dividend_events=[dividend],
+                txns=txns,
+            )
 
             self.assertEqual(len(results), 5)
             cumulative_returns = \
@@ -368,18 +433,22 @@ class TestDividendPerformance(unittest.TestCase):
                 ratio=2,
                 # declared date, when the algorithm finds out about
                 # the dividend
-                declared_date=events[1].dt,
-                # ex_date, when the algorithm is credited with the
-                # dividend
+                declared_date=events[0].dt,
+                # ex_date, the date before which the algorithm must hold stock
+                # to receive the dividend
                 ex_date=events[1].dt,
                 # pay date, when the algorithm receives the dividend.
                 pay_date=events[2].dt
             )
 
-            txn = create_txn(events[0], 10.0, 100)
-            events.insert(0, txn)
-            events.insert(1, dividend)
-            results = calculate_results(self, events)
+            txns = [create_txn(events[0], 10.0, 100)]
+
+            results = calculate_results(
+                self,
+                events,
+                dividend_events=[dividend],
+                txns=txns,
+            )
 
             self.assertEqual(len(results), 5)
             cumulative_returns = \
@@ -398,7 +467,7 @@ class TestDividendPerformance(unittest.TestCase):
                 [event['cumulative_perf']['ending_cash'] for event in results]
             self.assertEqual(cash_pos, [9000] * 5)
 
-    def test_post_ex_long_position_receives_no_dividend(self):
+    def test_long_position_purchased_on_ex_date_receives_no_dividend(self):
         # post some trades in the market
         events = factory.create_trade_history(
             1,
@@ -411,15 +480,20 @@ class TestDividendPerformance(unittest.TestCase):
         dividend = factory.create_dividend(
             1,
             10.00,
-            events[0].dt,
-            events[1].dt,
-            events[2].dt
+            events[0].dt,  # Declared date
+            events[1].dt,  # Exclusion date
+            events[2].dt   # Pay date
         )
 
-        events.insert(1, dividend)
-        txn = create_txn(events[3], 10.0, 100)
-        events.insert(4, txn)
-        results = calculate_results(self, events)
+        # Simulate a transaction being filled on the ex_date.
+        txns = [create_txn(events[1], 10.0, 100)]
+
+        results = calculate_results(
+            self,
+            events,
+            dividend_events=[dividend],
+            txns=txns,
+        )
 
         self.assertEqual(len(results), 5)
         cumulative_returns = \
@@ -428,10 +502,11 @@ class TestDividendPerformance(unittest.TestCase):
         daily_returns = [event['daily_perf']['returns'] for event in results]
         self.assertEqual(daily_returns, [0, 0, 0, 0, 0])
         cash_flows = [event['daily_perf']['capital_used'] for event in results]
-        self.assertEqual(cash_flows, [0, 0, -1000, 0, 0])
+        self.assertEqual(cash_flows, [0, -1000, 0, 0, 0])
         cumulative_cash_flows = \
             [event['cumulative_perf']['capital_used'] for event in results]
-        self.assertEqual(cumulative_cash_flows, [0, 0, -1000, -1000, -1000])
+        self.assertEqual(cumulative_cash_flows,
+                         [0, -1000, -1000, -1000, -1000])
 
     def test_selling_before_dividend_payment_still_gets_paid(self):
         # post some trades in the market
@@ -446,17 +521,21 @@ class TestDividendPerformance(unittest.TestCase):
         dividend = factory.create_dividend(
             1,
             10.00,
-            events[0].dt,
-            events[1].dt,
-            events[3].dt
+            events[0].dt,  # Declared date
+            events[1].dt,  # Exclusion date
+            events[3].dt   # Pay date
         )
 
         buy_txn = create_txn(events[0], 10.0, 100)
-        events.insert(1, buy_txn)
-        sell_txn = create_txn(events[3], 10.0, -100)
-        events.insert(4, sell_txn)
-        events.insert(0, dividend)
-        results = calculate_results(self, events)
+        sell_txn = create_txn(events[2], 10.0, -100)
+        txns = [buy_txn, sell_txn]
+
+        results = calculate_results(
+            self,
+            events,
+            dividend_events=[dividend],
+            txns=txns,
+        )
 
         self.assertEqual(len(results), 5)
         cumulative_returns = \
@@ -489,11 +568,15 @@ class TestDividendPerformance(unittest.TestCase):
         )
 
         buy_txn = create_txn(events[1], 10.0, 100)
-        events.insert(1, buy_txn)
-        sell_txn = create_txn(events[3], 10.0, -100)
-        events.insert(3, sell_txn)
-        events.insert(1, dividend)
-        results = calculate_results(self, events)
+        sell_txn = create_txn(events[2], 10.0, -100)
+        txns = [buy_txn, sell_txn]
+
+        results = calculate_results(
+            self,
+            events,
+            dividend_events=[dividend],
+            txns=txns,
+        )
 
         self.assertEqual(len(results), 6)
         cumulative_returns = \
@@ -525,14 +608,18 @@ class TestDividendPerformance(unittest.TestCase):
             1,
             10.00,
             events[0].dt,
-            events[1].dt,
+            events[0].dt,
             pay_date
         )
 
-        buy_txn = create_txn(events[1], 10.0, 100)
-        events.insert(2, buy_txn)
-        events.insert(1, dividend)
-        results = calculate_results(self, events)
+        txns = [create_txn(events[1], 10.0, 100)]
+
+        results = calculate_results(
+            self,
+            events,
+            dividend_events=[dividend],
+            txns=txns,
+        )
 
         self.assertEqual(len(results), 5)
         cumulative_returns = \
@@ -569,10 +656,14 @@ class TestDividendPerformance(unittest.TestCase):
             events[3].dt
         )
 
-        txn = create_txn(events[1], 10.0, -100)
-        events.insert(1, txn)
-        events.insert(0, dividend)
-        results = calculate_results(self, events)
+        txns = [create_txn(events[1], 10.0, -100)]
+
+        results = calculate_results(
+            self,
+            events,
+            dividend_events=[dividend],
+            txns=txns,
+        )
 
         self.assertEqual(len(results), 5)
         cumulative_returns = \
@@ -604,8 +695,11 @@ class TestDividendPerformance(unittest.TestCase):
             events[2].dt
         )
 
-        events.insert(1, dividend)
-        results = calculate_results(self, events)
+        results = calculate_results(
+            self,
+            events,
+            dividend_events=[dividend],
+        )
 
         self.assertEqual(len(results), 5)
         cumulative_returns = \
@@ -1161,13 +1255,13 @@ class TestPerformanceTracker(unittest.TestCase):
         # 19 20 21 22 23 24 25
         # 26 27 28 29 30 31
         start_dt = datetime.datetime(year=2008,
-                                     month=10,
-                                     day=9,
-                                     tzinfo=pytz.utc)
+                            month=10,
+                            day=9,
+                            tzinfo=pytz.utc)
         end_dt = datetime.datetime(year=2008,
-                                   month=10,
-                                   day=16,
-                                   tzinfo=pytz.utc)
+                          month=10,
+                          day=16,
+                          tzinfo=pytz.utc)
 
         trade_count = 6
         sid = 133
@@ -1243,10 +1337,10 @@ class TestPerformanceTracker(unittest.TestCase):
 
         # Extract events with transactions to use for verification.
         txns = [event for event in
-                events if event.type == DATASOURCE_TYPE.TRANSACTION]
+                events if event.type == zp.DATASOURCE_TYPE.TRANSACTION]
 
         orders = [event for event in
-                  events if event.type == DATASOURCE_TYPE.ORDER]
+                  events if event.type == zp.DATASOURCE_TYPE.ORDER]
 
         all_events = date_sorted_sources(events, benchmark_events)
 
@@ -1328,7 +1422,7 @@ class TestPerformanceTracker(unittest.TestCase):
             benchmark_event_1 = Event({
                 'dt': start_dt,
                 'returns': 0.01,
-                'type': DATASOURCE_TYPE.BENCHMARK
+                'type': zp.DATASOURCE_TYPE.BENCHMARK
             })
 
             foo_event_2 = factory.create_trade(
@@ -1338,7 +1432,7 @@ class TestPerformanceTracker(unittest.TestCase):
             benchmark_event_2 = Event({
                 'dt': start_dt + datetime.timedelta(minutes=1),
                 'returns': 0.02,
-                'type': DATASOURCE_TYPE.BENCHMARK
+                'type': zp.DATASOURCE_TYPE.BENCHMARK
             })
 
             events = [
diff --git a/zipline/algorithm.py b/zipline/algorithm.py
index ea3dbc54..b93c62ac 100644
--- a/zipline/algorithm.py
+++ b/zipline/algorithm.py
@@ -665,6 +665,13 @@ class TradingAlgorithm(object):
         """
         self.blotter.transact = transact
 
+    def update_dividends(self, dividend_frame):
+        """
+        Set DataFrame used to process dividends.  DataFrame columns should
+        contain at least the entries in zp.DIVIDEND_FIELDS.
+        """
+        self.perf_tracker.update_dividends(dividend_frame)
+
     @api_method
     def set_slippage(self, slippage):
         if not isinstance(slippage, SlippageModel):
diff --git a/zipline/finance/performance/period.py b/zipline/finance/performance/period.py
index 7b51effc..d943c5ed 100644
--- a/zipline/finance/performance/period.py
+++ b/zipline/finance/performance/period.py
@@ -75,8 +75,10 @@ import logbook
 
 import numpy as np
 import pandas as pd
-from collections import Counter, OrderedDict, defaultdict
-
+from collections import (
+    defaultdict,
+    OrderedDict,
+)
 from six import iteritems, itervalues
 
 import zipline.protocol as zp
@@ -123,6 +125,10 @@ class PerformancePeriod(object):
         self._positions_store = zp.Positions()
         self.serialize_positions = serialize_positions
 
+        self._unpaid_dividends = pd.DataFrame(
+            columns=zp.DIVIDEND_PAYMENT_FIELDS,
+        )
+
     def rollover(self):
         self.starting_value = self.ending_value
         self.starting_cash = self.ending_cash
@@ -142,14 +148,6 @@ class PerformancePeriod(object):
             self._position_last_sale_prices = \
                 self._position_last_sale_prices.append(pd.Series({sid: 0.0}))
 
-    def add_dividend(self, div):
-        # The dividend is received on midnight of the dividend
-        # declared date. We calculate the dividends based on the amount of
-        # stock owned on midnight of the ex dividend date. However, the cash
-        # is not dispersed until the payment date, which is
-        # included in the event.
-        self.positions[div.sid].add_dividend(div)
-
     def handle_split(self, split):
         if split.sid in self.positions:
             # Make the position object handle the split. It returns the
@@ -163,39 +161,82 @@ class PerformancePeriod(object):
             if leftover_cash > 0:
                 self.handle_cash_payment(leftover_cash)
 
-    def update_dividends(self, todays_date):
+    def earn_dividends(self, dividend_frame):
         """
-        Check the payment date and ex date against today's date
-        to determine if we are owed a dividend payment or if the
-        payment has been disbursed.
+        Given a frame of dividends whose ex_dates are all the next trading day,
+        calculate and store the cash and/or stock payments to be paid on each
+        dividend's pay date.
         """
-        cash_payments = 0.0
-        stock_payments = Counter()  # maps sid to number of shares paid
-        for sid, pos in iteritems(self.positions):
-            cash_payment, stock_payment = pos.update_dividends(todays_date)
-            cash_payments += cash_payment
-            stock_payments.update(stock_payment)
+        earned = dividend_frame.apply(self._maybe_earn_dividend, axis=1)\
+                               .dropna(how='all')
+        if len(earned) > 0:
+            # Store the earned dividends so that they can be paid on the
+            # dividends' pay_dates.
+            self._unpaid_dividends = pd.concat(
+                [self._unpaid_dividends, earned],
+            )
 
-        for stock, payment in iteritems(stock_payments):
+    def _maybe_earn_dividend(self, dividend):
+        """
+        Take a historical dividend record and return a Series with fields in
+        zipline.protocol.DIVIDEND_FIELDS (plus an 'id' field) representing
+        the cash/stock amount we are owed when the dividend is paid.
+        """
+        if dividend['sid'] in self.positions:
+            return self.positions[dividend['sid']].earn_dividend(dividend)
+        else:
+            return zp.dividend_payment()
+
+    def pay_dividends(self, dividend_frame):
+        """
+        Given a frame of dividends whose pay_dates are all the next trading
+        day, grant the cash and/or stock payments that were calculated on the
+        given dividends' ex dates.
+        """
+        payments = dividend_frame.apply(self._maybe_pay_dividend, axis=1)\
+                                 .dropna(how='all')
+
+        # Mark these dividends as paid by dropping them from our unpaid
+        # table.
+        self._unpaid_dividends.drop(payments.index)
+
+        # Add cash equal to the net cash payed from all dividends.  Note that
+        # "negative cash" is effectively paid if we're short a security,
+        # representing the fact that we're required to reimburse the owner of
+        # the stock for any dividends paid while borrowing.
+        net_cash_payment = payments['cash_amount'].fillna(0).sum()
+        if net_cash_payment:
+            self.handle_cash_payment(net_cash_payment)
+
+        # Add stock for any stock dividends paid.  Again, the values here may
+        # be negative in the case of short positions.
+        stock_payments = payments[payments['payment_sid'].notnull()]
+        for _, row in stock_payments.iterrows():
+            stock = row['payment_sid']
+            share_count = row['share_count']
             position = self.positions[stock]
-            position.amount += payment
+
+            position.amount += share_count
             self.ensure_position_index(stock)
             self._position_amounts[stock] = position.amount
             self._position_last_sale_prices[stock] = \
                 position.last_sale_price
 
-        # credit our cash balance with the dividend payments, or
-        # if we are short, debit our cash balance with the
-        # payments.
-        # debit our cumulative cash spent with the dividend
-        # payments, or credit our cumulative cash spent if we are
-        # short the stock.
-        self.handle_cash_payment(cash_payments)
-
-        # recalculate performance, including the dividend
-        # payments
+        # Recalculate performance after applying dividend benefits.
         self.calculate_performance()
 
+    def _maybe_pay_dividend(self, dividend):
+        """
+        Take a historical dividend record, look up any stored record of
+        cash/stock we are owed for that dividend, and return a Series
+        with fields drawn from zipline.protocol.DIVIDEND_PAYMENT_FIELDS.
+        """
+        try:
+            unpaid_dividend = self._unpaid_dividends.loc[dividend['guid']]
+            return unpaid_dividend
+        except KeyError:
+            return zp.dividend_payment()
+
     def handle_cash_payment(self, payment_amount):
         self.adjust_cash(payment_amount)
 
@@ -255,6 +296,9 @@ class PerformancePeriod(object):
     def execute_transaction(self, txn):
         # Update Position
         # ----------------
+
+        # NOTE: self.positions has defaultdict semantics, so this will create
+        # an empty position if one does not already exist.
         position = self.positions[txn.sid]
         position.update(txn)
         self.ensure_position_index(txn.sid)
diff --git a/zipline/finance/performance/position.py b/zipline/finance/performance/position.py
index 45c36d73..da2a4166 100644
--- a/zipline/finance/performance/position.py
+++ b/zipline/finance/performance/position.py
@@ -33,10 +33,13 @@ Position Tracking
 """
 
 from __future__ import division
-import logbook
-import math
+from math import (
+    copysign,
+    floor,
+)
 
-from collections import Counter
+import logbook
+import zipline.protocol as zp
 
 log = logbook.Logger('Performance')
 
@@ -44,69 +47,43 @@ log = logbook.Logger('Performance')
 class Position(object):
 
     def __init__(self, sid, amount=0, cost_basis=0.0,
-                 last_sale_price=0.0, last_sale_date=None,
-                 dividends=None):
+                 last_sale_price=0.0, last_sale_date=None):
+
         self.sid = sid
         self.amount = amount
         self.cost_basis = cost_basis  # per share
         self.last_sale_price = last_sale_price
         self.last_sale_date = last_sale_date
-        self.dividends = dividends or []
 
-    def update_dividends(self, midnight_utc):
+    def earn_dividend(self, dividend):
         """
-        midnight_utc is the 0 hour for the current (not yet open) trading day.
-        This method will be invoked at the end of the market
-        close handling, before the next market open.
+        Register the number of shares we held at this dividend's ex date so
+        that we can pay out the correct amount on the dividend's pay date.
         """
-        cash_payment = 0.0
-        stock_payment = Counter()  # maps sid to number of shares paid
-        unpaid_dividends = []
-        for dividend in self.dividends:
-            if midnight_utc == dividend.ex_date:
-                # if we own shares at midnight of the div_ex date
-                # we are entitled to the dividend.
-                dividend.amount_on_ex_date = self.amount
-                # stock dividend
-                if dividend.payment_sid:
-                    # e.g., 33.333
-                    raw_share_count = self.amount * float(dividend.ratio)
-                    # e.g., 33
-                    dividend.stock_payment = math.floor(raw_share_count)
-                else:
-                    dividend.stock_payment = None
-                # cash dividend
-                if dividend.net_amount:
-                    dividend.cash_payment = self.amount * dividend.net_amount
-                elif dividend.gross_amount:
-                    dividend.cash_payment = self.amount * dividend.gross_amount
-                else:
-                    dividend.cash_payment = None
+        assert dividend['sid'] == self.sid
+        out = {'guid': dividend['guid']}
 
-            if midnight_utc == dividend.pay_date:
-                # if it is the payment date, include this
-                # dividend's actual payment (calculated on
-                # ex_date)
-                if dividend.stock_payment:
-                    stock_payment[dividend.payment_sid] += \
-                        dividend.stock_payment
+        # stock dividend
+        if dividend['payment_sid']:
+            out['payment_sid'] = dividend['payment_sid']
+            out['share_count'] = floor(self.amount * float(dividend['ratio']))
 
-                if dividend.cash_payment:
-                    cash_payment += dividend.cash_payment
-            else:
-                unpaid_dividends.append(dividend)
+        # cash dividend
+        if dividend['net_amount']:
+            out['cash_amount'] = self.amount * dividend['net_amount']
+        elif dividend['gross_amount']:
+            out['cash_amount'] = self.amount * dividend['gross_amount']
 
-        self.dividends = unpaid_dividends
-        return cash_payment, stock_payment
+        payment_owed = zp.dividend_payment(out)
+        return payment_owed
 
-    def add_dividend(self, dividend):
-        self.dividends.append(dividend)
-
-    # Update the position by the split ratio, and return the
-    # resulting fractional share that will be converted into cash.
-
-    # Returns the unused cash.
     def handle_split(self, split):
+        """
+        Update the position by the split ratio, and return the resulting
+        fractional share that will be converted into cash.
+
+        Returns the unused cash.
+        """
         if self.sid != split.sid:
             raise Exception("updating split with the wrong sid!")
 
@@ -126,7 +103,7 @@ class Position(object):
         raw_share_count = self.amount / float(ratio)
 
         # e.g., 33
-        full_share_count = math.floor(raw_share_count)
+        full_share_count = floor(raw_share_count)
 
         # e.g., 0.333
         fractional_share_count = raw_share_count - full_share_count
@@ -160,8 +137,8 @@ class Position(object):
         if total_shares == 0:
             self.cost_basis = 0.0
         else:
-            prev_direction = math.copysign(1, self.amount)
-            txn_direction = math.copysign(1, txn.amount)
+            prev_direction = copysign(1, self.amount)
+            txn_direction = copysign(1, txn.amount)
 
             if prev_direction != txn_direction:
                 # we're covering a short or closing a position
diff --git a/zipline/finance/performance/tracker.py b/zipline/finance/performance/tracker.py
index 179c2cb9..d01572b3 100644
--- a/zipline/finance/performance/tracker.py
+++ b/zipline/finance/performance/tracker.py
@@ -60,6 +60,7 @@ Performance Tracking
 from __future__ import division
 import logbook
 
+import numpy as np
 import pandas as pd
 from pandas.tseries.tools import normalize_date
 
@@ -96,6 +97,9 @@ class PerformanceTracker(object):
 
         self.trading_days = all_trading_days[mask]
 
+        self.dividend_frame = pd.DataFrame()
+        self._dividend_count = 0
+
         self.perf_periods = []
 
         if self.emission_rate == 'daily':
@@ -188,6 +192,35 @@ class PerformanceTracker(object):
             self.saved_dt = date
             self.todays_performance.period_close = self.saved_dt
 
+    def update_dividends(self, new_dividends):
+        """
+        Update our dividend frame with new dividends.
+        """
+        # Mark each new dividend with a unique integer id.  This ensures that
+        # we can differentiate dividends whose date/sid fields are otherwise
+        # identical.
+        new_dividends['guid'] = np.arange(
+            self._dividend_count,
+            self._dividend_count + len(new_dividends),
+        )
+        self._dividend_count += len(new_dividends)
+
+        self.dividend_frame = pd.concat(
+            [self.dividend_frame, new_dividends]
+        ).sort(['pay_date', 'ex_date']).set_index('guid', drop=False)
+
+    def initialize_dividends_from_other(self, other):
+        """
+        Helper for copying dividends to a new PerformanceTracker while
+        preserving dividend count.  Useful if a simulation needs to create a
+        new PerformanceTracker mid-stream and wants to preserve stored dividend
+        info.
+
+        Note that this does not copy unpaid dividends.
+        """
+        self.dividend_frame = other.dividend_frame
+        self._dividend_count = other._dividend_count
+
     def update_performance(self):
         # calculate performance as of last trade
         for perf_period in self.perf_periods:
@@ -239,8 +272,7 @@ class PerformanceTracker(object):
                 perf_period.execute_transaction(event)
 
         elif event.type == zp.DATASOURCE_TYPE.DIVIDEND:
-            for perf_period in self.perf_periods:
-                perf_period.add_dividend(event)
+            log.info("Ignoring DIVIDEND event.")
 
         elif event.type == zp.DATASOURCE_TYPE.SPLIT:
             for perf_period in self.perf_periods:
@@ -256,6 +288,7 @@ class PerformanceTracker(object):
 
         elif event.type == zp.DATASOURCE_TYPE.CUSTOM:
             pass
+
         elif event.type == zp.DATASOURCE_TYPE.BENCHMARK:
             if (
                 self.sim_params.data_frequency == 'minute'
@@ -266,16 +299,62 @@ class PerformanceTracker(object):
                 # close, so that calculations are triggered at the right time.
                 # However, risk module uses midnight as the 'day'
                 # marker for returns, so adjust back to midgnight.
-                midnight = event.dt.replace(
-                    hour=0,
-                    minute=0,
-                    second=0,
-                    microsecond=0)
+                midnight = pd.tseries.tools.normalize_date(event.dt)
             else:
                 midnight = event.dt
 
             self.all_benchmark_returns[midnight] = event.returns
 
+    def check_upcoming_dividends(self, midnight_of_date_that_just_ended):
+        """
+        Check if we currently own any stocks with dividends whose ex_date is
+        the next trading day.  Track how much we should be payed on those
+        dividends' pay dates.
+
+        Then check if we are owed cash/stock for any dividends whose pay date
+        is the next trading day.  Apply all such benefits, then recalculate
+        performance.
+        """
+        if len(self.dividend_frame) == 0:
+            # We don't currently know about any dividends for this simulation
+            # period, so bail.
+            return
+
+        next_trading_day_idx = self.trading_days.get_loc(
+            midnight_of_date_that_just_ended,
+        ) + 1
+
+        if next_trading_day_idx < len(self.trading_days):
+            next_trading_day = self.trading_days[next_trading_day_idx]
+        else:
+            # Bail if the next trading day is outside our trading range, since
+            # we won't simulate the next day.
+            return
+
+        # Dividends whose ex_date is the next trading day.  We need to check if
+        # we own any of these stocks so we know to pay them out when the pay
+        # date comes.
+        ex_date_mask = (self.dividend_frame['ex_date'] == next_trading_day)
+        dividends_earnable = self.dividend_frame[ex_date_mask]
+
+        # Dividends whose pay date is the next trading day.  If we held any of
+        # these stocks on midnight before the ex_date, we need to pay these out
+        # now.
+        pay_date_mask = (self.dividend_frame['pay_date'] == next_trading_day)
+        dividends_payable = self.dividend_frame[pay_date_mask]
+
+        for period in self.perf_periods:
+            # TODO SS: There's no reason we should have to duplicate this
+            #          computation, but we do it currently because each perf
+            #          period maintains its own separate positiondict.  We
+            #          should eventually remove this duplication and give each
+            #          period a (preferably read-only) DataFrame of positions.
+            if len(dividends_earnable):
+                period.earn_dividends(dividends_earnable)
+
+            if len(dividends_payable):
+                period.pay_dividends(dividends_payable)
+
     def handle_minute_close(self, dt):
         self.update_performance()
         todays_date = normalize_date(dt)
@@ -291,29 +370,20 @@ class PerformanceTracker(object):
         bench_since_open = \
             self.intraday_risk_metrics.benchmark_cumulative_returns[dt]
 
-        # if we've reached market close, check on dividends
-        if dt == self.market_close:
-            for perf_period in self.perf_periods:
-                perf_period.update_dividends(todays_date)
-
         self.cumulative_risk_metrics.update(todays_date,
                                             self.todays_performance.returns,
                                             bench_since_open)
 
-        # if this is the close, save the returns objects for cumulative
-        # risk calculations
+        # if this is the close, save the returns objects for cumulative risk
+        # calculations and update dividends for the next day.
         if dt == self.market_close:
+            self.check_upcoming_dividends(todays_date)
             self.returns[todays_date] = self.todays_performance.returns
 
     def handle_intraday_market_close(self, new_mkt_open, new_mkt_close):
         """
         Function called at market close only when emitting at minutely
         frequency.
-
-        TODO_SS: Why dont' we call this if we're emitting at daily frequency
-                 but running with a minutely datasource?  Is that just not a
-                 valid combination? If so, why do we draw a distinction between
-                 emission rate and data frequency?
         """
 
         # update_performance should have been called in handle_minute_close
@@ -331,18 +401,16 @@ class PerformanceTracker(object):
         rate.
         """
         self.update_performance()
-        # add the return results from today to the returns series
-        todays_date = normalize_date(self.market_close)
-        self.cumulative_performance.update_dividends(todays_date)
-        self.todays_performance.update_dividends(todays_date)
+        completed_date = normalize_date(self.market_close)
 
-        self.returns[todays_date] = self.todays_performance.returns
+        # add the return results from today to the returns series
+        self.returns[completed_date] = self.todays_performance.returns
 
         # update risk metrics for cumulative performance
         self.cumulative_risk_metrics.update(
-            todays_date,
+            completed_date,
             self.todays_performance.returns,
-            self.all_benchmark_returns[todays_date])
+            self.all_benchmark_returns[completed_date])
 
         # increment the day counter before we move markers forward.
         self.day_count += 1.0
@@ -352,8 +420,8 @@ class PerformanceTracker(object):
         daily_update = self.to_dict()
 
         # On the last day of the test, don't create tomorrow's performance
-        # period.  We may not be able to find the next trading day if we're
-        # at the end of our historical data
+        # period.  We may not be able to find the next trading day if we're at
+        # the end of our historical data
         if self.market_close >= self.last_close:
             return daily_update
 
@@ -366,15 +434,7 @@ class PerformanceTracker(object):
         self.todays_performance.period_open = self.market_open
         self.todays_performance.period_close = self.market_close
 
-        # The dividend calculation for the daily needs to be made
-        # after the rollover. midnight_between is the last midnight
-        # hour between the close of markets and the next open. To
-        # make sure midnight_between matches identically with
-        # dividend data dates, it is in UTC.
-        midnight_between = self.market_open.replace(hour=0, minute=0, second=0,
-                                                    microsecond=0)
-        self.cumulative_performance.update_dividends(midnight_between)
-        self.todays_performance.update_dividends(midnight_between)
+        self.check_upcoming_dividends(completed_date)
 
         return daily_update
 
diff --git a/zipline/protocol.py b/zipline/protocol.py
index 7d78119d..46849537 100644
--- a/zipline/protocol.py
+++ b/zipline/protocol.py
@@ -14,6 +14,7 @@
 # limitations under the License.
 
 from six import iteritems, iterkeys
+import pandas as pd
 
 from . utils.protocol_utils import Enum
 
@@ -34,6 +35,50 @@ DATASOURCE_TYPE = Enum(
     'COMMISSION'
 )
 
+# Expected fields/index values for a dividend Series.
+DIVIDEND_FIELDS = [
+    'declared_date',
+    'ex_date',
+    'gross_amount',
+    'net_amount',
+    'pay_date',
+    'payment_sid',
+    'ratio',
+    'sid',
+]
+# Expected fields/index values for a dividend payment Series.
+DIVIDEND_PAYMENT_FIELDS = ['guid', 'payment_sid', 'cash_amount', 'share_count']
+
+
+def dividend_payment(data=None):
+    """
+    Take a dictionary whose values are in DIVIDEND_PAYMENT_FIELDS and return a
+    series representing the payment of a dividend.
+
+    Guids are assigned to each historical dividend in
+    PerformanceTracker.update_dividends. They are guaranteed to be unique
+    integers with the context of a single simulation. If @data is non-empty, a
+    guid is required to identify the historical dividend associated with this
+    payment.
+
+    Additionally, if @data is non-empty, either data['cash_amount'] should be
+    nonzero or data['payment_sid'] should be a security identifier and
+    data['share_count'] should be nonzero.
+
+    The returned Series is given its guid value as a name so that concatenating
+    payments results in a DataFrame indexed by guid.  (Note, however, that the
+    name value is not used to construct an index when this series is returned
+    by function called by `DataFrame.apply`.  In such a case, pandas preserves
+    the index of the DataFrame on which `apply` is being called.)
+
+    """
+    return pd.Series(
+        data=data,
+        name=data['guid'] if data is not None else None,
+        index=DIVIDEND_PAYMENT_FIELDS,
+        dtype=object,
+    )
+
 
 class Event(object):
 
@@ -62,6 +107,9 @@ class Event(object):
     def __repr__(self):
         return "Event({0})".format(self.__dict__)
 
+    def to_series(self, index=None):
+        return pd.Series(self.__dict__, index=index)
+
 
 class Order(Event):
     pass
diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py
index e2301eb8..d55d9281 100644
--- a/zipline/utils/factory.py
+++ b/zipline/utils/factory.py
@@ -143,7 +143,7 @@ def create_dividend(sid, payment, declared_date, ex_date, pay_date):
         'net_amount': payment,
         'payment_sid': None,
         'ratio': None,
-        'dt': pd.tslib.normalize_date(declared_date),
+        'declared_date': pd.tslib.normalize_date(declared_date),
         'ex_date': pd.tslib.normalize_date(ex_date),
         'pay_date': pd.tslib.normalize_date(pay_date),
         'type': DATASOURCE_TYPE.DIVIDEND,