diff --git a/catalyst/exchange/ccxt/ccxt_exchange.py b/catalyst/exchange/ccxt/ccxt_exchange.py index bd655761..0a4b8950 100644 --- a/catalyst/exchange/ccxt/ccxt_exchange.py +++ b/catalyst/exchange/ccxt/ccxt_exchange.py @@ -583,29 +583,62 @@ class CCXT(Exchange): The Catalyst order object """ - if order_status['status'] == 'canceled' \ - or (order_status['status'] == 'closed' - and order_status['filled'] == 0): + order_id = order_status['id'] + symbol = self.get_symbol(order_status['symbol'], source='ccxt') + asset = self.get_asset(symbol) + + s = order_status['status'] + amount = order_status['amount'] + filled = order_status['filled'] + + if s == 'canceled' or (s == 'closed' and filled == 0): status = ORDER_STATUS.CANCELLED - elif order_status['status'] == 'closed' and order_status['filled'] > 0: - log.debug('found executed order {}'.format(order_status)) + elif s == 'closed' and filled > 0: + if filled < amount: + log.warn( + 'order {id} is executed but only partially filled:' + ' {filled} {symbol} out of {amount}'.format( + id=order_status['status'], + filled=order_status['filled'], + symbol=asset.symbol, + amount=order_status['amount'], + ) + ) + else: + log.info( + 'order {id} executed in full: {filled} {symbol}'.format( + id=order_id, + filled=filled, + symbol=asset.symbol, + ) + ) + status = ORDER_STATUS.FILLED - elif order_status['status'] == 'open': + elif s == 'open': + status = ORDER_STATUS.OPEN + + elif filled > 0: + log.info( + 'order {id} partially filled: {filled} {symbol} out of ' + '{amount}, waiting for complete execution'.format( + id=order_id, + filled=filled, + symbol=asset.symbol, + amount=amount, + ) + ) status = ORDER_STATUS.OPEN else: log.warn( 'invalid state {} for order {}'.format( - order_status['status'], order_status['id'] + s, order_id ) ) status = ORDER_STATUS.OPEN - amount = order_status['amount'] - filled = order_status['filled'] - if order_status['side'] == 'sell': amount = -amount filled = -filled @@ -614,21 +647,16 @@ class CCXT(Exchange): order_type = order_status['type'] limit_price = price if order_type == 'limit' else None - stop_price = None # TODO: add support executed_price = order_status['cost'] / order_status['amount'] commission = order_status['fee'] date = from_ms_timestamp(order_status['timestamp']) - # order_id = str(order_status['info']['clientOrderId']) - order_id = order_status['id'] - - symbol = self.get_symbol(order_status['symbol'], source='ccxt') order = Order( dt=date, - asset=self.get_asset(symbol), + asset=asset, amount=amount, - stop=stop_price, + stop=None, limit=limit_price, filled=filled, id=order_id, diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index a09d86a1..bb3c5768 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -653,15 +653,7 @@ class Exchange: return df def _check_low_balance(self, currency, balances, amount): - free = balances[currency]['free'] \ - if currency in balances else None - - if free is None or free == 0: - raise BalanceNotFoundError( - currency=currency, - exchange=self.name, - balances=balances, - ) + free = balances[currency]['free'] if currency in balances else 0.0 if free < amount: return free, True @@ -683,12 +675,15 @@ class Exchange: Check balances amounts against the exchange. """ - log.debug('synchronizing portfolio with exchange {}'.format(self.name)) - free_cash = 0.0 if check_balances: + log.debug('fetching {} balances'.format(self.name)) balances = self.get_balances() - + log.debug( + 'got free balances for {} currencies'.format( + len(balances) + ) + ) if cash is not None: free_cash, is_lower = self._check_low_balance( currency=self.base_currency, @@ -718,8 +713,12 @@ class Exchange: ticker = tickers[asset] log.debug( - 'updating {} position with ticker: {}'.format( - asset.symbol, ticker + 'updating {symbol} position, last traded on {dt} for ' + '{price}{currency}'.format( + symbol=asset.symbol, + dt=ticker['last_traded'], + price=ticker['last_price'], + currency=asset.quote_currency, ) ) position.last_sale_price = ticker['last_price'] diff --git a/catalyst/exchange/exchange_algorithm.py b/catalyst/exchange/exchange_algorithm.py index a9045a9d..01c16f24 100644 --- a/catalyst/exchange/exchange_algorithm.py +++ b/catalyst/exchange/exchange_algorithm.py @@ -42,14 +42,17 @@ from catalyst.exchange.live_graph_clock import LiveGraphClock from catalyst.exchange.simple_clock import SimpleClock from catalyst.exchange.stats_utils import get_pretty_stats, stats_to_s3, \ stats_to_algo_folder +from catalyst.exchange.utils.serialization import portfolio_to_dict from catalyst.finance.execution import MarketOrder from catalyst.finance.performance import PerformanceTracker from catalyst.finance.performance.period import calc_period_stats +from catalyst.protocol import Positions, Position from catalyst.gens.tradesimulation import AlgorithmSimulator from catalyst.utils.api_support import api_method from catalyst.utils.input_validation import error_keywords, ensure_upper_case from catalyst.utils.math_utils import round_nearest from catalyst.utils.preprocess import preprocess +from catalyst.protocol import Portfolio log = logbook.Logger('exchange_algorithm', level=LOG_LEVEL) @@ -435,24 +438,47 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): def _create_generator(self, sim_params): if self.perf_tracker is None: - self.perf_tracker = PerformanceTracker( + tracker = self.perf_tracker = PerformanceTracker( sim_params=self.sim_params, trading_calendar=self.trading_calendar, env=self.trading_environment, ) - # Unpacking the perf_tracker and positions if available perf = get_algo_object( algo_name=self.algo_namespace, key='perf_tracker', ) if perf is not None: - positions = get_algo_object( + # Unpack the position and converting dict or object + p = get_algo_object( algo_name=self.algo_namespace, - key='positions', + key='portfolio', + how='json', ) - self.perf_tracker.period_start = perf['period_start'] - self.perf_tracker.position_tracker.positions = positions + portfolio = Portfolio() + portfolio.capital_used = p['capital_used'] + portfolio.starting_cash = p['starting_cash'] + portfolio.portfolio_value = p['portfolio_value'] + portfolio.pnl = p['pnl'] + portfolio.returns = p['returns'] + portfolio.cash = p['cash'] + portfolio.start_date = p['start_date'] + portfolio.positions_value = p['positions_value'] + + portfolio.positions = positions = Positions() + for p in p['positions']: + exchange = self.exchanges[p['exchange']] + asset = exchange.get_asset(p['symbol']) + positions[asset] = Position( + asset=asset, + amount=p['amount'], + cost_basis=p['cost_basis'], + last_sale_price=p['last_sale_price'], + last_sale_date=None, + ) + + tracker.period_start = perf['period_start'] + tracker.position_tracker.positions = portfolio.positions # Call the simulation trading algorithm for side-effects: # it creates the perf tracker @@ -675,20 +701,18 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): except Exception as e: log.warn('unable to calculate performance: {}'.format(e)) - # TODO: pickle does not seem to work in python 3 - # try: save_algo_object( algo_name=self.algo_namespace, key='perf_tracker', obj=self.perf_tracker.to_dict(emission_type=self.data_frequency), ) + portfolio = portfolio_to_dict(self.portfolio) save_algo_object( algo_name=self.algo_namespace, - key='positions', - obj=self.perf_tracker.position_tracker.positions, + key='portfolio', + obj=portfolio, + how='json', ) - # except Exception as e: - # log.warn('unable to save perf_tracker to disk: {}'.format(e)) self.current_day = data.current_dt.floor('1D') diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index da341237..81c302f6 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -14,6 +14,8 @@ from six.moves.urllib import request from catalyst.constants import DATE_FORMAT, SYMBOLS_URL from catalyst.exchange.exchange_errors import ExchangeSymbolsNotFound, \ InvalidHistoryFrequencyError, InvalidHistoryFrequencyAlias +from catalyst.exchange.utils.serialization import ExchangeJSONEncoder, \ + ExchangeJSONDecoder from catalyst.utils.paths import data_root, ensure_directory, \ last_modified_time @@ -108,20 +110,6 @@ def download_exchange_symbols(exchange_name, environ=None): return response -def symbols_parser(asset_def): - for key, value in asset_def.items(): - match = isinstance(value, string_types) \ - and re.search(r'(\d{4}-\d{2}-\d{2})', value) - - if match: - try: - asset_def[key] = pd.to_datetime(value, utc=True) - except ValueError: - pass - - return asset_def - - def get_exchange_symbols(exchange_name, is_local=False, environ=None): """ The de-serialized content of the exchange's symbols.json. @@ -147,7 +135,7 @@ def get_exchange_symbols(exchange_name, is_local=False, environ=None): if os.path.isfile(filename): with open(filename) as data_file: try: - data = json.load(data_file, object_hook=symbols_parser) + data = json.load(data_file, cls=ExchangeJSONDecoder) return data except ValueError: @@ -273,7 +261,7 @@ def get_algo_folder(algo_name, environ=None): return algo_folder -def get_algo_object(algo_name, key, environ=None, rel_path=None): +def get_algo_object(algo_name, key, environ=None, rel_path=None, how='pickle'): """ The de-serialized object of the algo name and key. @@ -297,14 +285,19 @@ def get_algo_object(algo_name, key, environ=None, rel_path=None): if rel_path is not None: folder = os.path.join(folder, rel_path) - filename = os.path.join(folder, key + '.p') + name = '{}.p'.format(key) if how == 'pickle' else '{}.json'.format(key) + filename = os.path.join(folder, name) if os.path.isfile(filename): - try: + if how == 'pickle': with open(filename, 'rb') as handle: return pickle.load(handle) - except Exception: - return None + + else: + with open(filename) as data_file: + data = json.load(data_file, cls=ExchangeJSONDecoder) + return data + else: return None @@ -332,7 +325,7 @@ def save_algo_object(algo_name, key, obj, environ=None, rel_path=None, if how == 'json': filename = os.path.join(folder, '{}.json'.format(key)) with open(filename, 'wt') as handle: - json.dump(obj, handle, indent=4, default=symbols_serial) + json.dump(obj, handle, indent=4, cls=ExchangeJSONEncoder) else: filename = os.path.join(folder, '{}.p'.format(key)) diff --git a/catalyst/exchange/utils/__init__.py b/catalyst/exchange/utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/catalyst/exchange/utils/serialization.py b/catalyst/exchange/utils/serialization.py new file mode 100644 index 00000000..fe7f734b --- /dev/null +++ b/catalyst/exchange/utils/serialization.py @@ -0,0 +1,68 @@ +import json +from json import JSONEncoder + +import pandas as pd +import re + +from six import string_types + +from catalyst.constants import DATE_TIME_FORMAT + + +class ExchangeJSONEncoder(json.JSONEncoder): + def default(self, obj): + if isinstance(obj, pd.Timestamp): + return obj.strftime(DATE_TIME_FORMAT) + + # Let the base class default method raise the TypeError + return JSONEncoder.default(self, obj) + + +class ExchangeJSONDecoder(json.JSONDecoder): + def __init__(self, *args, **kwargs): + json.JSONDecoder.__init__( + self, object_hook=self.object_hook, *args, **kwargs + ) + + def recursive_iter(self, obj): + if isinstance(obj, dict): + for key, value in obj.items(): + match = isinstance(value, string_types) and re.search( + r'(\d{4}-\d{2}-\d{2}).*', value + ) + if match: + try: + obj[key] = pd.to_datetime(value, utc=True) + except ValueError: + pass + + elif any(isinstance(obj, t) for t in (list, tuple)): + for item in obj: + self.recursive_iter(item) + + def object_hook(self, obj): + self.recursive_iter(obj) + return obj + + +def portfolio_to_dict(portfolio): + positions = portfolio.positions + for asset in portfolio.positions: + position = portfolio.positions[asset].to_dict() + + position['symbol'] = asset.symbol + position['exchange'] = asset.exchange + del position['sid'] + + positions.append(position) + + portfolio_dict = vars(portfolio) + del portfolio_dict['positions'] + + portfolio_dict['positions'] = positions + return portfolio_dict + + +def portfolio_from_dict(self, portfolio_data): + from catalyst.protocol import Portfolio + return Portfolio()