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

BLD: created a new sample algo for instructional materials. Fixed some minor issues in the process.

Browse Source
This commit is contained in:
fredfortier
2017-11-09 17:09:55 -05:00
parent a85b6c798a
commit 1103947af0
11 changed files with 708 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/__main__.py
+24 -2
View File
@@ -9,6 +9,7 @@ from six import text_type
from catalyst.data import bundles as bundles_module
from catalyst.exchange.exchange_bundle import ExchangeBundle
from catalyst.exchange.exchange_utils import delete_algo_folder
from catalyst.exchange.factory import get_exchange
from catalyst.utils.cli import Date, Timestamp
from catalyst.utils.run_algo import _run, load_extensions
@@ -495,8 +496,14 @@ def live(ctx,
default=False,
help='Show a progress indicator for every currency pair.'
)
@click.option(
'--validate/--no-validate`',
default=False,
help='Report potential anomalies found in data bundles.'
)
def ingest_exchange(exchange_name, data_frequency, start, end,
include_symbols, exclude_symbols, show_progress, verbose):
include_symbols, exclude_symbols, show_progress, verbose,
validate):
"""
Ingest data for the given exchange.
"""
@@ -515,10 +522,25 @@ def ingest_exchange(exchange_name, data_frequency, start, end,
start=start,
end=end,
show_progress=show_progress,
show_breakdown=verbose
show_breakdown=verbose,
show_report=validate
)
@main.command(name='clean-algo')
@click.option(
'-n',
'--algo-namespace',
help='The label of the algorithm to for which to clean the state.'
)
@click.pass_context
def clean_algo(ctx, algo_namespace):
click.echo(
'Deleting the state folder of algo: {}...'.format(algo_namespace)
)
delete_algo_folder(algo_namespace)
@main.command(name='clean-exchange')
@click.option(
'-x',
catalyst/constants.py
+3 -1
View File
@@ -2,4 +2,6 @@
import logbook
LOG_LEVEL = logbook.INFO
LOG_LEVEL = logbook.INFO
DATE_TIME_FORMAT = '%Y-%m-%d %H:%M'
catalyst/examples/momemtum.py
+194
View File
@@ -0,0 +1,194 @@
# For this example, we're going to write a simple momentum script. When the
# stock goes up quickly, we're going to buy; when it goes down quickly, we're
# going to sell. Hopefully we'll ride the waves.
import pandas as pd
# To run an algorithm in Catalyst, you need two functions: initialize and
# handle_data.
from logbook import Logger
from catalyst import run_algorithm
from catalyst.api import symbol, record, order_target_percent, \
get_open_orders
from catalyst.exchange import stats_utils
from catalyst.finance.execution import LimitOrder
# We give a name to the algorithm which Catalyst will use to persist its state.
# In this example, Catalyst will create the `.catalyst/data/live_algos`
# directory. If we stop and start the algorithm, Catalyst will resume its
# state using the files included in the folder.
algo_namespace = 'momentum'
log = Logger(algo_namespace)
def initialize(context):
# This initialize function sets any data or variables that you'll use in
# your algorithm. For instance, you'll want to define the trading pair (or
# trading pairs) you want to backtest. You'll also want to define any
# parameters or values you're going to use.
# In our example, we're looking at Ether in Bitcoin.
context.eth_btc = symbol('eth_usdt')
context.max_amount = 0.01
context.base_price = None
def handle_data(context, data):
# This handle_data function is where the real work is done. Our data is
# minute-level tick data, and each minute is called a frame. This function
# runs on each frame of the data.
# We're computing the volume-weighted-average-price of the security
# defined above, in the context.eth_btc variable. For this example, we're
# using three bars on the daily chart.
bars = data.history(
context.eth_btc,
fields=['close', 'volume'],
bar_count=3,
frequency='1D'
)
vwap = stats_utils.vwap(bars)
# We need a variable for the current price of the security to compare to
# the average.
current = data.current(context.eth_btc, fields=['close', 'volume'])
price = current['close']
log.info('{}: price: {}, vwap: {}'.format(data.current_dt, price, vwap))
# If base_price is not set, we use the current value. This is the
# price at the first bar which we reference to calculate price_change.
if context.base_price is None:
context.base_price = price
price_change = (price - context.base_price) / context.base_price
record(
price=price,
volume=current['volume'],
vwap=vwap,
price_change=price_change,
)
orders = get_open_orders(context.eth_btc)
if len(orders) > 0:
log.info('skipping bar until all open orders execute')
return
# Another powerful built-in feature of the Catalyst backtester is the
# portfolio object. The portfolio object tracks your positions, cash,
# cost basis of specific holdings, and more. In this line, we calculate
# how long or short our position is at this minute.
position_amount = context.portfolio.positions[context.eth_btc].amount
# This is the meat of the algorithm, placed in this if statement. If the
# price of the security is .5% less than the 3-day volume weighted average
# price AND we haven't reached our maximum short, then we call the order
# command and sell 100 shares. Similarly, if the stock is .5% higher than
# the 3-day average AND we haven't reached our maximum long, then we call
# the order command and buy 100 shares.
if price > vwap * 1.01 and position_amount < context.max_amount:
order_target_percent(
context.eth_btc, 1, style=LimitOrder(price * 1.02)
)
elif price < vwap * 0.995 and position_amount > 0:
order_target_percent(
context.eth_btc, 0, style=LimitOrder(price * 0.98)
)
def analyze(context=None, results=None):
import matplotlib.pyplot as plt
# The base currency of the algo exchange
base_currency = context.exchanges.values()[0].base_currency.upper()
# Plot the portfolio value over time.
ax1 = plt.subplot(611)
results.loc[:, 'portfolio_value'].plot(ax=ax1)
ax1.set_ylabel('Portfolio Value ({})'.format(base_currency))
# Plot the price increase or decrease over time.
ax2 = plt.subplot(612, sharex=ax1)
results.loc[:, 'price'].plot(ax=ax2)
ax2.set_ylabel('{asset} ({base})'.format(
asset=context.eth_btc.symbol, base=base_currency
))
# Compute indexes for buy and sell transactions
trans_list = results.transactions.values
all_trans = [t for sublist in trans_list for t in sublist]
all_trans.sort(key=lambda t: t['dt'])
# Transaction have an exact timestamp while stVats are daily.
# We adjust the time to the end of each period to place them on the graph.
for t in all_trans:
t['dt'] = t['dt'].replace(hour=23, minute=59)
buys = results.loc[[t['dt'] for t in all_trans if t['amount'] > 0], :]
sells = results.loc[[t['dt'] for t in all_trans if t['amount'] < 0], :]
ax2.plot(
buys.index,
results.loc[buys.index, 'price'],
'^',
markersize=10,
color='g',
)
ax2.plot(
sells.index,
results.loc[sells.index, 'price'],
'v',
markersize=10,
color='r',
)
ax4 = plt.subplot(613, sharex=ax1)
results.loc[:, ['starting_cash', 'cash']].plot(ax=ax4)
ax4.set_ylabel('Base Currency ({})'.format(base_currency))
results['algorithm'] = results.loc[:, 'algorithm_period_return']
ax5 = plt.subplot(614, sharex=ax1)
results.loc[:, ['algorithm', 'price_change']].plot(ax=ax5)
ax5.set_ylabel('Percent Change')
ax6 = plt.subplot(615, sharex=ax1)
results.loc[:, 'vwap'].plot(ax=ax6)
ax6.set_ylabel('VWAP')
ax6.plot(
buys.index,
results.loc[buys.index, 'vwap'],
'^',
markersize=10,
color='g',
)
ax6.plot(
sells.index,
results.loc[sells.index, 'vwap'],
'v',
markersize=10,
color='r',
)
plt.legend(loc=3)
# Show the plot.
plt.gcf().set_size_inches(18, 8)
plt.show()
pass
# Backtest
run_algorithm(
capital_base=1,
data_frequency='minute',
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='poloniex',
algo_namespace=algo_namespace,
base_currency='usdt',
start=pd.to_datetime('2017-5-15', utc=True),
end=pd.to_datetime('2017-5-20', utc=True),
)
catalyst/examples/rsi_profit_target.py
+276
View File
@@ -0,0 +1,276 @@
from datetime import timedelta
import pandas as pd
import numpy as np
import talib
from logbook import Logger
from catalyst.api import (
order,
symbol,
record,
get_open_orders,
)
from catalyst.exchange.stats_utils import crossover, crossunder
from catalyst.utils.run_algo import run_algorithm
algo_namespace = 'rsi'
log = Logger(algo_namespace)
def initialize(context):
log.info('initializing algo')
context.asset = symbol('eth_btc')
context.base_price = None
context.MAX_HOLDINGS = 0.2
context.RSI_OVERSOLD = 30
context.RSI_OVERSOLD_BBANDS = 45
context.RSI_OVERBOUGHT_BBANDS = 55
context.SLIPPAGE_ALLOWED = 0.03
context.TARGET = 0.15
context.STOP_LOSS = 0.1
context.STOP = 0.03
context.position = None
context.last_bar = None
context.errors = []
pass
def _handle_buy_sell_decision(context, data, signal, price):
orders = get_open_orders(context.asset)
if len(orders) > 0:
log.info('skipping bar until all open orders execute')
return
positions = context.portfolio.positions
if context.position is None and context.asset in positions:
position = positions[context.asset]
context.position = dict(
cost_basis=position['cost_basis'],
amount=position['amount'],
stop=None
)
action = None
if context.position is not None:
cost_basis = context.position['cost_basis']
amount = context.position['amount']
log.info(
'found {amount} positions with cost basis {cost_basis}'.format(
amount=amount,
cost_basis=cost_basis
)
)
stop = context.position['stop']
target = cost_basis * (1 + context.TARGET)
if price >= target:
context.position['cost_basis'] = price
context.position['stop'] = context.STOP
stop_target = context.STOP_LOSS if stop is None else context.STOP
if price < cost_basis * (1 - stop_target):
log.info('executing stop loss')
order(
asset=context.asset,
amount=-amount,
limit_price=price * (1 - context.SLIPPAGE_ALLOWED),
)
action = 0
context.position = None
else:
if signal == 'long':
log.info('opening position')
buy_amount = context.MAX_HOLDINGS / price
order(
asset=context.asset,
amount=buy_amount,
limit_price=price * (1 + context.SLIPPAGE_ALLOWED),
)
context.position = dict(
cost_basis=price,
amount=buy_amount,
stop=None
)
action = 0
def _handle_data_rsi_only(context, data):
price = data.current(context.asset, 'close')
log.info('got price {price}'.format(price=price))
if price is np.nan:
log.warn('no pricing data')
return
if context.base_price is None:
context.base_price = price
try:
prices = data.history(
context.asset,
fields='price',
bar_count=17,
frequency='30T'
)
except Exception as e:
log.warn('historical data not available: '.format(e))
return
rsi = talib.RSI(prices.values, timeperiod=16)[-1]
log.info('got rsi {}'.format(rsi))
signal = None
if rsi < context.RSI_OVERSOLD:
signal = 'long'
# Making sure that the price is still current
price = data.current(context.asset, 'close')
cash = context.portfolio.cash
log.info(
'base currency available: {cash}, cap: {cap}'.format(
cash=cash,
cap=context.MAX_HOLDINGS
)
)
volume = data.current(context.asset, 'volume')
price_change = (price - context.base_price) / context.base_price
record(
price=price,
price_change=price_change,
rsi=rsi,
volume=volume,
cash=cash,
starting_cash=context.portfolio.starting_cash,
leverage=context.account.leverage,
)
_handle_buy_sell_decision(context, data, signal, price)
def handle_data(context, data):
dt = data.current_dt
if context.last_bar is None or (
context.last_bar + timedelta(minutes=15)) <= dt:
context.last_bar = dt
else:
return
log.info('BAR {}'.format(dt))
try:
_handle_data_rsi_only(context, data)
except Exception as e:
log.warn('aborting the bar on error {}'.format(e))
context.errors.append(e)
if len(context.errors) > 0:
log.info('the errors:\n{}'.format(context.errors))
def analyze(context=None, results=None):
import matplotlib.pyplot as plt
base_currency = context.exchanges.values()[0].base_currency.upper()
# Plot the portfolio and asset data.
ax1 = plt.subplot(611)
results.loc[:, 'portfolio_value'].plot(ax=ax1)
ax1.set_ylabel('Portfolio Value ({})'.format(base_currency))
ax2 = plt.subplot(612, sharex=ax1)
results.loc[:, 'price'].plot(ax=ax2)
ax2.set_ylabel('{asset} ({base})'.format(
asset=context.asset.symbol, base=base_currency
))
trans = results.loc[[t != [] for t in results.transactions], :]
buys = trans.loc[[t[0]['amount'] > 0 for t in trans.transactions], :]
sells = trans.loc[[t[0]['amount'] < 0 for t in trans.transactions], :]
# buys = results.loc[results['action'] == 1, :]
# sells = results.loc[results['action'] == 0, :]
ax2.plot(
buys.index,
results.loc[buys.index, 'price'],
'^',
markersize=10,
color='g',
)
ax2.plot(
sells.index,
results.loc[sells.index, 'price'],
'v',
markersize=10,
color='r',
)
ax3 = plt.subplot(613, sharex=ax1)
results.loc[:, ['alpha', 'beta']].plot(ax=ax3)
ax3.set_ylabel('Alpha / Beta ')
ax4 = plt.subplot(614, sharex=ax1)
results.loc[:, ['starting_cash', 'cash']].plot(ax=ax4)
ax4.set_ylabel('Base Currency ({})'.format(base_currency))
results['algorithm'] = results.loc[:, 'algorithm_period_return']
ax5 = plt.subplot(615, sharex=ax1)
results.loc[:, ['algorithm', 'price_change']].plot(ax=ax5)
ax5.set_ylabel('Percent Change')
ax6 = plt.subplot(616, sharex=ax1)
results.loc[:, 'rsi'].plot(ax=ax6)
ax6.set_ylabel('RSI')
ax6.plot(
buys.index,
results.loc[buys.index, 'rsi'],
'^',
markersize=10,
color='g',
)
ax6.plot(
sells.index,
results.loc[sells.index, 'rsi'],
'v',
markersize=10,
color='r',
)
plt.legend(loc=3)
# Show the plot.
plt.gcf().set_size_inches(18, 8)
plt.show()
pass
run_algorithm(
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='bittrex',
live=True,
algo_namespace=algo_namespace,
base_currency='btc',
live_graph=False
)
# Backtest
# run_algorithm(
# capital_base=0.5,
# data_frequency='minute',
# initialize=initialize,
# handle_data=handle_data,
# analyze=analyze,
# exchange_name='poloniex',
# algo_namespace=algo_namespace,
# base_currency='btc',
# start=pd.to_datetime('2017-9-1', utc=True),
# end=pd.to_datetime('2017-10-1', utc=True),
# )
catalyst/examples/simple_loop.py
+5 -5
View File
@@ -7,7 +7,7 @@ from catalyst.api import symbol
def initialize(context):
print('initializing')
context.asset = symbol('eth_btc')
context.asset = symbol('swift_btc')
def handle_data(context, data):
@@ -20,8 +20,8 @@ def handle_data(context, data):
prices = data.history(
context.asset,
fields='price',
bar_count=16,
frequency='60T'
bar_count=15,
frequency='1D'
)
rsi = talib.RSI(prices.values, timeperiod=14)[-1]
print('got rsi: {}'.format(rsi))
@@ -31,13 +31,13 @@ def handle_data(context, data):
run_algorithm(
capital_base=250,
start=pd.to_datetime('2016-6-1', utc=True),
start=pd.to_datetime('2015-4-1', utc=True),
end=pd.to_datetime('2017-11-1', utc=True),
data_frequency='daily',
initialize=initialize,
handle_data=handle_data,
analyze=None,
exchange_name='bitfinex',
exchange_name='bittrex',
algo_namespace='simple_loop',
base_currency='btc'
)
catalyst/exchange/bundle_utils.py
+4 -1
View File
@@ -149,7 +149,7 @@ def get_periods(start_dt, end_dt, freq):
return len(get_periods_range(start_dt, end_dt, freq))
def get_start_dt(end_dt, bar_count, data_frequency):
def get_start_dt(end_dt, bar_count, data_frequency, include_first=True):
"""
The start date based on specified end date and data frequency.
@@ -168,6 +168,9 @@ def get_start_dt(end_dt, bar_count, data_frequency):
if periods > 1:
delta = get_delta(periods, data_frequency)
start_dt = end_dt - delta
if not include_first:
start_dt += get_delta(1, data_frequency)
else:
start_dt = end_dt
catalyst/exchange/exchange_bundle.py
+126 -83
View File
@@ -1,47 +1,24 @@
import os
import shutil
from functools import partial
from itertools import chain
from operator import is_not
import numpy as np
import pandas as pd
from catalyst.assets._assets import TradingPair
from datetime import datetime, timedelta
from logbook import Logger
from pandas.tslib import Timestamp
from pytz import UTC
from six import itervalues
from catalyst import get_calendar
from catalyst.constants import DATE_TIME_FORMAT
from catalyst.constants import LOG_LEVEL
from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
BcolzMinuteBarMetadata
from catalyst.exchange.bundle_utils import range_in_bundle, \
get_bcolz_chunk, get_delta, get_month_start_end, \
get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label
from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
BcolzExchangeBarWriter
from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
TempBundleNotFoundError, \
NoDataAvailableOnExchange, \
PricingDataNotLoadedError
from catalyst.exchange.exchange_utils import get_exchange_folder
from catalyst.utils.cli import maybe_show_progress
from catalyst.utils.paths import ensure_directory
import os
import shutil
from itertools import chain
import pandas as pd
from catalyst.assets._assets import TradingPair
from logbook import Logger
from pandas.tslib import Timestamp
from pytz import UTC
from six import itervalues
from catalyst import get_calendar
from catalyst.constants import LOG_LEVEL
from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
BcolzMinuteBarMetadata
from catalyst.exchange.bundle_utils import range_in_bundle, \
get_bcolz_chunk, get_delta, get_month_start_end, \
get_bcolz_chunk, get_month_start_end, \
get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label
from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
BcolzExchangeBarWriter
@@ -244,8 +221,91 @@ class ExchangeBundle:
if data_frequency == 'minute' \
else self.calendar.sessions_in_range(start_dt, end_dt)
def _spot_empty_periods(self, ohlcv_df, asset, data_frequency,
empty_rows_behavior):
problems = []
nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index
if len(nan_rows) > 0:
dates = []
for row_date in nan_rows.values:
row_date = pd.to_datetime(row_date, utc=True)
if row_date > asset.start_date:
dates.append(row_date)
if len(dates) > 0:
end_dt = asset.end_minute if data_frequency == 'minute' \
else asset.end_daily
problem = '{name} ({start_dt} to {end_dt}) has empty ' \
'periods: {dates}'.format(
name=asset.symbol,
start_dt=asset.start_date.strftime(DATE_TIME_FORMAT),
end_dt=end_dt.strftime(DATE_TIME_FORMAT),
dates=[date.strftime(DATE_TIME_FORMAT) for date in dates]
)
if empty_rows_behavior == 'warn':
log.warn(problem)
elif empty_rows_behavior == 'raise':
raise EmptyValuesInBundleError(
name=asset.symbol,
end_minute=end_dt,
dates=dates
)
else:
ohlcv_df.dropna(inplace=True)
else:
problem = None
problems.append(problem)
return problems
def _spot_duplicates(self, ohlcv_df, asset, data_frequency, threshold):
# TODO: work in progress
series = ohlcv_df.reset_index().groupby('close')['index'].apply(
np.array
)
ref_delta = timedelta(minutes=1) if data_frequency == 'minute' \
else timedelta(days=1)
dups = series.loc[lambda values: [len(x) > 10 for x in values]]
for index, dates in dups.iteritems():
prev_date = None
for date in dates:
if prev_date is not None:
delta = (date - prev_date) / 1e9
if delta == ref_delta.seconds:
log.info('pex')
prev_date = date
problems = []
for index, dates in dups.iteritems():
end_dt = asset.end_minute if data_frequency == 'minute' \
else asset.end_daily
problem = '{name} ({start_dt} to {end_dt}) has {threshold} ' \
'identical close values on: {dates}'.format(
name=asset.symbol,
start_dt=asset.start_date.strftime(DATE_TIME_FORMAT),
end_dt=end_dt.strftime(DATE_TIME_FORMAT),
threshold=threshold,
dates=[pd.to_datetime(date).strftime(DATE_TIME_FORMAT)
for date in dates]
)
problems.append(problem)
return problems
def ingest_df(self, ohlcv_df, data_frequency, asset, writer,
empty_rows_behavior='strip'):
empty_rows_behavior='warn', duplicates_threshold=None):
"""
Ingest a DataFrame of OHLCV data for a given market.
@@ -258,50 +318,16 @@ class ExchangeBundle:
empty_rows_behavior: str
"""
problems = []
if empty_rows_behavior is not 'ignore':
nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index
problems += self._spot_empty_periods(
ohlcv_df, asset, data_frequency, empty_rows_behavior
)
if len(nan_rows) > 0:
dates = []
previous_date = None
for row_date in nan_rows.values:
row_date = pd.to_datetime(row_date)
if previous_date is None:
dates.append(row_date)
else:
seq_date = previous_date + get_delta(1, data_frequency)
if row_date > seq_date:
dates.append(previous_date)
dates.append(row_date)
previous_date = row_date
dates.append(pd.to_datetime(nan_rows.values[-1]))
name = '{} from {} to {}'.format(
asset.symbol, ohlcv_df.index[0], ohlcv_df.index[-1]
)
if empty_rows_behavior == 'warn':
log.warn(
'\n{name} with end minute {end_minute} has empty rows '
'in ranges: {dates}'.format(
name=name,
end_minute=asset.end_minute,
dates=dates
)
)
elif empty_rows_behavior == 'raise':
raise EmptyValuesInBundleError(
name=name,
end_minute=asset.end_minute,
dates=dates
)
else:
ohlcv_df.dropna(inplace=True)
# if duplicates_threshold is not None:
# problems += self._spot_duplicates(
# ohlcv_df, asset, data_frequency, duplicates_threshold
# )
data = []
if not ohlcv_df.empty:
@@ -310,8 +336,11 @@ class ExchangeBundle:
self._write(data, writer, data_frequency)
return problems
def ingest_ctable(self, asset, data_frequency, period,
writer, empty_rows_behavior='strip', cleanup=False):
writer, empty_rows_behavior='strip',
duplicates_threshold=100, cleanup=False):
"""
Merge a ctable bundle chunk into the main bundle for the exchange.
@@ -327,8 +356,14 @@ class ExchangeBundle:
cleanup: bool
Remove the temp bundle directory after ingestion.
:return:
Returns
-------
list[str]
A list of problems which occurred during ingestion.
"""
problems = []
# Download and extract the bundle
path = get_bcolz_chunk(
exchange_name=self.exchange.name,
@@ -375,12 +410,13 @@ class ExchangeBundle:
start_dt, end_dt, data_frequency
)
df = get_df_from_arrays(arrays, periods)
self.ingest_df(
problems += self.ingest_df(
ohlcv_df=df,
data_frequency=data_frequency,
asset=asset,
writer=writer,
empty_rows_behavior=empty_rows_behavior
empty_rows_behavior=empty_rows_behavior,
duplicates_threshold=duplicates_threshold
)
if cleanup:
@@ -390,7 +426,7 @@ class ExchangeBundle:
)
shutil.rmtree(reader._rootdir)
return reader._rootdir
return filter(partial(is_not, None), problems)
def get_adj_dates(self, start, end, assets, data_frequency):
"""
@@ -528,7 +564,8 @@ class ExchangeBundle:
return chunks
def ingest_assets(self, assets, data_frequency, start_dt=None, end_dt=None,
show_progress=False, show_breakdown=False):
show_progress=False, show_breakdown=False,
show_report=False):
"""
Determine if data is missing from the bundle and attempt to ingest it.
@@ -562,6 +599,7 @@ class ExchangeBundle:
end_dt=end_dt
)
problems = []
# This is the common writer for the entire exchange bundle
# we want to give an end_date far in time
writer = self.get_writer(start_dt, end_dt, data_frequency)
@@ -577,7 +615,7 @@ class ExchangeBundle:
symbol=asset.symbol
)) as it:
for chunk in it:
self.ingest_ctable(
problems += self.ingest_ctable(
asset=chunk['asset'],
data_frequency=data_frequency,
period=chunk['period'],
@@ -601,7 +639,7 @@ class ExchangeBundle:
frequency=data_frequency,
)) as it:
for chunk in it:
self.ingest_ctable(
problems += self.ingest_ctable(
asset=chunk['asset'],
data_frequency=data_frequency,
period=chunk['period'],
@@ -610,9 +648,14 @@ class ExchangeBundle:
cleanup=True
)
if show_report and len(problems) > 0:
log.info('problems during ingestion:{}\n'.format(
'\n'.join(problems)
))
def ingest(self, data_frequency, include_symbols=None,
exclude_symbols=None, start=None, end=None,
show_progress=True, show_breakdown=True, environ=os.environ):
show_progress=True, show_breakdown=True, show_report=True):
"""
Inject data based on specified parameters.
@@ -631,7 +674,7 @@ class ExchangeBundle:
for frequency in data_frequency.split(','):
self.ingest_assets(assets, frequency, start, end,
show_progress, show_breakdown)
show_progress, show_breakdown, show_report)
def get_history_window_series_and_load(self,
assets,
@@ -749,7 +792,7 @@ class ExchangeBundle:
field,
data_frequency,
reset_reader=False):
start_dt = get_start_dt(end_dt, bar_count, data_frequency)
start_dt = get_start_dt(end_dt, bar_count, data_frequency, False)
start_dt, end_dt = self.get_adj_dates(
start_dt, end_dt, assets, data_frequency
)
catalyst/exchange/exchange_utils.py
+19
View File
@@ -2,6 +2,7 @@ import json
import os
import pickle
import re
import shutil
from datetime import date, datetime
import pandas as pd
@@ -158,6 +159,24 @@ def get_exchange_auth(exchange_name, environ=None):
return data
def delete_algo_folder(algo_name, environ=None):
"""
Delete the folder containing the algo state.
Parameters
----------
algo_name: str
environ:
Returns
-------
str
"""
folder = get_algo_folder(algo_name, environ)
shutil.rmtree(folder)
def get_algo_folder(algo_name, environ=None):
"""
The algorithm root folder of the algorithm.
catalyst/exchange/stats_utils.py
+32
View File
@@ -54,6 +54,38 @@ def crossunder(source, target):
return False
def vwap(df):
"""
Volume-weighted average price (VWAP) is a ratio generally used by
institutional investors and mutual funds to make buys and sells so as not
to disturb the market prices with large orders. It is the average share
price of a stock weighted against its trading volume within a particular
time frame, generally one day.
Read more: Volume Weighted Average Price - VWAP
https://www.investopedia.com/terms/v/vwap.asp#ixzz4xt922daE
Parameters
----------
df: pd.DataFrame
Returns
-------
"""
if 'close' not in df.columns or 'volume' not in df.columns:
raise ValueError('price data must include `volume` and `close`')
vol_sum = np.nansum(df['volume'].values)
try:
ret = np.nansum(df['close'].values * df['volume'].values) / vol_sum
except ZeroDivisionError:
ret = np.nan
return ret
def get_pretty_stats(stats_df, recorded_cols=None, num_rows=10):
"""
Format and print the last few rows of a statistics DataFrame.
catalyst/finance/execution.py
+14 -22
View File
@@ -77,6 +77,7 @@ class LimitOrder(ExecutionStyle):
Execution style representing an order to be executed at a price equal to or
better than a specified limit price.
"""
def __init__(self, limit_price, exchange=None):
"""
Store the given price.
@@ -99,6 +100,7 @@ class StopOrder(ExecutionStyle):
Execution style representing an order to be placed once the market price
reaches a specified stop price.
"""
def __init__(self, stop_price, exchange=None):
"""
Store the given price.
@@ -121,6 +123,7 @@ class StopLimitOrder(ExecutionStyle):
Execution style representing a limit order to be placed with a specified
limit price once the market reaches a specified stop price.
"""
def __init__(self, limit_price, stop_price, exchange=None):
"""
Store the given prices
@@ -144,31 +147,20 @@ class StopLimitOrder(ExecutionStyle):
def asymmetric_round_price_to_penny(price, prefer_round_down,
diff=(0.0095 - .005)):
"""
Asymmetric rounding function for adjusting prices to two places in a way
that "improves" the price. For limit prices, this means preferring to
round down on buys and preferring to round up on sells. For stop prices,
it means the reverse.
Modified the original function because we do not want to round
prices on crypto exchange.
If prefer_round_down == True:
When .05 below to .95 above a penny, use that penny.
If prefer_round_down == False:
When .95 below to .05 above a penny, use that penny.
Parameters
----------
price: float
Returns
-------
float
In math-speak:
If prefer_round_down: [<X-1>.0095, X.0195) -> round to X.01.
If not prefer_round_down: (<X-1>.0005, X.0105] -> round to X.01.
"""
# Subtracting an epsilon from diff to enforce the open-ness of the upper
# bound on buys and the lower bound on sells. Using the actual system
# epsilon doesn't quite get there, so use a slightly less epsilon-ey value.
epsilon = float_info.epsilon * 10
diff = diff - epsilon
# relies on rounding half away from zero, unlike numpy's bankers' rounding
rounded = round(price - (diff if prefer_round_down else -diff), 2)
if zp_math.tolerant_equals(rounded, 0.0):
return 0.0
return rounded
# TODO: consider overriding outside of the original function
return price
def check_stoplimit_prices(price, label):
tests/exchange/test_bundle.py
+11 -11
View File
@@ -42,17 +42,16 @@ class TestExchangeBundle:
def test_ingest_minute(self):
data_frequency = 'minute'
exchange_name = 'bitfinex'
exchange_name = 'poloniex'
exchange = get_exchange(exchange_name)
exchange_bundle = ExchangeBundle(exchange)
assets = [
exchange.get_asset('xmr_btc')
exchange.get_asset('eth_btc')
]
# start = pd.to_datetime('2017-09-01', utc=True)
start = pd.to_datetime('2016-01-01', utc=True)
end = pd.to_datetime('2017-9-30', utc=True)
start = pd.to_datetime('2016-03-01', utc=True)
end = pd.to_datetime('2017-11-1', utc=True)
log.info('ingesting exchange bundle {}'.format(exchange_name))
exchange_bundle.ingest(
@@ -122,8 +121,8 @@ class TestExchangeBundle:
def test_ingest_daily(self):
exchange_name = 'bitfinex'
data_frequency = 'daily'
include_symbols = 'btc_usd'
data_frequency = 'minute'
include_symbols = 'neo_btc'
# exchange_name = 'poloniex'
# data_frequency = 'daily'
@@ -422,7 +421,8 @@ class TestExchangeBundle:
data_frequency=data_frequency,
asset=asset,
writer=writer,
empty_rows_behavior='raise'
empty_rows_behavior='raise',
duplicates_behavior='raise'
)
bundle_series = bundle.get_history_window_series(
@@ -458,10 +458,10 @@ class TestExchangeBundle:
)
def bundle_to_csv(self):
exchange_name = 'bitfinex'
exchange_name = 'poloniex'
data_frequency = 'minute'
period = '2017-10'
symbol = 'neo_btc'
period = '2017-09'
symbol = 'eth_btc'
exchange = get_exchange(exchange_name)
asset = exchange.get_asset(symbol)