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

BUG: work around for issue #53 and possibly fixed issue #47

Browse Source
This commit is contained in:
fredfortier
2017-11-03 20:59:26 -04:00
parent 2bbc0c00cc
commit b476024612
5 changed files with 114 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/examples/simple_loop.py
+16 -16
View File
@@ -29,25 +29,25 @@ def handle_data(context, data):
print(e)
# run_algorithm(
# capital_base=250,
# start=pd.to_datetime('2017-1-1', utc=True),
# end=pd.to_datetime('2017-10-22', utc=True),
# data_frequency='daily',
# initialize=initialize,
# handle_data=handle_data,
# analyze=None,
# exchange_name='bitfinex',
# algo_namespace='simple_loop',
# base_currency='btc'
# )
run_algorithm(
capital_base=250,
start=pd.to_datetime('2017-1-1', utc=True),
end=pd.to_datetime('2017-10-22', utc=True),
data_frequency='daily',
initialize=initialize,
handle_data=handle_data,
analyze=None,
exchange_name='poloniex',
live=True,
exchange_name='bitfinex',
algo_namespace='simple_loop',
base_currency='eth',
live_graph=False
base_currency='btc'
)
# run_algorithm(
# initialize=initialize,
# handle_data=handle_data,
# analyze=None,
# exchange_name='poloniex',
# live=True,
# algo_namespace='simple_loop',
# base_currency='eth',
# live_graph=False
# )
catalyst/exchange/bundle_utils.py
+6
View File
@@ -223,6 +223,9 @@ def get_month_start_end(dt, first_day=None, last_day=None):
dt.year, dt.month, month_range[1], 23, 59, 0, 0
), utc=True)
if month_end > pd.Timestamp.utcnow():
month_end = pd.Timestamp.utcnow().floor('1D')
return month_start, month_end
@@ -247,6 +250,9 @@ def get_year_start_end(dt, first_day=None, last_day=None):
year_end = last_day if last_day \
else pd.to_datetime(date(dt.year, 12, 31), utc=True)
if year_end > pd.Timestamp.utcnow():
year_end = pd.Timestamp.utcnow().floor('1D')
return year_start, year_end
catalyst/exchange/exchange_bundle.py
+39 -39
View File
@@ -303,7 +303,7 @@ class ExchangeBundle:
:return:
"""
# Download and extract the bundle
path = get_bcolz_chunk(
exchange_name=self.exchange.name,
symbol=asset.symbol,
@@ -335,7 +335,7 @@ class ExchangeBundle:
))
if not arrays:
return path
return reader._rootdir
periods = self.get_calendar_periods_range(
start_dt, end_dt, data_frequency
@@ -351,11 +351,12 @@ class ExchangeBundle:
if cleanup:
log.debug(
'removing bundle folder following ingestion: {}'.format(path)
'removing bundle folder following ingestion: {}'.format(
reader._rootdir)
)
shutil.rmtree(path)
shutil.rmtree(reader._rootdir)
return path
return reader._rootdir
def get_adj_dates(self, start, end, assets, data_frequency):
"""
@@ -429,9 +430,7 @@ class ExchangeBundle:
get_start_end = get_month_start_end \
if data_frequency == 'minute' else get_year_start_end
start_dt, _ = get_start_end(start_dt)
_, end_dt = get_start_end(end_dt)
# Get a reader for the main bundle to verify if data exists
reader = self.get_reader(data_frequency)
chunks = dict()
@@ -462,7 +461,6 @@ class ExchangeBundle:
chunks[asset] = []
for index, dt in enumerate(dates):
period_start, period_end = get_start_end(
dt=dt,
first_day=dt if index == 0 else None,
@@ -481,17 +479,17 @@ class ExchangeBundle:
asset, range_start, period_end, reader
)
if not has_data:
chunks[asset].append(
dict(
asset=asset,
period_start=period_start,
period_end=period_end,
period=get_period_label(dt, data_frequency)
)
period = get_period_label(dt, data_frequency)
chunk = dict(
asset=asset,
period=period,
)
chunks[asset].append(chunk)
# We sort the chunks by end date to ingest most recent data first
chunks[asset].sort(key=lambda chunk: chunk['period_end'])
chunks[asset].sort(
key=lambda chunk: pd.to_datetime(chunk['period'])
)
return chunks
@@ -503,20 +501,26 @@ class ExchangeBundle:
Parameters
----------
assets: list[TradingPair]
data_frequency: str
start_dt: datetime
end_dt: datetime
show_progress: bool
asset_chunks: bool
"""
if start_dt is None:
start_dt = self.calendar.first_session
if end_dt is None:
end_dt = pd.Timestamp.utcnow()
start_dt, end_dt = self.get_adj_dates(
start_dt, end_dt, assets, data_frequency
)
get_start_end = get_month_start_end \
if data_frequency == 'minute' else get_year_start_end
# Assign the first and last day of the period
start_dt, _ = get_start_end(start_dt)
_, end_dt = get_start_end(end_dt)
chunks = self.prepare_chunks(
assets=assets,
data_frequency=data_frequency,
@@ -524,19 +528,9 @@ class ExchangeBundle:
end_dt=end_dt
)
# Since chunks are either monthly or yearly, it is possible that
# our ingestion data range is greater than specified. We adjust
# the boundaries to ensure that the writer can write all data.
all_chunks = list(chain.from_iterable(itervalues(chunks)))
for chunk in all_chunks:
if chunk['period_start'] < start_dt:
start_dt = chunk['period_start']
if chunk['period_end'] > end_dt:
end_dt = chunk['period_end']
# This is the common writer for the entire exchange bundle
# we want to give an end_date far in time
writer = self.get_writer(start_dt, end_dt, data_frequency)
if asset_chunks:
for asset in chunks:
with maybe_show_progress(
@@ -558,6 +552,12 @@ class ExchangeBundle:
cleanup=True
)
else:
all_chunks = list(chain.from_iterable(itervalues(chunks)))
# We sort the chunks by end date to ingest most recent data first
all_chunks.sort(
key=lambda chunk: pd.to_datetime(chunk['period'])
)
with maybe_show_progress(
all_chunks,
show_progress,
@@ -600,12 +600,12 @@ class ExchangeBundle:
show_progress)
def get_history_window_series_and_load(self,
assets, # type: List[TradingPair]
end_dt, # type: Timestamp
bar_count, # type: int
field, # type: str
data_frequency, # type: str
algo_end_dt=None # type: Timestamp
assets,
end_dt,
bar_count,
field,
data_frequency,
algo_end_dt=None
):
"""
Retrieve price data history, ingest missing data.
catalyst/exchange/exchange_utils.py
+1 -2
View File
@@ -284,13 +284,12 @@ def save_algo_df(algo_name, key, df, environ=None, rel_path=None):
----------
algo_name: str
key: str
df: DataFrame
df: pd.DataFrame
environ:
rel_path: str
"""
folder = get_algo_folder(algo_name, environ)
if rel_path is not None:
folder = os.path.join(folder, rel_path)
ensure_directory(folder)
tests/exchange/test_bundle.py
+52 -16
View File
@@ -122,18 +122,20 @@ class TestExchangeBundle:
pass
def test_ingest_daily(self):
# exchange_name = 'bitfinex'
# data_frequency = 'daily'
# include_symbols = 'neo_btc,bch_btc,eth_btc'
exchange_name = 'poloniex'
exchange_name = 'bitfinex'
data_frequency = 'daily'
include_symbols = 'eth_btc'
include_symbols = 'btc_usd'
start = pd.to_datetime('2017-1-1', utc=True)
end = pd.to_datetime('2017-10-16', utc=True)
periods = get_periods_range(start, end, data_frequency)
# exchange_name = 'poloniex'
# data_frequency = 'daily'
# include_symbols = 'eth_btc'
# start = pd.to_datetime('2017-1-1', utc=True)
# end = pd.to_datetime('2017-10-16', utc=True)
# periods = get_periods_range(start, end, data_frequency)
start = None
end = None
exchange = get_exchange(exchange_name)
exchange_bundle = ExchangeBundle(exchange)
@@ -153,12 +155,18 @@ class TestExchangeBundle:
assets.append(exchange.get_asset(pair_symbol))
reader = exchange_bundle.get_reader(data_frequency)
start_dt = reader.first_trading_day
end_dt = reader.last_available_dt
if data_frequency == 'daily':
end_dt = end_dt - pd.Timedelta(hours=23, minutes=59)
for asset in assets:
arrays = reader.load_raw_arrays(
sids=[asset.sid],
fields=['close'],
start_dt=start,
end_dt=end
start_dt=start_dt,
end_dt=end_dt
)
print('found {} rows for {} ingestion\n{}'.format(
len(arrays[0]), asset.symbol, arrays[0])
@@ -430,22 +438,50 @@ class TestExchangeBundle:
print('\n' + df_to_string(df))
pass
def bundle_to_csv(self):
def main_bundle_to_csv(self):
exchange_name = 'bitfinex'
data_frequency = 'daily'
period = '2016'
data_frequency = 'minute'
exchange = get_exchange(exchange_name)
bundle = ExchangeBundle(exchange)
asset = exchange.get_asset('eth_btc')
self._bundle_to_csv(
asset=asset,
exchange=exchange,
data_frequency=data_frequency,
filename='{}_{}_{}'.format(
exchange_name, data_frequency, asset.symbol
)
)
def bundle_to_csv(self):
exchange_name = 'bitfinex'
data_frequency = 'minute'
period = '2017-06'
symbol = 'etc_btc'
exchange = get_exchange(exchange_name)
asset = exchange.get_asset(symbol)
path = get_bcolz_chunk(
exchange_name=exchange.name,
symbol=asset.symbol,
data_frequency=data_frequency,
period=period
)
self._bundle_to_csv(
asset=asset,
exchange=exchange,
data_frequency=data_frequency,
path=path,
filename=period
)
def _bundle_to_csv(self, asset, exchange, data_frequency, filename,
path=None):
bundle = ExchangeBundle(exchange)
reader = bundle.get_reader(data_frequency, path=path)
start_dt = reader.first_trading_day
end_dt = reader.last_available_dt
@@ -475,7 +511,7 @@ class TestExchangeBundle:
)
ensure_directory(folder)
path = os.path.join(folder, period + '.csv')
path = os.path.join(folder, filename + '.csv')
log.info('creating csv file: {}'.format(path))
print('HEAD\n{}'.format(df.head(10)))