BLD: completed data market place first integration and created an algo to test it

catalyst/api.pyi

@@ -34,6 +34,7 @@ def attach_pipeline(pipeline, name, chunks=None):
     :func:`catalyst.api.pipeline_output`
     """
 
+
 def batch_market_order(share_counts):
     """Place a batch market order for multiple assets.
 
@@ -48,6 +49,7 @@ def batch_market_order(share_counts):
         Index of ids for newly-created orders.
     """
 
+
 def cancel_order(order_param):
     """Cancel an open order.
 
@@ -57,7 +59,9 @@ def cancel_order(order_param):
         The order_id or order object to cancel.
     """
 
-def continuous_future(root_symbol_str, offset=0, roll='volume', adjustment='mul'):
+
+def continuous_future(root_symbol_str, offset=0, roll='volume',
+                      adjustment='mul'):
     """Create a specifier for a continuous contract.
 
     Parameters
@@ -81,7 +85,10 @@ def continuous_future(root_symbol_str, offset=0, roll='volume', adjustment='mul'
         The continuous future specifier.
     """
 
-def fetch_csv(url, pre_func=None, post_func=None, date_column='date', date_format=None, timezone='UTC', symbol=None, mask=True, symbol_column=None, special_params_checker=None, **kwargs):
+
+def fetch_csv(url, pre_func=None, post_func=None, date_column='date',
+              date_format=None, timezone='UTC', symbol=None, mask=True,
+              symbol_column=None, special_params_checker=None, **kwargs):
     """Fetch a csv from a remote url and register the data so that it is
     queryable from the ``data`` object.
 
@@ -125,6 +132,7 @@ def fetch_csv(url, pre_func=None, post_func=None, date_column='date', date_forma
         A requests source that will pull data from the url specified.
     """
 
+
 def future_symbol(symbol):
     """Lookup a futures contract with a given symbol.
 
@@ -144,6 +152,7 @@ def future_symbol(symbol):
         Raised when no contract named 'symbol' is found.
     """
 
+
 def get_datetime(tz=None):
     """
 Returns the current simulation datetime.
@@ -159,6 +168,7 @@ dt : datetime
     The current simulation datetime converted to ``tz``.
     """
 
+
 def get_environment(field='platform'):
     """Query the execution environment.
 
@@ -198,6 +208,7 @@ def get_environment(field='platform'):
         Raised when ``field`` is not a valid option.
     """
 
+
 def get_order(order_id):
     """Lookup an order based on the order id returned from one of the
     order functions.
@@ -213,10 +224,12 @@ def get_order(order_id):
         The order object.
     """
 
+
 def history(bar_count, frequency, field, ffill=True):
     """DEPRECATED: use ``data.history`` instead.
     """
 
+
 def order(asset, amount, limit_price=None, stop_price=None, style=None):
     """Place an order.
 
@@ -258,7 +271,9 @@ def order(asset, amount, limit_price=None, stop_price=None, style=None):
     :func:`catalyst.api.order_percent`
     """
 
-def order_percent(asset, percent, limit_price=None, stop_price=None, style=None):
+
+def order_percent(asset, percent, limit_price=None, stop_price=None,
+                  style=None):
     """Place an order in the specified asset corresponding to the given
     percent of the current portfolio value.
 
@@ -293,6 +308,7 @@ def order_percent(asset, percent, limit_price=None, stop_price=None, style=None)
     :func:`catalyst.api.order_value`
     """
 
+
 def order_target(asset, target, limit_price=None, stop_price=None, style=None):
     """Place an order to adjust a position to a target number of shares. If
     the position doesn't already exist, this is equivalent to placing a new
@@ -344,7 +360,9 @@ def order_target(asset, target, limit_price=None, stop_price=None, style=None):
     :func:`catalyst.api.order_target_value`
     """
 
-def order_target_percent(asset, target, limit_price=None, stop_price=None, style=None):
+
+def order_target_percent(asset, target, limit_price=None, stop_price=None,
+                         style=None):
     """Place an order to adjust a position to a target percent of the
     current portfolio value. If the position doesn't already exist, this is
     equivalent to placing a new order. If the position does exist, this is
@@ -396,7 +414,9 @@ def order_target_percent(asset, target, limit_price=None, stop_price=None, style
     :func:`catalyst.api.order_target_value`
     """
 
-def order_target_value(asset, target, limit_price=None, stop_price=None, style=None):
+
+def order_target_value(asset, target, limit_price=None, stop_price=None,
+                       style=None):
     """Place an order to adjust a position to a target value. If
     the position doesn't already exist, this is equivalent to placing a new
     order. If the position does exist, this is equivalent to placing an
@@ -448,6 +468,7 @@ def order_target_value(asset, target, limit_price=None, stop_price=None, style=N
     :func:`catalyst.api.order_target_percent`
     """
 
+
 def order_value(asset, value, limit_price=None, stop_price=None, style=None):
     """Place an order by desired value rather than desired number of
     shares.
@@ -488,6 +509,7 @@ def order_value(asset, value, limit_price=None, stop_price=None, style=None):
     :func:`catalyst.api.order_percent`
     """
 
+
 def pipeline_output(name):
     """Get the results of the pipeline that was attached with the name:
     ``name``.
@@ -514,6 +536,7 @@ def pipeline_output(name):
     :meth:`catalyst.pipeline.engine.PipelineEngine.run_pipeline`
     """
 
+
 def record(*args, **kwargs):
     """Track and record values each day.
 
@@ -529,7 +552,9 @@ def record(*args, **kwargs):
     :func:`~catalyst.run_algorithm`.
     """
 
-def schedule_function(func, date_rule=None, time_rule=None, half_days=True, calendar=None):
+
+def schedule_function(func, date_rule=None, time_rule=None, half_days=True,
+                      calendar=None):
     """Schedules a function to be called according to some timed rules.
 
     Parameters
@@ -549,6 +574,7 @@ def schedule_function(func, date_rule=None, time_rule=None, half_days=True, cale
     :class:`catalyst.api.time_rules`
     """
 
+
 def set_asset_restrictions(restrictions, on_error='fail'):
     """Set a restriction on which assets can be ordered.
 
@@ -562,6 +588,7 @@ def set_asset_restrictions(restrictions, on_error='fail'):
     catalyst.finance.asset_restrictions.Restrictions
     """
 
+
 def set_benchmark(benchmark):
     """Set the benchmark asset.
 
@@ -576,6 +603,7 @@ def set_benchmark(benchmark):
     automatically reinvested.
     """
 
+
 def set_cancel_policy(cancel_policy):
     """Sets the order cancellation policy for the simulation.
 
@@ -590,6 +618,7 @@ def set_cancel_policy(cancel_policy):
     :class:`catalyst.api.NeverCancel`
     """
 
+
 def set_commission(commission):
     """Sets the commission model for the simulation.
 
@@ -605,6 +634,7 @@ def set_commission(commission):
     :class:`catalyst.finance.commission.PerDollar`
     """
 
+
 def set_do_not_order_list(restricted_list, on_error='fail'):
     """Set a restriction on which assets can be ordered.
 
@@ -614,11 +644,13 @@ def set_do_not_order_list(restricted_list, on_error='fail'):
         The assets that cannot be ordered.
     """
 
+
 def set_long_only(on_error='fail'):
     """Set a rule specifying that this algorithm cannot take short
     positions.
     """
 
+
 def set_max_leverage(max_leverage):
     """Set a limit on the maximum leverage of the algorithm.
 
@@ -629,6 +661,7 @@ def set_max_leverage(max_leverage):
         be no maximum.
     """
 
+
 def set_max_order_count(max_count, on_error='fail'):
     """Set a limit on the number of orders that can be placed in a single
     day.
@@ -639,7 +672,9 @@ def set_max_order_count(max_count, on_error='fail'):
         The maximum number of orders that can be placed on any single day.
     """
 
-def set_max_order_size(asset=None, max_shares=None, max_notional=None, on_error='fail'):
+
+def set_max_order_size(asset=None, max_shares=None, max_notional=None,
+                       on_error='fail'):
     """Set a limit on the number of shares and/or dollar value of any single
     order placed for sid.  Limits are treated as absolute values and are
     enforced at the time that the algo attempts to place an order for sid.
@@ -658,7 +693,9 @@ def set_max_order_size(asset=None, max_shares=None, max_notional=None, on_error=
         The maximum value that can be ordered at one time.
     """
 
-def set_max_position_size(asset=None, max_shares=None, max_notional=None, on_error='fail'):
+
+def set_max_position_size(asset=None, max_shares=None, max_notional=None,
+                          on_error='fail'):
     """Set a limit on the number of shares and/or dollar value held for the
     given sid. Limits are treated as absolute values and are enforced at
     the time that the algo attempts to place an order for sid. This means
@@ -681,6 +718,7 @@ def set_max_position_size(asset=None, max_shares=None, max_notional=None, on_err
         The maximum value to hold for an asset.
     """
 
+
 def set_slippage(slippage):
     """Set the slippage model for the simulation.
 
@@ -694,6 +732,7 @@ def set_slippage(slippage):
     :class:`catalyst.finance.slippage.SlippageModel`
     """
 
+
 def set_symbol_lookup_date(dt):
     """Set the date for which symbols will be resolved to their assets
     (symbols may map to different firms or underlying assets at
@@ -705,6 +744,7 @@ def set_symbol_lookup_date(dt):
         The new symbol lookup date.
     """
 
+
 def sid(sid):
     """Lookup an Asset by its unique asset identifier.
 
@@ -724,6 +764,7 @@ def sid(sid):
         When a requested ``sid`` does not map to any asset.
     """
 
+
 def symbol(symbol_str):
     """Lookup an Equity by its ticker symbol.
 
@@ -748,6 +789,7 @@ def symbol(symbol_str):
     :func:`catalyst.api.set_symbol_lookup_date`
     """
 
+
 def symbols(*args):
     """Lookup multuple Equities as a list.
 
@@ -773,3 +815,21 @@ def symbols(*args):
     :func:`catalyst.api.set_symbol_lookup_date`
     """
 
+
+def get_data_source(data_source_name, data_frequency=None,
+                    start=None, end=None):
+    """
+    Lookup a data source from the marketplace
+
+    Parameters
+    ----------
+    self
+    data_source_name
+    data_frequency
+    start
+    end
+
+    Returns
+    -------
+
+    """

catalyst/constants.py

@@ -15,6 +15,9 @@ SYMBOLS_URL = 'https://s3.amazonaws.com/enigmaco/catalyst-exchanges/' \
 DATE_TIME_FORMAT = '%Y-%m-%d %H:%M'
 DATE_FORMAT = '%Y-%m-%d'
 
-ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
+try:
+    ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
+except Exception as e:
+    print('unable to get catalyst path: {}'.format(e))
 
 AUTO_INGEST = False

catalyst/examples/mean_reversion_by_marketcap.py

@@ -0,0 +1,233 @@
+# For this example, we're going to write a simple momentum script.  When the
+# stock goes up quickly, we're going to buy; when it goes down quickly, we're
+# going to sell.  Hopefully we'll ride the waves.
+import os
+import tempfile
+import time
+
+import pandas as pd
+import talib
+from logbook import Logger
+
+from catalyst import run_algorithm
+from catalyst.api import symbol, record, order_target_percent, get_data_source
+from catalyst.exchange.utils.stats_utils import set_print_settings, \
+    get_pretty_stats
+# We give a name to the algorithm which Catalyst will use to persist its state.
+# In this example, Catalyst will create the `.catalyst/data/live_algos`
+# directory. If we stop and start the algorithm, Catalyst will resume its
+# state using the files included in the folder.
+from catalyst.utils.paths import ensure_directory
+
+NAMESPACE = 'mean_reversion_simple'
+log = Logger(NAMESPACE)
+
+
+# To run an algorithm in Catalyst, you need two functions: initialize and
+# handle_data.
+
+def initialize(context):
+    # This initialize function sets any data or variables that you'll use in
+    # your algorithm.  For instance, you'll want to define the trading pair (or
+    # trading pairs) you want to backtest.  You'll also want to define any
+    # parameters or values you're going to use.
+
+    # In our example, we're looking at Neo in Ether.
+    df = get_data_source(
+        'marketcap', start=context.datetime
+    )  # type: pd.DataFrame
+
+    # Keep only the top coins by market cap
+    df = df.loc[df['market_cap_usd'].isin(df['market_cap_usd'].nlargest(100))]
+
+    set_print_settings()
+
+    df.sort_values(by=['market_cap_usd'], ascending=True, inplace=True)
+    print('the marketplace data:\n{}'.format(df))
+
+    # Pick the 5 assets with the lowest market cap for trading
+    quote_currency = 'eth'
+    exchange = context.exchanges[next(iter(context.exchanges))]
+    symbols = [a.symbol for a in exchange.assets
+               if a.start_date < context.datetime]
+    context.assets = []
+    for currency, price in df['market_cap_usd'].iteritems():
+        if len(context.assets) >= 5:
+            break
+
+        s = '{}_{}'.format(currency.decode('utf-8'), quote_currency)
+        if s in symbols:
+            context.assets.append(symbol(s))
+
+    context.base_price = None
+    context.current_day = None
+
+    context.RSI_OVERSOLD = 55
+    context.RSI_OVERBOUGHT = 60
+    context.CANDLE_SIZE = '5T'
+
+    context.start_time = time.time()
+
+
+def handle_data(context, data):
+    # This handle_data function is where the real work is done.  Our data is
+    # minute-level tick data, and each minute is called a frame.  This function
+    # runs on each frame of the data.
+
+    # We flag the first period of each day.
+    # Since cryptocurrencies trade 24/7 the `before_trading_starts` handle
+    # would only execute once. This method works with minute and daily
+    # frequencies.
+    today = data.current_dt.floor('1D')
+    if today != context.current_day:
+        context.traded_today = dict()
+        context.current_day = today
+
+    # Preparing dictionaries for asset-level data points
+    volumes = dict()
+    rsis = dict()
+    price_values = dict()
+    cash = context.portfolio.cash
+
+    for asset in context.assets:
+        # We're computing the volume-weighted-average-price of the security
+        # defined above, in the context.assets variable.  For this example,
+        #  we're using three bars on the 15 min bars.
+
+        # The frequency attribute determine the bar size. We use this
+        # convention for the frequency alias:
+        # http://pandas.pydata.org/pandas-docs/stable/timeseries.html#offset-aliases
+        prices = data.history(
+            asset,
+            fields='close',
+            bar_count=50,
+            frequency=context.CANDLE_SIZE
+        )
+
+        # Ta-lib calculates various technical indicator based on price and
+        # volume arrays.
+
+        # In this example, we are comp
+        rsi = talib.RSI(prices.values, timeperiod=14)
+
+        # We need a variable for the current price of the security to compare
+        #  to the average. Since we are requesting two fields, data.current()
+        # returns a DataFrame with
+        current = data.current(asset, fields=['close', 'volume'])
+        price = current['close']
+
+        # If base_price is not set, we use the current value. This is the
+        # price at the first bar which we reference to calculate price_change.
+        # if asset not in context.base_price:
+        #     context.base_price[asset] = price
+        #
+        # base_price = context.base_price[asset]
+        # price_change = (price - base_price) / base_price
+
+        # Tracking the relevant data
+        volumes[asset] = current['volume']
+        rsis[asset] = rsi[-1]
+        price_values[asset] = price
+        # price_changes[asset] = price_change
+
+        # We are trying to avoid over-trading by limiting our trades to
+        # one per day.
+        if asset in context.traded_today:
+            continue
+
+        # Exit if we cannot trade
+        if not data.can_trade(asset):
+            continue
+
+        # Another powerful built-in feature of the Catalyst backtester is the
+        # portfolio object.  The portfolio object tracks your positions, cash,
+        # cost basis of specific holdings, and more.  In this line, we
+        # calculate how long or short our position is at this minute.
+        pos_amount = context.portfolio.positions[asset].amount
+
+        if rsi[-1] <= context.RSI_OVERSOLD and pos_amount == 0:
+            log.info(
+                '{}: buying - price: {}, rsi: {}'.format(
+                    data.current_dt, price, rsi[-1]
+                )
+            )
+            # Set a style for limit orders,
+            limit_price = price * 1.005
+            target = 1.0 / len(context.assets)
+            order_target_percent(
+                asset, target, limit_price=limit_price
+            )
+            context.traded_today[asset] = True
+
+        elif rsi[-1] >= context.RSI_OVERBOUGHT and pos_amount > 0:
+            log.info(
+                '{}: selling - price: {}, rsi: {}'.format(
+                    data.current_dt, price, rsi[-1]
+                )
+            )
+            limit_price = price * 0.995
+            order_target_percent(
+                asset, 0, limit_price=limit_price
+            )
+            context.traded_today[asset] = True
+
+    # Now that we've collected all current data for this frame, we use
+    # the record() method to save it. This data will be available as
+    # a parameter of the analyze() function for further analysis.
+    record(
+        current_price=price_values,
+        volume=volumes,
+        rsi=rsis,
+        cash=cash,
+    )
+
+
+def analyze(context=None, perf=None):
+    stats = get_pretty_stats(perf)
+    print('the algo stats:\n{}'.format(stats))
+    pass
+
+
+if __name__ == '__main__':
+    # The execution mode: backtest or live
+    live = False
+
+    if live:
+        run_algorithm(
+            capital_base=0.1,
+            initialize=initialize,
+            handle_data=handle_data,
+            analyze=analyze,
+            exchange_name='poloniex',
+            live=True,
+            algo_namespace=NAMESPACE,
+            base_currency='btc',
+            live_graph=False,
+            simulate_orders=False,
+            stats_output=None,
+        )
+
+    else:
+        folder = os.path.join(
+            tempfile.gettempdir(), 'catalyst', NAMESPACE
+        )
+        ensure_directory(folder)
+
+        timestr = time.strftime('%Y%m%d-%H%M%S')
+        out = os.path.join(folder, '{}.p'.format(timestr))
+        # catalyst run -f catalyst/examples/mean_reversion_simple.py \
+        #    -x bitfinex -s 2017-10-1 -e 2017-11-10 -c usdt -n mean-reversion \
+        #   --data-frequency minute --capital-base 10000
+        run_algorithm(
+            capital_base=100,
+            data_frequency='minute',
+            initialize=initialize,
+            handle_data=handle_data,
+            analyze=analyze,
+            exchange_name='poloniex',
+            algo_namespace=NAMESPACE,
+            base_currency='eth',
+            start=pd.to_datetime('2017-10-01', utc=True),
+            end=pd.to_datetime('2017-10-15', utc=True),
+        )
+        log.info('saved perf stats: {}'.format(out))

catalyst/exchange/exchange_algorithm.py

@@ -43,6 +43,7 @@ from catalyst.finance.execution import MarketOrder
 from catalyst.finance.performance import PerformanceTracker
 from catalyst.finance.performance.period import calc_period_stats
 from catalyst.gens.tradesimulation import AlgorithmSimulator
+from catalyst.marketplace.marketplace import Marketplace
 from catalyst.utils.api_support import api_method
 from catalyst.utils.input_validation import error_keywords, ensure_upper_case
 from catalyst.utils.math_utils import round_nearest
@@ -92,6 +93,8 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm):
             attempts=self.attempts,
         )
 
+        self._marketplace = None
+
     @staticmethod
     def __convert_order_params_for_blotter(limit_price, stop_price, style):
         """
@@ -167,6 +170,16 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm):
         """
         return round_nearest(amount, asset.min_trade_size)
 
+    @api_method
+    def get_data_source(self, data_source_name, data_frequency=None,
+                        start=None, end=None):
+        if self._marketplace is None:
+            self._marketplace = Marketplace()
+
+        return self._marketplace.get_data_source(
+            data_source_name, data_frequency, start, end,
+        )
+
     @api_method
     @preprocess(symbol_str=ensure_upper_case)
     def symbol(self, symbol_str, exchange_name=None):

catalyst/marketplace/marketplace.py

@@ -1,19 +1,22 @@
 import json
 import os
-
-import bcolz
-import pandas as pd
 import shutil
 
+import bcolz
+import logbook
+import pandas as pd
+import six
 from web3 import Web3, HTTPProvider
 
+from catalyst.constants import ROOT_DIR, LOG_LEVEL
 from catalyst.exchange.utils.stats_utils import set_print_settings
-from catalyst.constants import ROOT_DIR
 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
+from catalyst.marketplace.utils.path_utils import get_data_source, \
+    get_bundle_folder, get_data_source_folder
 
+# TODO: host our own node on aws?
 REMOTE_NODE = 'http://localhost:7545'
+# TODO: read from GitHub
 CONTRACT_PATH = os.path.join(
     ROOT_DIR, '..', 'marketplace', 'build', 'contracts', 'Marketplace.json'
 )
@@ -21,6 +24,8 @@ CONTRACT_ADDRESS = Web3.toChecksumAddress(
     '0xe2b6cf3863240892d59664d209a28289a73ef644'
 )
 
+log = logbook.Logger('Marketplace', level=LOG_LEVEL)
+
 
 class Marketplace:
     def __init__(self):
@@ -43,21 +48,33 @@ class Marketplace:
                 desc='The marketcap value in USD.',
                 start_date=pd.to_datetime('2017-01-01'),
                 end_date=pd.to_datetime('2018-01-15'),
+                data_frequencies=['daily'],
             ),
             dict(
                 name='GitHub',
                 desc='The rate of development activity on GitHub.',
                 start_date=pd.to_datetime('2017-01-01'),
                 end_date=pd.to_datetime('2018-01-15'),
+                data_frequencies=['daily', 'hour'],
             ),
             dict(
                 name='Influencers',
                 desc='Tweets and related sentiments by selected influencers.',
                 start_date=pd.to_datetime('2017-01-01'),
                 end_date=pd.to_datetime('2018-01-15'),
+                data_frequencies=['daily', 'hour', 'minute'],
             ),
         ]
 
+    def get_data_source_def(self, data_source_name):
+        data_source_name = data_source_name.lower()
+        dsm = self.get_data_sources_map()
+
+        ds = six.next(
+            (d for d in dsm if d['name'].lower() == data_source_name), None
+        )
+        return ds
+
     def list(self):
         subscribers = self.contract.call(
             {'from': self.default_account}
@@ -136,6 +153,35 @@ class Marketplace:
 
         pass
 
+    def get_data_source(self, data_source_name, data_frequency=None,
+                        start=None, end=None):
+        data_source_name = data_source_name.lower()
+
+        if data_frequency is None:
+            ds_def = self.get_data_source_def(data_source_name)
+            freqs = ds_def['data_frequencies']
+            data_frequency = freqs[0]
+
+            if len(freqs) > 1:
+                log.warn(
+                    'no data frequencies specified for data source {}, '
+                    'selected the first one by default: {}'.format(
+                        data_source_name, data_frequency
+                    )
+                )
+
+        # TODO: filter ctable by start and end date
+        bundle_folder = get_bundle_folder(data_source_name, data_frequency)
+        z = bcolz.ctable(rootdir=bundle_folder, mode='r')
+
+        df = z.todataframe()  # type: pd.DataFrame
+        df.set_index(['date', 'symbol'], drop=False, inplace=True)
+
+        if start and end is None:
+            df = df.xs(start, level=0)
+
+        return df
+
     def clean(self, data_source_name, data_frequency=None):
         data_source_name = data_source_name.lower()
 
@@ -143,7 +189,7 @@ class Marketplace:
             folder = get_data_source_folder(data_source_name)
 
         else:
-            forlder = get_bundle_folder(data_source_name, data_frequency)
+            folder = get_bundle_folder(data_source_name, data_frequency)
 
         shutil.rmtree(folder)
         pass

catalyst/marketplace/utils/bundle_utils.py

@@ -18,9 +18,9 @@ def merge_bundles(zsource, ztarget):
     """
     # 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_source.set_index('date', drop=False, inplace=True)
     df_target = ztarget.todataframe()
-    df_target.set_index('last_updated', drop=False, inplace=True)
+    df_target.set_index('date', drop=False, inplace=True)
 
     df = df_target.merge(
         right=df_source,

catalyst/marketplace/utils/path_utils.py

@@ -34,8 +34,7 @@ def get_data_source_folder(data_source_name, environ=None):
 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)
+    bundle_folder = os.path.join(data_source_folder, data_frequency)
 
     ensure_directory(bundle_folder)
 

tests/marketplace/test_marketplace.py

@@ -16,10 +16,17 @@ class TestMarketplace(WithLogger, ZiplineTestCase):
 
     def test_ingest(self):
         marketplace = Marketplace()
+        ds_def = marketplace.get_data_source_def('Marketcap')
+
         marketplace.ingest(
             data_source_name='Marketcap',
-            data_frequency='finest',
-            start=pd.Timestamp.utcnow(),
+            data_frequency=ds_def['data_frequencies'][0],
+            start=pd.to_datetime('2017-10-01'),
             force_download=True,
         )
         pass
+
+    def test_clean(self):
+        marketplace = Marketplace()
+        marketplace.clean('marketcap')
+        pass