wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-03 12:40:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fixed issues with daily bundles

Browse Source
This commit is contained in:
fredfortier
2017-10-17 18:29:48 -04:00
parent d21cc36bef
commit 4649b31d89
5 changed files with 108 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/examples/simple_loop.py
+6 -5
View File
@@ -7,7 +7,7 @@ from catalyst.api import symbol
def initialize(context):
print('initializing')
context.asset = symbol('gno_btc')
context.asset = symbol('etc_btc')
def handle_data(context, data):
@@ -19,18 +19,19 @@ def handle_data(context, data):
prices = data.history(
context.asset,
fields='price',
bar_count=50,
frequency='1m'
bar_count=15,
frequency='1d'
)
rsi = talib.RSI(prices.values, timeperiod=14)[-1]
print('got rsi: {}'.format(rsi))
pass
run_algorithm(
capital_base=250,
start=pd.to_datetime('2017-9-5', utc=True),
start=pd.to_datetime('2017-9-01', utc=True),
end=pd.to_datetime('2017-9-30', utc=True),
data_frequency='minute',
data_frequency='daily',
initialize=initialize,
handle_data=handle_data,
analyze=None,
catalyst/exchange/data_portal_exchange.py
+20 -9
View File
@@ -299,26 +299,36 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
raise InvalidHistoryFrequencyError(frequency=data_frequency)
try:
values = reader.load_raw_arrays(
# values = reader.load_raw_arrays(
# fields=[field],
# start_dt=dts[0],
# end_dt=dts[-1],
# sids=[asset.sid for asset in assets],
# )[0]
values = bundle.get_raw_arrays(
assets=assets,
fields=[field],
start_dt=dts[0],
end_dt=dts[-1],
sids=[asset.sid for asset in assets],
data_frequency=data_frequency
)[0]
except Exception:
first_trading_day = self._get_first_trading_day(assets)
symbols = [asset.symbol.encode('utf-8') for asset in assets]
symbol_list = ','.join(symbols)
raise PricingDataNotLoadedError(
field=field,
first_trading_day=self._get_first_trading_day(assets),
first_trading_day=first_trading_day,
exchange=exchange.name,
symbols=[asset.symbol.encode('utf-8') for asset in assets],
symbols=symbols,
symbol_list=symbol_list
)
series = dict()
for index, asset in enumerate(assets):
asset_values = []
for value in values:
asset_values.append(value[index])
asset_values = values[:, index]
value_series = pd.Series(asset_values, index=dts)
series[asset] = value_series
@@ -333,7 +343,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
first_trading_day=first_trading_day,
exchange=assets[0].exchange,
symbols=[asset.symbol.encode('utf-8') for asset in assets],
dt = dt,
dt=dt,
)
def get_exchange_spot_value(self, exchange, assets, field, dt,
@@ -358,7 +368,8 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
first_trading_day=self._get_first_trading_day(assets),
exchange=exchange.name,
symbols=[asset.symbol.encode('utf-8') for asset in assets],
symbol_list = ''.join([asset.symbol.encode('utf-8') for asset in assets])
symbol_list=''.join(
[asset.symbol.encode('utf-8') for asset in assets])
)
return values
catalyst/exchange/exchange.py
+8 -97
View File
@@ -419,98 +419,6 @@ class Exchange:
return value
@deprecated
def get_history(self, assets, end_dt, bar_count, data_frequency,
fallback_exchange=True):
"""
Retrieve OHLCV bars from the Catalyst and/or exchange API.
If Catalyst does not have the full data set, retrieve the missing
portion from the exchange API if the exchanges supports
historical data.
:param assets: list[TradingPair]
The TradingPair asset.
:param data_frequency: str
The bar frequency: daily or minute
:param bar_count: int
The number of bars desired.
:param end: datetime
The last trading date of the last bar.
:return:
"""
candles = dict()
for asset in assets:
candles[asset] = self.get_asset_history(
asset=asset,
end=end_dt,
bar_count=bar_count,
data_frequency=data_frequency,
fallback_exchange=fallback_exchange
)
return candles
@deprecated
def get_asset_history(self, asset, end, bar_count, data_frequency,
fallback_exchange=True):
"""
Retrieve the OHLVC bars of a single asset.
:param asset: TradingPair
The TradingPair asset.
:param data_frequency: str
The bar frequency: daily or minute
:param bar_count: int
The number of bars desired.
:param end: datetime
The last trading date of the last bar.
:return:
"""
start = get_start_dt(end, bar_count, data_frequency)
exchange_start = None
catalyst_end = None
if asset.end_minute is not None and start < asset.end_minute:
catalyst_start = start
if end <= asset.end_minute:
catalyst_end = end
else:
catalyst_end = asset.end_minute
delta = timedelta(minutes=1) \
if data_frequency == 'minute' else timedelta(days=1)
exchange_start = catalyst_end + delta
exchange_end = end
else:
exchange_end = end
exchange_start = start
data = []
if catalyst_end is not None:
# TODO: support multiple assets in the Catalyst API.
candles = bundle_utils.get_history(
exchange_name=self.name,
data_frequency=data_frequency,
symbol=asset.symbol, # TODO: use Catalyst symbol
start=catalyst_start,
end=catalyst_end
)
data += candles
if exchange_start is not None and fallback_exchange:
candles = self.get_candles(
data_frequency=data_frequency,
assets=[asset],
bar_count=bar_count
)
data += candles[asset]
return data
def get_history_window(self,
assets,
end_dt,
@@ -565,10 +473,12 @@ class Exchange:
raise InvalidHistoryFrequencyError(frequency)
if unit.lower() == 'd':
data_frequency = 'daily'
if data_frequency != 'daily':
raise InvalidHistoryFrequencyError(frequency=frequency)
elif unit.lower() == 'm':
data_frequency = 'minute'
if data_frequency != 'minute':
raise InvalidHistoryFrequencyError(frequency=frequency)
else:
raise InvalidHistoryFrequencyError(frequency)
@@ -635,12 +545,12 @@ class Exchange:
writer=writer
)
reader = bundle.get_reader(data_frequency)
values = reader.load_raw_arrays(
values = bundle.get_raw_arrays(
assets=assets,
fields=[field],
start_dt=start_dt,
end_dt=end_dt,
sids=[asset.sid for asset in assets],
data_frequency=data_frequency
)[0]
series = dict()
@@ -648,6 +558,7 @@ class Exchange:
all_dates = []
asset_values = []
# TODO: use numpy to avoid the loop
date = start_dt
for value in values:
all_dates.append(date)
catalyst/exchange/exchange_bundle.py
+71 -65
View File
@@ -78,40 +78,40 @@ class ExchangeBundle:
return start, end
def get_reader(self, data_frequency):
def get_reader(self, data_frequency, path=None):
"""
Get a data writer object, either a new object or from cache
:return: BcolzMinuteBarReader or BcolzDailyBarReader
"""
if data_frequency in self._readers \
and self._readers[data_frequency] is not None:
return self._readers[data_frequency]
if path is None:
root = get_exchange_folder(self.exchange.name)
path = BUNDLE_NAME_TEMPLATE.format(
root=root,
frequency=data_frequency
)
root = get_exchange_folder(self.exchange.name)
input_dir = BUNDLE_NAME_TEMPLATE.format(
root=root,
frequency=data_frequency
)
if path in self._readers and self._readers[path] is not None:
return self._readers[path]
self._readers[data_frequency] = None
self._readers[path] = None
if data_frequency == 'minute':
try:
self._readers[data_frequency] = BcolzMinuteBarReader(input_dir)
self._readers[path] = BcolzMinuteBarReader(path)
except IOError:
log.debug('no reader data found in {}'.format(input_dir))
log.debug('no reader data found in {}'.format(path))
elif data_frequency == 'daily':
try:
self._readers[data_frequency] = BcolzDailyBarReader(input_dir)
self._readers[path] = BcolzDailyBarReader(path)
except IOError:
log.debug('no reader data found in {}'.format(input_dir))
log.debug('no reader data found in {}'.format(path))
else:
raise InvalidHistoryFrequencyError(
frequency=data_frequency
)
return self._readers[data_frequency]
return self._readers[path]
def update_metadata(self, writer, start_dt, end_dt):
pass
@@ -122,21 +122,21 @@ class ExchangeBundle:
:return: BcolzMinuteBarWriter or BcolzDailyBarWriter
"""
key = data_frequency
if key in self._writers:
return self._writers[key]
root = get_exchange_folder(self.exchange.name)
output_dir = BUNDLE_NAME_TEMPLATE.format(
path = BUNDLE_NAME_TEMPLATE.format(
root=root,
frequency=data_frequency
)
ensure_directory(output_dir)
if path in self._writers:
return self._writers[path]
ensure_directory(path)
if data_frequency == 'minute':
if len(os.listdir(output_dir)) > 0:
if len(os.listdir(path)) > 0:
metadata = BcolzMinuteBarMetadata.read(output_dir)
metadata = BcolzMinuteBarMetadata.read(path)
write_metadata = False
if start_dt < metadata.start_session:
@@ -152,9 +152,9 @@ class ExchangeBundle:
else:
end_session = metadata.end_session
self._writers[key] = \
self._writers[path] = \
BcolzMinuteBarWriter(
output_dir,
path,
metadata.calendar,
start_session,
end_session,
@@ -164,8 +164,8 @@ class ExchangeBundle:
write_metadata=write_metadata
)
else:
self._writers[key] = BcolzMinuteBarWriter(
rootdir=output_dir,
self._writers[path] = BcolzMinuteBarWriter(
rootdir=path,
calendar=self.calendar,
minutes_per_day=self.minutes_per_day,
start_session=start_dt,
@@ -176,8 +176,8 @@ class ExchangeBundle:
elif data_frequency == 'daily':
end_session = end_dt.floor('1d')
self._writers[key] = BcolzDailyBarWriter(
filename=output_dir,
self._writers[path] = BcolzDailyBarWriter(
filename=path,
calendar=self.calendar,
start_session=start_dt,
end_session=end_session
@@ -187,7 +187,7 @@ class ExchangeBundle:
frequency=data_frequency
)
return self._writers[key]
return self._writers[path]
def filter_existing_assets(self, assets, start_dt, end_dt, data_frequency):
"""
@@ -315,6 +315,30 @@ class ExchangeBundle:
return data
def get_raw_arrays(self, assets, start_dt, end_dt, fields, data_frequency,
path=None):
reader = self.get_reader(data_frequency, path)
if data_frequency == 'minute':
values = reader.load_raw_arrays(
fields=fields,
start_dt=start_dt,
end_dt=end_dt,
sids=[asset.sid for asset in assets],
)
else:
# Note that the parameters convention is totally different
# from the minute reader.
values = reader.load_raw_arrays(
columns=fields,
start_date=start_dt,
end_date=end_dt,
assets=assets
)
return values
def download_bundle(self, name):
"""
@@ -347,41 +371,23 @@ class ExchangeBundle:
period=period
)
sid = asset.sid
if data_frequency == 'minute':
reader = BcolzMinuteBarReader(path)
# TODO: is this the optimal approach?
# Ensures that we read exact range which we want to write
start_dt = writer._start_session
end_dt = writer._end_session
start = reader.first_trading_day
end = reader.last_available_dt
periods = self.calendar.minutes_in_range(start_dt, end_dt) \
if data_frequency == 'minute' \
else self.calendar.sessions_in_range(start_dt, end_dt)
periods = self.calendar.minutes_in_range(start, end)
arrays = reader.load_raw_arrays(
fields=['open', 'high', 'low', 'close', 'volume'],
start_dt=start,
end_dt=end,
sids=[sid]
)
elif data_frequency == 'daily':
reader = BcolzDailyBarReader(path)
start = writer._start_session
end = writer._end_session
periods = self.calendar.sessions_in_range(start, end)
# Note that the parameters convention is totally different
# from the minute reader.
arrays = reader.load_raw_arrays(
columns=['open', 'high', 'low', 'close', 'volume'],
start_date=start,
end_date=end,
assets=[asset]
)
else:
raise InvalidHistoryFrequencyError(frequency=data_frequency)
arrays = self.get_raw_arrays(
assets=[asset],
fields=['open', 'high', 'low', 'close', 'volume'],
start_dt=start_dt,
end_dt=end_dt,
data_frequency=data_frequency,
path=path
)
ohlcv = dict(
open=arrays[0].flatten(),
@@ -442,7 +448,7 @@ class ExchangeBundle:
data = []
if not df.empty:
df.sort_index(inplace=True)
data.append((sid, df))
data.append((asset.sid, df))
self._write(data, writer, data_frequency)
@@ -573,5 +579,5 @@ class ExchangeBundle:
data_frequency=data_frequency,
period=chunk['period'],
writer=writer,
empty_rows_behavior='ignore'
empty_rows_behavior='strip'
)
tests/exchange/test_bundle.py
+3 -3
View File
@@ -15,15 +15,15 @@ class ExchangeBundleTestCase:
exchange_name = 'poloniex'
# start = pd.to_datetime('2017-09-01', utc=True)
start = pd.to_datetime('2017-1-1', utc=True)
start = pd.to_datetime('2017-9-1', utc=True)
end = pd.to_datetime('2017-9-30', utc=True)
exchange_bundle = ExchangeBundle(get_exchange(exchange_name))
log.info('ingesting exchange bundle {}'.format(exchange_name))
exchange_bundle.ingest(
data_frequency='minute',
include_symbols='gno_btc',
data_frequency='daily',
include_symbols='etc_btc',
# include_symbols=None,
exclude_symbols=None,
start=start,