BLD: ingesting and cleaning marketplace data sources

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
+

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

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

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

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