mirror of
https://github.com/wassname/catalyst.git
synced 2026-09-09 11:19:23 +08:00
MAINT: Remove future/equity distinction.
In the data portal, remove methods that make a distinction between future and equity asset type. Instead rely on the pricing reader dispatching. In support of incoming work which will upsample equity history arrays to the larger future calendar. Also, remove perf tracker tests which were using an equity reader/writer, to be added back in later.
This commit is contained in:
Eddie Hebert
+33
-132
@@ -137,7 +137,7 @@ class DataPortal(object):
|
||||
|
||||
self._equity_daily_reader = equity_daily_reader
|
||||
if self._equity_daily_reader is not None:
|
||||
self._equity_history_loader = DailyHistoryLoader(
|
||||
self._history_loader = DailyHistoryLoader(
|
||||
self.trading_calendar,
|
||||
self._equity_daily_reader,
|
||||
self._adjustment_reader
|
||||
@@ -157,17 +157,16 @@ class DataPortal(object):
|
||||
}
|
||||
}
|
||||
|
||||
if self._equity_minute_reader is not None:
|
||||
self._equity_daily_aggregator = DailyHistoryAggregator(
|
||||
self.trading_calendar.schedule.market_open,
|
||||
self._equity_minute_reader,
|
||||
self.trading_calendar
|
||||
)
|
||||
self._equity_minute_history_loader = MinuteHistoryLoader(
|
||||
self.trading_calendar,
|
||||
self._equity_minute_reader,
|
||||
self._adjustment_reader
|
||||
)
|
||||
self._daily_aggregator = DailyHistoryAggregator(
|
||||
self.trading_calendar.schedule.market_open,
|
||||
self._equity_minute_reader,
|
||||
self.trading_calendar
|
||||
)
|
||||
self._minute_history_loader = MinuteHistoryLoader(
|
||||
self.trading_calendar,
|
||||
self._equity_minute_reader,
|
||||
self._adjustment_reader
|
||||
)
|
||||
|
||||
self._first_trading_day = first_trading_day
|
||||
|
||||
@@ -522,9 +521,9 @@ class DataPortal(object):
|
||||
)
|
||||
|
||||
def _get_daily_data(self, asset, column, dt):
|
||||
reader = self._pricing_readers[type(asset)]['daily']
|
||||
if column == "last_traded":
|
||||
last_traded_dt = \
|
||||
self._equity_daily_reader.get_last_traded_dt(asset, dt)
|
||||
last_traded_dt = reader.get_last_traded_dt(asset, dt)
|
||||
|
||||
if pd.isnull(last_traded_dt):
|
||||
return pd.NaT
|
||||
@@ -533,7 +532,7 @@ class DataPortal(object):
|
||||
elif column in OHLCV_FIELDS:
|
||||
# don't forward fill
|
||||
try:
|
||||
val = self._equity_daily_reader.spot_price(asset, dt, column)
|
||||
val = reader.spot_price(asset, dt, column)
|
||||
if val == -1:
|
||||
if column == "volume":
|
||||
return 0
|
||||
@@ -547,7 +546,7 @@ class DataPortal(object):
|
||||
found_dt = dt
|
||||
while True:
|
||||
try:
|
||||
value = self._equity_daily_reader.spot_price(
|
||||
value = reader.spot_price(
|
||||
asset, found_dt, "close"
|
||||
)
|
||||
if value != -1:
|
||||
@@ -592,88 +591,16 @@ class DataPortal(object):
|
||||
index=days_for_window,
|
||||
columns=None)
|
||||
|
||||
future_data = []
|
||||
eq_assets = []
|
||||
|
||||
for asset in assets:
|
||||
if isinstance(asset, Future):
|
||||
future_data.append(self._get_history_daily_window_future(
|
||||
asset, days_for_window, end_dt, field_to_use
|
||||
))
|
||||
else:
|
||||
eq_assets.append(asset)
|
||||
eq_data = self._get_history_daily_window_equities(
|
||||
eq_assets, days_for_window, end_dt, field_to_use
|
||||
data = self._get_history_daily_window_data(
|
||||
assets, days_for_window, end_dt, field_to_use
|
||||
)
|
||||
if future_data:
|
||||
# TODO: This case appears to be uncovered by testing.
|
||||
data = np.concatenate(eq_data, np.array(future_data).T)
|
||||
else:
|
||||
data = eq_data
|
||||
return pd.DataFrame(
|
||||
data,
|
||||
index=days_for_window,
|
||||
columns=assets
|
||||
)
|
||||
|
||||
def _get_history_daily_window_future(self, asset, days_for_window,
|
||||
end_dt, column):
|
||||
# Since we don't have daily bcolz files for futures (yet), use minute
|
||||
# bars to calculate the daily values.
|
||||
data = []
|
||||
data_groups = []
|
||||
|
||||
# get all the minutes for the days NOT including today
|
||||
for day in days_for_window[:-1]:
|
||||
minutes = self.sessions_in_range.minutes_for_session(day)
|
||||
|
||||
values_for_day = np.zeros(len(minutes), dtype=np.float64)
|
||||
|
||||
for idx, minute in enumerate(minutes):
|
||||
minute_val = self._get_minute_spot_value_future(
|
||||
asset, column, minute
|
||||
)
|
||||
|
||||
values_for_day[idx] = minute_val
|
||||
|
||||
data_groups.append(values_for_day)
|
||||
|
||||
# get the minutes for today
|
||||
last_day_minutes = pd.date_range(
|
||||
start=self.trading_calendar.open_and_close_for_session(end_dt)[0],
|
||||
end=end_dt,
|
||||
freq="T"
|
||||
)
|
||||
|
||||
values_for_last_day = np.zeros(len(last_day_minutes), dtype=np.float64)
|
||||
|
||||
for idx, minute in enumerate(last_day_minutes):
|
||||
minute_val = self._get_minute_spot_value_future(
|
||||
asset, column, minute
|
||||
)
|
||||
|
||||
values_for_last_day[idx] = minute_val
|
||||
|
||||
data_groups.append(values_for_last_day)
|
||||
|
||||
for group in data_groups:
|
||||
if len(group) == 0:
|
||||
continue
|
||||
|
||||
if column == 'volume':
|
||||
data.append(np.sum(group))
|
||||
elif column == 'open':
|
||||
data.append(group[0])
|
||||
elif column == 'close':
|
||||
data.append(group[-1])
|
||||
elif column == 'high':
|
||||
data.append(np.amax(group))
|
||||
elif column == 'low':
|
||||
data.append(np.amin(group))
|
||||
|
||||
return data
|
||||
|
||||
def _get_history_daily_window_equities(
|
||||
def _get_history_daily_window_data(
|
||||
self, assets, days_for_window, end_dt, field_to_use):
|
||||
ends_at_midnight = end_dt.hour == 0 and end_dt.minute == 0
|
||||
|
||||
@@ -697,19 +624,19 @@ class DataPortal(object):
|
||||
)
|
||||
|
||||
if field_to_use == 'open':
|
||||
minute_value = self._equity_daily_aggregator.opens(
|
||||
minute_value = self._daily_aggregator.opens(
|
||||
assets, end_dt)
|
||||
elif field_to_use == 'high':
|
||||
minute_value = self._equity_daily_aggregator.highs(
|
||||
minute_value = self._daily_aggregator.highs(
|
||||
assets, end_dt)
|
||||
elif field_to_use == 'low':
|
||||
minute_value = self._equity_daily_aggregator.lows(
|
||||
minute_value = self._daily_aggregator.lows(
|
||||
assets, end_dt)
|
||||
elif field_to_use == 'close':
|
||||
minute_value = self._equity_daily_aggregator.closes(
|
||||
minute_value = self._daily_aggregator.closes(
|
||||
assets, end_dt)
|
||||
elif field_to_use == 'volume':
|
||||
minute_value = self._equity_daily_aggregator.volumes(
|
||||
minute_value = self._daily_aggregator.volumes(
|
||||
assets, end_dt)
|
||||
|
||||
# append the partial day.
|
||||
@@ -871,40 +798,14 @@ class DataPortal(object):
|
||||
-------
|
||||
A numpy array with requested values.
|
||||
"""
|
||||
if isinstance(assets, Future):
|
||||
return self._get_minute_window_for_future([assets], field,
|
||||
minutes_for_window)
|
||||
else:
|
||||
# TODO: Make caller accept assets.
|
||||
window = self._get_minute_window_for_equities(assets, field,
|
||||
minutes_for_window)
|
||||
return window
|
||||
return self._get_minute_window_data(assets, field, minutes_for_window)
|
||||
|
||||
def _get_minute_window_for_future(self, asset, field, minutes_for_window):
|
||||
# THIS IS TEMPORARY. For now, we are only exposing futures within
|
||||
# equity trading hours (9:30 am to 4pm, Eastern). The easiest way to
|
||||
# do this is to simply do a spot lookup for each desired minute.
|
||||
return_data = np.zeros(len(minutes_for_window), dtype=np.float64)
|
||||
for idx, minute in enumerate(minutes_for_window):
|
||||
return_data[idx] = \
|
||||
self._get_minute_spot_value_future(asset, field, minute)
|
||||
|
||||
# Note: an improvement could be to find the consecutive runs within
|
||||
# minutes_for_window, and use them to read the underlying ctable
|
||||
# more efficiently.
|
||||
|
||||
# Once futures are on 24-hour clock, then we can just grab all the
|
||||
# requested minutes in one shot from the ctable.
|
||||
|
||||
# no adjustments for futures, yay.
|
||||
return return_data
|
||||
|
||||
def _get_minute_window_for_equities(
|
||||
def _get_minute_window_data(
|
||||
self, assets, field, minutes_for_window):
|
||||
return self._equity_minute_history_loader.history(assets,
|
||||
minutes_for_window,
|
||||
field,
|
||||
False)
|
||||
return self._minute_history_loader.history(assets,
|
||||
minutes_for_window,
|
||||
field,
|
||||
False)
|
||||
|
||||
def _apply_all_adjustments(self, data, asset, dts, field,
|
||||
price_adj_factor=1.0):
|
||||
@@ -1018,10 +919,10 @@ class DataPortal(object):
|
||||
return_array[:] = np.NAN
|
||||
|
||||
if bar_count != 0:
|
||||
data = self._equity_history_loader.history(assets,
|
||||
days_in_window,
|
||||
field,
|
||||
extra_slot)
|
||||
data = self._history_loader.history(assets,
|
||||
days_in_window,
|
||||
field,
|
||||
extra_slot)
|
||||
if extra_slot:
|
||||
return_array[:len(return_array) - 1, :] = data
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user