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

BLD: made some adjustments to the client when testing the Binance data bundle

Browse Source
This commit is contained in:
Frederic Fortier
2018-03-03 23:09:59 -05:00
parent 964c90176b
commit 0d051a9496
5 changed files with 89 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/constants.py
+3 -1
View File
@@ -11,8 +11,10 @@ LOG_LEVEL = int(os.environ.get('CATALYST_LOG_LEVEL', logbook.INFO))
SYMBOLS_URL = 'https://s3.amazonaws.com/enigmaco/catalyst-exchanges/' \
'{exchange}/symbols.json'
EXCHANGE_CONFIG_URL = 'https://s3.amazonaws.com/enigmaco/exchanges/' \
EXCHANGE_CONFIG_URL = 'https://s3.amazonaws.com/enigmaco/ohlcv/' \
'{exchange}/config.json'
BUNDLE_URL = 'https://s3.amazonaws.com/enigmaco/ohlcv/' \
'{exchange}/{data_frequency}/{name}.tar.gz'
DATE_TIME_FORMAT = '%Y-%m-%d %H:%M'
DATE_FORMAT = '%Y-%m-%d'
catalyst/exchange/exchange_bundle.py
+20 -13
View File
@@ -28,7 +28,8 @@ from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
from catalyst.exchange.utils.bundle_utils import range_in_bundle, \
get_bcolz_chunk, get_df_from_arrays, get_assets
from catalyst.exchange.utils.datetime_utils import get_start_dt, \
get_period_label, get_month_start_end, get_year_start_end
get_period_label, get_month_start_end, get_year_start_end, get_period, \
timestr_to_dt
from catalyst.exchange.utils.exchange_utils import get_exchange_folder
from catalyst.utils.cli import maybe_show_progress
from catalyst.utils.paths import ensure_directory
@@ -513,8 +514,8 @@ class ExchangeBundle:
continue
dates = pd.date_range(
start=get_period_label(adj_start, data_frequency),
end=get_period_label(adj_end, data_frequency),
start=get_period(adj_start, data_frequency),
end=get_period(adj_end, data_frequency),
freq='MS' if data_frequency == 'minute' else 'AS',
tz=UTC
)
@@ -553,7 +554,9 @@ class ExchangeBundle:
# We sort the chunks by end date to ingest most recent data first
chunks[asset].sort(
key=lambda chunk: pd.to_datetime(chunk['period'])
key=lambda chunk: timestr_to_dt(
chunk['period'], data_frequency
)
)
return chunks
@@ -608,7 +611,8 @@ class ExchangeBundle:
exchange=self.exchange_name,
frequency=data_frequency,
symbol=asset.symbol
)) as it:
)
) as it:
for chunk in it:
problems += self.ingest_ctable(
asset=chunk['asset'],
@@ -623,16 +627,19 @@ class ExchangeBundle:
# We sort the chunks by end date to ingest most recent data first
all_chunks.sort(
key=lambda chunk: pd.to_datetime(chunk['period'])
key=lambda chunk: timestr_to_dt(
chunk['period'], data_frequency
)
)
with maybe_show_progress(
all_chunks,
show_progress,
label='Ingesting {frequency} price data on '
'{exchange}'.format(
exchange=self.exchange_name,
frequency=data_frequency,
)) as it:
all_chunks,
show_progress,
label='Ingesting {frequency} price data on '
'{exchange}'.format(
exchange=self.exchange_name,
frequency=data_frequency,
)
) as it:
for chunk in it:
problems += self.ingest_ctable(
asset=chunk['asset'],
catalyst/exchange/utils/bundle_utils.py
+6 -5
View File
@@ -5,6 +5,7 @@ from datetime import datetime
import numpy as np
import pandas as pd
from catalyst.constants import BUNDLE_URL
from catalyst.data.bundles.core import download_without_progress
from catalyst.exchange.utils.exchange_utils import get_exchange_bundles_folder
import os
@@ -48,10 +49,11 @@ def get_bcolz_chunk(exchange_name, symbol, data_frequency, period):
path = os.path.join(root, name)
if not os.path.isdir(path):
url = 'https://s3.amazonaws.com/enigmaco/catalyst-bundles/' \
'exchange-{exchange}/{name}.tar.gz'.format(
url = BUNDLE_URL.format(
exchange=exchange_name,
name=name)
data_frequency=data_frequency,
name=name,
)
bytes = download_without_progress(url)
with tarfile.open('r', fileobj=bytes) as tar:
@@ -75,8 +77,7 @@ def get_df_from_arrays(arrays, periods):
"""
ohlcv = dict()
for index, field in enumerate(
['open', 'high', 'low', 'close', 'volume']):
for index, field in enumerate(['open', 'high', 'low', 'close', 'volume']):
ohlcv[field] = arrays[index].flatten()
df = pd.DataFrame(
catalyst/exchange/utils/datetime_utils.py
+26
View File
@@ -164,6 +164,12 @@ def get_start_dt(end_dt, bar_count, data_frequency, include_first=True):
return start_dt
def timestr_to_dt(timestr, data_frequency):
dt_format = '%Y' if data_frequency == 'daily' else '%Y%m'
dt = pd.to_datetime(timestr, format=dt_format, utc=True)
return dt
def get_period_label(dt, data_frequency):
"""
The period label for the specified date and frequency.
@@ -177,6 +183,26 @@ def get_period_label(dt, data_frequency):
-------
str
"""
if data_frequency == 'minute':
return '{}{:02d}'.format(dt.year, dt.month)
else:
return '{}'.format(dt.year)
def get_period(dt, data_frequency):
"""
The period label for the specified date and frequency.
Parameters
----------
dt: datetime
data_frequency: str
Returns
-------
str
"""
if data_frequency == 'minute':
return '{}-{:02d}'.format(dt.year, dt.month)
tests/exchange/test_bundle.py
+34 -16
View File
@@ -11,10 +11,11 @@ from catalyst.exchange.exchange_bundle import ExchangeBundle, \
BUNDLE_NAME_TEMPLATE
from catalyst.exchange.utils.bundle_utils import get_bcolz_chunk, \
get_df_from_arrays
from exchange.utils.datetime_utils import get_start_dt
from catalyst.exchange.utils.datetime_utils import get_start_dt
from catalyst.exchange.utils.exchange_utils import get_exchange_folder
from catalyst.exchange.utils.factory import get_exchange
from catalyst.exchange.utils.stats_utils import df_to_string
from catalyst.exchange.utils.stats_utils import df_to_string, \
set_print_settings
from catalyst.utils.paths import ensure_directory
log = getLogger('test_exchange_bundle')
@@ -42,16 +43,16 @@ class TestExchangeBundle:
def test_ingest_minute(self):
data_frequency = 'minute'
exchange_name = 'poloniex'
exchange_name = 'binance'
exchange = get_exchange(exchange_name)
exchange_bundle = ExchangeBundle(exchange)
exchange_bundle = ExchangeBundle(exchange_name)
assets = [
exchange.get_asset('eth_btc')
exchange.get_asset('eng_eth')
]
start = pd.to_datetime('2016-03-01', utc=True)
end = pd.to_datetime('2017-11-1', utc=True)
start = pd.to_datetime('2018-02-01', utc=True)
end = pd.to_datetime('2018-02-10', utc=True)
log.info('ingesting exchange bundle {}'.format(exchange_name))
exchange_bundle.ingest(
@@ -61,7 +62,8 @@ class TestExchangeBundle:
exclude_symbols=None,
start=start,
end=end,
show_progress=True
show_progress=False,
show_breakdown=False
)
reader = exchange_bundle.get_reader(data_frequency)
@@ -72,9 +74,15 @@ class TestExchangeBundle:
start_dt=start,
end_dt=end
)
print('found {} rows for {} ingestion\n{}'.format(
len(arrays[0]), asset.symbol, arrays[0])
periods = exchange_bundle.get_calendar_periods_range(
start, end, data_frequency
)
dx = get_df_from_arrays(arrays[0], periods)
set_print_settings()
print('found {} rows for last ingestion:\n{}\n{}'.format(
len(dx), dx.head(1000), dx.tail(1000)
))
pass
def test_ingest_minute_all(self):
@@ -222,9 +230,14 @@ class TestExchangeBundle:
start_dt=start,
end_dt=end
)
print('found {} rows for {} ingestion\n{}'.format(
len(arrays[0]), asset.symbol, arrays[0])
periods = exchange_bundle.get_calendar_periods_range(
start, end, data_frequency
)
dx = get_df_from_arrays(arrays, periods)
print('found {} rows for last ingestion'.format(
len(dx)
))
pass
def test_daily_data_to_minute_table(self):
@@ -290,17 +303,22 @@ class TestExchangeBundle:
for asset in assets:
sid = asset.sid
daily_values = reader.load_raw_arrays(
arrays = reader.load_raw_arrays(
fields=['open', 'high', 'low', 'close', 'volume'],
start_dt=start,
end_dt=end,
sids=[sid],
)
print('found {} rows for last ingestion'.format(
len(daily_values[0]))
periods = exchange_bundle.get_calendar_periods_range(
start, end, data_frequency
)
pass
dx = get_df_from_arrays(arrays, periods)
print('found {} rows for last ingestion'.format(
len(dx)
))
pass
def test_minute_bundle(self):
# exchange_name = 'poloniex'