From 68546a0d8d605d35fb11f9a21c6377ea611e65cc Mon Sep 17 00:00:00 2001 From: fredfortier Date: Tue, 19 Sep 2017 03:49:34 -0400 Subject: [PATCH] Experimenting with simpler bundle and data portal approach (works in unit testing) --- catalyst/exchange/bitfinex/bitfinex.py | 13 +- catalyst/exchange/data_portal_exchange.py | 37 ++++- catalyst/exchange/exchange.py | 12 +- catalyst/exchange/exchange_bundle.py | 184 ++++++++++++++++++++++ tests/exchange/test_bundle.py | 60 +++++++ tests/exchange/test_data_portal.py | 35 +++- 6 files changed, 328 insertions(+), 13 deletions(-) create mode 100644 catalyst/exchange/exchange_bundle.py create mode 100644 tests/exchange/test_bundle.py diff --git a/catalyst/exchange/bitfinex/bitfinex.py b/catalyst/exchange/bitfinex/bitfinex.py index a72b6167..0397c2d8 100644 --- a/catalyst/exchange/bitfinex/bitfinex.py +++ b/catalyst/exchange/bitfinex/bitfinex.py @@ -4,6 +4,7 @@ import hmac import json import re import time +import datetime import numpy as np import pandas as pd @@ -47,6 +48,7 @@ class Bitfinex(Exchange): self._portfolio = portfolio self.minute_writer = None self.minute_reader = None + self.num_candles_limit = 100 def _request(self, operation, data, version='v1'): payload_object = { @@ -224,8 +226,7 @@ class Bitfinex(Exchange): # TODO: fetch account data and keep in cache return None - def get_candles(self, data_frequency, assets, bar_count=None, - start_date=None): + def get_candles(self, data_frequency, assets, bar_count=None, end_dt=None): """ Retrieve OHLVC candles from Bitfinex @@ -281,6 +282,14 @@ class Bitfinex(Exchange): if bar_count: is_list = True url += '/hist?limit={}'.format(int(bar_count)) + + if end_dt is not None: + epoch = datetime.datetime.utcfromtimestamp(0) + epoch = epoch.replace(tzinfo=pytz.UTC) + + end_ms = (end_dt - epoch).total_seconds() * 1000.0 + url += '&end={0:f}'.format(end_ms) + else: is_list = False url += '/last' diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index 10a85ba8..3c1dd01d 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -13,15 +13,20 @@ import abc from time import sleep + +import os import pandas as pd from catalyst.assets._assets import TradingPair from logbook import Logger from catalyst.data.data_portal import DataPortal +from catalyst.data.minute_bars import BcolzMinuteBarReader from catalyst.exchange.exchange_errors import ( ExchangeRequestError, ExchangeBarDataError ) +from catalyst.data.bundles.core import load +from catalyst.exchange.exchange_utils import get_exchange_minute_writer_root log = Logger('DataPortalExchange') @@ -230,8 +235,14 @@ class DataPortalExchangeLive(DataPortalExchangeBase): class DataPortalExchangeBacktest(DataPortalExchangeBase): - def __init__(self, exchanges, *args, **kwargs): - super(self.__class__, self).__init__(exchanges, *args, **kwargs) + def __init__(self, *args, **kwargs): + + super(DataPortalExchangeBacktest, self).__init__(*args, **kwargs) + + self.minute_readers = dict() + for exchange_name in self.exchanges: + root = get_exchange_minute_writer_root(exchange_name) + self.minute_readers[exchange_name] = BcolzMinuteBarReader(root) def get_exchange_history_window(self, exchange, @@ -254,7 +265,23 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): def get_exchange_spot_value(self, exchange, assets, field, dt, data_frequency): - exchange_spot_values = exchange.get_spot_value( - assets, field, dt, data_frequency) - return exchange_spot_values + if data_frequency == 'minute': + reader = self.minute_readers[exchange.name] + else: + raise ValueError('Unsupported frequency') + + values = [] + for asset in assets: + try: + value = reader.get_value( + sid=asset.sid, + dt=dt, + field=field + ) + values.append(value) + except Exception as e: + log.warn('minute data not found: {}'.format(e)) + values.append(None) + + return values diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index cf22b80a..cf26c9ff 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -35,6 +35,7 @@ class Exchange: self.minute_writer = None self.minute_reader = None self.base_currency = None + self.num_candles_limit = 100 @property def positions(self): @@ -96,6 +97,15 @@ class Exchange: return symbols + def get_assets(self, symbols): + assets = [] + + for symbol in symbols: + asset = self.get_asset(symbol) + assets.append(asset) + + return assets + def get_asset(self, symbol): """ Find an Asset on the current exchange based on its Catalyst symbol @@ -323,7 +333,7 @@ class Exchange: ) try: - #TODO: use victor's modified branch using int64 + # TODO: use victor's modified branch using int64 self.minute_writer.write_sid( sid=asset.sid, df=df diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py new file mode 100644 index 00000000..1605f279 --- /dev/null +++ b/catalyst/exchange/exchange_bundle.py @@ -0,0 +1,184 @@ +from datetime import timedelta + +import pandas as pd +from logbook import Logger + +from catalyst.data.minute_bars import BcolzMinuteOverlappingData +from catalyst.exchange.bitfinex.bitfinex import Bitfinex +from catalyst.exchange.bittrex.bittrex import Bittrex +from catalyst.exchange.exchange_errors import ExchangeNotFoundError +from catalyst.exchange.exchange_utils import get_exchange_auth +from catalyst.utils.cli import maybe_show_progress + + +def _cachpath(symbol, type_): + return '-'.join([symbol, type_]) + + +log = Logger('exchange_bundle') + + +def fetch_candles_chunk(exchange, assets, data_frequency, end_dt, bar_count): + candles = exchange.get_candles( + data_frequency=data_frequency, + assets=assets, + bar_count=bar_count, + end_dt=end_dt + ) + + series = dict() + + for asset in assets: + asset_candles = candles[asset] + + asset_df = pd.DataFrame(asset_candles) + asset_df.set_index('last_traded', inplace=True, drop=True) + asset_df.sort_index(inplace=True) + + series[asset] = asset_df + + return series + + +def exchange_bundle(exchange_name, symbols, start=None, end=None): + """Create a data bundle ingest function for the specified exchange. + + Parameters + ---------- + exchange_name: str + The name of the exchange + symbols : iterable[str] + The ticker symbols to load data for. + start : datetime, optional + The start date to query for. By default this pulls the full history + for the calendar. + end : datetime, optional + The end date to query for. By default this pulls the full history + for the calendar. + + Returns + ------- + ingest : callable + The bundle ingest function for the given set of symbols. + + Examples + -------- + This code should be added to ~/.catalyst/extension.py + + .. code-block:: python + + from catalyst.data.bundles import register + + symbols = ( + 'eth_btc', + 'etc_btc', + 'neo_btc', + ) + register('bitfinex_bundle', exchange_bundle('bitfinex', symbols)) + + Notes + ----- + The sids for each symbol will be the index into the symbols sequence. + """ + # strict this in memory so that we can reiterate over it + symbols = tuple(symbols) + + def ingest(environ, + asset_db_writer, + minute_bar_writer, # unused + daily_bar_writer, + adjustment_writer, + calendar, + start_session, + end_session, + cache, + show_progress, + output_dir, + # pass these as defaults to make them 'nonlocal' in py2 + start=start, + end=end): + + # TODO: I don't understand this session vs dates idea + if start is None: + start = start_session + if end is None: + end = None + + log.info('ingesting data from {} to {}'.format(start, end)) + + exchange_auth = get_exchange_auth(exchange_name) + if exchange_name == 'bitfinex': + exchange = Bitfinex( + key=exchange_auth['key'], + secret=exchange_auth['secret'], + base_currency=None, # TODO: make optional at the exchange + portfolio=None + ) + elif exchange_name == 'bittrex': + exchange = Bittrex( + key=exchange_auth['key'], + secret=exchange_auth['secret'], + base_currency=None, + portfolio=None + ) + else: + raise ExchangeNotFoundError(exchange_name=exchange_name) + + assets = exchange.get_assets(symbols) + + delta = end - start + delta_minutes = delta.total_seconds() / 60 + if delta_minutes > exchange.num_candles_limit: + bar_count = exchange.num_candles_limit + + chunks = [] + last_chunk_date = end + while last_chunk_date > start + timedelta(minutes=bar_count): + # TODO: account for the partial last bar + chunk = dict(end=last_chunk_date, bar_count=bar_count) + chunks.append(chunk) + + last_chunk_date = \ + last_chunk_date - timedelta(minutes=(bar_count + 1)) + + chunks.reverse() + + else: + chunks = [dict(end=end, bar_count=delta_minutes)] + + with maybe_show_progress( + chunks, + show_progress, + label='Fetching {} candles: '.format(exchange_name)) as it: + + for chunk in it: + asset_df = fetch_candles_chunk( + exchange=exchange, + assets=assets, + data_frequency='1m', + end_dt=chunk['end'], + bar_count=chunk['bar_count'] + ) + + data = [] + for asset in asset_df: + df = asset_df[asset] + sid = asset.sid + data.append((sid, df)) + + try: + log.debug( + 'writing chunk: {sid} start: {start} end: {end}'.format( + sid=sid, + start=chunk['end'] - timedelta( + minutes=chunk['bar_count']), + end=chunk['end'] + ) + ) + minute_bar_writer.write(data, show_progress=show_progress) + except KeyError: + minute_bar_writer.write(data, show_progress=show_progress) + except BcolzMinuteOverlappingData as e: + log.warn('Unable to write chunk {}: {}'.format(chunk, e)) + + return ingest diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py new file mode 100644 index 00000000..784aecea --- /dev/null +++ b/tests/exchange/test_bundle.py @@ -0,0 +1,60 @@ +from datetime import timedelta + +import os +import pandas as pd +from logging import Logger + +from catalyst import get_calendar + +from catalyst.data.minute_bars import BcolzMinuteBarWriter +from catalyst.exchange.exchange_bundle import exchange_bundle +from catalyst.exchange.exchange_utils import get_exchange_minute_writer_root + +log = Logger('test_exchange_bundle') + + +class ExchangeBundleTestCase: + def test_ingest(self): + exchange_name = 'bitfinex' + + start = pd.Timestamp.utcnow() - timedelta(days=2) + end = pd.Timestamp.utcnow() + open_calendar = get_calendar('OPEN') + + root = get_exchange_minute_writer_root(exchange_name) + filename = os.path.join(root, 'metadata.json') + + if os.path.isfile(filename): + minute_bar_writer = BcolzMinuteBarWriter.open(root, end) + else: + # TODO: need to be able to write more precise numbers + minute_bar_writer = BcolzMinuteBarWriter( + rootdir=root, + calendar=open_calendar, + minutes_per_day=1440, + start_session=start.floor('1d'), + end_session=end, + write_metadata=True + ) + + ingest = exchange_bundle( + exchange_name=exchange_name, + symbols=['btc_usd'] + ) + + ingest( + environ=os.environ, + asset_db_writer=None, # TODO: nice to have + minute_bar_writer=minute_bar_writer, + daily_bar_writer=None, # TODO: add later + adjustment_writer=None, # Not applicable to crypto + calendar=open_calendar, + start_session=start, + end_session=end, + cache=dict(), + show_progress=True, + output_dir=exchange_name, # TODO: not sure + start=start, + end=end + ) + pass diff --git a/tests/exchange/test_data_portal.py b/tests/exchange/test_data_portal.py index b569799d..08980162 100644 --- a/tests/exchange/test_data_portal.py +++ b/tests/exchange/test_data_portal.py @@ -1,7 +1,10 @@ +from datetime import timedelta + import pandas as pd from catalyst import get_calendar from logbook import Logger +from catalyst.data.minute_bars import BcolzMinuteBarReader from catalyst.exchange.asset_finder_exchange import AssetFinderExchange from catalyst.exchange.bitfinex.bitfinex import Bitfinex from catalyst.exchange.bittrex.bittrex import Bittrex @@ -33,12 +36,19 @@ class ExchangeDataPortalTestCase: open_calendar = get_calendar('OPEN') asset_finder = AssetFinderExchange() + self.data_portal_live = DataPortalExchangeLive( exchanges=dict(bitfinex=self.bitfinex, bittrex=self.bittrex), asset_finder=asset_finder, trading_calendar=open_calendar, first_trading_day=pd.to_datetime('today', utc=True) ) + self.data_portal_backtest = DataPortalExchangeBacktest( + exchanges=dict(bitfinex=self.bitfinex, bittrex=self.bittrex), + asset_finder=asset_finder, + trading_calendar=open_calendar, + first_trading_day=pd.to_datetime('today', utc=True) + ) def test_get_history_window_live(self): asset_finder = self.data_portal_live.asset_finder @@ -49,11 +59,11 @@ class ExchangeDataPortalTestCase: ] now = pd.Timestamp.utcnow() data = self.data_portal_live.get_history_window( - assets, - now, - 10, - '1m', - 'price') + assets, + now, + 10, + '1m', + 'price') pass def test_get_spot_value_live(self): @@ -67,3 +77,18 @@ class ExchangeDataPortalTestCase: value = self.data_portal_live.get_spot_value( assets, 'price', now, '1m') pass + + def test_get_spot_value_backtest(self): + asset_finder = self.data_portal_backtest.asset_finder + + assets = [ + asset_finder.lookup_symbol('btc_usd', self.bitfinex), + ] + + date = pd.Timestamp.utcnow() - timedelta(hours=2) + value = self.data_portal_backtest.get_spot_value( + assets, 'close', date, 'minute') + pass + + def test_get_history_window_backtest(self): + pass