diff --git a/catalyst/__main__.py b/catalyst/__main__.py
index 638d096e..c1872d3f 100644
--- a/catalyst/__main__.py
+++ b/catalyst/__main__.py
@@ -9,6 +9,7 @@ from six import text_type
 
 from catalyst.data import bundles as bundles_module
 from catalyst.exchange.exchange_bundle import ExchangeBundle
+from catalyst.exchange.exchange_utils import delete_algo_folder
 from catalyst.exchange.factory import get_exchange
 from catalyst.utils.cli import Date, Timestamp
 from catalyst.utils.run_algo import _run, load_extensions
@@ -495,8 +496,14 @@ def live(ctx,
     default=False,
     help='Show a progress indicator for every currency pair.'
 )
+@click.option(
+    '--validate/--no-validate`',
+    default=False,
+    help='Report potential anomalies found in data bundles.'
+)
 def ingest_exchange(exchange_name, data_frequency, start, end,
-                    include_symbols, exclude_symbols, show_progress, verbose):
+                    include_symbols, exclude_symbols, show_progress, verbose,
+                    validate):
     """
     Ingest data for the given exchange.
     """
@@ -515,10 +522,25 @@ def ingest_exchange(exchange_name, data_frequency, start, end,
         start=start,
         end=end,
         show_progress=show_progress,
-        show_breakdown=verbose
+        show_breakdown=verbose,
+        show_report=validate
     )
 
 
+@main.command(name='clean-algo')
+@click.option(
+    '-n',
+    '--algo-namespace',
+    help='The label of the algorithm to for which to clean the state.'
+)
+@click.pass_context
+def clean_algo(ctx, algo_namespace):
+    click.echo(
+        'Deleting the state folder of algo: {}...'.format(algo_namespace)
+    )
+    delete_algo_folder(algo_namespace)
+
+
 @main.command(name='clean-exchange')
 @click.option(
     '-x',
diff --git a/catalyst/constants.py b/catalyst/constants.py
index c5f8d4ef..da6a957f 100644
--- a/catalyst/constants.py
+++ b/catalyst/constants.py
@@ -2,4 +2,6 @@
 
 import logbook
 
-LOG_LEVEL = logbook.INFO
\ No newline at end of file
+LOG_LEVEL = logbook.INFO
+
+DATE_TIME_FORMAT = '%Y-%m-%d %H:%M'
\ No newline at end of file
diff --git a/catalyst/examples/momemtum.py b/catalyst/examples/momemtum.py
new file mode 100644
index 00000000..dd928fd6
--- /dev/null
+++ b/catalyst/examples/momemtum.py
@@ -0,0 +1,194 @@
+# 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 pandas as pd
+# To run an algorithm in Catalyst, you need two functions: initialize and 
+# handle_data.
+from logbook import Logger
+
+from catalyst import run_algorithm
+from catalyst.api import symbol, record, order_target_percent, \
+    get_open_orders
+from catalyst.exchange import stats_utils
+from catalyst.finance.execution import LimitOrder
+
+# 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.
+algo_namespace = 'momentum'
+log = Logger(algo_namespace)
+
+
+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 Ether in Bitcoin.
+    context.eth_btc = symbol('eth_usdt')
+    context.max_amount = 0.01
+    context.base_price = None
+
+
+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're computing the volume-weighted-average-price of the security 
+    # defined above, in the context.eth_btc variable.  For this example, we're 
+    # using three bars on the daily chart.
+    bars = data.history(
+        context.eth_btc,
+        fields=['close', 'volume'],
+        bar_count=3,
+        frequency='1D'
+    )
+    vwap = stats_utils.vwap(bars)
+
+    # We need a variable for the current price of the security to compare to
+    # the average.
+    current = data.current(context.eth_btc, fields=['close', 'volume'])
+    price = current['close']
+    log.info('{}: price: {}, vwap: {}'.format(data.current_dt, price, vwap))
+
+    # 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 context.base_price is None:
+        context.base_price = price
+    price_change = (price - context.base_price) / context.base_price
+
+    record(
+        price=price,
+        volume=current['volume'],
+        vwap=vwap,
+        price_change=price_change,
+    )
+
+    orders = get_open_orders(context.eth_btc)
+    if len(orders) > 0:
+        log.info('skipping bar until all open orders execute')
+        return
+
+    # 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.   
+    position_amount = context.portfolio.positions[context.eth_btc].amount
+
+    # This is the meat of the algorithm, placed in this if statement.  If the
+    # price of the security is .5% less than the 3-day volume weighted average
+    # price AND we haven't reached our maximum short, then we call the order
+    # command and sell 100 shares.  Similarly, if the stock is .5% higher than
+    # the 3-day average AND we haven't reached our maximum long, then we call
+    # the order command and buy 100 shares.
+    if price > vwap * 1.01 and position_amount < context.max_amount:
+        order_target_percent(
+            context.eth_btc, 1, style=LimitOrder(price * 1.02)
+        )
+
+    elif price < vwap * 0.995 and position_amount > 0:
+        order_target_percent(
+            context.eth_btc, 0, style=LimitOrder(price * 0.98)
+        )
+
+
+def analyze(context=None, results=None):
+    import matplotlib.pyplot as plt
+
+    # The base currency of the algo exchange
+    base_currency = context.exchanges.values()[0].base_currency.upper()
+
+    # Plot the portfolio value over time.
+    ax1 = plt.subplot(611)
+    results.loc[:, 'portfolio_value'].plot(ax=ax1)
+    ax1.set_ylabel('Portfolio Value ({})'.format(base_currency))
+
+    # Plot the price increase or decrease over time.
+    ax2 = plt.subplot(612, sharex=ax1)
+    results.loc[:, 'price'].plot(ax=ax2)
+    ax2.set_ylabel('{asset} ({base})'.format(
+        asset=context.eth_btc.symbol, base=base_currency
+    ))
+
+    # Compute indexes for buy and sell transactions
+    trans_list = results.transactions.values
+    all_trans = [t for sublist in trans_list for t in sublist]
+    all_trans.sort(key=lambda t: t['dt'])
+
+    # Transaction have an exact timestamp while stVats are daily.
+    # We adjust the time to the end of each period to place them on the graph.
+    for t in all_trans:
+        t['dt'] = t['dt'].replace(hour=23, minute=59)
+
+    buys = results.loc[[t['dt'] for t in all_trans if t['amount'] > 0], :]
+    sells = results.loc[[t['dt'] for t in all_trans if t['amount'] < 0], :]
+
+    ax2.plot(
+        buys.index,
+        results.loc[buys.index, 'price'],
+        '^',
+        markersize=10,
+        color='g',
+    )
+    ax2.plot(
+        sells.index,
+        results.loc[sells.index, 'price'],
+        'v',
+        markersize=10,
+        color='r',
+    )
+
+    ax4 = plt.subplot(613, sharex=ax1)
+    results.loc[:, ['starting_cash', 'cash']].plot(ax=ax4)
+    ax4.set_ylabel('Base Currency ({})'.format(base_currency))
+
+    results['algorithm'] = results.loc[:, 'algorithm_period_return']
+
+    ax5 = plt.subplot(614, sharex=ax1)
+    results.loc[:, ['algorithm', 'price_change']].plot(ax=ax5)
+    ax5.set_ylabel('Percent Change')
+
+    ax6 = plt.subplot(615, sharex=ax1)
+    results.loc[:, 'vwap'].plot(ax=ax6)
+    ax6.set_ylabel('VWAP')
+
+    ax6.plot(
+        buys.index,
+        results.loc[buys.index, 'vwap'],
+        '^',
+        markersize=10,
+        color='g',
+    )
+    ax6.plot(
+        sells.index,
+        results.loc[sells.index, 'vwap'],
+        'v',
+        markersize=10,
+        color='r',
+    )
+
+    plt.legend(loc=3)
+
+    # Show the plot.
+    plt.gcf().set_size_inches(18, 8)
+    plt.show()
+    pass
+
+
+# Backtest
+run_algorithm(
+    capital_base=1,
+    data_frequency='minute',
+    initialize=initialize,
+    handle_data=handle_data,
+    analyze=analyze,
+    exchange_name='poloniex',
+    algo_namespace=algo_namespace,
+    base_currency='usdt',
+    start=pd.to_datetime('2017-5-15', utc=True),
+    end=pd.to_datetime('2017-5-20', utc=True),
+)
diff --git a/catalyst/examples/rsi_profit_target.py b/catalyst/examples/rsi_profit_target.py
new file mode 100644
index 00000000..a426c402
--- /dev/null
+++ b/catalyst/examples/rsi_profit_target.py
@@ -0,0 +1,276 @@
+from datetime import timedelta
+
+import pandas as pd
+import numpy as np
+import talib
+from logbook import Logger
+
+from catalyst.api import (
+    order,
+    symbol,
+    record,
+    get_open_orders,
+)
+from catalyst.exchange.stats_utils import crossover, crossunder
+from catalyst.utils.run_algo import run_algorithm
+
+algo_namespace = 'rsi'
+log = Logger(algo_namespace)
+
+
+def initialize(context):
+    log.info('initializing algo')
+    context.asset = symbol('eth_btc')
+    context.base_price = None
+
+    context.MAX_HOLDINGS = 0.2
+    context.RSI_OVERSOLD = 30
+    context.RSI_OVERSOLD_BBANDS = 45
+    context.RSI_OVERBOUGHT_BBANDS = 55
+    context.SLIPPAGE_ALLOWED = 0.03
+
+    context.TARGET = 0.15
+    context.STOP_LOSS = 0.1
+    context.STOP = 0.03
+    context.position = None
+
+    context.last_bar = None
+
+    context.errors = []
+    pass
+
+
+def _handle_buy_sell_decision(context, data, signal, price):
+    orders = get_open_orders(context.asset)
+    if len(orders) > 0:
+        log.info('skipping bar until all open orders execute')
+        return
+
+    positions = context.portfolio.positions
+    if context.position is None and context.asset in positions:
+        position = positions[context.asset]
+        context.position = dict(
+            cost_basis=position['cost_basis'],
+            amount=position['amount'],
+            stop=None
+        )
+
+    action = None
+    if context.position is not None:
+        cost_basis = context.position['cost_basis']
+        amount = context.position['amount']
+        log.info(
+            'found {amount} positions with cost basis {cost_basis}'.format(
+                amount=amount,
+                cost_basis=cost_basis
+            )
+        )
+        stop = context.position['stop']
+
+        target = cost_basis * (1 + context.TARGET)
+        if price >= target:
+            context.position['cost_basis'] = price
+            context.position['stop'] = context.STOP
+
+        stop_target = context.STOP_LOSS if stop is None else context.STOP
+        if price < cost_basis * (1 - stop_target):
+            log.info('executing stop loss')
+            order(
+                asset=context.asset,
+                amount=-amount,
+                limit_price=price * (1 - context.SLIPPAGE_ALLOWED),
+            )
+            action = 0
+            context.position = None
+
+    else:
+        if signal == 'long':
+            log.info('opening position')
+            buy_amount = context.MAX_HOLDINGS / price
+            order(
+                asset=context.asset,
+                amount=buy_amount,
+                limit_price=price * (1 + context.SLIPPAGE_ALLOWED),
+            )
+            context.position = dict(
+                cost_basis=price,
+                amount=buy_amount,
+                stop=None
+            )
+            action = 0
+
+
+def _handle_data_rsi_only(context, data):
+    price = data.current(context.asset, 'close')
+    log.info('got price {price}'.format(price=price))
+
+    if price is np.nan:
+        log.warn('no pricing data')
+        return
+
+    if context.base_price is None:
+        context.base_price = price
+
+    try:
+        prices = data.history(
+            context.asset,
+            fields='price',
+            bar_count=17,
+            frequency='30T'
+        )
+    except Exception as e:
+        log.warn('historical data not available: '.format(e))
+        return
+
+    rsi = talib.RSI(prices.values, timeperiod=16)[-1]
+    log.info('got rsi {}'.format(rsi))
+
+    signal = None
+    if rsi < context.RSI_OVERSOLD:
+        signal = 'long'
+
+    # Making sure that the price is still current
+    price = data.current(context.asset, 'close')
+    cash = context.portfolio.cash
+    log.info(
+        'base currency available: {cash}, cap: {cap}'.format(
+            cash=cash,
+            cap=context.MAX_HOLDINGS
+        )
+    )
+    volume = data.current(context.asset, 'volume')
+    price_change = (price - context.base_price) / context.base_price
+    record(
+        price=price,
+        price_change=price_change,
+        rsi=rsi,
+        volume=volume,
+        cash=cash,
+        starting_cash=context.portfolio.starting_cash,
+        leverage=context.account.leverage,
+    )
+
+    _handle_buy_sell_decision(context, data, signal, price)
+
+
+def handle_data(context, data):
+    dt = data.current_dt
+
+    if context.last_bar is None or (
+                context.last_bar + timedelta(minutes=15)) <= dt:
+        context.last_bar = dt
+    else:
+        return
+
+    log.info('BAR {}'.format(dt))
+    try:
+        _handle_data_rsi_only(context, data)
+    except Exception as e:
+        log.warn('aborting the bar on error {}'.format(e))
+        context.errors.append(e)
+
+    if len(context.errors) > 0:
+        log.info('the errors:\n{}'.format(context.errors))
+
+
+def analyze(context=None, results=None):
+    import matplotlib.pyplot as plt
+
+    base_currency = context.exchanges.values()[0].base_currency.upper()
+    # Plot the portfolio and asset data.
+    ax1 = plt.subplot(611)
+    results.loc[:, 'portfolio_value'].plot(ax=ax1)
+    ax1.set_ylabel('Portfolio Value ({})'.format(base_currency))
+
+    ax2 = plt.subplot(612, sharex=ax1)
+    results.loc[:, 'price'].plot(ax=ax2)
+    ax2.set_ylabel('{asset} ({base})'.format(
+        asset=context.asset.symbol, base=base_currency
+    ))
+
+    trans = results.loc[[t != [] for t in results.transactions], :]
+    buys = trans.loc[[t[0]['amount'] > 0 for t in trans.transactions], :]
+    sells = trans.loc[[t[0]['amount'] < 0 for t in trans.transactions], :]
+    # buys = results.loc[results['action'] == 1, :]
+    # sells = results.loc[results['action'] == 0, :]
+
+    ax2.plot(
+        buys.index,
+        results.loc[buys.index, 'price'],
+        '^',
+        markersize=10,
+        color='g',
+    )
+    ax2.plot(
+        sells.index,
+        results.loc[sells.index, 'price'],
+        'v',
+        markersize=10,
+        color='r',
+    )
+
+    ax3 = plt.subplot(613, sharex=ax1)
+    results.loc[:, ['alpha', 'beta']].plot(ax=ax3)
+    ax3.set_ylabel('Alpha / Beta ')
+
+    ax4 = plt.subplot(614, sharex=ax1)
+    results.loc[:, ['starting_cash', 'cash']].plot(ax=ax4)
+    ax4.set_ylabel('Base Currency ({})'.format(base_currency))
+
+    results['algorithm'] = results.loc[:, 'algorithm_period_return']
+
+    ax5 = plt.subplot(615, sharex=ax1)
+    results.loc[:, ['algorithm', 'price_change']].plot(ax=ax5)
+    ax5.set_ylabel('Percent Change')
+
+    ax6 = plt.subplot(616, sharex=ax1)
+    results.loc[:, 'rsi'].plot(ax=ax6)
+    ax6.set_ylabel('RSI')
+
+    ax6.plot(
+        buys.index,
+        results.loc[buys.index, 'rsi'],
+        '^',
+        markersize=10,
+        color='g',
+    )
+    ax6.plot(
+        sells.index,
+        results.loc[sells.index, 'rsi'],
+        'v',
+        markersize=10,
+        color='r',
+    )
+
+    plt.legend(loc=3)
+
+    # Show the plot.
+    plt.gcf().set_size_inches(18, 8)
+    plt.show()
+    pass
+
+
+run_algorithm(
+    initialize=initialize,
+    handle_data=handle_data,
+    analyze=analyze,
+    exchange_name='bittrex',
+    live=True,
+    algo_namespace=algo_namespace,
+    base_currency='btc',
+    live_graph=False
+)
+
+# Backtest
+# run_algorithm(
+#     capital_base=0.5,
+#     data_frequency='minute',
+#     initialize=initialize,
+#     handle_data=handle_data,
+#     analyze=analyze,
+#     exchange_name='poloniex',
+#     algo_namespace=algo_namespace,
+#     base_currency='btc',
+#     start=pd.to_datetime('2017-9-1', utc=True),
+#     end=pd.to_datetime('2017-10-1', utc=True),
+# )
diff --git a/catalyst/examples/simple_loop.py b/catalyst/examples/simple_loop.py
index 3927c4d2..1e6c1bdc 100644
--- a/catalyst/examples/simple_loop.py
+++ b/catalyst/examples/simple_loop.py
@@ -7,7 +7,7 @@ from catalyst.api import symbol
 
 def initialize(context):
     print('initializing')
-    context.asset = symbol('eth_btc')
+    context.asset = symbol('swift_btc')
 
 
 def handle_data(context, data):
@@ -20,8 +20,8 @@ def handle_data(context, data):
         prices = data.history(
             context.asset,
             fields='price',
-            bar_count=16,
-            frequency='60T'
+            bar_count=15,
+            frequency='1D'
         )
         rsi = talib.RSI(prices.values, timeperiod=14)[-1]
         print('got rsi: {}'.format(rsi))
@@ -31,13 +31,13 @@ def handle_data(context, data):
 
 run_algorithm(
     capital_base=250,
-    start=pd.to_datetime('2016-6-1', utc=True),
+    start=pd.to_datetime('2015-4-1', utc=True),
     end=pd.to_datetime('2017-11-1', utc=True),
     data_frequency='daily',
     initialize=initialize,
     handle_data=handle_data,
     analyze=None,
-    exchange_name='bitfinex',
+    exchange_name='bittrex',
     algo_namespace='simple_loop',
     base_currency='btc'
 )
diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py
index 2210aa7b..9b0c8c49 100644
--- a/catalyst/exchange/bundle_utils.py
+++ b/catalyst/exchange/bundle_utils.py
@@ -149,7 +149,7 @@ def get_periods(start_dt, end_dt, freq):
     return len(get_periods_range(start_dt, end_dt, freq))
 
 
-def get_start_dt(end_dt, bar_count, data_frequency):
+def get_start_dt(end_dt, bar_count, data_frequency, include_first=True):
     """
     The start date based on specified end date and data frequency.
 
@@ -168,6 +168,9 @@ def get_start_dt(end_dt, bar_count, data_frequency):
     if periods > 1:
         delta = get_delta(periods, data_frequency)
         start_dt = end_dt - delta
+
+        if not include_first:
+            start_dt += get_delta(1, data_frequency)
     else:
         start_dt = end_dt
 
diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py
index c317812a..2a92799f 100644
--- a/catalyst/exchange/exchange_bundle.py
+++ b/catalyst/exchange/exchange_bundle.py
@@ -1,47 +1,24 @@
 import os
 import shutil
+from functools import partial
 from itertools import chain
+from operator import is_not
 
+import numpy as np
 import pandas as pd
 from catalyst.assets._assets import TradingPair
+from datetime import datetime, timedelta
 from logbook import Logger
-from pandas.tslib import Timestamp
 from pytz import UTC
 from six import itervalues
 
 from catalyst import get_calendar
+from catalyst.constants import DATE_TIME_FORMAT
 from catalyst.constants import LOG_LEVEL
 from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
     BcolzMinuteBarMetadata
 from catalyst.exchange.bundle_utils import range_in_bundle, \
-    get_bcolz_chunk, get_delta, get_month_start_end, \
-    get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label
-from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
-    BcolzExchangeBarWriter
-from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
-    TempBundleNotFoundError, \
-    NoDataAvailableOnExchange, \
-    PricingDataNotLoadedError
-from catalyst.exchange.exchange_utils import get_exchange_folder
-from catalyst.utils.cli import maybe_show_progress
-from catalyst.utils.paths import ensure_directory
-import os
-import shutil
-from itertools import chain
-
-import pandas as pd
-from catalyst.assets._assets import TradingPair
-from logbook import Logger
-from pandas.tslib import Timestamp
-from pytz import UTC
-from six import itervalues
-
-from catalyst import get_calendar
-from catalyst.constants import LOG_LEVEL
-from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
-    BcolzMinuteBarMetadata
-from catalyst.exchange.bundle_utils import range_in_bundle, \
-    get_bcolz_chunk, get_delta, get_month_start_end, \
+    get_bcolz_chunk, get_month_start_end, \
     get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label
 from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
     BcolzExchangeBarWriter
@@ -244,8 +221,91 @@ class ExchangeBundle:
             if data_frequency == 'minute' \
             else self.calendar.sessions_in_range(start_dt, end_dt)
 
+    def _spot_empty_periods(self, ohlcv_df, asset, data_frequency,
+                            empty_rows_behavior):
+        problems = []
+
+        nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index
+        if len(nan_rows) > 0:
+            dates = []
+            for row_date in nan_rows.values:
+                row_date = pd.to_datetime(row_date, utc=True)
+                if row_date > asset.start_date:
+                    dates.append(row_date)
+
+            if len(dates) > 0:
+                end_dt = asset.end_minute if data_frequency == 'minute' \
+                    else asset.end_daily
+
+                problem = '{name} ({start_dt} to {end_dt}) has empty ' \
+                          'periods: {dates}'.format(
+                    name=asset.symbol,
+                    start_dt=asset.start_date.strftime(DATE_TIME_FORMAT),
+                    end_dt=end_dt.strftime(DATE_TIME_FORMAT),
+                    dates=[date.strftime(DATE_TIME_FORMAT) for date in dates]
+                )
+                if empty_rows_behavior == 'warn':
+                    log.warn(problem)
+
+                elif empty_rows_behavior == 'raise':
+                    raise EmptyValuesInBundleError(
+                        name=asset.symbol,
+                        end_minute=end_dt,
+                        dates=dates
+                    )
+
+                else:
+                    ohlcv_df.dropna(inplace=True)
+
+            else:
+                problem = None
+
+            problems.append(problem)
+
+        return problems
+
+    def _spot_duplicates(self, ohlcv_df, asset, data_frequency, threshold):
+        # TODO: work in progress
+        series = ohlcv_df.reset_index().groupby('close')['index'].apply(
+            np.array
+        )
+
+        ref_delta = timedelta(minutes=1) if data_frequency == 'minute' \
+            else timedelta(days=1)
+
+        dups = series.loc[lambda values: [len(x) > 10 for x in values]]
+
+        for index, dates in dups.iteritems():
+            prev_date = None
+            for date in dates:
+                if prev_date is not None:
+                    delta = (date - prev_date) / 1e9
+                    if delta == ref_delta.seconds:
+                        log.info('pex')
+
+                prev_date = date
+
+        problems = []
+        for index, dates in dups.iteritems():
+            end_dt = asset.end_minute if data_frequency == 'minute' \
+                else asset.end_daily
+
+            problem = '{name} ({start_dt} to {end_dt}) has {threshold} ' \
+                      'identical close values on: {dates}'.format(
+                name=asset.symbol,
+                start_dt=asset.start_date.strftime(DATE_TIME_FORMAT),
+                end_dt=end_dt.strftime(DATE_TIME_FORMAT),
+                threshold=threshold,
+                dates=[pd.to_datetime(date).strftime(DATE_TIME_FORMAT)
+                       for date in dates]
+            )
+
+            problems.append(problem)
+
+        return problems
+
     def ingest_df(self, ohlcv_df, data_frequency, asset, writer,
-                  empty_rows_behavior='strip'):
+                  empty_rows_behavior='warn', duplicates_threshold=None):
         """
         Ingest a DataFrame of OHLCV data for a given market.
 
@@ -258,50 +318,16 @@ class ExchangeBundle:
         empty_rows_behavior: str
 
         """
+        problems = []
         if empty_rows_behavior is not 'ignore':
-            nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index
+            problems += self._spot_empty_periods(
+                ohlcv_df, asset, data_frequency, empty_rows_behavior
+            )
 
-            if len(nan_rows) > 0:
-                dates = []
-                previous_date = None
-                for row_date in nan_rows.values:
-                    row_date = pd.to_datetime(row_date)
-
-                    if previous_date is None:
-                        dates.append(row_date)
-
-                    else:
-                        seq_date = previous_date + get_delta(1, data_frequency)
-
-                        if row_date > seq_date:
-                            dates.append(previous_date)
-                            dates.append(row_date)
-
-                    previous_date = row_date
-
-                dates.append(pd.to_datetime(nan_rows.values[-1]))
-
-                name = '{} from {} to {}'.format(
-                    asset.symbol, ohlcv_df.index[0], ohlcv_df.index[-1]
-                )
-                if empty_rows_behavior == 'warn':
-                    log.warn(
-                        '\n{name} with end minute {end_minute} has empty rows '
-                        'in ranges: {dates}'.format(
-                            name=name,
-                            end_minute=asset.end_minute,
-                            dates=dates
-                        )
-                    )
-
-                elif empty_rows_behavior == 'raise':
-                    raise EmptyValuesInBundleError(
-                        name=name,
-                        end_minute=asset.end_minute,
-                        dates=dates
-                    )
-                else:
-                    ohlcv_df.dropna(inplace=True)
+        # if duplicates_threshold is not None:
+        #     problems += self._spot_duplicates(
+        #         ohlcv_df, asset, data_frequency, duplicates_threshold
+        #     )
 
         data = []
         if not ohlcv_df.empty:
@@ -310,8 +336,11 @@ class ExchangeBundle:
 
         self._write(data, writer, data_frequency)
 
+        return problems
+
     def ingest_ctable(self, asset, data_frequency, period,
-                      writer, empty_rows_behavior='strip', cleanup=False):
+                      writer, empty_rows_behavior='strip',
+                      duplicates_threshold=100, cleanup=False):
         """
         Merge a ctable bundle chunk into the main bundle for the exchange.
 
@@ -327,8 +356,14 @@ class ExchangeBundle:
         cleanup: bool
             Remove the temp bundle directory after ingestion.
 
-        :return:
+        Returns
+        -------
+        list[str]
+            A list of problems which occurred during ingestion.
+
         """
+        problems = []
+
         # Download and extract the bundle
         path = get_bcolz_chunk(
             exchange_name=self.exchange.name,
@@ -375,12 +410,13 @@ class ExchangeBundle:
             start_dt, end_dt, data_frequency
         )
         df = get_df_from_arrays(arrays, periods)
-        self.ingest_df(
+        problems += self.ingest_df(
             ohlcv_df=df,
             data_frequency=data_frequency,
             asset=asset,
             writer=writer,
-            empty_rows_behavior=empty_rows_behavior
+            empty_rows_behavior=empty_rows_behavior,
+            duplicates_threshold=duplicates_threshold
         )
 
         if cleanup:
@@ -390,7 +426,7 @@ class ExchangeBundle:
             )
             shutil.rmtree(reader._rootdir)
 
-        return reader._rootdir
+        return filter(partial(is_not, None), problems)
 
     def get_adj_dates(self, start, end, assets, data_frequency):
         """
@@ -528,7 +564,8 @@ class ExchangeBundle:
         return chunks
 
     def ingest_assets(self, assets, data_frequency, start_dt=None, end_dt=None,
-                      show_progress=False, show_breakdown=False):
+                      show_progress=False, show_breakdown=False,
+                      show_report=False):
         """
         Determine if data is missing from the bundle and attempt to ingest it.
 
@@ -562,6 +599,7 @@ class ExchangeBundle:
             end_dt=end_dt
         )
 
+        problems = []
         # This is the common writer for the entire exchange bundle
         # we want to give an end_date far in time
         writer = self.get_writer(start_dt, end_dt, data_frequency)
@@ -577,7 +615,7 @@ class ExchangeBundle:
                             symbol=asset.symbol
                         )) as it:
                     for chunk in it:
-                        self.ingest_ctable(
+                        problems += self.ingest_ctable(
                             asset=chunk['asset'],
                             data_frequency=data_frequency,
                             period=chunk['period'],
@@ -601,7 +639,7 @@ class ExchangeBundle:
                         frequency=data_frequency,
                     )) as it:
                 for chunk in it:
-                    self.ingest_ctable(
+                    problems += self.ingest_ctable(
                         asset=chunk['asset'],
                         data_frequency=data_frequency,
                         period=chunk['period'],
@@ -610,9 +648,14 @@ class ExchangeBundle:
                         cleanup=True
                     )
 
+        if show_report and len(problems) > 0:
+            log.info('problems during ingestion:{}\n'.format(
+                '\n'.join(problems)
+            ))
+
     def ingest(self, data_frequency, include_symbols=None,
                exclude_symbols=None, start=None, end=None,
-               show_progress=True, show_breakdown=True, environ=os.environ):
+               show_progress=True, show_breakdown=True, show_report=True):
         """
         Inject data based on specified parameters.
 
@@ -631,7 +674,7 @@ class ExchangeBundle:
 
         for frequency in data_frequency.split(','):
             self.ingest_assets(assets, frequency, start, end,
-                               show_progress, show_breakdown)
+                               show_progress, show_breakdown, show_report)
 
     def get_history_window_series_and_load(self,
                                            assets,
@@ -749,7 +792,7 @@ class ExchangeBundle:
                                   field,
                                   data_frequency,
                                   reset_reader=False):
-        start_dt = get_start_dt(end_dt, bar_count, data_frequency)
+        start_dt = get_start_dt(end_dt, bar_count, data_frequency, False)
         start_dt, end_dt = self.get_adj_dates(
             start_dt, end_dt, assets, data_frequency
         )
diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py
index b4f828fe..554825b4 100644
--- a/catalyst/exchange/exchange_utils.py
+++ b/catalyst/exchange/exchange_utils.py
@@ -2,6 +2,7 @@ import json
 import os
 import pickle
 import re
+import shutil
 from datetime import date, datetime
 
 import pandas as pd
@@ -158,6 +159,24 @@ def get_exchange_auth(exchange_name, environ=None):
             return data
 
 
+def delete_algo_folder(algo_name, environ=None):
+    """
+    Delete the folder containing the algo state.
+
+    Parameters
+    ----------
+    algo_name: str
+    environ:
+
+    Returns
+    -------
+    str
+
+    """
+    folder = get_algo_folder(algo_name, environ)
+    shutil.rmtree(folder)
+
+
 def get_algo_folder(algo_name, environ=None):
     """
     The algorithm root folder of the algorithm.
diff --git a/catalyst/exchange/stats_utils.py b/catalyst/exchange/stats_utils.py
index b7bfda98..cf5afc00 100644
--- a/catalyst/exchange/stats_utils.py
+++ b/catalyst/exchange/stats_utils.py
@@ -54,6 +54,38 @@ def crossunder(source, target):
         return False
 
 
+def vwap(df):
+    """
+    Volume-weighted average price (VWAP) is a ratio generally used by
+    institutional investors and mutual funds to make buys and sells so as not
+    to disturb the market prices with large orders. It is the average share
+    price of a stock weighted against its trading volume within a particular
+    time frame, generally one day.
+
+    Read more: Volume Weighted Average Price - VWAP
+    https://www.investopedia.com/terms/v/vwap.asp#ixzz4xt922daE
+
+    Parameters
+    ----------
+    df: pd.DataFrame
+
+    Returns
+    -------
+
+    """
+    if 'close' not in df.columns or 'volume' not in df.columns:
+        raise ValueError('price data must include `volume` and `close`')
+
+    vol_sum = np.nansum(df['volume'].values)
+
+    try:
+        ret = np.nansum(df['close'].values * df['volume'].values) / vol_sum
+    except ZeroDivisionError:
+        ret = np.nan
+
+    return ret
+
+
 def get_pretty_stats(stats_df, recorded_cols=None, num_rows=10):
     """
     Format and print the last few rows of a statistics DataFrame.
diff --git a/catalyst/finance/execution.py b/catalyst/finance/execution.py
index fe7c1c89..8fec0af6 100644
--- a/catalyst/finance/execution.py
+++ b/catalyst/finance/execution.py
@@ -77,6 +77,7 @@ class LimitOrder(ExecutionStyle):
     Execution style representing an order to be executed at a price equal to or
     better than a specified limit price.
     """
+
     def __init__(self, limit_price, exchange=None):
         """
         Store the given price.
@@ -99,6 +100,7 @@ class StopOrder(ExecutionStyle):
     Execution style representing an order to be placed once the market price
     reaches a specified stop price.
     """
+
     def __init__(self, stop_price, exchange=None):
         """
         Store the given price.
@@ -121,6 +123,7 @@ class StopLimitOrder(ExecutionStyle):
     Execution style representing a limit order to be placed with a specified
     limit price once the market reaches a specified stop price.
     """
+
     def __init__(self, limit_price, stop_price, exchange=None):
         """
         Store the given prices
@@ -144,31 +147,20 @@ class StopLimitOrder(ExecutionStyle):
 def asymmetric_round_price_to_penny(price, prefer_round_down,
                                     diff=(0.0095 - .005)):
     """
-    Asymmetric rounding function for adjusting prices to two places in a way
-    that "improves" the price.  For limit prices, this means preferring to
-    round down on buys and preferring to round up on sells.  For stop prices,
-    it means the reverse.
+    Modified the original function because we do not want to round
+    prices on crypto exchange.
 
-    If prefer_round_down == True:
-        When .05 below to .95 above a penny, use that penny.
-    If prefer_round_down == False:
-        When .95 below to .05 above a penny, use that penny.
+    Parameters
+    ----------
+    price: float
+
+    Returns
+    -------
+    float
 
-    In math-speak:
-    If prefer_round_down: [<X-1>.0095, X.0195) -> round to X.01.
-    If not prefer_round_down: (<X-1>.0005, X.0105] -> round to X.01.
     """
-    # Subtracting an epsilon from diff to enforce the open-ness of the upper
-    # bound on buys and the lower bound on sells.  Using the actual system
-    # epsilon doesn't quite get there, so use a slightly less epsilon-ey value.
-    epsilon = float_info.epsilon * 10
-    diff = diff - epsilon
-
-    # relies on rounding half away from zero, unlike numpy's bankers' rounding
-    rounded = round(price - (diff if prefer_round_down else -diff), 2)
-    if zp_math.tolerant_equals(rounded, 0.0):
-        return 0.0
-    return rounded
+    # TODO: consider overriding outside of the original function
+    return price
 
 
 def check_stoplimit_prices(price, label):
diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py
index 2cf98b08..9492438e 100644
--- a/tests/exchange/test_bundle.py
+++ b/tests/exchange/test_bundle.py
@@ -42,17 +42,16 @@ class TestExchangeBundle:
 
     def test_ingest_minute(self):
         data_frequency = 'minute'
-        exchange_name = 'bitfinex'
+        exchange_name = 'poloniex'
 
         exchange = get_exchange(exchange_name)
         exchange_bundle = ExchangeBundle(exchange)
         assets = [
-            exchange.get_asset('xmr_btc')
+            exchange.get_asset('eth_btc')
         ]
 
-        # start = pd.to_datetime('2017-09-01', utc=True)
-        start = pd.to_datetime('2016-01-01', utc=True)
-        end = pd.to_datetime('2017-9-30', utc=True)
+        start = pd.to_datetime('2016-03-01', utc=True)
+        end = pd.to_datetime('2017-11-1', utc=True)
 
         log.info('ingesting exchange bundle {}'.format(exchange_name))
         exchange_bundle.ingest(
@@ -122,8 +121,8 @@ class TestExchangeBundle:
 
     def test_ingest_daily(self):
         exchange_name = 'bitfinex'
-        data_frequency = 'daily'
-        include_symbols = 'btc_usd'
+        data_frequency = 'minute'
+        include_symbols = 'neo_btc'
 
         # exchange_name = 'poloniex'
         # data_frequency = 'daily'
@@ -422,7 +421,8 @@ class TestExchangeBundle:
                 data_frequency=data_frequency,
                 asset=asset,
                 writer=writer,
-                empty_rows_behavior='raise'
+                empty_rows_behavior='raise',
+                duplicates_behavior='raise'
             )
 
         bundle_series = bundle.get_history_window_series(
@@ -458,10 +458,10 @@ class TestExchangeBundle:
         )
 
     def bundle_to_csv(self):
-        exchange_name = 'bitfinex'
+        exchange_name = 'poloniex'
         data_frequency = 'minute'
-        period = '2017-10'
-        symbol = 'neo_btc'
+        period = '2017-09'
+        symbol = 'eth_btc'
 
         exchange = get_exchange(exchange_name)
         asset = exchange.get_asset(symbol)