diff --git a/catalyst/exchange/utils/bundle_utils.py b/catalyst/exchange/utils/bundle_utils.py index e9207511..992cabbb 100644 --- a/catalyst/exchange/utils/bundle_utils.py +++ b/catalyst/exchange/utils/bundle_utils.py @@ -1,6 +1,8 @@ import calendar import os import tarfile + +import bcolz from datetime import timedelta, datetime, date import numpy as np @@ -355,3 +357,4 @@ def get_assets(exchange, include_symbols, exclude_symbols): else: return all_assets + diff --git a/catalyst/marketplace/marketplace.py b/catalyst/marketplace/marketplace.py index e22d3f49..c71cef1c 100644 --- a/catalyst/marketplace/marketplace.py +++ b/catalyst/marketplace/marketplace.py @@ -1,12 +1,17 @@ import json import os + +import bcolz import pandas as pd +import shutil from web3 import Web3, HTTPProvider from catalyst.exchange.utils.stats_utils import set_print_settings from catalyst.constants import ROOT_DIR -from catalyst.marketplace.utils.paths import get_temp_bundles_folder +from catalyst.marketplace.utils.bundle_utils import merge_bundles +from catalyst.marketplace.utils.path_utils import get_temp_bundles_folder, \ + get_data_source, get_bundle_folder, get_data_source_folder REMOTE_NODE = 'http://localhost:7545' CONTRACT_PATH = os.path.join( @@ -114,9 +119,31 @@ class Marketplace: pass def ingest(self, data_source_name, data_frequency=None, start=None, - end=None): - temp_folder = get_temp_bundles_folder(data_source_name) + end=None, force_download=False): + data_source_name = data_source_name.lower() + + period = start.strftime('%Y-%m-%d') + tmp_folder = get_data_source(data_source_name, period, force_download) + + bundle_folder = get_bundle_folder(data_source_name, data_frequency) + if os.listdir(bundle_folder): + zsource = bcolz.ctable(rootdir=tmp_folder, mode='r') + ztarget = bcolz.ctable(rootdir=bundle_folder, mode='r') + merge_bundles(zsource, ztarget) + + else: + os.rename(tmp_folder, bundle_folder) + pass - def clean(self, data_source_name): + def clean(self, data_source_name, data_frequency=None): + data_source_name = data_source_name.lower() + + if data_frequency is None: + folder = get_data_source_folder(data_source_name) + + else: + forlder = get_bundle_folder(data_source_name, data_frequency) + + shutil.rmtree(folder) pass diff --git a/catalyst/marketplace/utils/bundle_utils.py b/catalyst/marketplace/utils/bundle_utils.py new file mode 100644 index 00000000..916722da --- /dev/null +++ b/catalyst/marketplace/utils/bundle_utils.py @@ -0,0 +1,36 @@ +import bcolz +import os +import pandas as pd +import shutil + + +def merge_bundles(zsource, ztarget): + """ + Merge + Parameters + ---------- + zsource + ztarget + + Returns + ------- + + """ + # TODO: find a way to do this iteratively instead of in-memory + df_source = zsource.todataframe() + df_source.set_index('last_updated', drop=False, inplace=True) + df_target = ztarget.todataframe() + df_target.set_index('last_updated', drop=False, inplace=True) + + df = df_target.merge( + right=df_source, + how='right', + ) # type: pd.DataFrame + + dirname = os.path.basename(ztarget.rootdir) + bak_dir = ztarget.rootdir.replace(dirname, '.{}'.format(dirname)) + os.rename(ztarget.rootdir, bak_dir) + + z = bcolz.ctable.fromdataframe(df=df, rootdir=ztarget.rootdir) + shutil.rmtree(bak_dir) + return z diff --git a/catalyst/marketplace/utils/paths.py b/catalyst/marketplace/utils/path_utils.py similarity index 61% rename from catalyst/marketplace/utils/paths.py rename to catalyst/marketplace/utils/path_utils.py index 008ed2b7..6d301880 100644 --- a/catalyst/marketplace/utils/paths.py +++ b/catalyst/marketplace/utils/path_utils.py @@ -1,6 +1,8 @@ import os import tarfile +import shutil + from catalyst.data.bundles.core import download_without_progress from catalyst.utils.paths import data_root, ensure_directory @@ -29,6 +31,17 @@ def get_data_source_folder(data_source_name, environ=None): return data_source_folder +def get_bundle_folder(data_source_name, data_frequency, environ=None): + data_source_folder = get_data_source_folder(data_source_name, environ) + + subfolder = data_frequency if data_frequency is not None else 'data' + bundle_folder = os.path.join(data_source_folder, subfolder) + + ensure_directory(bundle_folder) + + return bundle_folder + + def get_temp_bundles_folder(data_source_name, environ=None): """ The temp folder for bundle downloads by algo name. @@ -51,7 +64,7 @@ def get_temp_bundles_folder(data_source_name, environ=None): return temp_bundles -def get_data_source(data_source_name, period): +def get_data_source(data_source_name, period, force_download=False): """ Download and extract a bcolz bundle. @@ -65,23 +78,30 @@ def get_data_source(data_source_name, period): Returns ------- str - Filename: bitfinex-daily-neo_eth-2017-10.tar.gz """ root = get_temp_bundles_folder(data_source_name) - name = '{data_source}-{period}'.format( + name = '{data_source}_{period}'.format( data_source=data_source_name, period=period, ) path = os.path.join(root, name) - if not os.path.isdir(path): - url = 'http://127.0.0.1:8080/{data_source}/{name}.tar.gz'.format( - data_source=data_source_name, - name=name, - ) - bytes = download_without_progress(url) - with tarfile.open('r', fileobj=bytes) as tar: - tar.extractall(path) + if os.path.isdir(path): + if force_download: + shutil.rmtree(path) + + else: + return path + + ensure_directory(path) + + url = 'http://127.0.0.1:8080/{data_source}/{name}.tar.gz'.format( + data_source=data_source_name, + name=name, + ) + bytes = download_without_progress(url) + with tarfile.open('r', fileobj=bytes) as tar: + tar.extractall(path) return path diff --git a/tests/marketplace/test_marketplace.py b/tests/marketplace/test_marketplace.py index 3feeb9a6..68990791 100644 --- a/tests/marketplace/test_marketplace.py +++ b/tests/marketplace/test_marketplace.py @@ -17,8 +17,9 @@ class TestMarketplace(WithLogger, ZiplineTestCase): def test_ingest(self): marketplace = Marketplace() marketplace.ingest( - data_source_name='GitHub', + data_source_name='Marketcap', data_frequency='finest', start=pd.Timestamp.utcnow(), + force_download=True, ) pass