wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-13 12:00:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

Working on multi-exchange implementation (not fully tested)

Browse Source
This commit is contained in:
fredfortier
2017-09-07 23:54:11 -04:00
parent 8850657f26
commit 36881b03e2
5 changed files with 226 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/exchange/algorithm_exchange.py
+101 -33
View File
@@ -23,6 +23,7 @@ import numpy as np
import logbook
import pandas as pd
from catalyst.utils.preprocess import preprocess
import catalyst.protocol as zp
from catalyst.algorithm import TradingAlgorithm
@@ -45,7 +46,7 @@ from catalyst.gens.tradesimulation import AlgorithmSimulator
from catalyst.utils.api_support import (
api_method,
disallowed_in_before_trading_start)
from catalyst.utils.input_validation import error_keywords
from catalyst.utils.input_validation import error_keywords, ensure_upper_case
log = logbook.Logger("ExchangeTradingAlgorithm")
@@ -57,7 +58,7 @@ class ExchangeAlgorithmExecutor(AlgorithmSimulator):
class ExchangeTradingAlgorithm(TradingAlgorithm):
def __init__(self, *args, **kwargs):
self.exchange = kwargs.pop('exchange', None)
self.exchanges = kwargs.pop('exchanges', None)
self.algo_namespace = kwargs.pop('algo_namespace', None)
self.live_graph = kwargs.pop('live_graph', None)
@@ -83,6 +84,7 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
self.stats_minutes = 5
super(self.__class__, self).__init__(*args, **kwargs)
# TODO: fix precision before re-enabling
# self._create_minute_writer()
signal.signal(signal.SIGINT, self.signal_handler)
@@ -97,6 +99,7 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
writer = BcolzMinuteBarWriter.open(
root, self.sim_params.end_session)
else:
# TODO: need to be able to write more precise numbers
writer = BcolzMinuteBarWriter(
rootdir=root,
calendar=self.trading_calendar,
@@ -163,13 +166,11 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
if self.live_graph:
self._clock = LiveGraphClock(
self.sim_params.sessions,
time_skew=self.exchange.time_skew,
context=self
)
else:
self._clock = SimpleClock(
self.sim_params.sessions,
time_skew=self.exchange.time_skew
)
return self._clock
@@ -202,27 +203,31 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
portfolio directly.
:return:
"""
return self.exchange.portfolio
# TODO: build cumulative portfolio
return self.perf_tracker.get_portfolio(False)
def updated_account(self):
return self.exchange.account
return self.perf_tracker.get_account(False)
def _synchronize_portfolio(self, attempt_index=0):
try:
self.exchange.synchronize_portfolio()
for exchange_name in self.exchanges:
exchange = self.exchanges[exchange_name]
# Applying the updated last_sales_price to the positions
# in the performance tracker. This seems a bit redundant
# but it will make sense when we have multiple exchange portfolios
# feeding into the same performance tracker.
tracker = self.perf_tracker.todays_performance.position_tracker
for asset in self.exchange.portfolio.positions:
position = self.exchange.portfolio.positions[asset]
tracker.update_position(
asset=asset,
last_sale_date=position.last_sale_date,
last_sale_price=position.last_sale_price
)
exchange.synchronize_portfolio()
# Applying the updated last_sales_price to the positions
# in the performance tracker. This seems a bit redundant
# but it will make sense when we have multiple exchange portfolios
# feeding into the same performance tracker.
tracker = self.perf_tracker.todays_performance.position_tracker
for asset in exchange.portfolio.positions:
position = exchange.portfolio.positions[asset]
tracker.update_position(
asset=asset,
last_sale_date=position.last_sale_date,
last_sale_price=position.last_sale_price
)
except ExchangeRequestError as e:
log.warn(
'update portfolio attempt {}: {}'.format(attempt_index, e)
@@ -239,7 +244,14 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
def _check_open_orders(self, attempt_index=0):
try:
return self.exchange.check_open_orders()
orders = list()
for exchange_name in self.exchanges:
exchange = self.exchanges[exchange_name]
exchange_orders = exchange.check_open_orders()
orders += exchange_orders
return orders
except ExchangeRequestError as e:
log.warn(
'check open orders attempt {}: {}'.format(attempt_index, e)
@@ -429,11 +441,13 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
log.warn('unable to save minute perfs to disk: {}'.format(e))
try:
save_algo_object(
algo_name=self.algo_namespace,
key='portfolio_{}'.format(self.exchange.name),
obj=self.exchange.portfolio
)
for exchange_name in self.exchanges:
exchange = self.exchanges[exchange_name]
save_algo_object(
algo_name=self.algo_namespace,
key='portfolio_{}'.format(exchange_name),
obj=exchange.portfolio
)
except Exception as e:
log.warn('unable to save portfolio to disk: {}'.format(e))
@@ -445,9 +459,10 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
style=None,
attempt_index=0):
try:
return self.exchange.order(asset, amount, limit_price,
stop_price,
style)
exchange = self.exchanges[asset.exchange]
return exchange.order(asset, amount, limit_price,
stop_price,
style)
except ExchangeRequestError as e:
log.warn(
'order attempt {}: {}'.format(attempt_index, e)
@@ -500,7 +515,18 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
def _get_open_orders(self, asset=None, attempt_index=0):
try:
return self.exchange.get_open_orders(asset)
if asset:
exchange = self.exchanges[asset.exchange]
return exchange.get_open_orders(asset)
else:
open_orders = []
for exchange_name in self.exchanges:
exchange = self.exchanges[exchange_name]
exchange_orders = exchange.get_open_orders()
open_orders.append(exchange_orders)
return open_orders
except ExchangeRequestError as e:
log.warn(
'open orders attempt {}: {}'.format(attempt_index, e)
@@ -522,12 +548,54 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
return self._get_open_orders(asset)
@api_method
def get_order(self, order_id):
return self.exchange.get_order(order_id)
def get_order(self, order_id, exchange_name):
exchange = self.exchanges[exchange_name]
return exchange.get_order(order_id)
@api_method
def cancel_order(self, order_param):
def cancel_order(self, order_param, exchange_name):
exchange = self.exchanges[exchange_name]
order_id = order_param
if isinstance(order_param, zp.Order):
order_id = order_param.id
self.exchange.cancel_order(order_id)
exchange.cancel_order(order_id)
@api_method
@preprocess(symbol_str=ensure_upper_case)
def symbol(self, symbol_str, exchange_name=None):
"""Lookup an Equity by its ticker symbol.
Parameters
----------
symbol_str : str
The ticker symbol for the equity to lookup.
Returns
-------
equity : Equity
The equity that held the ticker symbol on the current
symbol lookup date.
Raises
------
SymbolNotFound
Raised when the symbols was not held on the current lookup date.
See Also
--------
:func:`catalyst.api.set_symbol_lookup_date`
"""
# If the user has not set the symbol lookup date,
# use the end_session as the date for sybmol->sid resolution.
_lookup_date = self._symbol_lookup_date \
if self._symbol_lookup_date is not None \
else self.sim_params.end_session
exchange = self.exchanges[exchange_name]
return self.asset_finder.lookup_symbol(
symbol_str,
as_of_date=_lookup_date,
exchange=exchange
)
catalyst/exchange/asset_finder_exchange.py
+9 -8
View File
@@ -4,8 +4,7 @@ log = Logger('AssetFinderExchange')
class AssetFinderExchange(object):
def __init__(self, exchange):
self.exchange = exchange
def __init__(self):
self._asset_cache = {}
@property
@@ -47,7 +46,8 @@ class AssetFinderExchange(object):
log.info('fetching asset: {}'.format(sid))
return list()
def lookup_symbol(self, symbol, as_of_date, fuzzy=False):
def lookup_symbol(self, symbol, as_of_date, exchange,
fuzzy=False):
"""Lookup an asset by symbol.
Parameters
@@ -81,11 +81,12 @@ class AssetFinderExchange(object):
there are multiple candidates for the given ``symbol`` on the
``as_of_date``.
"""
log.debug('looking up symbol: {}'.format(symbol))
log.debug('looking up symbol: {} {}'.format(symbol, exchange.name))
if symbol in self._asset_cache:
return self._asset_cache[symbol]
key = ','.join([exchange.name, symbol])
if key in self._asset_cache:
return self._asset_cache[key]
else:
asset = self.exchange.get_asset(symbol)
self._asset_cache[symbol] = asset
asset = exchange.get_asset(symbol)
self._asset_cache[key] = asset
return asset
catalyst/exchange/data_portal_exchange.py
+65 -13
View File
@@ -12,7 +12,8 @@
# limitations under the License.
from time import sleep
import pandas as pd
from catalyst.assets._assets import TradingPair
from logbook import Logger
from catalyst.data.data_portal import DataPortal
@@ -25,8 +26,8 @@ log = Logger('DataPortalExchange')
class DataPortalExchange(DataPortal):
def __init__(self, exchange, *args, **kwargs):
self.exchange = exchange
def __init__(self, exchanges, *args, **kwargs):
self.exchanges = exchanges
# TODO: put somewhere accessible by each algo
self.retry_get_history_window = 5
@@ -45,14 +46,43 @@ class DataPortalExchange(DataPortal):
ffill=True,
attempt_index=0):
try:
return self.exchange.get_history_window(
assets,
end_dt,
bar_count,
frequency,
field,
data_frequency,
ffill)
exchange_assets = dict()
for asset in assets:
if asset.exchange not in exchange_assets:
exchange_assets[asset.exchange] = list()
exchange_assets[asset.exchange].append(asset)
if len(exchange_assets) > 1:
df_list = []
for exchange_name in exchange_assets:
exchange = self.exchanges[exchange_name]
assets = exchange_assets[exchange_name]
df = exchange.get_history_window(
assets,
end_dt,
bar_count,
frequency,
field,
data_frequency,
ffill)
df_list.append(df)
return pd.concat(df_list)
else:
exchange = self.exchanges[exchange_assets.keys()[0]]
return exchange.get_history_window(
assets,
end_dt,
bar_count,
frequency,
field,
data_frequency,
ffill)
except ExchangeRequestError as e:
log.warn(
'get history attempt {}: {}'.format(attempt_index, e)
@@ -93,8 +123,30 @@ class DataPortalExchange(DataPortal):
def _get_spot_value(self, assets, field, dt, data_frequency,
attempt_index=0):
try:
return self.exchange.get_spot_value(assets, field, dt,
data_frequency)
if isinstance(assets, TradingPair):
exchange = self.exchanges[assets.exchange]
return exchange.get_spot_value(
assets, field, dt, data_frequency)
else:
exchange_assets = dict()
for asset in assets:
if asset.exchange not in exchange_assets:
exchange_assets[asset.exchange] = list()
exchange_assets[asset.exchange].append(asset)
spot_values = []
for exchange_name in exchange_assets:
exchange = self.exchanges[exchange_name]
assets = exchange_assets[exchange_name]
exchange_spot_values = exchange.get_spot_value(
assets, field, dt, data_frequency)
spot_values += exchange_spot_values
return spot_values
except ExchangeRequestError as e:
log.warn(
'get spot value attempt {}: {}'.format(attempt_index, e)
catalyst/exchange/exchange_errors.py
+7
View File
@@ -34,6 +34,13 @@ class ExchangeTransactionError(ZiplineError):
).strip()
class ExchangeNotFoundError(ZiplineError):
msg = (
'Exchange {exchange_name} not found. Please specify exchanges '
'supported by Catalyst and verify spelling for accuracy.'
).strip()
class ExchangeAuthNotFound(ZiplineError):
msg = (
'Please create an auth.json file containing the api token and key for '
catalyst/utils/run_algo.py
+44 -33
View File
@@ -45,7 +45,7 @@ from catalyst.exchange.exchange_portfolio import ExchangePortfolio
from catalyst.exchange.exchange_errors import (
ExchangeRequestError,
ExchangeRequestErrorTooManyAttempts,
BaseCurrencyNotFoundError)
BaseCurrencyNotFoundError, ExchangeNotFoundError)
from catalyst.exchange.exchange_utils import get_exchange_auth, \
get_algo_object
from logbook import Logger
@@ -150,37 +150,42 @@ def _run(handle_data,
if live and exchange is not None:
exchange_name = exchange
start = pd.Timestamp.utcnow()
end = start + timedelta(minutes=1439)
exchange_list = [x.strip().lower() for x in exchange.split(',')]
portfolio = get_algo_object(
algo_name=algo_namespace,
key='portfolio_{}'.format(exchange_name),
environ=environ
)
if portfolio is None:
portfolio = ExchangePortfolio(
start_date=pd.Timestamp.utcnow()
exchanges = dict()
for exchange_name in exchange_list:
portfolio = get_algo_object(
algo_name=algo_namespace,
key='portfolio_{}'.format(exchange_name),
environ=environ
)
exchange_auth = get_exchange_auth(exchange_name)
if exchange_name == 'bitfinex':
exchange = Bitfinex(
key=exchange_auth['key'],
secret=exchange_auth['secret'],
base_currency=base_currency,
portfolio=portfolio
)
elif exchange_name == 'bittrex':
exchange = Bittrex(
key=exchange_auth['key'],
secret=exchange_auth['secret'],
base_currency=base_currency,
portfolio=portfolio
)
else:
raise NotImplementedError(
'exchange not supported: %s' % exchange_name)
if portfolio is None:
portfolio = ExchangePortfolio(
start_date=pd.Timestamp.utcnow()
)
exchange_auth = get_exchange_auth(exchange_name)
if exchange_name == 'bitfinex':
exchanges[exchange_name] = Bitfinex(
key=exchange_auth['key'],
secret=exchange_auth['secret'],
base_currency=base_currency,
portfolio=portfolio
)
elif exchange_name == 'bittrex':
exchanges[exchange_name] = Bittrex(
key=exchange_auth['key'],
secret=exchange_auth['secret'],
base_currency=base_currency,
portfolio=portfolio
)
else:
raise ExchangeNotFoundError(exchange_name=exchange_name)
open_calendar = get_calendar('OPEN')
sim_params = create_simulation_parameters(
@@ -197,23 +202,24 @@ def _run(handle_data,
exchange_tz='UTC',
asset_db_path=None
)
env.asset_finder = AssetFinderExchange(exchange)
env.asset_finder = AssetFinderExchange()
data = DataPortalExchange(
exchange=exchange,
exchanges=exchanges,
asset_finder=env.asset_finder,
trading_calendar=open_calendar,
first_trading_day=pd.to_datetime('today', utc=True)
)
choose_loader = None
def fetch_capital_base(attempt_index=0):
def fetch_capital_base(exchange, attempt_index=0):
"""
Fetch the base currency amount required to bootstrap
the algorithm against the exchange.
The algorithm cannot continue without this value.
:param exchange: the targeted exchange
:param attempt_index:
:return capital_base: the amount of base currency available for
trading
@@ -240,10 +246,15 @@ def _run(handle_data,
exchange=exchange_name
)
capital_base = 0
for exchange_name in exchanges:
exchange = exchanges[exchange_name]
capital_base += fetch_capital_base(exchange)
sim_params = create_simulation_parameters(
start=start,
end=end,
capital_base=fetch_capital_base(),
capital_base=capital_base,
emission_rate='minute',
data_frequency='minute'
)
@@ -339,9 +350,9 @@ def _run(handle_data,
choose_loader = None
TradingAlgorithmClass = (
partial(ExchangeTradingAlgorithm, exchange=exchange,
partial(ExchangeTradingAlgorithm, exchanges=exchanges,
algo_namespace=algo_namespace, live_graph=live_graph)
if live and exchange else TradingAlgorithm)
if live and exchanges else TradingAlgorithm)
perf = TradingAlgorithmClass(
namespace=namespace,