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

BUG: Apply latest adjustment for minute 1d

Browse Source 
Fix behavior in minute mode history with frequency `1d`, where on the
day immediately following an adjustment action, the overnight adjustment
would not apply. (However the adjustment would be applied after a 1 day
lag.)

The root cause of the bug was that the history data for minute mode when
using `1d` stitches together a sliding window of the daily data for
previous  and the current minute. That daily data sliding window and
corresponding adjustments was being read as if the data was being viewed
from on the last day of the window; however in this case the data is
being viewed from the day after the window has completed. The difference
in view points requires the adjustments to popped and applied by the
adjusted array one index earlier. The fix uses the `extra_slot` value as
signifier on whether the data is being viewed on the following day and
then accordingly adjusts the index of the mulitpy object.

Also, change the split and merger test data ratios to have different values,
to ensure that different adjustment values are applied; as opposed to
doubling up on just one of the values.
This commit is contained in:
Eddie Hebert
2016-06-02 15:22:25 -04:00
parent 6979ae8d6a
commit b450ab841f
3 changed files with 258 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_history.py
+152 -23
View File
@@ -158,7 +158,7 @@ class WithHistory(WithDataPortal):
return pd.DataFrame([
{
'effective_date': str_to_seconds('2015-01-06'),
'ratio': 0.5,
'ratio': 0.25,
'sid': cls.SPLIT_ASSET_SID,
},
{
@@ -173,7 +173,7 @@ class WithHistory(WithDataPortal):
return pd.DataFrame([
{
'effective_date': str_to_seconds('2015-01-06'),
'ratio': 0.5,
'ratio': 0.25,
'sid': cls.MERGER_ASSET_SID,
},
{
@@ -482,14 +482,15 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase):
# Start values are crafted so that the thousands place are equal when
# adjustments are applied correctly.
# The splits and mergers are defined as 2:1 splits, so the prices
# approximate that adjustment by halving the thousands place each day.
# The splits and mergers are defined as 4:1 then 2:1 ratios, so the
# prices approximate that adjustment by quartering and then halving
# the thousands place.
data[cls.MERGER_ASSET_SID] = data[cls.SPLIT_ASSET_SID] = pd.concat((
create_minute_df_for_asset(
cls.env,
pd.Timestamp('2015-01-05', tz='UTC'),
pd.Timestamp('2015-01-05', tz='UTC'),
start_val=4000),
start_val=8000),
create_minute_df_for_asset(
cls.env,
pd.Timestamp('2015-01-06', tz='UTC'),
@@ -499,6 +500,11 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase):
cls.env,
pd.Timestamp('2015-01-07', tz='UTC'),
pd.Timestamp('2015-01-07', tz='UTC'),
start_val=1000),
create_minute_df_for_asset(
cls.env,
pd.Timestamp('2015-01-08', tz='UTC'),
pd.Timestamp('2015-01-08', tz='UTC'),
start_val=1000)
))
asset3 = cls.asset_finder.retrieve_asset(3)
@@ -546,6 +552,129 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase):
with self.assertRaises(HistoryInInitialize):
test_algo.initialize()
def test_daily_splits_and_mergers(self):
# self.SPLIT_ASSET and self.MERGER_ASSET had splits/mergers
# on 1/6 and 1/7
jan5 = pd.Timestamp('2015-01-05', tz='UTC')
for asset in [self.SPLIT_ASSET, self.MERGER_ASSET]:
# before any of the adjustments, 1/4 and 1/5
window1 = self.data_portal.get_history_window(
[asset],
self.env.get_open_and_close(jan5)[1],
2,
'1d',
'close'
)[asset]
np.testing.assert_array_equal(np.array([np.nan, 8389]), window1)
# straddling the first event
window2 = self.data_portal.get_history_window(
[asset],
pd.Timestamp('2015-01-06 14:35', tz='UTC'),
2,
'1d',
'close'
)[asset]
# Value from 1/5 should be quartered
np.testing.assert_array_equal(
[2097.25,
# Split occurs. The value of the thousands place should
# match.
2004],
window2
)
# straddling both events!
window3 = self.data_portal.get_history_window(
[asset],
pd.Timestamp('2015-01-07 14:35', tz='UTC'),
3,
'1d',
'close'
)[asset]
np.testing.assert_array_equal(
[1048.625, 1194.50, 1004.0],
window3
)
# after last event
window4 = self.data_portal.get_history_window(
[asset],
pd.Timestamp('2015-01-08 14:40', tz='UTC'),
2,
'1d',
'close'
)[asset]
# should not be adjusted
np.testing.assert_array_equal([1389, 1009], window4)
def test_daily_dividends(self):
# self.DIVIDEND_ASSET had dividends on 1/6 and 1/7
jan5 = pd.Timestamp('2015-01-05', tz='UTC')
asset = self.DIVIDEND_ASSET
# before any of the dividends
window1 = self.data_portal.get_history_window(
[asset],
self.env.get_open_and_close(jan5)[1],
2,
'1d',
'close'
)[asset]
np.testing.assert_array_equal(np.array([nan, 391]), window1)
# straddling the first event
window2 = self.data_portal.get_history_window(
[asset],
pd.Timestamp('2015-01-06 14:35', tz='UTC'),
2,
'1d',
'close'
)[asset]
np.testing.assert_array_equal(
[383.18, # 391 (last close) * 0.98 (first div)
# Dividend occurs prior.
396],
window2
)
# straddling both events!
window3 = self.data_portal.get_history_window(
[asset],
pd.Timestamp('2015-01-07 14:35', tz='UTC'),
3,
'1d',
'close'
)[asset]
np.testing.assert_array_equal(
[367.853, # 391 (last close) * 0.98 * 0.96 (both)
749.76, # 781 (last_close) * 0.96 (second div)
786], # no adjustment
window3
)
# after last event
window4 = self.data_portal.get_history_window(
[asset],
pd.Timestamp('2015-01-08 14:40', tz='UTC'),
2,
'1d',
'close'
)[asset]
# should not be adjusted, should be 787 to 791
np.testing.assert_array_equal([1171, 1181], window4)
def test_minute_before_assets_trading(self):
# since asset2 and asset3 both started trading on 1/5/2015, let's do
# some history windows that are completely before that
@@ -728,7 +857,7 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase):
)[asset]
np.testing.assert_array_equal(
np.array(range(4380, 4390)), window1)
np.array(range(8380, 8390)), window1)
# straddling the first event
window2 = self.data_portal.get_history_window(
@@ -741,11 +870,11 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase):
# five minutes from 1/5 should be halved
np.testing.assert_array_equal(
[2192.5,
2193,
2193.5,
2194,
2194.5,
[2096.25,
2096.5,
2096.75,
2097,
2097.25,
# Split occurs. The value of the thousands place should
# match.
2000,
@@ -765,9 +894,9 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase):
'close'
)[asset]
# first five minutes should be 4385-4390, but quartered
# first five minutes should be 4385-4390, but eigthed
np.testing.assert_array_equal(
[1096.25, 1096.5, 1096.75, 1097, 1097.25],
[1048.125, 1048.25, 1048.375, 1048.5, 1048.625],
window3[0:5]
)
@@ -872,12 +1001,12 @@ class MinuteEquityHistoryTestCase(WithHistory, ZiplineTestCase):
bar_data = BarData(self.data_portal, lambda: current_dt, 'minute')
adj_expected = {
'open': np.arange(4381, 4391) / 2.0,
'high': np.arange(4382, 4392) / 2.0,
'low': np.arange(4379, 4389) / 2.0,
'close': np.arange(4380, 4390) / 2.0,
'volume': np.arange(4380, 4390) * 100 * 2.0,
'price': np.arange(4380, 4390) / 2.0,
'open': np.arange(8381, 8391) / 4.0,
'high': np.arange(8382, 8392) / 4.0,
'low': np.arange(8379, 8389) / 4.0,
'close': np.arange(8380, 8390) / 4.0,
'volume': np.arange(8380, 8390) * 100 * 4.0,
'price': np.arange(8380, 8390) / 4.0,
}
expected = {
@@ -1390,7 +1519,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase):
)[asset]
# first value should be halved, second value unadjusted
np.testing.assert_array_equal([1, 3], window2)
np.testing.assert_array_equal([0.5, 3], window2)
window2_volume = self.data_portal.get_history_window(
[asset],
@@ -1402,7 +1531,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase):
if asset == self.SPLIT_ASSET:
# first value should be doubled, second value unadjusted
np.testing.assert_array_equal(window2_volume, [400, 300])
np.testing.assert_array_equal(window2_volume, [800, 300])
elif asset == self.MERGER_ASSET:
np.testing.assert_array_equal(window2_volume, [200, 300])
@@ -1415,7 +1544,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase):
'close'
)[asset]
np.testing.assert_array_equal([0.5, 1.5, 4], window3)
np.testing.assert_array_equal([0.25, 1.5, 4], window3)
window3_volume = self.data_portal.get_history_window(
[asset],
@@ -1426,7 +1555,7 @@ class DailyEquityHistoryTestCase(WithHistory, ZiplineTestCase):
)[asset]
if asset == self.SPLIT_ASSET:
np.testing.assert_array_equal(window3_volume, [800, 600, 400])
np.testing.assert_array_equal(window3_volume, [1600, 600, 400])
elif asset == self.MERGER_ASSET:
np.testing.assert_array_equal(window3_volume, [200, 300, 400])
zipline/data/data_portal.py
+4 -2
View File
@@ -1336,7 +1336,8 @@ class DataPortal(object):
self, assets, field, minutes_for_window):
return self._equity_minute_history_loader.history(assets,
minutes_for_window,
field)
field,
False)
def _apply_all_adjustments(self, data, asset, dts, field,
price_adj_factor=1.0):
@@ -1452,7 +1453,8 @@ class DataPortal(object):
if bar_count != 0:
data = self._equity_history_loader.history(assets,
days_in_window,
field)
field,
extra_slot)
if extra_slot:
return_array[:len(return_array) - 1, :] = data
else:
zipline/data/us_equity_loader.py
+102 -15
View File
@@ -103,7 +103,8 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
def _array(self, start, end, assets, field):
pass
def _get_adjustments_in_range(self, asset, dts, field):
def _get_adjustments_in_range(self, asset, dts, field,
is_perspective_after):
"""
Get the Float64Multiply objects to pass to an AdjustedArrayWindow.
@@ -126,6 +127,12 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
The days for which adjustment data is needed.
field : str
OHLCV field for which to get the adjustments.
is_perspective_after : bool
see: `USEquityHistoryLoader.history`
If True, the index at which the Multiply object is registered to
be popped is calculated so that it applies to the last slot in the
sliding window when the adjustment occurs immediately after the dt
that slot represents.
Returns
-------
@@ -142,30 +149,42 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
dt = m[0]
if start < dt <= end:
end_loc = dts.searchsorted(dt)
adj_loc = end_loc
if is_perspective_after:
# Set adjustment pop location so that it applies
# to last value if adjustment occurs immediately after
# the last slot.
adj_loc -= 1
mult = Float64Multiply(0,
end_loc - 1,
0,
0,
m[1])
try:
adjs[end_loc].append(mult)
adjs[adj_loc].append(mult)
except KeyError:
adjs[end_loc] = [mult]
adjs[adj_loc] = [mult]
divs = self._adjustments_reader.get_adjustments_for_sid(
'dividends', sid)
for d in divs:
dt = d[0]
if start < dt <= end:
end_loc = dts.searchsorted(dt)
adj_loc = end_loc
if is_perspective_after:
# Set adjustment pop location so that it applies
# to last value if adjustment occurs immediately after
# the last slot.
adj_loc -= 1
mult = Float64Multiply(0,
end_loc - 1,
0,
0,
d[1])
try:
adjs[end_loc].append(mult)
adjs[adj_loc].append(mult)
except KeyError:
adjs[end_loc] = [mult]
adjs[adj_loc] = [mult]
splits = self._adjustments_reader.get_adjustments_for_sid(
'splits', sid)
for s in splits:
@@ -176,18 +195,25 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
ratio = s[1]
if start < dt <= end:
end_loc = dts.searchsorted(dt)
adj_loc = end_loc
if is_perspective_after:
# Set adjustment pop location so that it applies
# to last value if adjustment occurs immediately after
# the last slot.
adj_loc -= 1
mult = Float64Multiply(0,
end_loc - 1,
0,
0,
ratio)
try:
adjs[end_loc].append(mult)
adjs[adj_loc].append(mult)
except KeyError:
adjs[end_loc] = [mult]
adjs[adj_loc] = [mult]
return adjs
def _ensure_sliding_windows(self, assets, dts, field):
def _ensure_sliding_windows(self, assets, dts, field,
is_perspective_after):
"""
Ensure that there is a Float64Multiply window for each asset that can
provide data for the given parameters.
@@ -207,6 +233,8 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
in the calendar.
field : str
The OHLCV field for which to retrieve data.
is_perspective_after : bool
see: `USEquityHistoryLoader.history`
Returns
-------
@@ -218,10 +246,11 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
size = len(dts)
asset_windows = {}
needed_assets = []
for asset in assets:
try:
asset_windows[asset] = self._window_blocks[field].get(
(asset, size), end)
(asset, size, is_perspective_after), end)
except KeyError:
needed_assets.append(asset)
@@ -245,7 +274,7 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
for i, asset in enumerate(needed_assets):
if self._adjustments_reader:
adjs = self._get_adjustments_in_range(
asset, prefetch_dts, field)
asset, prefetch_dts, field, is_perspective_after)
else:
adjs = {}
window = Float64Window(
@@ -257,13 +286,14 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
)
sliding_window = SlidingWindow(window, size, start_ix, offset)
asset_windows[asset] = sliding_window
self._window_blocks[field].set((asset, size),
sliding_window,
prefetch_end)
self._window_blocks[field].set(
(asset, size, is_perspective_after),
sliding_window,
prefetch_end)
return [asset_windows[asset] for asset in assets]
def history(self, assets, dts, field):
def history(self, assets, dts, field, is_perspective_after):
"""
A window of pricing data with adjustments applied assuming that the
end of the window is the day before the current simulation time.
@@ -278,13 +308,70 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)):
in the calendar.
field : str
The OHLCV field for which to retrieve data.
is_perspective_after : bool
True, if the window is being viewed immediately after the last dt
in the sliding window.
False, if the window is viewed on the last dt.
This flag is used for handling the case where the last dt in the
requested window immediately precedes a corporate action, e.g.:
- is_perspective_after is True
When the viewpoint is after the last dt in the window, as when a
daily history window is accessed from a simulation that uses a
minute data frequency, the history call to this loader will not
include the current simulation dt. At that point in time, the raw
data for the last day in the window will require adjustment, so the
most recent adjustment with respect to the simulation time is
applied to the last dt in the requested window.
An example equity which has a 0.5 split ratio dated for 05-27,
with the dts for a history call of 5 bars with a '1d' frequency at
05-27 9:31. Simulation frequency is 'minute'.
(In this case this function is called with 4 daily dts, and the
calling function is responsible for stitching back on the
'current' dt)
| | | | | last dt | <-- viewer is here |
| | 05-23 | 05-24 | 05-25 | 05-26 | 05-27 9:31 |
| raw | 10.10 | 10.20 | 10.30 | 10.40 | |
| adj | 5.05 | 5.10 | 5.15 | 5.25 | |
The adjustment is applied to the last dt, 05-26, and all previous
dts.
- is_perspective_after is False, daily
When the viewpoint is the same point in time as the last dt in the
window, as when a daily history window is accessed from a
simulation that uses a daily data frequency, the history call will
include the current dt. At that point in time, the raw data for the
last day in the window will be post-adjustment, so no adjustment
is applied to the last dt.
An example equity which has a 0.5 split ratio dated for 05-27,
with the dts for a history call of 5 bars with a '1d' frequency at
05-27 0:00. Simulation frequency is 'daily'.
| | | | | | <-- viewer is here |
| | | | | | last dt |
| | 05-23 | 05-24 | 05-25 | 05-26 | 05-27 |
| raw | 10.10 | 10.20 | 10.30 | 10.40 | 5.25 |
| adj | 5.05 | 5.10 | 5.15 | 5.20 | 5.25 |
Adjustments are applied 05-23 through 05-26 but not to the last dt,
05-27
Returns
-------
out : np.ndarray with shape(len(days between start, end), len(assets))
"""
block = self._ensure_sliding_windows(assets, dts, field)
block = self._ensure_sliding_windows(assets,
dts,
field,
is_perspective_after)
end_ix = self._calendar.get_loc(dts[-1])
return hstack([window.get(end_ix) for window in block])