wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-06-27 20:04:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Testing related adjustments

Browse Source
This commit is contained in:
fredfortier
2017-10-16 03:09:13 -04:00
parent 403f951c77
commit 1732b4a985
12 changed files with 257 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/examples/arbitrage_with_interface.py
+6 -6
View File
@@ -9,7 +9,7 @@ from catalyst.api import (
from catalyst.exchange.stats_utils import get_pretty_stats
from catalyst.utils.run_algo import run_algorithm
algo_namespace = 'arbitrage_neo_eth'
algo_namespace = 'arbitrage_eth_btc'
log = Logger(algo_namespace)
@@ -19,10 +19,10 @@ def initialize(context):
# The context contains a new "exchanges" attribute which is a dictionary
# of exchange objects by exchange name. This allow easy access to the
# exchanges.
context.buying_exchange = context.exchanges['bittrex']
context.buying_exchange = context.exchanges['poloniex']
context.selling_exchange = context.exchanges['bitfinex']
context.trading_pair_symbol = 'neo_eth'
context.trading_pair_symbol = 'eth_btc'
context.trading_pairs = dict()
# Note the second parameter of the symbol() method
@@ -30,7 +30,7 @@ def initialize(context):
# the exchange information. This allow all other operations using
# the TradingPair to target the correct exchange.
context.trading_pairs[context.buying_exchange] = \
symbol('neo_eth', context.buying_exchange.name)
symbol('eth_btc', context.buying_exchange.name)
context.trading_pairs[context.selling_exchange] = \
symbol(context.trading_pair_symbol, context.selling_exchange.name)
@@ -267,9 +267,9 @@ run_algorithm(
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='bittrex,bitfinex',
exchange_name='poloniex,bitfinex',
live=True,
algo_namespace=algo_namespace,
base_currency='eth',
base_currency='btc',
live_graph=False
)
catalyst/examples/simple_loop.py
+12 -2
View File
@@ -1,4 +1,5 @@
import pandas as pd
import talib
from catalyst import run_algorithm
from catalyst.api import symbol
@@ -15,11 +16,20 @@ def handle_data(context, data):
price = data.current(context.asset, 'close')
print('got price {price}'.format(price=price))
prices = data.history(
context.asset,
fields='price',
bar_count=50,
frequency='1m'
)
rsi = talib.RSI(prices.values, timeperiod=14)[-1]
print('got rsi: {}'.format(rsi))
run_algorithm(
capital_base=250,
start=pd.to_datetime('2017-5-1', utc=True),
end=pd.to_datetime('2017-5-31', utc=True),
start=pd.to_datetime('2017-9-5', utc=True),
end=pd.to_datetime('2017-9-30', utc=True),
data_frequency='minute',
initialize=initialize,
handle_data=handle_data,
catalyst/exchange/bitfinex/bitfinex.py
+24
View File
@@ -661,3 +661,27 @@ class Bitfinex(Exchange):
return time.strftime('%Y-%m-%d',
time.gmtime(int(response.json()[-1][0] / 1000)))
def get_orderbook(self, asset, order_type='all'):
exchange_symbol = asset.exchange_symbol
try:
self.ask_request()
response = self._request(
'book/{}'.format(exchange_symbol), None)
data = response.json()
except Exception as e:
raise ExchangeRequestError(error=e)
# TODO: filter by type
result = dict()
for order_type in data:
result[order_type] = []
for entry in data[order_type]:
result[order_type].append(dict(
rate=float(entry['price']),
quantity=float(entry['amount'])
))
return result
catalyst/exchange/bittrex/bittrex.py
+21 -15
View File
@@ -65,13 +65,19 @@ class Bittrex(Exchange):
log.debug('retrieving wallet balances')
self.ask_request()
balances = self.api.getbalances()
except Exception as e:
raise ExchangeRequestError(error=e)
std_balances = dict()
for balance in balances:
currency = balance['Currency'].lower()
std_balances[currency] = balance['Available']
try:
for balance in balances:
currency = balance['Currency'].lower()
std_balances[currency] = balance['Available']
except TypeError:
raise ExchangeRequestError(error=balances)
return std_balances
def create_order(self, asset, amount, is_buy, style):
@@ -349,29 +355,29 @@ class Bittrex(Exchange):
json.dump(symbol_map, f, sort_keys=True, indent=2,
separators=(',', ':'))
def get_orderbook(self, asset, type='all'):
if type == 'all':
type = 'both'
elif type == 'bid':
type = 'buy'
elif type == 'ask':
type = 'sell'
def get_orderbook(self, asset, order_type='all'):
if order_type == 'all':
order_type = 'both'
elif order_type == 'bid':
order_type = 'buy'
elif order_type == 'ask':
order_type = 'sell'
else:
raise ValueError('invalid type')
exchange_symbol = asset.exchange_symbol
data = self.api.getorderbook(market=exchange_symbol, type=type)
data = self.api.getorderbook(market=exchange_symbol, type=order_type)
result = dict()
for exchange_type in data:
if exchange_type == 'buy':
type = 'bid'
order_type = 'bids'
elif exchange_type == 'sell':
type = 'ask'
order_type = 'asks'
result[type] = []
result[order_type] = []
for entry in data[exchange_type]:
result[type].append(dict(
result[order_type].append(dict(
rate=entry['Rate'],
quantity=entry['Quantity']
))
catalyst/exchange/bundle_utils.py
+81 -149
View File
@@ -71,86 +71,6 @@ def get_bcolz_chunk(exchange_name, symbol, data_frequency, period):
return path
def get_history(exchange_name, data_frequency, symbol, start=None, end=None):
"""
History API provides OHLCV data for any of the supported exchanges up to yesterday.
:param exchange_name: string
Required: The name identifier of the exchange (e.g. bitfinex, bittrex, poloniex).
:param data_frequency: string
Required: The bar frequency (minute or daily)
:param symbol: string
Required: The trading pair symbol, using Catalyst naming convention
:param start: datetime
Optional: The start date.
:param end: datetime
Optional: The end date.
:return ohlcv: list[dict[string, float]]
Each row contains the following dictionary for the resulting bars:
'ts' : int, the timestamp in seconds
'open' : float
'high' : float
'low' : float
'close' : float
'volume' : float
Notes
=====
Using seconds for the start and end dates for ease of use in the
function query parameters.
Sometimes, one minute goes by without completing a trade of the given
trading pair on the given exchange. To minimize the payload size, we
don't return identical sequential bars. Post-processing code will
forward fill missing bars outside of this function.
"""
start_seconds = get_seconds_from_date(start) if start else None
end_seconds = get_seconds_from_date(end) if end else None
if exchange_name not in EXCHANGE_NAMES:
raise ValueError(
'get_history function only supports the following exchanges: {}'.format(
list(EXCHANGE_NAMES)))
if data_frequency != 'daily' and data_frequency != 'minute':
raise ValueError(
'get_history currently only supports daily and minute data.'
)
url = '{api_url}/candles?exchange={exchange}&market={symbol}&freq={data_frequency}'.format(
api_url=API_URL,
exchange=exchange_name,
symbol=symbol,
data_frequency=data_frequency,
)
if start_seconds:
url += '&start={}'.format(start_seconds)
if end_seconds:
url += '&end={}'.format(end_seconds)
try:
response = requests.get(url)
except Exception as e:
raise ValueError(e)
data = response.json()
if 'error' in data:
raise ApiCandlesError(error=data['error'])
for candle in data:
last_traded = pd.Timestamp.utcfromtimestamp(candle['ts'])
last_traded = last_traded.replace(tzinfo=pytz.UTC)
candle['last_traded'] = last_traded
return data
def get_delta(periods, data_frequency):
return timedelta(minutes=periods) \
if data_frequency == 'minute' else timedelta(days=periods)
@@ -270,75 +190,6 @@ def range_in_bundle(asset, start_dt, end_dt, reader):
return has_data
@deprecated
def get_history_mock(exchange_name, data_frequency, symbol, start_ms, end_ms,
exchanges):
"""
Mocking the history API written by Victor by proxying the request
to Bitfinex.
:param exchange_name: string
The name identifier of the exchange (e.g. bitfinex).
Only bitfinex is supported in this mock function.
:param data_frequency: string
The bar frequency (minute or daily)
:param symbol: string
The trading pair symbol.
:param start_ms: float
The start date in milliseconds.
:param end_ms: float
The end date in milliseconds.
:param exchanges: MOCK ONLY
This won't be required in production mode since the exchange object
will be retrieved on the server.
:return ohlcv: list[dict[string, float]]
The open, high, low, volume collection for the resulting bars.
Notes
=====
Using milliseconds for the start and end dates for ease of use in
URL query parameters.
Sometimes, one minute goes by without completing a trade of the given
trading pair on the given exchange. To minimize the payload size, we
don't return identical sequential bars. Post-processing code will
forward fill missing bars outside of this function.
"""
if exchange_name != 'bitfinex':
raise ValueError('get_history mock function only works with bitfinex')
exchange = exchanges[exchange_name]
assets = [exchange.get_asset(symbol=symbol)]
start = get_date_from_ms(start_ms)
end = get_date_from_ms(end_ms)
delta = end - start
periods = delta.seconds % 3600 / 60.0 \
if data_frequency == 'minute' else delta.days
candles = exchange.get_candles(
data_frequency=data_frequency,
assets=assets,
bar_count=periods,
start_dt=start,
end_dt=end
)
ohlcv = []
for candle in candles:
ohlcv.append(dict(
open=candle['open'],
high=candle['high'],
low=candle['low'],
close=candle['close'],
volume=candle['volume'],
last_traded=candle['last_traded']
))
return ohlcv
def find_most_recent_time(bundle_name):
"""
Find most recent "time folder" for a given bundle.
@@ -368,3 +219,84 @@ def find_most_recent_time(bundle_name):
return most_recent_bundle.keys()[0]
else:
return None
@deprecated
def get_history(exchange_name, data_frequency, symbol, start=None, end=None):
"""
History API provides OHLCV data for any of the supported exchanges up to yesterday.
:param exchange_name: string
Required: The name identifier of the exchange (e.g. bitfinex, bittrex, poloniex).
:param data_frequency: string
Required: The bar frequency (minute or daily)
:param symbol: string
Required: The trading pair symbol, using Catalyst naming convention
:param start: datetime
Optional: The start date.
:param end: datetime
Optional: The end date.
:return ohlcv: list[dict[string, float]]
Each row contains the following dictionary for the resulting bars:
'ts' : int, the timestamp in seconds
'open' : float
'high' : float
'low' : float
'close' : float
'volume' : float
Notes
=====
Using seconds for the start and end dates for ease of use in the
function query parameters.
Sometimes, one minute goes by without completing a trade of the given
trading pair on the given exchange. To minimize the payload size, we
don't return identical sequential bars. Post-processing code will
forward fill missing bars outside of this function.
"""
start_seconds = get_seconds_from_date(start) if start else None
end_seconds = get_seconds_from_date(end) if end else None
if exchange_name not in EXCHANGE_NAMES:
raise ValueError(
'get_history function only supports the following exchanges: {}'.format(
list(EXCHANGE_NAMES)))
if data_frequency != 'daily' and data_frequency != 'minute':
raise ValueError(
'get_history currently only supports daily and minute data.'
)
url = '{api_url}/candles?exchange={exchange}&market={symbol}&freq={data_frequency}'.format(
api_url=API_URL,
exchange=exchange_name,
symbol=symbol,
data_frequency=data_frequency,
)
if start_seconds:
url += '&start={}'.format(start_seconds)
if end_seconds:
url += '&end={}'.format(end_seconds)
try:
response = requests.get(url)
except Exception as e:
raise ValueError(e)
data = response.json()
if 'error' in data:
raise ApiCandlesError(error=data['error'])
for candle in data:
last_traded = pd.Timestamp.utcfromtimestamp(candle['ts'])
last_traded = last_traded.replace(tzinfo=pytz.UTC)
candle['last_traded'] = last_traded
return data
catalyst/exchange/exchange.py
+37 -12
View File
@@ -13,7 +13,7 @@ from logbook import Logger
from catalyst.data.data_portal import BASE_FIELDS
from catalyst.exchange import bundle_utils
from catalyst.exchange.bundle_utils import get_start_dt, \
get_delta, get_trailing_candles_dt
get_delta, get_trailing_candles_dt, get_periods
from catalyst.exchange.exchange_bundle import ExchangeBundle
from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \
InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \
@@ -505,9 +505,7 @@ class Exchange:
candles = self.get_candles(
data_frequency=data_frequency,
assets=[asset],
bar_count=bar_count,
start_dt=exchange_start if bar_count > 1 else None,
end_dt=exchange_end
bar_count=bar_count
)
data += candles[asset]
@@ -588,8 +586,28 @@ class Exchange:
if len(missing_assets) > 0:
writer = bundle.get_writer(start_dt, end_dt, data_frequency)
chunks = bundle.prepare_chunks(
assets=assets,
data_frequency=data_frequency,
start_dt=start_dt,
end_dt=end_dt
)
for chunk in chunks:
log.debug('ingesting chunk for pair {}, period {}'.format(
chunk['asset'],
chunk['period']
))
self.ingest_ctable(
asset=chunk['asset'],
data_frequency=data_frequency,
period=chunk['period'],
writer=writer
)
# Adding bars too recent to be contained in the consolidated
# exchanges bundles. We go directly against the exchange
# to retrieve the candles.
for asset in missing_assets:
# TODO: use this only for data too recent to be in a bundle
trailing_candles_dt = get_trailing_candles_dt(
asset=asset,
start_dt=start_dt,
@@ -598,18 +616,20 @@ class Exchange:
)
if trailing_candles_dt is not None:
trailing_bar_count = \
get_periods(start_dt, end_dt, data_frequency)
# The get_history method supports multiple asset
candles = self.get_candles(
data_frequency=data_frequency,
assets=[asset],
bar_count=bar_count,
start_dt=trailing_candles_dt,
bar_count=trailing_bar_count,
end_dt=end_dt
)
bundle.ingest_candles(
candles=candles,
bar_count=adj_bar_count,
bar_count=trailing_bar_count,
end_dt=end_dt,
data_frequency=data_frequency,
writer=writer
@@ -866,7 +886,7 @@ class Exchange:
@abstractmethod
def get_candles(self, data_frequency, assets, bar_count=None,
start_date=None):
start_dt=None, end_dt=None):
"""
Retrieve OHLCV candles for the given assets
@@ -878,7 +898,7 @@ class Exchange:
The number of bar desired. (default 1)
:param end_dt: datetime, optional
The last bar date.
:param start_date: datetime, optional
:param start_dt: datetime, optional
The first bar date.
:return dict[TradingPair, dict[str, Object]]: OHLCV data
@@ -915,9 +935,14 @@ class Exchange:
pass
@abc.abstractmethod
def get_orderbook(self):
def get_orderbook(self, asset, order_type):
"""
Retrieve the account parameters.
Retrieve the the orderbook for the given trading pair.
:param asset: TradingPair
:param order_type: str
The type of orders: bid, ask or all
:return:
"""
pass
catalyst/exchange/exchange_bundle.py
+42 -25
View File
@@ -245,7 +245,6 @@ class ExchangeBundle:
# session_label when using a newly created writer
del self._writers[data_frequency]
# TODO: these are the dates of the chunk, not the job
writer = self.get_writer(writer._start_session,
writer._end_session, data_frequency)
writer.write(
@@ -255,9 +254,13 @@ class ExchangeBundle:
)
def ingest_candles(self, candles, bar_count, end_dt, data_frequency,
writer, previous_candle=dict()):
writer, previous_candle=dict()):
"""
Retrieve the specified OHLCV chunk and write it to the bundle
Ingest candles obtained via the get_candles API of an exchange.
Since exchange APIs generally only do not return candles when there
are no transactions in the period, we ffill values using the
previous candle to ensure that each period has a candle.
:param bar_count:
:param end_dt:
@@ -268,7 +271,6 @@ class ExchangeBundle:
:return:
"""
num_candles = 0
data = []
for asset in candles:
@@ -357,9 +359,8 @@ class ExchangeBundle:
start = reader.first_trading_day
# TODO: temp workaround, remove when the bundles are fixed
# end = reader.last_available_dt
end = reader.last_available_dt - timedelta(days=1)
end = reader.last_available_dt
periods = self.calendar.minutes_in_range(start, end)
@@ -407,33 +408,25 @@ class ExchangeBundle:
return path
def ingest(self, data_frequency, include_symbols=None,
exclude_symbols=None, start=None, end=None,
show_progress=True, environ=os.environ):
def prepare_chunks(self, assets, data_frequency, start_dt, end_dt):
"""
:param assets:
:param data_frequency:
:param include_symbols:
:param exclude_symbols:
:param start:
:param end:
:param show_progress:
:param environ:
:param start_dt:
:param end_dt:
:return:
"""
assets = self.get_assets(include_symbols, exclude_symbols)
start, end = self.get_adj_dates(start, end, assets, data_frequency)
reader = self.get_reader(data_frequency)
chunks = []
for asset in assets:
try:
asset_start, asset_end = \
self.get_adj_dates(start, end, [asset], data_frequency)
self.get_adj_dates(start_dt, end_dt, [asset],
data_frequency)
except ValueError:
dt += timedelta(days=1)
continue
sessions = self.calendar.sessions_in_range(asset_start, asset_end)
@@ -451,11 +444,11 @@ class ExchangeBundle:
datetime(dt.year, dt.month, 1, 0, 0, 0, 0),
utc=True)
# TODO: workaround, remove when bundles are fixed
month_end = pd.to_datetime(
datetime(dt.year, dt.month, month_range[1] - 1,
23, 59, 0, 0),
utc=True)
datetime(
dt.year, dt.month, month_range[1], 23, 59, 0, 0),
utc=True
)
if month_end > asset_end:
month_end = asset_end
@@ -477,7 +470,32 @@ class ExchangeBundle:
chunks.sort(key=lambda chunk: chunk['period_end'])
return chunks
def ingest(self, data_frequency, include_symbols=None,
exclude_symbols=None, start=None, end=None,
show_progress=True, environ=os.environ):
"""
:param data_frequency:
:param include_symbols:
:param exclude_symbols:
:param start:
:param end:
:param show_progress:
:param environ:
:return:
"""
assets = self.get_assets(include_symbols, exclude_symbols)
start, end = self.get_adj_dates(start, end, assets, data_frequency)
writer = self.get_writer(start, end, data_frequency)
chunks = self.prepare_chunks(
assets=assets,
data_frequency=data_frequency,
start_dt=start,
end_dt=end
)
with maybe_show_progress(
chunks,
show_progress,
@@ -485,7 +503,6 @@ class ExchangeBundle:
exchange=self.exchange.name,
frequency=data_frequency
)) as it:
for chunk in it:
self.ingest_ctable(
asset=chunk['asset'],
catalyst/exchange/poloniex/poloniex.py
+21 -17
View File
@@ -173,7 +173,8 @@ class Poloniex(Exchange):
# TODO: fetch account data and keep in cache
return None
def get_candles(self, data_frequency, assets, bar_count=None):
def get_candles(self, data_frequency, assets, bar_count=None,
start_dt=None, end_dt=None):
"""
Retrieve OHLVC candles from Poloniex
@@ -187,9 +188,10 @@ class Poloniex(Exchange):
'5m', '15m', '30m', '2h', '4h', '1D'
"""
# TODO: use BcolzMinuteBarReader to read from cache
# TODO: implement end_dt and start_dt filters
if (
data_frequency == '5m' or data_frequency == 'minute'): # TODO: Polo does not have '1m'
data_frequency == '5m' or data_frequency == 'minute'): # TODO: Polo does not have '1m'
frequency = 300
elif (data_frequency == '15m'):
frequency = 900
@@ -231,6 +233,9 @@ class Poloniex(Exchange):
)
def ohlc_from_candle(candle):
last_traded = pd.Timestamp.utcfromtimestamp(candle['date'])
last_traded = last_traded.replace(tzinfo=pytz.UTC)
ohlc = dict(
open=np.float64(candle['open']),
high=np.float64(candle['high']),
@@ -238,7 +243,7 @@ class Poloniex(Exchange):
close=np.float64(candle['close']),
volume=np.float64(candle['volume']),
price=np.float64(candle['close']),
last_traded=pd.Timestamp.utcfromtimestamp(candle['date'])
last_traded=last_traded
)
return ohlc
@@ -608,24 +613,23 @@ class Poloniex(Exchange):
return transactions
def get_orderbook(self, asset, type='all'):
def get_orderbook(self, asset, order_type='all'):
exchange_symbol = asset.exchange_symbol
data = self.api.returnOrderBook(market=exchange_symbol)
result = dict()
for exchange_type in data:
if exchange_type == 'bids':
type = 'bid'
elif exchange_type == 'asks':
type = 'ask'
else:
for order_type in data:
# TODO: filter by type
if order_type != 'asks' and order_type != 'bids':
continue
result[type] = []
for entry in data[exchange_type]:
result[order_type] = []
for entry in data[order_type]:
if len(entry) == 2:
result[type].append(dict(
rate=float(entry[0]),
quantity=float(entry[1])
))
result[order_type].append(
dict(
rate=float(entry[0]),
quantity=float(entry[1])
)
)
return result
catalyst/utils/run_algo.py
+4 -1
View File
@@ -228,8 +228,11 @@ def _run(handle_data,
balances = exchange.get_balances()
except ExchangeRequestError as e:
if attempt_index < 20:
log.warn('exchange error when retrieving balances, {} '
'trying again in 5 seconds'.format(e))
sleep(5)
return fetch_capital_base(attempt_index + 1)
return fetch_capital_base(exchange, attempt_index + 1)
else:
raise ExchangeRequestErrorTooManyAttempts(
attempts=attempt_index,
tests/exchange/test_bitfinex.py
+6
View File
@@ -69,3 +69,9 @@ class BitfinexTestCase(BaseExchangeTestCase):
log.info('testing exchange balances')
balances = self.exchange.get_balances()
pass
def test_orderbook(self):
log.info('testing order book for bitfinex')
asset = self.exchange.get_asset('eth_btc')
orderbook = self.exchange.get_orderbook(asset)
pass
tests/exchange/test_bundle.py
+1 -1
View File
@@ -14,7 +14,7 @@ class ExchangeBundleTestCase:
# start = pd.to_datetime('2017-09-01', utc=True)
start = pd.to_datetime('2017-1-1', utc=True)
end = pd.to_datetime('2017-6-30', utc=True)
end = pd.to_datetime('2017-9-30', utc=True)
exchange_bundle = ExchangeBundle(get_exchange(exchange_name))
tests/exchange/test_poloniex.py
+2 -1
View File
@@ -84,7 +84,8 @@ class PoloniexTestCase(BaseExchangeTestCase):
pass
def test_orderbook(self):
log.info('testing order book for bittrex')
log.info('testing order book for poloniex')
asset = self.exchange.get_asset('eth_btc')
orderbook = self.exchange.get_orderbook(asset)
pass