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

Working on history implementation and improving error handling.

Browse Source
This commit is contained in:
Frederic Fortier
2017-08-19 15:51:57 -04:00
parent 30e15a7b8b
commit 14047fde15
7 changed files with 337 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/examples/buy_and_hold_live.py → catalyst/examples/buy_the_dip_live.py
+30 -6
View File
@@ -7,8 +7,10 @@ from catalyst.api import (
record,
get_open_orders,
)
from catalyst.exchange.exchange_errors import ExchangeRequestError
log = Logger('buy_and_hold_live')
algo_namespace = 'buy_the_dip_live'
log = Logger(algo_namespace)
def initialize(context):
@@ -18,9 +20,28 @@ def initialize(context):
context.TARGET_POSITIONS = 100
context.BUY_INCREMENT = 1
context.retry_check_open_orders = 2
context.retry_update_portfolio = 2
context.retry_order = 2
def handle_data(context, data):
log.info('handling bar {data}'.format(data=data))
# ohlc = data.history(context.asset,
# fields='price',
# bar_count=20,
# frequency='1d'
# )
# ohlc = data.history([context.asset, symbol('iot_usd')],
# fields='price',
# bar_count=20,
# frequency='1d'
# )
ohlc = data.history([context.asset, symbol('iot_usd')],
fields=['price', 'volume'],
bar_count=20,
frequency='1d'
)
cash = context.portfolio.cash
log.info('base currency available: {cash}'.format(cash=cash))
@@ -82,10 +103,13 @@ def handle_data(context, data):
leverage=context.account.leverage,
)
context.perf_tracker.update_performance()
log.info('the performance:\n{}'.format(
context.perf_tracker.to_dict('minute')
))
try:
context.perf_tracker.update_performance()
log.info('the performance:\n{}'.format(
context.perf_tracker.to_dict('minute')
))
except Exception as e:
log.warn('unable to calculate performance: {}'.format(e))
pass
@@ -101,5 +125,5 @@ run_algorithm(
capital_base=100000,
exchange_conn=exchange_conn,
live=True,
algo_namespace='buy_and_hold_live'
algo_namespace=algo_namespace
)
catalyst/exchange/algorithm_exchange.py
+58 -5
View File
@@ -24,6 +24,9 @@ from catalyst.utils.api_support import (
disallowed_in_before_trading_start)
from catalyst.utils.calendars.trading_calendar import days_at_time
from catalyst.exchange.exchange_errors import (
ExchangeRequestError
)
log = logbook.Logger("ExchangeTradingAlgorithm")
@@ -38,6 +41,11 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
self.exchange = kwargs.pop('exchange', None)
self.orders = {}
self.retry_check_open_orders = 5
self.retry_update_portfolio = 5
self.retry_order = 1
self.retry_delay = 5
super(self.__class__, self).__init__(*args, **kwargs)
log.info('exchange trading algorithm successfully initialized')
@@ -113,9 +121,32 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
def updated_account(self):
return self.exchange.account
def _update_portfolio(self, attempt_index=0):
try:
self.exchange.update_portfolio()
except ExchangeRequestError as e:
log.warn(
'update portfolio attempt {}: {}'.format(attempt_index, e)
)
if attempt_index < self.retry_update_portfolio:
time.sleep(self.retry_delay)
self._update_portfolio(attempt_index + 1)
def _check_open_orders(self, attempt_index=0):
try:
return self.exchange.check_open_orders()
except ExchangeRequestError as e:
log.warn(
'check open orders attempt {}: {}'.format(attempt_index, e)
)
if attempt_index < self.retry_check_open_orders:
time.sleep(self.retry_delay)
return self._check_open_orders(attempt_index + 1)
def handle_data(self, data):
self.exchange.update_portfolio()
transactions = self.exchange.check_open_orders()
self._update_portfolio()
transactions = self._check_open_orders()
for transaction in transactions:
self.perf_tracker.process_transaction(transaction)
@@ -127,6 +158,27 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
# every bar no matter if the algorithm places an order or not.
self.validate_account_controls()
def _order(self,
asset,
amount,
limit_price=None,
stop_price=None,
style=None,
attempt_index=0):
try:
return self.exchange.order(asset, amount, limit_price,
stop_price,
style)
except ExchangeRequestError as e:
log.warn(
'order attempt {}: {}'.format(attempt_index, e)
)
if attempt_index < self.retry_order:
time.sleep(self.retry_delay)
return self._order(
asset, amount, limit_price, stop_price, style,
attempt_index + 1)
@api_method
@disallowed_in_before_trading_start(OrderInBeforeTradingStart())
def order(self,
@@ -136,11 +188,12 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
stop_price=None,
style=None):
amount, style = self._calculate_order(asset, amount,
limit_price, stop_price, style)
limit_price, stop_price,
style)
order_id = self.exchange.order(asset, amount, limit_price, stop_price,
style)
order_id = self._order(asset, amount, limit_price, stop_price, style)
order = self.portfolio.open_orders[order_id]
self.perf_tracker.process_order(order)
return order
catalyst/exchange/bitfinex.py
+88 -96
View File
@@ -7,19 +7,26 @@ import json
import time
import requests
import pandas as pd
import collections
from datetime import timedelta, datetime
from catalyst.protocol import Portfolio, Account
# from websocket import create_connection
from catalyst.exchange.exchange import Exchange
from logbook import Logger
from catalyst.assets._assets import Asset
from catalyst.finance.order import ORDER_STATUS
from catalyst.exchange.exchange_order import ExchangeOrder
from catalyst.finance.execution import (MarketOrder,
LimitOrder,
StopOrder,
StopLimitOrder)
from catalyst.data.data_portal import BASE_FIELDS
from catalyst.exchange.exchange_portfolio import ExchangePortfolio
from catalyst.errors import (
IncompatibleHistoryFrequency,
)
from catalyst.exchange.exchange_errors import (
ExchangeRequestError,
InvalidHistoryFrequencyError
)
BITFINEX_URL = 'https://api.bitfinex.com'
ASSETS = '{ "USDT_BTC": {"symbol":"btc_usd", "start_date": "2010-01-01"}, "ltcusd": {"symbol":"ltc_usd", "start_date": "2010-01-01"}, "ltcbtc": {"symbol":"ltc_btc", "start_date": "2010-01-01"}, "ethusd": {"symbol":"eth_usd", "start_date": "2010-01-01"}, "ethbtc": {"symbol":"eth_btc", "start_date": "2010-01-01"}, "etcbtc": {"symbol":"etc_btc", "start_date": "2010-01-01"}, "etcusd": {"symbol":"etc_usd", "start_date": "2010-01-01"}, "rrtusd": {"symbol":"rrt_usd", "start_date": "2010-01-01"}, "rrtbtc": {"symbol":"rrt_btc", "start_date": "2010-01-01"}, "zecusd": {"symbol":"zec_usd", "start_date": "2010-01-01"}, "zecbtc": {"symbol":"zec_btc", "start_date": "2010-01-01"}, "xmrusd": {"symbol":"xmr_usd", "start_date": "2010-01-01"}, "xmrbtc": {"symbol":"xmr_btc", "start_date": "2010-01-01"}, "dshusd": {"symbol":"dsh_usd", "start_date": "2010-01-01"}, "dshbtc": {"symbol":"dsh_btc", "start_date": "2010-01-01"}, "bccbtc": {"symbol":"bcc_btc", "start_date": "2010-01-01"}, "bcubtc": {"symbol":"bcu_btc", "start_date": "2010-01-01"}, "bccusd": {"symbol":"bcc_usd", "start_date": "2010-01-01"}, "bcuusd": {"symbol":"bcu_usd", "start_date": "2010-01-01"}, "xrpusd": {"symbol":"xrp_usd", "start_date": "2010-01-01"}, "xrpbtc": {"symbol":"xrp_btc", "start_date": "2010-01-01"}, "iotusd": {"symbol":"iot_usd", "start_date": "2010-01-01"}, "iotbtc": {"symbol":"iot_btc", "start_date": "2010-01-01"}, "ioteth": {"symbol":"iot_eth", "start_date": "2010-01-01"}, "eosusd": {"symbol":"eos_usd", "start_date": "2010-01-01"}, "eosbtc": {"symbol":"eos_btc", "start_date": "2010-01-01"}, "eoseth": {"symbol":"eos_eth", "start_date": "2010-01-01"} }'
@@ -175,8 +182,8 @@ class Bitfinex(Exchange):
response = self._request('balances', None)
balances = response.json()
if 'message' in balances:
raise ValueError(
'unable to fetch balance %s' % balances['message']
raise ExchangeRequestError(
error='unable to fetch balance {}'.format(balances['message'])
)
base_position = None
@@ -187,7 +194,7 @@ class Bitfinex(Exchange):
if position is None:
raise ValueError(
'Base currency %s not found in portfolio' % self.base_currency
error='Base currency %s not found in portfolio' % self.base_currency
)
portfolio = self.store.portfolio
@@ -265,99 +272,81 @@ class Bitfinex(Exchange):
def subscribe_to_market_data(self, symbol):
pass
def get_spot_value(self, assets, field, dt=None, data_frequency='minute'):
"""
Public API method that returns a scalar value representing the value
of the desired asset's field at either the given dt.
def get_candles(self, data_frequency, assets,
end_dt=None, bar_count=None, limit=None):
Parameters
----------
assets : Asset, ContinuousFuture, or iterable of same.
The asset or assets whose data is desired.
field : {'open', 'high', 'low', 'close', 'volume',
'price', 'last_traded'}
The desired field of the asset.
dt : pd.Timestamp
The timestamp for the desired value.
data_frequency : str
The frequency of the data to query; i.e. whether the data is
'daily' or 'minute' bars
Returns
-------
value : float, int, or pd.Timestamp
The spot value of ``field`` for ``asset`` The return type is based
on the ``field`` requested. If the field is one of 'open', 'high',
'low', 'close', or 'price', the value will be a float. If the
``field`` is 'volume' the value will be a int. If the ``field`` is
'last_traded' the value will be a Timestamp.
Bitfinex timeframes
-------------------
Available values: '1m', '5m', '15m', '30m', '1h', '3h', '6h', '12h',
'1D', '7D', '14D', '1M'
"""
if field not in BASE_FIELDS:
raise KeyError('Invalid column: ' + str(field))
if isinstance(assets, collections.Iterable):
values = list()
for asset in assets:
value = self.get_single_spot_value(
asset, field, data_frequency)
values.append(value)
return values
else:
return self.get_single_spot_value(
assets, field, data_frequency)
def get_single_spot_value(self, asset, field, data_frequency):
symbol = self._get_v2_symbol(asset)
log.debug(
'fetching spot value {field} for symbol {symbol}'.format(
symbol=symbol,
field=field
)
)
if data_frequency == 'minute':
# TODO: support all available frequencies
start_dt = None
if data_frequency == 'minute' or data_frequency == '1m':
frequency = '1m'
elif data_frequency == 'daily':
if bar_count and end_dt:
start_dt = end_dt - timedelta(minutes=bar_count)
elif data_frequency == 'daily' or data_frequency == '1d':
frequency = '1D'
if bar_count and end_dt:
start_dt = end_dt - timedelta(days=bar_count)
else:
raise NotImplementedError(
'Unsupported frequency %s' % data_frequency
raise InvalidHistoryFrequencyError(
frequency=data_frequency
)
response = requests.get(
'{url}/v2/candles/trade:{frequency}:{symbol}/last'.format(
# Making sure that assets are iterable
asset_list = [assets] if isinstance(assets, Asset) else assets
ohlc_list = dict()
for asset in asset_list:
symbol = self._get_v2_symbol(asset)
url = '{url}/v2/candles/trade:{frequency}:{symbol}'.format(
url=self.url,
frequency=frequency,
symbol=symbol
)
)
candles = response.json()
if 'message' in candles:
raise ValueError(
'Unable to retrieve candles: %s' % candles['message']
)
if start_dt and end_dt:
is_list = True
url += '/hist?start={start}&end={end}'.format(
start=time.mktime(start_dt.timetuple()) * 1000,
end=time.mktime(end_dt.timetuple()) * 1000,
)
else:
is_list = False
url += '/last'
ohlc = dict(
open=candles[1],
high=candles[3],
low=candles[4],
close=candles[2],
volume=candles[5],
price=candles[2],
last_traded=pd.Timestamp.utcfromtimestamp(candles[0] / 1000.0),
)
response = requests.get(url)
candles = response.json()
if field not in ohlc:
raise KeyError('Invalid column: %s' % field)
if 'message' in candles:
raise ExchangeRequestError(
error='Unable to retrieve candles: {}'.format(
candles['message'])
)
def ohlc_from_candle(candle):
return dict(
open=candle[1],
high=candle[3],
low=candle[4],
close=candle[2],
volume=candle[5],
price=candle[2],
last_traded=pd.Timestamp.utcfromtimestamp(
candle[0] / 1000.0),
)
if is_list:
ohlc_bars = []
for candle in candles:
ohlc = ohlc_from_candle(candle)
ohlc_bars.append(ohlc)
ohlc_list[asset] = ohlc_bars
else:
ohlc = ohlc_from_candle(candles)
ohlc_list[asset] = ohlc
return ohlc_list[assets] \
if isinstance(assets, Asset) else ohlc_list
return ohlc[field]
def order(self, asset, amount, limit_price, stop_price, style):
"""Place an order.
@@ -460,9 +449,9 @@ class Bitfinex(Exchange):
response = self._request('order/new', req)
exchange_order = response.json()
if 'message' in exchange_order:
raise ValueError(
'unable to create Bitfinex order %s' % exchange_order[
'message']
raise ExchangeRequestError(
error='unable to create Bitfinex order {}'.format(
exchange_order['message'])
)
order_id = exchange_order['id']
@@ -501,9 +490,9 @@ class Bitfinex(Exchange):
response = self._request('orders', None)
order_statuses = response.json()
if 'message' in order_statuses:
raise ValueError(
'Unable to retrieve open orders: %s' % order_statuses[
'message']
raise ExchangeRequestError(
error='Unable to retrieve open orders: {}'.format(
order_statuses['message'])
)
orders = list()
@@ -533,8 +522,9 @@ class Bitfinex(Exchange):
order_status = response.json()
if 'message' in order_status:
raise ValueError(
'Unable to retrieve order status: %s' % order_status['message']
raise ExchangeRequestError(
error='Unable to retrieve order status: {}'.format(
order_status['message'])
)
return self._create_order(order_status)
@@ -552,8 +542,9 @@ class Bitfinex(Exchange):
response = self._request('order/cancel', {'order_id': order_id})
status = response.json()
if 'message' in status:
raise ValueError(
'Unable to cancel order: %s %s' % (order_id, status['message'])
raise ExchangeRequestError(
error='Unable to cancel order: {} {}'.format(
order_id, status['message'])
)
def tickers(self, assets):
@@ -576,8 +567,9 @@ class Bitfinex(Exchange):
tickers = request.json()
if 'message' in tickers:
raise ValueError(
'Unable to retrieve tickers: %s' % tickers['message']
raise ExchangeRequestError(
error='Unable to retrieve tickers: {}'.format(
tickers['message'])
)
formatted_tickers = []
catalyst/exchange/data_portal_exchange.py
+2 -7
View File
@@ -33,7 +33,7 @@ class DataPortalExchange(DataPortal):
field,
data_frequency,
ffill=True):
history_window = super(self.__class__, self).get_history_window(
return self.exchange.get_history_window(
assets,
end_dt,
bar_count,
@@ -42,12 +42,6 @@ class DataPortalExchange(DataPortal):
data_frequency,
ffill)
# The returned dataframe contains today's value as a NaN because
# end_dt points to the current wall clock. We drop today's
# value to be in sync with the simulation's behavior.
today = pd.Timestamp.utcnow()
return history_window[history_window.index.date != today]
def get_spot_value(self, assets, field, dt, data_frequency):
return self.exchange.get_spot_value(assets, field, dt, data_frequency)
@@ -55,4 +49,5 @@ class DataPortalExchange(DataPortal):
perspective_dt,
data_frequency,
spot_value=None):
# TODO: does this pertain to cryptocurrencies?
raise NotImplementedError("get_adjusted_value is not implemented yet!")
catalyst/exchange/exchange.py
+137
View File
@@ -1,11 +1,13 @@
import abc
import json
from abc import ABCMeta, abstractmethod, abstractproperty
import collections
import pandas as pd
from catalyst.assets._assets import Asset
from catalyst.finance.order import ORDER_STATUS
from catalyst.finance.transaction import Transaction
from catalyst.data.data_portal import BASE_FIELDS
from catalyst.errors import (
MultipleSymbolsFound,
@@ -265,6 +267,141 @@ class Exchange:
"""
pass
@abstractmethod
def get_candles(self, data_frequency, assets,
end_dt=None, bar_count=None, limit=None):
"""
Retrieve OHLC candles
"""
pass
def get_spot_value(self, assets, field, dt=None, data_frequency='minute'):
"""
Public API method that returns a scalar value representing the value
of the desired asset's field at either the given dt.
Parameters
----------
assets : Asset, ContinuousFuture, or iterable of same.
The asset or assets whose data is desired.
field : {'open', 'high', 'low', 'close', 'volume',
'price', 'last_traded'}
The desired field of the asset.
dt : pd.Timestamp
The timestamp for the desired value.
data_frequency : str
The frequency of the data to query; i.e. whether the data is
'daily' or 'minute' bars
Returns
-------
value : float, int, or pd.Timestamp
The spot value of ``field`` for ``asset`` The return type is based
on the ``field`` requested. If the field is one of 'open', 'high',
'low', 'close', or 'price', the value will be a float. If the
``field`` is 'volume' the value will be a int. If the ``field`` is
'last_traded' the value will be a Timestamp.
Bitfinex timeframes
-------------------
Available values: '1m', '5m', '15m', '30m', '1h', '3h', '6h', '12h',
'1D', '7D', '14D', '1M'
"""
if field not in BASE_FIELDS:
raise KeyError('Invalid column: ' + str(field))
if isinstance(assets, collections.Iterable):
values = list()
for asset in assets:
value = self.get_single_spot_value(
asset, field, data_frequency)
values.append(value)
return values
else:
return self.get_single_spot_value(
assets, field, data_frequency)
def get_single_spot_value(self, asset, field, data_frequency):
log.debug(
'fetching spot value {field} for symbol {symbol}'.format(
symbol=asset.symbol,
field=field
)
)
ohlc = self.get_candles(data_frequency, asset)
if field not in ohlc:
raise KeyError('Invalid column: %s' % field)
return ohlc[field]
def get_history_window(self,
assets,
end_dt,
bar_count,
frequency,
fields,
data_frequency,
ffill=True):
"""
:param assets:
:param end_dt:
:param bar_count:
:param frequency:
:param fields:
:param data_frequency:
:param ffill:
:return df:
If a single security and a single field were passed into data.history,
a pandas Series is returned, indexed by date.
If multiple securities and single field are passed in, the returned
pandas DataFrame is indexed by date, and has assets as columns.
If a single security and multiple fields are passed in, the returned
pandas DataFrame is indexed by date, and has fields as columns.
If multiple assets and multiple fields are passed in, the returned
pandas Panel is indexed by field, has date as the major axis, and
securities as the minor axis.
"""
candles = self.get_candles(
data_frequency=frequency,
assets=assets,
bar_count=bar_count,
end_dt=end_dt
)
def get_single_field_series(candles):
return pd.Series(
map(lambda candle: candle[fields], candles),
index=map(lambda candle: candle['last_traded'], candles)
)
df = None
if len(assets) == 1:
if type(fields) is str:
asset = assets[0]
df = get_single_field_series(candles[asset])
else:
raise NotImplementedError()
else:
if type(fields) is str:
series = []
for asset in assets:
item = get_single_field_series(candles[asset])
series.append(item)
df = pd.concat(series, axis=1)
else:
raise NotImplementedError()
return df
@abc.abstractmethod
def tickers(self, date, pairs):
return
catalyst/exchange/exchange_errors.py
+13
View File
@@ -0,0 +1,13 @@
from catalyst.errors import ZiplineError
class ExchangeRequestError(ZiplineError):
msg = (
'Request failed: {error}'
).strip()
class InvalidHistoryFrequencyError(ZiplineError):
msg = (
'History frequency {frequency} not supported by the exchange.'
).strip()
catalyst/exchange/exchange_utils.py
+9
View File
@@ -0,0 +1,9 @@
import os
from catalyst.utils.paths import data_root
def get_exchange_folder(environ=None):
if not environ:
environ = os.environ
root = data_root(environ)