diff --git a/catalyst/examples/buy_low_sell_high_neo_with_interface.py b/catalyst/examples/buy_low_sell_high_neo_with_interface.py new file mode 100644 index 00000000..c453786b --- /dev/null +++ b/catalyst/examples/buy_low_sell_high_neo_with_interface.py @@ -0,0 +1,170 @@ +import talib +from logbook import Logger +import pandas as pd + +from catalyst.api import ( + order, + order_target_percent, + symbol, + record, + get_open_orders, +) +from catalyst.exchange.stats_utils import get_pretty_stats +from catalyst.utils.run_algo import run_algorithm + +algo_namespace = 'buy_low_sell_high_neo' +log = Logger(algo_namespace) + + +def initialize(context): + log.info('initializing algo') + context.asset = symbol('neo_btc', 'bitfinex') + + context.TARGET_POSITIONS = 50 + context.PROFIT_TARGET = 0.1 + context.SLIPPAGE_ALLOWED = 0.02 + + context.retry_check_open_orders = 10 + context.retry_update_portfolio = 10 + context.retry_order = 5 + + context.errors = [] + pass + + +def _handle_data(context, data): + prices = data.history( + context.asset, + fields='price', + bar_count=20, + frequency='30m' + ) + rsi = talib.RSI(prices.values, timeperiod=14)[-1] + log.info('got rsi: {}'.format(rsi)) + + # Buying more when RSI is low, this should lower our cost basis + if rsi <= 30: + buy_increment = 1 + elif rsi <= 40: + buy_increment = 0.5 + elif rsi <= 70: + buy_increment = 0.1 + else: + buy_increment = None + + cash = context.portfolio.cash + log.info('base currency available: {cash}'.format(cash=cash)) + + price = data.current(context.asset, 'close') + log.info('got price {price}'.format(price=price)) + + if price is None: + log.warn('no pricing data') + return + + record(price=price, rsi=rsi) + + orders = get_open_orders(context.asset) + if orders: + log.info('skipping bar until all open orders execute') + return + + is_buy = False + cost_basis = None + if context.asset in context.portfolio.positions: + position = context.portfolio.positions[context.asset] + + cost_basis = position.cost_basis + log.info( + 'found {amount} positions with cost basis {cost_basis}'.format( + amount=position.amount, + cost_basis=cost_basis + ) + ) + + if position.amount >= context.TARGET_POSITIONS: + log.info('reached positions target: {}'.format(position.amount)) + return + + if price < cost_basis: + is_buy = True + elif position.amount > 0 and \ + price > cost_basis * (1 + context.PROFIT_TARGET): + profit = (price * position.amount) - (cost_basis * position.amount) + log.info('closing position, taking profit: {}'.format(profit)) + order_target_percent( + asset=context.asset, + target=0, + limit_price=price * (1 - context.SLIPPAGE_ALLOWED), + ) + else: + log.info('no buy or sell opportunity found') + else: + is_buy = True + + if is_buy: + if buy_increment is None: + log.info('the rsi is too high to consider buying {}'.format(rsi)) + return + + if price * buy_increment > cash: + log.info('not enough base currency to consider buying') + return + + log.info( + 'buying position cheaper than cost basis {} < {}'.format( + price, + cost_basis + ) + ) + order( + asset=context.asset, + amount=buy_increment, + limit_price=price * (1 + context.SLIPPAGE_ALLOWED) + ) + + +def handle_data(context, data): + log.info('handling bar {}'.format(data.current_dt)) + # try: + _handle_data(context, data) + # except Exception as e: + # log.warn('aborting the bar on error {}'.format(e)) + # context.errors.append(e) + + log.info('completed bar {}, total execution errors {}'.format( + data.current_dt, + len(context.errors) + )) + + if len(context.errors) > 0: + log.info('the errors:\n{}'.format(context.errors)) + + +def analyze(context, stats): + log.info('the daily stats:\n{}'.format(get_pretty_stats(stats))) + pass + + +# run_algorithm( +# initialize=initialize, +# handle_data=handle_data, +# analyze=analyze, +# exchange_name='bittrex,bitfinex', +# live=True, +# algo_namespace=algo_namespace, +# base_currency='eth', +# live_graph=True +# ) +run_algorithm( + capital_base=10000, + start=pd.to_datetime('2017-09-10', utc=True), + end=pd.to_datetime('2017-09-15', utc=True), + data_frequency='minute', + initialize=initialize, + handle_data=handle_data, + analyze=analyze, + exchange_name='bitfinex', + algo_namespace=algo_namespace, + base_currency='btc' +) diff --git a/catalyst/exchange/exchange_algorithm.py b/catalyst/exchange/exchange_algorithm.py index d20bed49..2d902572 100644 --- a/catalyst/exchange/exchange_algorithm.py +++ b/catalyst/exchange/exchange_algorithm.py @@ -63,6 +63,16 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): super(ExchangeTradingAlgorithmBase, self).__init__(*args, **kwargs) + def round_order(self, amount): + """ + We need fractions with cryptocurrencies + + :param amount: + :return: + """ + # TODO: is this good enough? Victor has a better solution. + return amount + @api_method @preprocess(symbol_str=ensure_upper_case) def symbol(self, symbol_str, exchange_name=None): @@ -595,16 +605,6 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): amount, asset.symbol, asset.exchange)) return None - def round_order(self, amount): - """ - We need fractions with cryptocurrencies - - :param amount: - :return: - """ - # TODO: is this good enough? Victor has a better solution. - return amount - @api_method def batch_market_order(self, share_counts): raise NotImplementedError() diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 147c4d9d..056a6421 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -30,44 +30,34 @@ def fetch_candles_chunk(exchange, assets, data_frequency, end_dt, bar_count): start_dt=calc_start_dt, end_dt=end_dt ) + return candles - series = dict() - - for asset in assets: - asset_candles = candles[asset] - - candle_start_dt = None - candle_end_dt = None - for candle in asset_candles: - last_traded = candle['last_traded'] - - if candle_start_dt is None or candle_start_dt > last_traded: - candle_start_dt = last_traded - - if candle_end_dt is None or candle_end_dt < last_traded: - candle_end_dt = last_traded - - if candle_end_dt < end_dt: - asset_candles.append( - dict( - open=None, - high=None, - close=None, - low=None, - volume=None, - last_traded=end_dt - ) - ) - - asset_df = pd.DataFrame(asset_candles) - if not asset_df.empty: - asset_df.set_index('last_traded', inplace=True, drop=True) - asset_df.sort_index(inplace=True) - asset_df = asset_df.resample('1T').ffill() - - series[asset] = asset_df - - return series + # series = dict() + # + # for asset in assets: + # asset_candles = candles[asset] + # + # candle_start_dt = None + # candle_end_dt = None + # for candle in asset_candles: + # last_traded = candle['last_traded'] + # + # if candle_start_dt is None or candle_start_dt > last_traded: + # candle_start_dt = last_traded + # + # if candle_end_dt is None or candle_end_dt < last_traded: + # candle_end_dt = last_traded + # + # + # asset_df = pd.DataFrame(asset_candles) + # if not asset_df.empty: + # asset_df.set_index('last_traded', inplace=True, drop=True) + # asset_df.sort_index(inplace=True) + # asset_df = asset_df.resample('1T').ffill() + # + # series[asset] = asset_df + # + # return series def process_bar_data(exchange, assets, writer, data_frequency, @@ -121,47 +111,73 @@ def process_bar_data(exchange, assets, writer, data_frequency, frequency=data_frequency )) as it: + previous_candle = dict() for chunk in it: - assets_candles_dict = fetch_candles_chunk( + chunk_end = chunk['end'] + chunk_start = chunk_end - timedelta(minutes=chunk['bar_count']) + + candles = fetch_candles_chunk( exchange=exchange, assets=assets, data_frequency=frequency, - end_dt=chunk['end'], + end_dt=chunk_end, bar_count=chunk['bar_count'] ) log.debug('requests counter {}'.format(exchange.request_cpt)) - if not assets_candles_dict.keys(): - log.debug( - 'no data: {symbols} on {exchange}, date {end}'.format( - symbols=assets, - exchange=exchange.name, - end=chunk['end'] - ) - ) - continue - num_candles = 0 data = [] - for asset in assets_candles_dict: - df = assets_candles_dict[asset] - sid = asset.sid + for asset in candles: + asset_candles = candles[asset] + if not asset_candles: + log.debug( + 'no data: {symbols} on {exchange}, date {end}'.format( + symbols=assets, + exchange=exchange.name, + end=chunk_end + ) + ) + continue - num_candles += len(df.values) - data.append((sid, df)) + all_dates = [] + all_candles = [] + date = chunk_start + while date <= chunk_end: + + previous = previous_candle[asset] \ + if asset in previous_candle else None + + candle = next((candle for candle in asset_candles \ + if candle['last_traded'] == date), previous) + + if candle is not None: + all_dates.append(date) + all_candles.append(candle) + + previous_candle[asset] = candle + + date += timedelta(minutes=1) + + df = pd.DataFrame(all_candles, index=all_dates) + if not df.empty: + df.sort_index(inplace=True) + + sid = asset.sid + num_candles += len(df.values) + + data.append((sid, df)) try: - log.info( + log.debug( 'writing {num_candles} candles from {start} to {end}'.format( num_candles=num_candles, - start=chunk['end'] - \ - timedelta(minutes=chunk['bar_count']), - end=chunk['end'] + start=chunk_start, + end=chunk_end ) ) for pair in data: - log.info('data for sid {}\n{}\n{}'.format( + log.debug('data for sid {}\n{}\n{}'.format( pair[0], pair[1].head(2), pair[1].tail(2))) writer.write( diff --git a/tests/exchange/test_data_portal.py b/tests/exchange/test_data_portal.py index 87357c2b..25d123cc 100644 --- a/tests/exchange/test_data_portal.py +++ b/tests/exchange/test_data_portal.py @@ -90,7 +90,7 @@ class ExchangeDataPortalTestCase: asset_finder.lookup_symbol('neo_btc', self.bitfinex), ] - date = pd.to_datetime('2017-09-10 9:00', utc=True) + date = pd.to_datetime('2017-09-10', utc=True) value = self.data_portal_backtest.get_spot_value( assets, 'close', date, 'minute') pass