mirror of
https://github.com/wassname/catalyst.git
synced 2026-07-13 03:17:15 +08:00
Experimenting with simpler bundle and data portal approach (works in unit testing)
This commit is contained in:
fredfortier
@@ -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'
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user