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

BLD: improved serialization of portfolio data

Browse Source
This commit is contained in:
Frederic Fortier
2017-12-26 19:46:26 -05:00
parent 869ef8ec87
commit 3a7128ad4f
6 changed files with 176 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/exchange/ccxt/ccxt_exchange.py
+45 -17
View File
@@ -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,
catalyst/exchange/exchange.py
+13 -14
View File
@@ -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']
catalyst/exchange/exchange_algorithm.py
+36 -12
View File
@@ -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')
catalyst/exchange/exchange_utils.py
+14 -21
View File
@@ -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))
catalyst/exchange/utils/__init__.py
View File
catalyst/exchange/utils/serialization.py
+68
View File
@@ -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()