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

Implemented and tested the history() method in backtest mode.

Browse Source
This commit is contained in:
fredfortier
2017-09-22 23:17:38 -04:00
parent ddecd6bb48
commit bc65c10fc6
5 changed files with 167 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/examples/buy_low_sell_high_neo_with_interface.py
+19 -20
View File
@@ -20,7 +20,7 @@ def initialize(context):
log.info('initializing algo')
context.asset = symbol('neo_btc', 'bitfinex')
context.TARGET_POSITIONS = 50
context.TARGET_POSITIONS = 50000
context.PROFIT_TARGET = 0.1
context.SLIPPAGE_ALLOWED = 0.02
@@ -33,25 +33,24 @@ def initialize(context):
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))
prices = data.history(
context.asset,
fields='price',
bar_count=30,
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
buy_increment = 0.1
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))
@@ -159,8 +158,8 @@ def analyze(context, stats):
# live_graph=True
# )
run_algorithm(
capital_base=10000,
start=pd.to_datetime('2017-09-10', utc=True),
capital_base=250,
start=pd.to_datetime('2017-09-08', utc=True),
end=pd.to_datetime('2017-09-15', utc=True),
data_frequency='minute',
initialize=initialize,
catalyst/exchange/data_portal_exchange.py
+93 -22
View File
@@ -28,7 +28,8 @@ from catalyst.data.us_equity_pricing import BcolzDailyBarReader
from catalyst.exchange.exchange_errors import (
ExchangeRequestError,
ExchangeBarDataError,
BundleNotFoundError)
BundleNotFoundError, PricingDataBeforeTradingError,
PricingDataNotLoadedError)
from catalyst.utils.paths import data_path
log = Logger('DataPortalExchange')
@@ -85,7 +86,8 @@ class DataPortalExchangeBase(DataPortal):
else:
exchange = self.exchanges[exchange_assets.keys()[0]]
return exchange.get_history_window(
return self.get_exchange_history_window(
exchange,
assets,
end_dt,
bar_count,
@@ -123,6 +125,10 @@ class DataPortalExchangeBase(DataPortal):
field,
data_frequency=None,
ffill=True):
if field == 'price':
field = 'close'
return self._get_history_window(assets,
end_dt,
bar_count,
@@ -251,12 +257,14 @@ class DataPortalExchangeLive(DataPortalExchangeBase):
class DataPortalExchangeBacktest(DataPortalExchangeBase):
def __init__(self, *args, **kwargs):
super(DataPortalExchangeBacktest, self).__init__(*args, **kwargs)
self.daily_bar_readers = dict()
self.minute_bar_readers = dict()
self.five_minute_bar_readers = dict()
self.history_loaders = dict()
self.minute_history_loaders = dict()
for exchange_name in self.exchanges:
name = 'exchange_{}'.format(exchange_name)
time_folder = \
@@ -289,6 +297,13 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
except IOError:
self.minute_bar_readers[exchange_name] = None
def _get_first_trading_day(self, assets):
first_date = None
for asset in assets:
if first_date is None or asset.start_date > first_date:
first_date = asset.start_date
return first_date
@staticmethod
def find_most_recent_time(bundle_name):
try:
@@ -320,30 +335,76 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
field,
data_frequency,
ffill=True):
# TODO: implement in the bundle
df = exchange.get_history_window(
assets,
end_dt,
bar_count,
frequency,
field,
data_frequency,
ffill)
if data_frequency == 'minute' or data_frequency == '1m':
reader = self.minute_bar_readers[exchange.name]
dts = self.trading_calendar.minutes_window(
end_dt, -bar_count
)
self.ensure_after_first_day(dts[0], assets)
elif data_frequency == '5-minute' or data_frequency == '5m':
reader = self.five_minute_bar_readers[exchange.name]
elif data_frequency == 'daily' or data_frequency == '1d':
reader = self.daily_bar_readers[exchange.name]
session = self.trading_calendar.minute_to_session_label(end_dt)
dts = self._get_days_for_window(session, bar_count)
self.ensure_after_first_day(dts[0], assets)
else:
raise ValueError('Unsupported frequency')
try:
values = reader.load_raw_arrays(
[field],
dts[0],
dts[-1],
assets,
)[0]
except Exception:
raise PricingDataNotLoadedError(
field=field,
first_trading_day=self._get_first_trading_day(assets),
exchange=exchange.name,
symbols=[asset.symbol for asset in assets],
)
series = dict()
for index, asset in enumerate(assets):
asset_values = []
for value in values:
asset_values.append(value[index])
value_series = pd.Series(asset_values, index=dts)
series[asset] = value_series
df = pd.DataFrame(series)
return df
def ensure_after_first_day(self, dt, assets):
first_trading_day = self._get_first_trading_day(assets)
if dt < first_trading_day:
raise PricingDataBeforeTradingError(
first_trading_day=first_trading_day,
exchange=assets[0].exchange,
symbols=[asset.symbol for asset in assets],
)
def get_exchange_spot_value(self, exchange, assets, field, dt,
data_frequency):
if data_frequency == 'minute':
if data_frequency == 'minute' or data_frequency == '1m':
reader = self.minute_bar_readers[exchange.name]
elif data_frequency == '5-minute':
elif data_frequency == '5-minute' or data_frequency == '5m':
reader = self.five_minute_bar_readers[exchange.name]
elif data_frequency == 'daily':
elif data_frequency == 'daily' or data_frequency == '1d':
reader = self.daily_bar_readers[exchange.name]
else:
raise ValueError('Unsupported frequency')
if isinstance(assets, TradingPair):
self.ensure_after_first_day(dt, [assets])
try:
value = reader.get_value(
sid=assets.sid,
@@ -351,10 +412,16 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
field=field
)
return value
except Exception as e:
log.warn('minute data not found: {}'.format(e))
return None
except Exception:
raise PricingDataNotLoadedError(
field=field,
first_trading_day=self._get_first_trading_day([assets]),
exchange=exchange.name,
symbols=assets.symbol,
)
else:
self.ensure_after_first_day(dt, assets)
values = []
for asset in assets:
try:
@@ -364,8 +431,12 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
field=field
)
values.append(value)
except Exception as e:
log.warn('minute data not found: {}'.format(e))
values.append(None)
except Exception:
raise PricingDataNotLoadedError(
field=field,
first_trading_day=self._get_first_trading_day(assets),
exchange=exchange.name,
symbols=[asset.symbol for asset in assets],
)
return values
catalyst/exchange/exchange_blotter.py
+22 -37
View File
@@ -1,18 +1,16 @@
from catalyst.assets._assets import TradingPair
from logbook import Logger
from catalyst.finance.blotter import Blotter
from catalyst.finance.commission import PerShare, CommissionModel
from catalyst.finance.slippage import VolumeShareSlippage, SlippageModel, \
LiquidityExceeded
from catalyst.assets._assets import TradingPair
# It seems like we need to accept greate slippage risk in cryptos
# Orders won't often close at Equity levels.
# TODO: consider adjusting dynamically based on trading pair
from catalyst.finance.commission import CommissionModel
from catalyst.finance.slippage import SlippageModel
from catalyst.finance.transaction import Transaction
log = Logger('exchange_blotter')
# It seems like we need to accept greater slippage risk in cryptos
# Orders won't often close at Equity levels.
# TODO: consider adjusting dynamically based on trading pair
DEFAULT_SLIPPAGE_SPREAD = 0.02
DEFAULT_MAKER_FEE = 0.001
DEFAULT_TAKER_FEE = 0.002
@@ -37,8 +35,7 @@ class TradingPairFeeSchedule(CommissionModel):
def __repr__(self):
return (
'{class_name}(maker_fee={maker_fee}, '
'taker_fee={taker_fee})'
.format(
'taker_fee={taker_fee})'.format(
class_name=self.__class__.__name__,
maker_fee=self.maker_fee,
taker_fee=self.taker_fee,
@@ -83,44 +80,32 @@ class TradingPairFixedSlippage(SlippageModel):
def simulate(self, data, asset, orders_for_asset):
self._volume_for_bar = 0
volume = data.current(asset, "volume")
if volume == 0:
return
price = data.current(asset, 'close')
# can use the close price, since we verified there's volume in this
# bar.
price = data.current(asset, "close")
dt = data.current_dt
for order in orders_for_asset:
if order.open_amount == 0:
continue
order.check_triggers(price, dt)
if not order.triggered:
log.debug('order has not reached the trigger at current '
'price {}'.format(price))
continue
transaction = None
try:
execution_price, execution_volume = \
self.process_order(data, order)
execution_price, execution_volume = self.process_order(data, order)
if execution_price is not None:
transaction = Transaction(
asset=order.asset,
amount=abs(execution_volume),
dt=data.current_dt,
price=execution_price,
order_id=order.id
)
transaction = Transaction(
asset=order.asset,
amount=abs(execution_volume),
dt=dt,
price=execution_price,
order_id=order.id
)
except LiquidityExceeded:
break
if transaction:
self._volume_for_bar += abs(transaction.amount)
yield order, transaction
self._volume_for_bar += abs(transaction.amount)
yield order, transaction
def process_order(self, data, order):
price = data.current(order.asset, 'close')
@@ -130,11 +115,11 @@ class TradingPairFixedSlippage(SlippageModel):
adj_price = price * (1 + self.spread)
else:
# Sell order
adj_price = price & (1 - self.spread)
adj_price = price * (1 - self.spread)
log.debug('added slippage to price: {} => {}'.format(price, adj_price))
return (adj_price, order.amount)
return adj_price, order.amount
class ExchangeBlotter(Blotter):
catalyst/exchange/exchange_errors.py
+14
View File
@@ -147,3 +147,17 @@ class BundleNotFoundError(ZiplineError):
'Please ingest data using the command '
'`catalyst ingest -b exchange_{exchange}`. '
'See catalyst documentation for details.').strip()
class PricingDataBeforeTradingError(ZiplineError):
msg = ('Pricing data for trading pairs {symbols} on exchange {exchange} '
'starts on {first_trading_day}.').strip()
class PricingDataNotLoadedError(ZiplineError):
msg = ('Pricing data {field} for trading pairs {symbols} trading on '
'exchange {exchange} since {first_trading_day} is unavailable. '
'The bundle data is either out-of-date or has not been loaded yet.'
'Please ingest data using the command '
'`catalyst ingest -b exchange_{exchange}`. '
'See catalyst documentation for details.').strip()
tests/exchange/test_data_portal.py
+19 -6
View File
@@ -50,11 +50,10 @@ class ExchangeDataPortalTestCase:
exchanges=dict(bitfinex=self.bitfinex),
asset_finder=asset_finder,
trading_calendar=open_calendar,
first_trading_day=pd.to_datetime('2017-09-10', utc=True)
first_trading_day=None # will set dynamically based on assets
)
def test_get_history_window_live(self):
asset_finder = self.data_portal_live.asset_finder
assets = [
@@ -82,8 +81,24 @@ class ExchangeDataPortalTestCase:
assets, 'price', now, '1m')
pass
def test_get_spot_value_backtest(self):
def test_get_history_window_backtest(self):
asset_finder = self.data_portal_live.asset_finder
assets = [
asset_finder.lookup_symbol('neo_btc', self.bitfinex),
]
date = pd.to_datetime('2017-09-10', utc=True)
data = self.data_portal_backtest.get_history_window(
assets,
date,
10,
'1m',
'close',
'minute')
pass
def test_get_spot_value_backtest(self):
asset_finder = self.data_portal_backtest.asset_finder
assets = [
@@ -92,8 +107,6 @@ class ExchangeDataPortalTestCase:
date = pd.to_datetime('2017-09-10', utc=True)
value = self.data_portal_backtest.get_spot_value(
assets, 'close', date, 'minute')
assets[0], 'close', date, 'minute')
pass
def test_get_history_window_backtest(self):
pass