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

WIP: Generic Data Bundles

Browse Source 
* Split up inheritance structure of Bundle classes
This commit is contained in:
Conner Fromknecht
2017-07-13 01:34:55 -07:00
parent e8e2346c60
commit cabea94973
4 changed files with 208 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/data/bundles/bundle.py → catalyst/data/bundles/base.py
+1 -4
View File
@@ -30,10 +30,7 @@ from catalyst.utils.memoize import lazyval
logbook.StderrHandler().push_application()
log = logbook.Logger(__name__)
class AbstractBundle(object):
def __init__(self):
pass
class BaseBundle(object):
@lazyval
def name(self):
raise NotImplementedError()
catalyst/data/bundles/base_pricing.py
+56
View File
@@ -0,0 +1,56 @@
#
# Copyright 2017 Enigma MPC, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from catalyst.data.bundles.bundle import BaseBundle
from catalyst.utils.memoize import lazyval
class BasePricingBundle(BaseBundle):
@lazyval
def md_dtypes(self):
return [
('symbol', 'object'),
('start_date', 'datetime64[ns]'),
('end_date', 'datetime64[ns]'),
('ac_date', 'datetime64[ns]'),
]
@lazyval
def dtypes(self):
return [
('date', 'datetime64[ns]'),
('open', 'float64'),
('high', 'float64'),
('low', 'float64'),
('close', 'float64'),
('volume', 'float64'),
]
class BaseCryptoPricingBundle(BasePricingBundle):
@lazyval
def calendar_name(self):
return 'OPEN'
@lazyval
def minutes_per_day(self):
return 1440
class BaseEquityPricingBundle(BasePricingBundle):
@lazyval
def calendar_name(self):
return 'NYSE'
@lazyval
def minutes_per_day(self):
return 390
catalyst/data/bundles/poloniex.py
+151 -32
View File
@@ -1,37 +1,156 @@
import tarfile
#
# Copyright 2017 Enigma MPC, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from . import core as bundles
from datetime import datetime
POLONIEX_BUNDLE_URL = (
'https://www.dropbox.com/s/9naqffawnq8o4r2/poloniex-bundle.tar?dl=1'
)
import pandas as pd
#@bundles.register(
# 'poloniex',
# create_writers=False,
# calendar_name='OPEN',
# minutes_per_day=1440)
def poloniex_bundle(environ,
asset_db_writer,
minute_bar_writer,
daily_bar_writer,
adjustment_writer,
calendar,
start_session,
end_session,
cache,
show_progress,
output_dir):
if show_progress:
data = bundles.download_with_progress(
POLONIEX_BUNDLE_URL,
chunk_size=bundles.ONE_MEGABYTE,
label="Downloading Bundle: poloniex",
from six.moves.urllib.parse import urlencode
from catalyst.data.bundles.core import register_bundle
from catalyst.data.bundles.pricing_bundle import BaseCryptoPricingBundle
from catalyst.utils.memoize import lazyval
class PoloniexBundle(BaseCryptoPricingBundle):
@lazyval
def name(self):
return 'poloniex'
@lazyval
def exchange(self):
return 'POLO'
@lazyval
def frequencies(self):
return set((
'daily',
))
@lazyval
def tar_url(self):
return (
'https://www.dropbox.com/s/9naqffawnq8o4r2/'
'poloniex-bundle.tar?dl=1'
)
else:
data = bundles.download_without_progress(POLONIEX_BUNDLE_URL)
with tarfile.open('r', fileobj=data) as tar:
if show_progress:
print("Writing data to %s." % output_dir)
tar.extractall(output_dir)
@lazyval
def wait_time(self):
return pd.Timedelta(milliseconds=170)
def fetch_raw_metadata_frame(self, api_key, page_number):
if page_number > 1:
return pd.DataFrame([])
raw = pd.read_json(
self._format_metadata_url(
api_key,
page_number,
),
orient='index',
)
raw = raw.sort_index().reset_index()
raw.rename(
columns={'index':'symbol'},
inplace=True,
)
return raw
def post_process_symbol_metadata(self, metadata, data):
start_date = data.index[0].tz_localize(None)
end_date = data.index[-1].tz_localize(None)
ac_date = end_date + pd.Timedelta(days=1)
return (
metadata.symbol,
start_date,
end_date,
ac_date,
)
def fetch_raw_symbol_frame(self,
api_key,
symbol,
start_date,
end_date,
frequency):
raw = pd.read_json(
self._format_data_url(
api_key,
symbol,
start_date,
end_date,
frequency,
),
orient='records',
)
raw.set_index('date', inplace=True)
scale = 1000.0
raw.loc[:, 'open'] /= scale
raw.loc[:, 'high'] /= scale
raw.loc[:, 'low'] /= scale
raw.loc[:, 'close'] /= scale
raw.loc[:, 'volume'] *= scale
return raw
'''
HELPER METHODS
'''
def _format_metadata_url(self, api_key, page_number):
query_params = [
('command', 'returnTicker'),
]
return self._format_polo_query(query_params)
def _format_data_url(self,
api_key,
symbol,
start_date,
end_date,
data_frequency):
period_map = {
'daily': 86400,
'5-minute': 300,
'minute': 60,
}
try:
period = period_map[data_frequency]
except KeyError:
return None
query_params = [
('command', 'returnChartData'),
('currencyPair', symbol),
('start', start_date.value / 10**9),
('end', end_date.value / 10**9),
('period', period),
]
return self._format_polo_query(query_params)
def _format_polo_query(self, query_params):
return 'https://poloniex.com/public?{query}'.format(
query=urlencode(query_params),
)
register_bundle(PoloniexBundle)
catalyst/data/bundles/poloniex_bundle.py
-187
View File
@@ -1,187 +0,0 @@
#
# Copyright 2017 Enigma MPC, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import datetime
import pandas as pd
from six.moves.urllib.parse import urlencode
from catalyst.data.bundles.core import register_bundle
from catalyst.data.bundles.bundle import AbstractBundle
from catalyst.utils.memoize import lazyval
class PoloniexBundle(AbstractBundle):
def __init__(self):
super(self.__class__, self).__init__()
@lazyval
def name(self):
return 'poloniex'
@lazyval
def exchange(self):
return 'POLO'
@lazyval
def calendar_name(self):
return 'OPEN'
@lazyval
def minutes_per_day(self):
return 1440
@lazyval
def frequencies(self):
return set((
'daily',
))
@lazyval
def md_dtypes(self):
return [
('symbol', 'object'),
('start_date', 'datetime64[ns]'),
('end_date', 'datetime64[ns]'),
('ac_date', 'datetime64[ns]'),
]
@lazyval
def dtypes(self):
return [
('date', 'datetime64[ns]'),
('open', 'float64'),
('high', 'float64'),
('low', 'float64'),
('close', 'float64'),
('volume', 'float64'),
]
@lazyval
def tar_url(self):
return (
'https://www.dropbox.com/s/9naqffawnq8o4r2/'
'poloniex-bundle.tar?dl=1'
)
@lazyval
def wait_time(self):
return pd.Timedelta(milliseconds=170)
def fetch_raw_metadata_frame(self, api_key, page_number):
if page_number > 1:
return pd.DataFrame([])
raw = pd.read_json(
self._format_metadata_url(
api_key,
page_number,
),
orient='index',
)
raw = raw.sort_index().reset_index()
raw.rename(
columns={'index':'symbol'},
inplace=True,
)
return raw
def post_process_symbol_metadata(self, metadata, data):
start_date = data.index[0].tz_localize(None)
end_date = data.index[-1].tz_localize(None)
ac_date = end_date + pd.Timedelta(days=1)
return (
metadata.symbol,
start_date,
end_date,
ac_date,
)
def fetch_raw_symbol_frame(self,
api_key,
symbol,
start_date,
end_date,
frequency):
raw = pd.read_json(
self._format_data_url(
api_key,
symbol,
start_date,
end_date,
frequency,
),
orient='records',
)
raw.set_index('date', inplace=True)
scale = 1000.0
raw.loc[:, 'open'] /= scale
raw.loc[:, 'high'] /= scale
raw.loc[:, 'low'] /= scale
raw.loc[:, 'close'] /= scale
raw.loc[:, 'volume'] *= scale
return raw
'''
HELPER METHODS
'''
def _format_metadata_url(self, api_key, page_number):
query_params = [
('command', 'returnTicker'),
]
return self._format_polo_query(query_params)
def _format_data_url(self,
api_key,
symbol,
start_date,
end_date,
data_frequency):
period_map = {
'daily': 86400,
'5-minute': 300,
'minute': 60,
}
try:
period = period_map[data_frequency]
except KeyError:
return None
query_params = [
('command', 'returnChartData'),
('currencyPair', symbol),
('start', start_date.value / 10**9),
('end', end_date.value / 10**9),
('period', period),
]
return self._format_polo_query(query_params)
def _format_polo_query(self, query_params):
return 'https://poloniex.com/public?{query}'.format(
query=urlencode(query_params),
)
register_bundle(PoloniexBundle)