Compare commits

..
3 Commits
6 changed files with 48 additions and 51 deletions
+1 -3
View File
@@ -215,13 +215,11 @@ cpdef _read_bcolz_data(ctable_t table,
else: else:
continue continue
if column_name in ['open', 'high', 'low', 'close']: if column_name in ['open', 'high', 'low', 'close', 'volume']:
where_nan = (outbuf == 0) where_nan = (outbuf == 0)
outbuf_as_float = outbuf.astype(float64) * .000000001 outbuf_as_float = outbuf.astype(float64) * .000000001
outbuf_as_float[where_nan] = NAN outbuf_as_float[where_nan] = NAN
results.append(outbuf_as_float) results.append(outbuf_as_float)
elif column_name != 'volume':
results.append(outbuf.astype(uint32))
else: else:
results.append(outbuf) results.append(outbuf)
return results return results
+1 -1
View File
@@ -491,7 +491,7 @@ class BaseBundle(object):
data_frequency, data_frequency,
) )
raw_data.index = pd.to_datetime(raw_data.index, utc=True) raw_data.index = pd.to_datetime(raw_data.index, utc=True)
raw_data.index = raw_data.index.tz_localize('UTC') #raw_data.index = raw_data.index.tz_localize('UTC')
# Filter incoming data to fit start and end sessions. # Filter incoming data to fit start and end sessions.
raw_data = raw_data[ raw_data = raw_data[
+4 -1
View File
@@ -18,6 +18,7 @@ from numpy import (
full, full,
nan, nan,
int64, int64,
float64,
zeros zeros
) )
from six import iteritems, with_metaclass from six import iteritems, with_metaclass
@@ -70,7 +71,9 @@ class AssetDispatchBarReader(with_metaclass(ABCMeta)):
return self._dt_window_size(start_dt, end_dt), num_sids return self._dt_window_size(start_dt, end_dt), num_sids
def _make_raw_array_out(self, field, shape): def _make_raw_array_out(self, field, shape):
if field != 'volume' and field != 'sid': if field == 'volume':
out = zeros(shape, dtype=float64)
elif field != 'sid':
out = full(shape, nan) out = full(shape, nan)
else: else:
out = zeros(shape, dtype=int64) out = zeros(shape, dtype=int64)
+35 -35
View File
@@ -39,7 +39,7 @@ from catalyst.data._minute_bar_internal import (
from catalyst.gens.sim_engine import NANOS_IN_MINUTE from catalyst.gens.sim_engine import NANOS_IN_MINUTE
from catalyst.data.bar_reader import BarReader, NoDataOnDate from catalyst.data.bar_reader import BarReader, NoDataOnDate
from catalyst.data.us_equity_pricing import check_uint32_safe from catalyst.data.us_equity_pricing import check_uint64_safe
from catalyst.utils.calendars import get_calendar from catalyst.utils.calendars import get_calendar
from catalyst.utils.cli import maybe_show_progress from catalyst.utils.cli import maybe_show_progress
from catalyst.utils.memoize import lazyval from catalyst.utils.memoize import lazyval
@@ -52,7 +52,7 @@ FUTURES_MINUTES_PER_DAY = 1440
DEFAULT_EXPECTEDLEN = US_EQUITIES_MINUTES_PER_DAY * 252 * 15 DEFAULT_EXPECTEDLEN = US_EQUITIES_MINUTES_PER_DAY * 252 * 15
OHLC_RATIO = 1000 OHLC_RATIO = 100000000
class BcolzMinuteOverlappingData(Exception): class BcolzMinuteOverlappingData(Exception):
@@ -114,15 +114,15 @@ def _sid_subdir_path(sid):
def convert_cols(cols, scale_factor, sid, invalid_data_behavior): def convert_cols(cols, scale_factor, sid, invalid_data_behavior):
"""Adapt OHLCV columns into uint32 columns. """Adapt OHLCV columns into uint64 columns.
Parameters Parameters
---------- ----------
cols : dict cols : dict
A dict mapping each column name (open, high, low, close, volume) A dict mapping each column name (open, high, low, close, volume)
to a float column to convert to uint32. to a float column to convert to uint64.
scale_factor : int scale_factor : int
Factor to use to scale float values before converting to uint32. Factor to use to scale float values before converting to uint64.
sid : int sid : int
Sid of the relevant asset, for logging. Sid of the relevant asset, for logging.
invalid_data_behavior : str invalid_data_behavior : str
@@ -135,6 +135,7 @@ def convert_cols(cols, scale_factor, sid, invalid_data_behavior):
scaled_highs = np.nan_to_num(cols['high']) * scale_factor scaled_highs = np.nan_to_num(cols['high']) * scale_factor
scaled_lows = np.nan_to_num(cols['low']) * scale_factor scaled_lows = np.nan_to_num(cols['low']) * scale_factor
scaled_closes = np.nan_to_num(cols['close']) * scale_factor scaled_closes = np.nan_to_num(cols['close']) * scale_factor
scaled_volumes = np.nan_to_num(cols['volume']) * scale_factor
exclude_mask = np.zeros_like(scaled_opens, dtype=bool) exclude_mask = np.zeros_like(scaled_opens, dtype=bool)
@@ -143,11 +144,12 @@ def convert_cols(cols, scale_factor, sid, invalid_data_behavior):
('high', scaled_highs), ('high', scaled_highs),
('low', scaled_lows), ('low', scaled_lows),
('close', scaled_closes), ('close', scaled_closes),
('volume', scaled_volumes),
]: ]:
max_val = scaled_col.max() max_val = scaled_col.max()
try: try:
check_uint32_safe(max_val, col_name) check_uint64_safe(max_val, col_name)
except ValueError: except ValueError:
if invalid_data_behavior == 'raise': if invalid_data_behavior == 'raise':
raise raise
@@ -155,20 +157,20 @@ def convert_cols(cols, scale_factor, sid, invalid_data_behavior):
if invalid_data_behavior == 'warn': if invalid_data_behavior == 'warn':
logger.warn( logger.warn(
'Values for sid={}, col={} contain some too large for ' 'Values for sid={}, col={} contain some too large for '
'uint32 (max={}), filtering them out', 'uint64 (max={}), filtering them out',
sid, col_name, max_val, sid, col_name, max_val,
) )
# We want to exclude all rows that have an unsafe value in # We want to exclude all rows that have an unsafe value in
# this column. # this column.
exclude_mask &= (scaled_col >= np.iinfo(np.uint32).max) exclude_mask &= (scaled_col >= np.iinfo(np.uint64).max)
# Convert all cols to uint32. # Convert all cols to uint32.
opens = scaled_opens.astype(np.uint32) opens = scaled_opens.astype(np.uint64)
highs = scaled_highs.astype(np.uint32) highs = scaled_highs.astype(np.uint64)
lows = scaled_lows.astype(np.uint32) lows = scaled_lows.astype(np.uint64)
closes = scaled_closes.astype(np.uint32) closes = scaled_closes.astype(np.uint64)
volumes = cols['volume'].astype(np.uint32) volumes = scaled_volumes.astype(np.uint64)
# Exclude rows with unsafe values by setting to zero. # Exclude rows with unsafe values by setting to zero.
opens[exclude_mask] = 0 opens[exclude_mask] = 0
@@ -288,7 +290,7 @@ class BcolzMinuteBarMetadata(object):
ohlc_ratio : int ohlc_ratio : int
The default ratio by which to multiply the pricing data to The default ratio by which to multiply the pricing data to
convert the floats from floats to an integer to fit within convert the floats from floats to an integer to fit within
the np.uint32. If ohlc_ratios_per_sid is None or does not the np.uint64. If ohlc_ratios_per_sid is None or does not
contain a mapping for a given sid, this ratio is used. contain a mapping for a given sid, this ratio is used.
ohlc_ratios_per_sid : dict ohlc_ratios_per_sid : dict
A dict mapping each sid in the output to the factor by A dict mapping each sid in the output to the factor by
@@ -372,13 +374,13 @@ class BcolzMinuteBarWriter(object):
The last trading session in the data set. The last trading session in the data set.
default_ohlc_ratio : int, optional default_ohlc_ratio : int, optional
The default ratio by which to multiply the pricing data to The default ratio by which to multiply the pricing data to
convert from floats to integers that fit within np.uint32. If convert from floats to integers that fit within np.uint64. If
ohlc_ratios_per_sid is None or does not contain a mapping for a ohlc_ratios_per_sid is None or does not contain a mapping for a
given sid, this ratio is used. Default is OHLC_RATIO (1000). given sid, this ratio is used. Default is OHLC_RATIO (10^8).
ohlc_ratios_per_sid : dict, optional ohlc_ratios_per_sid : dict, optional
A dict mapping each sid in the output to the ratio by which to A dict mapping each sid in the output to the ratio by which to
multiply the pricing data to convert the floats from floats to multiply the pricing data to convert the floats from floats to
an integer to fit within the np.uint32. an integer to fit within the np.uint64.
expectedlen : int, optional expectedlen : int, optional
The expected length of the dataset, used when creating the initial The expected length of the dataset, used when creating the initial
bcolz ctable. bcolz ctable.
@@ -401,11 +403,9 @@ class BcolzMinuteBarWriter(object):
Each individual asset's data is stored as a bcolz table with a column for Each individual asset's data is stored as a bcolz table with a column for
each pricing field: (open, high, low, close, volume) each pricing field: (open, high, low, close, volume)
The open, high, low, and close columns are integers which are 1000 times The open, high, low, close and volume columns are integers which are 10^8 times
the quoted price, so that the data can represented and stored as an the quoted price, so that the data can represented and stored as an
np.uint32, supporting market prices quoted up to the thousands place. np.uint64, supporting market prices quoted up to the 1/10^8-th place.
volume is a np.uint32 with no mutation of the tens place.
The 'index' for each individual asset are a repeating period of minutes of The 'index' for each individual asset are a repeating period of minutes of
length `minutes_per_day` starting from each market open. length `minutes_per_day` starting from each market open.
@@ -573,7 +573,7 @@ class BcolzMinuteBarWriter(object):
if not os.path.exists(sid_containing_dirname): if not os.path.exists(sid_containing_dirname):
# Other sids may have already created the containing directory. # Other sids may have already created the containing directory.
os.makedirs(sid_containing_dirname) os.makedirs(sid_containing_dirname)
initial_array = np.empty(0, np.uint32) initial_array = np.empty(0, np.uint64)
table = ctable( table = ctable(
rootdir=path, rootdir=path,
columns=[ columns=[
@@ -610,7 +610,7 @@ class BcolzMinuteBarWriter(object):
minute_offset = len(table) % self._minutes_per_day minute_offset = len(table) % self._minutes_per_day
num_to_prepend = numdays * self._minutes_per_day - minute_offset num_to_prepend = numdays * self._minutes_per_day - minute_offset
prepend_array = np.zeros(num_to_prepend, np.uint32) prepend_array = np.zeros(num_to_prepend, np.uint64)
# Fill all OHLCV with zeros. # Fill all OHLCV with zeros.
table.append([prepend_array] * 5) table.append([prepend_array] * 5)
table.flush() table.flush()
@@ -815,11 +815,11 @@ class BcolzMinuteBarWriter(object):
minutes_count = all_minutes_in_window.size minutes_count = all_minutes_in_window.size
open_col = np.zeros(minutes_count, dtype=np.uint32) open_col = np.zeros(minutes_count, dtype=np.uint64)
high_col = np.zeros(minutes_count, dtype=np.uint32) high_col = np.zeros(minutes_count, dtype=np.uint64)
low_col = np.zeros(minutes_count, dtype=np.uint32) low_col = np.zeros(minutes_count, dtype=np.uint64)
close_col = np.zeros(minutes_count, dtype=np.uint32) close_col = np.zeros(minutes_count, dtype=np.uint64)
vol_col = np.zeros(minutes_count, dtype=np.uint32) vol_col = np.zeros(minutes_count, dtype=np.uint64)
dt_ixs = np.searchsorted(all_minutes_in_window.values, dt_ixs = np.searchsorted(all_minutes_in_window.values,
dts.astype('datetime64[ns]')) dts.astype('datetime64[ns]'))
@@ -1125,8 +1125,8 @@ class BcolzMinuteBarReader(MinuteBarReader):
else: else:
return np.nan return np.nan
if field != 'volume': #if field != 'volume':
value *= self._ohlc_ratio_inverse_for_sid(sid) value *= self._ohlc_ratio_inverse_for_sid(sid)
return value return value
def get_last_traded_dt(self, asset, dt): def get_last_traded_dt(self, asset, dt):
@@ -1248,7 +1248,7 @@ class BcolzMinuteBarReader(MinuteBarReader):
if field != 'volume': if field != 'volume':
out = np.full(shape, np.nan) out = np.full(shape, np.nan)
else: else:
out = np.zeros(shape, dtype=np.uint32) out = np.zeros(shape, dtype=np.float64)
for i, sid in enumerate(sids): for i, sid in enumerate(sids):
carray = self._open_minute_file(field, sid) carray = self._open_minute_file(field, sid)
@@ -1262,11 +1262,11 @@ class BcolzMinuteBarReader(MinuteBarReader):
where = values != 0 where = values != 0
# first slice down to len(where) because we might not have # first slice down to len(where) because we might not have
# written data for all the minutes requested # written data for all the minutes requested
if field != 'volume': #if field != 'volume':
out[:len(where), i][where] = ( out[:len(where), i][where] = (
values[where] * self._ohlc_ratio_inverse_for_sid(sid)) values[where] * self._ohlc_ratio_inverse_for_sid(sid))
else: #else:
out[:len(where), i][where] = values[where] # out[:len(where), i][where] = values[where]
results.append(out) results.append(out)
return results return results
+6 -10
View File
@@ -441,7 +441,7 @@ class BcolzDailyBarWriter(object):
dates = raw_data.index.values.astype('datetime64[s]') dates = raw_data.index.values.astype('datetime64[s]')
check_uint32_safe(dates.max().view(np.int64), 'day') check_uint32_safe(dates.max().view(np.int64), 'day')
processed['day'] = dates.astype('uint32') processed['day'] = dates.astype('uint32')
processed['volume'] = raw_data.volume.astype('uint64') processed['volume'] = (raw_data.volume * PRICE_ADJUSTMENT_FACTOR).astype('uint64')
return ctable.fromdataframe(processed) return ctable.fromdataframe(processed)
@@ -494,9 +494,8 @@ class BcolzDailyBarReader(SessionBarReader):
The data in these columns is interpreted as follows: The data in these columns is interpreted as follows:
- Price columns ('open', 'high', 'low', 'close') are interpreted as 1000 * - Price columns ('open', 'high', 'low', 'close') and Volume are interpreted
as-traded dollar value. as 10^9 * as-traded dollar value.
- Volume is interpreted as as-traded volume.
- Day is interpreted as seconds since midnight UTC, Jan 1, 1970. - Day is interpreted as seconds since midnight UTC, Jan 1, 1970.
- Id is the asset id of the row. - Id is the asset id of the row.
@@ -762,13 +761,10 @@ class BcolzDailyBarReader(SessionBarReader):
""" """
ix = self.sid_day_index(sid, dt) ix = self.sid_day_index(sid, dt)
price = self._spot_col(field)[ix] price = self._spot_col(field)[ix]
if field != 'volume': if field != 'volume' and price == 0:
if price == 0: return nan
return nan
else:
return price / PRICE_ADJUSTMENT_FACTOR
else: else:
return price return price / PRICE_ADJUSTMENT_FACTOR
class PanelBarReader(SessionBarReader): class PanelBarReader(SessionBarReader):
+1 -1
View File
@@ -1,7 +1,7 @@
# Incompatible with earlier PIP versions # Incompatible with earlier PIP versions
pip>=7.1.0 pip>=7.1.0
# bcolz fails to install if this is not in the build_requires. # bcolz fails to install if this is not in the build_requires.
setuptools>18.0 setuptools>36.0
# Logging # Logging
Logbook==0.12.5 Logbook==0.12.5