diff --git a/catalyst/__main__.py b/catalyst/__main__.py index 1b08f69d..03669486 100644 --- a/catalyst/__main__.py +++ b/catalyst/__main__.py @@ -767,12 +767,18 @@ def bundles(): @main.group() @click.pass_context def marketplace(ctx): + """Access the Enigma Data Marketplace to:\n + - Register and Publish new datasets (seller-side)\n + - Subscribe and Ingest premium datasets (buyer-side)\n + """ pass @marketplace.command() @click.pass_context def ls(ctx): + """List all available datasets. + """ click.echo('Listing of available data sources on the marketplace:', sys.stdout) marketplace = Marketplace() @@ -787,6 +793,8 @@ def ls(ctx): ) @click.pass_context def subscribe(ctx, dataset): + """Subscribe to an exisiting dataset. + """ if dataset is None: ctx.fail("must specify a dataset to subscribe to with '--dataset'\n" "List available dataset on the marketplace with " @@ -825,6 +833,8 @@ def subscribe(ctx, dataset): ) @click.pass_context def ingest(ctx, dataset, data_frequency, start, end): + """Ingest a dataset (requires subscription). + """ if dataset is None: ctx.fail("must specify a dataset to clean with '--dataset'\n" "List available dataset on the marketplace with " @@ -842,8 +852,10 @@ def ingest(ctx, dataset, data_frequency, start, end): ) @click.pass_context def clean(ctx, dataset): + """Clean/Remove local data for a given dataset. + """ if dataset is None: - ctx.fail("must specify a dataset to ingest with '--dataset'\n" + ctx.fail("must specify a dataset to clean up with '--dataset'\n" "List available dataset on the marketplace with " "'catalyst marketplace ls'") click.echo('Cleaning data source: {}'.format(dataset), sys.stdout) @@ -855,6 +867,8 @@ def clean(ctx, dataset): @marketplace.command() @click.pass_context def register(ctx): + """Register a new dataset. + """ marketplace = Marketplace() marketplace.register() @@ -878,6 +892,8 @@ def register(ctx): ) @click.pass_context def publish(ctx, dataset, datadir, watch): + """Publish data for a registered dataset. + """ marketplace = Marketplace() if dataset is None: ctx.fail("must specify a dataset to publish data for " diff --git a/catalyst/constants.py b/catalyst/constants.py index 27b22971..3da77a44 100644 --- a/catalyst/constants.py +++ b/catalyst/constants.py @@ -30,20 +30,20 @@ AUTH_SERVER = 'https://data.enigma.co' # TODO: switch to mainnet ETH_REMOTE_NODE = 'https://ropsten.infura.io/' -# TODO: move to MASTER branch on github + MARKETPLACE_CONTRACT = 'https://raw.githubusercontent.com/enigmampc/' \ - 'catalyst/develop/catalyst/marketplace/' \ + 'catalyst/master/catalyst/marketplace/' \ 'contract_marketplace_address.txt' MARKETPLACE_CONTRACT_ABI = 'https://raw.githubusercontent.com/enigmampc/' \ - 'catalyst/develop/catalyst/marketplace/' \ + 'catalyst/master/catalyst/marketplace/' \ 'contract_marketplace_abi.json' # TODO: switch to mainnet ENIGMA_CONTRACT = 'https://raw.githubusercontent.com/enigmampc/catalyst/' \ - 'develop/catalyst/marketplace/' \ + 'master/catalyst/marketplace/' \ 'contract_enigma_address.txt' ENIGMA_CONTRACT_ABI = 'https://raw.githubusercontent.com/enigmampc/' \ - 'catalyst/develop/catalyst/marketplace/' \ + 'catalyst/master/catalyst/marketplace/' \ 'contract_enigma_abi.json' diff --git a/catalyst/examples/portfolio_optimization.py b/catalyst/examples/portfolio_optimization.py index 26c957b6..c6da1f1f 100644 --- a/catalyst/examples/portfolio_optimization.py +++ b/catalyst/examples/portfolio_optimization.py @@ -66,7 +66,7 @@ def handle_data(context, data): # Define portfolio optimization parameters n_portfolios = 50000 results_array = np.zeros((3 + context.nassets, n_portfolios)) - for p in xrange(n_portfolios): + for p in range(n_portfolios): weights = np.random.random(context.nassets) weights /= np.sum(weights) w = np.asmatrix(weights) diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 003a2a8a..bbe462b3 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -1,5 +1,4 @@ import abc -import pytz from abc import ABCMeta, abstractmethod, abstractproperty from datetime import timedelta from time import sleep @@ -14,7 +13,8 @@ from catalyst.exchange.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \ SymbolNotFoundOnExchange, \ PricingDataNotLoadedError, \ - NoDataAvailableOnExchange, NoValueForField, LastCandleTooEarlyError, \ + NoDataAvailableOnExchange, NoValueForField, \ + NoCandlesReceivedFromExchange, \ TickerNotFoundError, NotEnoughCashError from catalyst.exchange.utils.datetime_utils import get_delta, \ get_periods_range, \ @@ -257,7 +257,8 @@ class Exchange: elif data_frequency is not None: applies = ( ( - data_frequency == 'minute' and a.end_minute is not None) + data_frequency == 'minute' and + a.end_minute is not None) or ( data_frequency == 'daily' and a.end_daily is not None) ) @@ -485,49 +486,52 @@ class Exchange: freq, candle_size, unit, data_frequency = get_frequency( frequency, data_frequency, supported_freqs=['T', 'D', 'H'] ) + + # we want to avoid receiving empty candles + # so we request more than needed + # TODO: consider defining a const per asset + # and/or some retry mechanism (in each iteration request more data) + requested_bar_count = bar_count + 30 # The get_history method supports multiple asset candles = self.get_candles( freq=freq, assets=assets, - bar_count=bar_count, + bar_count=requested_bar_count, end_dt=end_dt if not is_current else None, ) - series = dict() + # candles sanity check - verify no empty candles were received: for asset in candles: - if candles[asset]: - first_candle = candles[asset][0] - asset_series = self.get_series_from_candles( - candles=candles[asset], - start_dt=first_candle['last_traded'], + if not candles[asset]: + raise NoCandlesReceivedFromExchange( + bar_count=requested_bar_count, end_dt=end_dt, - data_frequency=frequency, - field=field, - ) + asset=asset, + exchange=self.name) - delta_candle_size = candle_size * 60 if unit == 'H' else candle_size - # Checking to make sure that the dates match - delta = get_delta(delta_candle_size, data_frequency) - adj_end_dt = end_dt - delta - last_traded = asset_series.index[-1] + series = get_candles_df(candles=candles, + field=field, + freq=frequency, + bar_count=requested_bar_count, + end_dt=end_dt) - if last_traded < adj_end_dt: - raise LastCandleTooEarlyError( - last_traded=last_traded, - end_dt=adj_end_dt, - exchange=self.name, - ) - else: # empty candle received - # because other assets are tz-aware, we need its tz to be set as well - asset_series = pd.Series([], index=pd.DatetimeIndex([], tz=pytz.utc)) - - - series[asset] = asset_series + # TODO: consider how to approach this edge case + # delta_candle_size = candle_size * 60 if unit == 'H' else candle_size + # Checking to make sure that the dates match + # delta = get_delta(delta_candle_size, data_frequency) + # adj_end_dt = end_dt - delta + # last_traded = asset_series.index[-1] + # if last_traded < adj_end_dt: + # raise LastCandleTooEarlyError( + # last_traded=last_traded, + # end_dt=adj_end_dt, + # exchange=self.name, + # ) df = pd.DataFrame(series) - #df.dropna(inplace=True) # commented out due to issue 236 + df.dropna(inplace=True) - return df + return df.tail(bar_count) def get_history_window_with_bundle(self, assets, @@ -575,7 +579,8 @@ class Exchange: A dataframe containing the requested data. """ - # TODO: this function needs some work, we're currently using it just for benchmark data + # TODO: this function needs some work, + # we're currently using it just for benchmark data freq, candle_size, unit, data_frequency = get_frequency( frequency, data_frequency ) @@ -601,7 +606,7 @@ class Exchange: start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency) trailing_dt = \ series[asset].index[-1] + get_delta(1, data_frequency) \ - if asset in series else start_dt + if asset in series else start_dt # The get_history method supports multiple asset # Use the original frequency to let each api optimize diff --git a/catalyst/exchange/exchange_algorithm.py b/catalyst/exchange/exchange_algorithm.py index 641916f9..04e2352c 100644 --- a/catalyst/exchange/exchange_algorithm.py +++ b/catalyst/exchange/exchange_algorithm.py @@ -164,6 +164,25 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): style) return amount, style + def _calculate_order_target_amount(self, asset, target): + """ + removes order amounts so we won't run into issues + when two orders are placed one after the other. + it then proceeds to removing positions amount at TradingAlgorithm + :param asset: + :param target: + :return: target + """ + if asset in self.blotter.open_orders: + for open_order in self.blotter.open_orders[asset]: + current_amount = open_order.amount + target -= current_amount + + target = super(ExchangeTradingAlgorithmBase, self). \ + _calculate_order_target_amount(asset, target) + + return target + def round_order(self, amount, asset): """ We need fractions with cryptocurrencies diff --git a/catalyst/exchange/exchange_blotter.py b/catalyst/exchange/exchange_blotter.py index c82b2f00..3d85149e 100644 --- a/catalyst/exchange/exchange_blotter.py +++ b/catalyst/exchange/exchange_blotter.py @@ -68,7 +68,7 @@ class TradingPairFeeSchedule(CommissionModel): multiplier = maker \ if ((order.amount > 0 and order.limit < transaction.price) or (order.amount < 0 and order.limit > transaction.price)) \ - and order.limit_reached else taker + and order.limit_reached else taker fee = cost * multiplier return fee diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py index 91fba0da..46f74f9c 100644 --- a/catalyst/exchange/exchange_errors.py +++ b/catalyst/exchange/exchange_errors.py @@ -324,6 +324,13 @@ class BalanceTooLowError(ZiplineError): ).strip() +class NoCandlesReceivedFromExchange(ZiplineError): + msg = ( + 'Although requesting {bar_count} candles until {end_dt} of asset {asset}, ' + 'an empty list of candles was received for {exchange}.' + ).strip() + + class MarketsNotFoundError(ZiplineError): msg = ( 'Exchange {exchange} contains no valid market so it is unusable in ' diff --git a/catalyst/exchange/utils/exchange_utils.py b/catalyst/exchange/utils/exchange_utils.py index 420e1278..45d1664a 100644 --- a/catalyst/exchange/utils/exchange_utils.py +++ b/catalyst/exchange/utils/exchange_utils.py @@ -17,7 +17,6 @@ from catalyst.exchange.utils.serialization_utils import ExchangeJSONEncoder, \ ExchangeJSONDecoder, ConfigJSONEncoder from catalyst.utils.paths import data_root, ensure_directory, \ last_modified_time -from catalyst.exchange.utils.datetime_utils import get_periods_range def get_sid(symbol): @@ -673,24 +672,15 @@ def transform_candles_to_df(candles): return pd.DataFrame(candles).set_index('last_traded') -def get_candles_df(candles, field, freq, bar_count, end_dt=None): +def get_candles_df(candles, field, freq, bar_count, end_dt): all_series = dict() for asset in candles: asset_df = transform_candles_to_df(candles[asset]) - rounded_end_dt = end_dt.round(freq) - - periods = get_periods_range( - start_dt=None, end_dt=rounded_end_dt, - freq=freq, periods=bar_count - ) - - if rounded_end_dt > end_dt: - periods = periods[:-1] - elif rounded_end_dt <= end_dt: - periods = periods[1:] - - # periods = pd.date_range(end=end_dt, periods=bar_count, freq=freq) + rounded_end_dt = end_dt.floor(freq) + periods = pd.date_range(end=rounded_end_dt, + periods=bar_count, + freq=freq) asset_df = forward_fill_df_if_needed(asset_df, periods) all_series[asset] = pd.Series(asset_df[field]) diff --git a/catalyst/marketplace/marketplace.py b/catalyst/marketplace/marketplace.py index a1ac263c..e972b750 100644 --- a/catalyst/marketplace/marketplace.py +++ b/catalyst/marketplace/marketplace.py @@ -23,7 +23,7 @@ from catalyst.exchange.utils.stats_utils import set_print_settings from catalyst.marketplace.marketplace_errors import ( MarketplacePubAddressEmpty, MarketplaceDatasetNotFound, MarketplaceNoAddressMatch, MarketplaceHTTPRequest, - MarketplaceNoCSVFiles) + MarketplaceNoCSVFiles, MarketplaceRequiresPython3) from catalyst.marketplace.utils.auth_utils import get_key_secret, \ get_signed_headers from catalyst.marketplace.utils.bundle_utils import merge_bundles @@ -44,7 +44,10 @@ log = logbook.Logger('Marketplace', level=LOG_LEVEL) class Marketplace: def __init__(self): global Web3 - from web3 import Web3, HTTPProvider + try: + from web3 import Web3, HTTPProvider + except ImportError: + raise MarketplaceRequiresPython3() self.addresses = get_user_pubaddr() @@ -60,7 +63,8 @@ class Marketplace: contract_url = urllib.urlopen(MARKETPLACE_CONTRACT) self.mkt_contract_address = Web3.toChecksumAddress( - contract_url.readline().strip()) + contract_url.readline().decode( + contract_url.info().get_content_charset()).strip()) abi_url = urllib.urlopen(MARKETPLACE_CONTRACT_ABI) abi = json.load(abi_url) @@ -73,7 +77,8 @@ class Marketplace: contract_url = urllib.urlopen(ENIGMA_CONTRACT) self.eng_contract_address = Web3.toChecksumAddress( - contract_url.readline().strip()) + contract_url.readline().decode( + contract_url.info().get_content_charset()).strip()) abi_url = urllib.urlopen(ENIGMA_CONTRACT_ABI) abi = json.load(abi_url) @@ -148,13 +153,13 @@ class Marketplace: 'Gas Price:\t\t[Accept the default value]\n' 'Nonce:\t\t\t{nonce}\n' 'Data:\t\t\t{data}\n'.format( - _from=from_address, - to=tx['to'], - value=tx['value'], - gas=tx['gas'], - nonce=tx['nonce'], - data=tx['data'], ) - ) + _from=from_address, + to=tx['to'], + value=tx['value'], + gas=tx['gas'], + nonce=tx['nonce'], + data=tx['data'], ) + ) signed_tx = input('Copy and Paste the "Signed Transaction" ' 'field here:\n') @@ -259,14 +264,14 @@ class Marketplace: 'buy: {} ENG. Get enough ENG to cover the costs of the ' 'monthly\nsubscription for what you are trying to buy, ' 'and try again.'.format( - address, from_grains(balance), price)) + address, from_grains(balance), price)) return while True: agree_pay = input('Please confirm that you agree to pay {} ENG ' 'for a monthly subscription to the dataset "{}" ' 'starting today. [default: Y] '.format( - price, dataset)) or 'y' + price, dataset)) or 'y' if agree_pay.lower() not in ('y', 'n'): print("Please answer Y or N.") else: @@ -369,7 +374,7 @@ class Marketplace: 'You can now ingest this dataset anytime during the ' 'next month by running the following command:\n' 'catalyst marketplace ingest --dataset={}'.format( - dataset, address, dataset)) + dataset, address, dataset)) def process_temp_bundle(self, ds_name, path): """ @@ -426,10 +431,10 @@ class Marketplace: print('Your subscription to dataset "{}" expired on {} UTC.' 'Please renew your subscription by running:\n' 'catalyst marketplace subscribe --dataset={}'.format( - ds_name, - pd.to_datetime(check_sub[4], unit='s', utc=True), - ds_name) - ) + ds_name, + pd.to_datetime(check_sub[4], unit='s', utc=True), + ds_name) + ) if 'key' in self.addresses[address_i]: key = self.addresses[address_i]['key'] @@ -621,7 +626,7 @@ class Marketplace: ) except Exception as e: - print('Unable to subscribe to data source: {}'.format(e)) + print('Unable to register the requested dataset: {}'.format(e)) return self.check_transaction(tx_hash) diff --git a/catalyst/marketplace/marketplace_errors.py b/catalyst/marketplace/marketplace_errors.py index b6be1c3b..488c204f 100644 --- a/catalyst/marketplace/marketplace_errors.py +++ b/catalyst/marketplace/marketplace_errors.py @@ -9,7 +9,8 @@ def silent_except_hook(exctype, excvalue, exctraceback): MarketplaceNoAddressMatch, MarketplaceHTTPRequest, MarketplaceNoCSVFiles, MarketplaceContractDataNoMatch, MarketplaceSubscriptionExpired, MarketplaceJSONError, - MarketplaceWalletNotSupported, MarketplaceEmptySignature]: + MarketplaceWalletNotSupported, MarketplaceEmptySignature, + MarketplaceRequiresPython3]: fn = traceback.extract_tb(exctraceback)[-1][0] ln = traceback.extract_tb(exctraceback)[-1][1] print("Error traceback: {1} (line {2})\n" @@ -86,3 +87,11 @@ class MarketplaceJSONError(ZiplineError): 'The configuration file {file} is malformed. Please correct ' 'the following error:\n{error}' ) + + +class MarketplaceRequiresPython3(ZiplineError): + msg = ( + '\nCatalyst requires Python3 to access the Enigma Data Marketplace.\n' + 'If you want to use the Data Marketplace, you need to reinstall ' + 'Catalyst\nwith Python3. See the documentation website for additional ' + 'information.') diff --git a/etc/python2.7-environment.yml b/etc/python2.7-environment.yml index b2e2486c..3835d7d4 100644 --- a/etc/python2.7-environment.yml +++ b/etc/python2.7-environment.yml @@ -23,7 +23,9 @@ dependencies: - bottleneck==1.2.1 - chardet==3.0.4 - ccxt==1.10.1094 - - web3==4.0.0b7 +# The Enigma Data Marketplace requires Python3 because it depends on +# web3, which requires Python3, as building its dependencies breaks in Python2 +# - web3==4.0.0b7 - requests-toolbelt==0.8.0 - click==6.7 - contextlib2==0.5.5 diff --git a/etc/requirements.txt b/etc/requirements.txt index bac9666d..5be47e3d 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -84,5 +84,5 @@ tables==3.3.0 ccxt==1.10.1094 boto3==1.4.8 redo==1.6 -web3==4.0.0b7 +web3==4.0.0b11; python_version > '3.4' requests-toolbelt==0.8.0 diff --git a/etc/requirements_marketplace.txt b/etc/requirements_marketplace.txt deleted file mode 100644 index 2a56c200..00000000 --- a/etc/requirements_marketplace.txt +++ /dev/null @@ -1,2 +0,0 @@ -web3==4.0.0b7 -requests-toolbelt==0.8.0 diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index cbb713d2..4779d25d 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -51,8 +51,10 @@ class TestExchangeBundle: exchange.get_asset('eng_eth') ] - start = pd.to_datetime('2018-02-01', utc=True) - end = pd.to_datetime('2018-02-10', utc=True) + # start = pd.to_datetime('2018-02-01', utc=True) + # end = pd.to_datetime('2018-02-28', utc=True) + start = None + end = None log.info('ingesting exchange bundle {}'.format(exchange_name)) exchange_bundle.ingest( @@ -60,8 +62,8 @@ class TestExchangeBundle: include_symbols=','.join([asset.symbol for asset in assets]), # include_symbols=None, exclude_symbols=None, - start=start, - end=end, + start=None, + end=None, show_progress=False, show_breakdown=False ) diff --git a/tests/exchange/test_exchange_utils.py b/tests/exchange/test_exchange_utils.py index ddc15cc9..2d3d1efe 100644 --- a/tests/exchange/test_exchange_utils.py +++ b/tests/exchange/test_exchange_utils.py @@ -2,6 +2,7 @@ from catalyst.exchange.utils.exchange_utils import transform_candles_to_df, \ forward_fill_df_if_needed, get_candles_df from catalyst.testing.fixtures import WithLogger, ZiplineTestCase +from datetime import timedelta from pandas import Timestamp, DataFrame, concat import numpy as np @@ -15,9 +16,59 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase): new_df.index.name = None return new_df + @classmethod + def verify_forward_fill_df_if_needed(cls, candles, periods, expected_df): + observed_df = forward_fill_df_if_needed( + transform_candles_to_df(candles), + periods) + assert (expected_df.equals(observed_df)) + + @classmethod + def verify_get_candles_df(cls, assets, candles, end_fixed_dt, + expected_df, check_next_candle=False): + # run on all the fields + for field in ['volume', 'open', 'close', 'high', 'low']: + + field_dt = cls.get_specific_field_from_df(expected_df, + field, + assets[0]) + # run on several timestamps + for delta in range(5): + end_dt = end_fixed_dt + timedelta(minutes=delta) + assert (field_dt.equals(get_candles_df({assets[0]: candles}, + field, '5T', 3, + end_dt=end_dt))) + + field_dt_a1 = cls.get_specific_field_from_df(expected_df, + field, + assets[0]) + field_dt_a2 = cls.get_specific_field_from_df(expected_df, + field, + assets[1]) + observed_df = get_candles_df({assets[0]: candles, + assets[1]: candles}, + field, '5T', 3, + end_dt=end_dt) + + assert (observed_df.equals(concat([field_dt_a1, field_dt_a2], + axis=1))) + + if check_next_candle: + # one candle forward + end_dt = end_fixed_dt + timedelta(minutes=6) + observed_df = get_candles_df({assets[0]: candles, + assets[1]: candles}, + field, '5T', 3, + end_dt=end_dt) + + assert (not observed_df.equals(concat([field_dt_a1, + field_dt_a2], + axis=1))) + assert (concat([field_dt_a1, field_dt_a2], + axis=1)[1:].equals(observed_df[:-1])) + def test_get_candles_df(self): - asset = 'btc_usdt' - asset2 = 'eth_usdt' + assets = ['btc_usdt', 'eth_usdt'] # test forward fill in the end candles = [{'high': 595, 'volume': 10, 'low': 594, @@ -51,20 +102,12 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase): Timestamp('2018-03-01 09:50:00+0000', tz='UTC'), Timestamp('2018-03-01 09:55:00+0000', tz='UTC')] - observed_df = forward_fill_df_if_needed( - transform_candles_to_df(candles), - periods) expected_df = transform_candles_to_df(expected) - assert (expected_df.equals(observed_df)) - - for field in ['volume', 'open', 'close', 'high', 'low']: - field_dt = self.get_specific_field_from_df(expected_df, - field, - asset) - assert (field_dt.equals(get_candles_df({asset: candles}, - field, '5T', 3, - end_dt=periods[2]))) + self.verify_forward_fill_df_if_needed(candles, periods, + expected_df) + self.verify_get_candles_df(assets, candles, periods[2], + expected_df, True) # test forward fill in the middle candles = [{'high': 595, 'volume': 10, 'low': 594, @@ -94,28 +137,9 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase): tz='UTC') }] - df = transform_candles_to_df(candles) - observed_df = forward_fill_df_if_needed(df, periods) expected_df = transform_candles_to_df(expected) - - assert (expected_df.equals(observed_df)) - - for field in ['volume', 'open', 'close', 'high', 'low']: - # test several assets as well - observed_df = get_candles_df({asset: candles, - asset2: candles}, - field, '5T', 3, - end_dt=periods[2]) - - field_dt_a1 = self.get_specific_field_from_df(expected_df, - field, - asset) - field_dt_a2 = self.get_specific_field_from_df(expected_df, - field, - asset2) - - assert(observed_df.equals(concat([field_dt_a1, field_dt_a2], - axis=1))) + self.verify_forward_fill_df_if_needed(candles, periods, expected_df) + self.verify_get_candles_df(assets, candles, periods[2], expected_df) # test "forward fill" at the beginning candles = [{'high': 595, 'volume': 10, 'low': 594, @@ -145,18 +169,7 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase): tz='UTC') }] - df = transform_candles_to_df(candles) - observed_df = forward_fill_df_if_needed(df, periods) expected_df = transform_candles_to_df(expected) - - assert (expected_df.equals(observed_df)) + self.verify_forward_fill_df_if_needed(candles, periods, expected_df) # Not the same due to dropna - commenting out for now - """ - for field in ['volume', 'open', 'close', 'high', 'low']: - field_dt = self.get_specific_field_from_df(observed_df, - field, - asset) - assert(field_dt.equals(get_candles_df({asset:candles}, - field, '5T', 3, - end_dt=periods[2]))) - """ + # self.verify_get_candles_df(assets, candles, periods[2], expected_df)