diff --git a/catalyst/exchange/algorithm_exchange.py b/catalyst/exchange/algorithm_exchange.py index a9e9bbc0..a7198227 100644 --- a/catalyst/exchange/algorithm_exchange.py +++ b/catalyst/exchange/algorithm_exchange.py @@ -34,6 +34,7 @@ from catalyst.exchange.exchange_errors import ( ExchangeTransactionError ) from catalyst.finance.performance.period import calc_period_stats +from catalyst.exchange.exchange_utils import save_algo_object, get_algo_object log = logbook.Logger("ExchangeTradingAlgorithm") @@ -46,8 +47,9 @@ class ExchangeAlgorithmExecutor(AlgorithmSimulator): class ExchangeTradingAlgorithm(TradingAlgorithm): def __init__(self, *args, **kwargs): self.exchange = kwargs.pop('exchange', None) + self.algo_namespace = kwargs.pop('algo_namespace', None) self.orders = {} - self.minute_perfs = [] + self.minute_perfs = None self.is_running = True self.retry_check_open_orders = 5 @@ -254,6 +256,13 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): if not self.is_running: return + if self.minute_perfs is None: + perfs = get_algo_object( + algo_name=self.algo_namespace, + key='minute_perfs' + ) + self.minute_perfs = perfs if perfs is not None else [] + self._update_portfolio() transactions = self._check_open_orders() @@ -282,6 +291,24 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): except Exception as e: log.warn('unable to calculate performance: {}'.format(e)) + try: + save_algo_object( + algo_name=self.algo_namespace, + key='minute_perfs', + obj=self.minute_perfs + ) + except Exception as e: + log.warn('unable to save minute perfs to disk: {}'.format(e)) + + try: + save_algo_object( + algo_name=self.algo_namespace, + key='portfolio', + obj=self.exchange.portfolio + ) + except Exception as e: + log.warn('unable to save portfolio to disk: {}'.format(e)) + def _order(self, asset, amount, diff --git a/catalyst/exchange/bitfinex.py b/catalyst/exchange/bitfinex.py index e7812af4..211876e6 100644 --- a/catalyst/exchange/bitfinex.py +++ b/catalyst/exchange/bitfinex.py @@ -41,7 +41,7 @@ warning_logger = Logger('AlgoWarning') class Bitfinex(Exchange): - def __init__(self, key, secret, base_currency, store): + def __init__(self, key, secret, base_currency, portfolio=None): self.url = BITFINEX_URL self.key = key self.secret = secret @@ -50,7 +50,7 @@ class Bitfinex(Exchange): self.assets = {} self.load_assets(get_exchange_symbols(self.name)) self.base_currency = base_currency - self.store = store + self._portfolio = portfolio def _request(self, operation, data, version='v1'): payload_object = { @@ -203,8 +203,10 @@ class Bitfinex(Exchange): error='Base currency %s not found in portfolio' % self.base_currency ) - portfolio = self.store.portfolio + portfolio = self._portfolio portfolio.cash = float(base_position['available']) + if portfolio.starting_cash is None: + portfolio.starting_cash = portfolio.cash if portfolio.positions: assets = portfolio.positions.keys() @@ -228,19 +230,18 @@ class Bitfinex(Exchange): :return: """ - if self.store.portfolio is None: - portfolio = ExchangePortfolio( - store=self.store, - start_date=pd.Timestamp.utcnow() - ) - self.store.portfolio = portfolio - self.update_portfolio() + # if self._portfolio is None: + # portfolio = ExchangePortfolio( + # start_date=pd.Timestamp.utcnow() + # ) + # self.store.portfolio = portfolio + # self.update_portfolio() + # + # portfolio.starting_cash = portfolio.cash + # else: + # portfolio = self.store.portfolio - portfolio.starting_cash = portfolio.cash - else: - portfolio = self.store.portfolio - - return portfolio + return self._portfolio @property def account(self): diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py index 63eb13b2..14b06111 100644 --- a/catalyst/exchange/exchange_errors.py +++ b/catalyst/exchange/exchange_errors.py @@ -48,6 +48,12 @@ class ExchangeSymbolsNotFound(ZiplineError): ).strip() +class AlgoPickleNotFound(ZiplineError): + msg = ( + 'Pickle not found for algo {algo} in path {filename}' + ).strip() + + class InvalidHistoryFrequencyError(ZiplineError): msg = ( 'History frequency {frequency} not supported by the exchange.' diff --git a/catalyst/exchange/exchange_portfolio.py b/catalyst/exchange/exchange_portfolio.py index 192a7ebb..3f6e9899 100644 --- a/catalyst/exchange/exchange_portfolio.py +++ b/catalyst/exchange/exchange_portfolio.py @@ -5,44 +5,9 @@ from logbook import Logger log = Logger('ExchangePortfolio') -class PortfolioMemoryStore(object): - def __init__(self, algo_namespace): - self.algo_namespace = algo_namespace - self._portfolio = None - - @property - def portfolio(self): - """ - This is a mock store, the portfolio will always be None initially. - The goal is to retrieve a persisted portfolio using the - algo_namespace attribute so the algorithm can resume. - - :return: - """ - if self._portfolio is not None: - return self._portfolio - else: - return None - - @portfolio.setter - def portfolio(self, portfolio): - self._portfolio = portfolio - self.commit() - - def commit(self): - """ - The goal is to persist the portfolio somewhere so that the - algo can resume if it stops during execution. - - :return: - """ - log.debug('persisting updated portfolio') - - class ExchangePortfolio(Portfolio): - def __init__(self, store, start_date, starting_cash=0.0): + def __init__(self, start_date, starting_cash=None): self.capital_used = 0.0 - self.store = store self.starting_cash = starting_cash self.portfolio_value = starting_cash self.pnl = 0.0 @@ -56,9 +21,6 @@ class ExchangePortfolio(Portfolio): def calculate_pnl(self): log.debug('calculating pnl') - def update(self): - self.store.commit() - def create_order(self, order): log.debug('creating order {}'.format(order.id)) self.open_orders[order.id] = order @@ -72,7 +34,6 @@ class ExchangePortfolio(Portfolio): order_position.amount += order.amount log.debug('open order added to portfolio') - self.update() def execute_order(self, order): log.debug('executing order {}'.format(order.id)) @@ -98,7 +59,6 @@ class ExchangePortfolio(Portfolio): order_position.cost_basis = order.executed_price log.debug('updated portfolio with executed order') - self.update() def remove_order(self, order): log.info('removing cancelled order {}'.format(order.id)) @@ -115,4 +75,3 @@ class ExchangePortfolio(Portfolio): order_position.amount -= order.amount log.debug('removed order from portfolio') - self.update() diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index 9ebe637b..d6a69802 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -1,9 +1,10 @@ import os import urllib import json +import pickle from catalyst.utils.paths import data_root, ensure_directory from catalyst.exchange.exchange_errors import ExchangeAuthNotFound, \ - ExchangeSymbolsNotFound + ExchangeSymbolsNotFound, AlgoPickleNotFound SYMBOLS_URL = 'https://raw.githubusercontent.com/enigmampc/catalyst/' \ 'live-trading/catalyst/exchange/symbols/{exchange}.json' @@ -60,3 +61,36 @@ def get_exchange_auth(exchange_name, environ=None): exchange=exchange_name, filename=filename ) + + +def get_algo_folder(algo_name, environ=None): + if not environ: + environ = os.environ + + root = data_root(environ) + algo_folder = os.path.join(root, 'live_algos', algo_name) + ensure_directory(algo_folder) + + return algo_folder + + +def get_algo_object(algo_name, key, environ=None): + algo_folder = get_algo_folder(algo_name, environ) + filename = os.path.join(algo_folder, key + '.p') + + if os.path.isfile(filename): + try: + with open(filename, 'rb') as handle: + return pickle.load(handle) + except Exception as e: + return None + else: + return None + + +def save_algo_object(algo_name, key, obj, environ=None): + algo_folder = get_algo_folder(algo_name, environ) + filename = os.path.join(algo_folder, key + '.p') + + with open(filename, 'wb') as handle: + pickle.dump(obj, handle, protocol=pickle.HIGHEST_PROTOCOL) diff --git a/catalyst/utils/run_algo.py b/catalyst/utils/run_algo.py index 082f3c54..de6b06b5 100644 --- a/catalyst/utils/run_algo.py +++ b/catalyst/utils/run_algo.py @@ -39,13 +39,13 @@ from catalyst.exchange.algorithm_exchange import ExchangeTradingAlgorithm from catalyst.exchange.data_portal_exchange import DataPortalExchange from catalyst.exchange.bitfinex import Bitfinex from catalyst.exchange.asset_finder_exchange import AssetFinderExchange -from catalyst.exchange.exchange_portfolio import PortfolioMemoryStore +from catalyst.exchange.exchange_portfolio import ExchangePortfolio from catalyst.exchange.exchange_errors import ( ExchangeRequestError, ExchangeRequestErrorTooManyAttempts ) from catalyst.exchange.exchange_utils import get_exchange_auth, \ - get_exchange_folder + get_algo_object from logbook import Logger log = Logger('run_algo') @@ -254,7 +254,7 @@ def _run(handle_data, ) choose_loader = None - def fetch_portfolio(attempt_index=0): + def update_portfolio(attempt_index=0): """ Fetch the portfolio for the exchange We can't continue on error because it is required to bootstrap @@ -263,18 +263,19 @@ def _run(handle_data, :return: """ try: + exchange.update_portfolio() return exchange.portfolio except ExchangeRequestError as e: if attempt_index < 20: sleep(5) - return fetch_portfolio(attempt_index + 1) + return update_portfolio(attempt_index + 1) else: raise ExchangeRequestErrorTooManyAttempts( attempts=attempt_index, error=e ) - portfolio = fetch_portfolio() + portfolio = update_portfolio() sim_params = create_simulation_parameters( start=start, end=end, @@ -287,7 +288,8 @@ def _run(handle_data, choose_loader = None TradingAlgorithmClass = ( - partial(ExchangeTradingAlgorithm, exchange=exchange) + partial(ExchangeTradingAlgorithm, exchange=exchange, + algo_namespace=algo_namespace) if exchange else TradingAlgorithm) perf = TradingAlgorithmClass( @@ -484,14 +486,23 @@ def run_algorithm(initialize, ) else: if exchange_name is not None: - store = PortfolioMemoryStore(algo_namespace) + portfolio = get_algo_object( + algo_name=algo_namespace, + key='portfolio', + environ=environ + ) + if portfolio is None: + portfolio = ExchangePortfolio( + start_date=pd.Timestamp.utcnow() + ) + exchange_auth = get_exchange_auth(exchange_name) if exchange_name == 'bitfinex': exchange = Bitfinex( key=exchange_auth['key'], secret=exchange_auth['secret'].encode('UTF-8'), base_currency=base_currency, - store=store + portfolio=portfolio ) else: raise NotImplementedError(