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

Merge branch 'develop': adds 1-min OHLCV data resolution, fractional coins and 9 decimals of price resolution

Browse Source
This commit is contained in:
Victor Grau Serrat
2017-09-20 15:20:30 -06:00
parent ad2d0e9253 b354837b83
commit 09bf875d6c
21 changed files with 627 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/__main__.py
+10 -2
View File
@@ -210,6 +210,12 @@ def ipython_only(option):
help='The base currency used to calculate statistics '
'(e.g. usd, btc, eth).',
)
@click.option(
'--live-graph/--no-live-graph',
is_flag=True,
default=False,
help='Display live graph.',
)
@click.pass_context
def run(ctx,
algofile,
@@ -227,7 +233,8 @@ def run(ctx,
live,
exchange_name,
algo_namespace,
base_currency):
base_currency,
live_graph):
"""Run a backtest for the given algorithm.
"""
@@ -283,7 +290,8 @@ def run(ctx,
live=live,
exchange=exchange_name,
algo_namespace=algo_namespace,
base_currency=base_currency
base_currency=base_currency,
live_graph=live_graph
)
if output == '-':
catalyst/algorithm.py
+7 -10
View File
@@ -125,6 +125,7 @@ from catalyst.utils.factory import create_simulation_parameters
from catalyst.utils.math_utils import (
tolerant_equals,
round_if_near_integer,
round_nearest
)
from catalyst.utils.pandas_utils import clear_dataframe_indexer_caches
from catalyst.utils.preprocess import preprocess
@@ -1488,7 +1489,7 @@ class TradingAlgorithm(object):
def _calculate_order(self, asset, amount,
limit_price=None, stop_price=None, style=None):
amount = self.round_order(amount)
amount = self.round_order(amount, asset)
# Raises a ZiplineError if invalid parameters are detected.
self.validate_order_params(asset,
@@ -1505,16 +1506,13 @@ class TradingAlgorithm(object):
return amount, style
@staticmethod
def round_order(amount):
def round_order(amount, asset):
"""
Convert number of shares to an integer.
By default, truncates to the integer share count that's either within
.0001 of amount or closer to zero.
E.g. 3.9999 -> 4.0; 5.5 -> 5.0; -5.5 -> -5.0
Converts the number of shares to the smallest tradable lot size for
the asset being ordered.
"""
return int(round_if_near_integer(amount))
return round_nearest(amount, asset.min_trade_size)
def validate_order_params(self,
asset,
@@ -1550,7 +1548,6 @@ class TradingAlgorithm(object):
self.updated_portfolio(),
self.get_datetime(),
self.trading_client.current_data)
@staticmethod
def __convert_order_params_for_blotter(limit_price, stop_price, style):
"""
catalyst/assets/_assets.pyx
+11 -4
View File
@@ -59,6 +59,7 @@ cdef class Asset:
cdef readonly object exchange
cdef readonly object exchange_full
cdef readonly object min_trade_size
_kwargnames = frozenset({
'sid',
@@ -70,6 +71,7 @@ cdef class Asset:
'auto_close_date',
'exchange',
'exchange_full',
'min_trade_size',
})
def __init__(self,
@@ -81,7 +83,8 @@ cdef class Asset:
object end_date=None,
object first_traded=None,
object auto_close_date=None,
object exchange_full=None):
object exchange_full=None,
object min_trade_size=None):
self.sid = sid
self.sid_hash = hash(sid)
@@ -94,6 +97,7 @@ cdef class Asset:
self.end_date = end_date
self.first_traded = first_traded
self.auto_close_date = auto_close_date
self.min_trade_size = min_trade_size
def __int__(self):
return self.sid
@@ -148,7 +152,8 @@ cdef class Asset:
def __repr__(self):
attrs = ('symbol', 'asset_name', 'exchange',
'start_date', 'end_date', 'first_traded', 'auto_close_date')
'start_date', 'end_date', 'first_traded', 'auto_close_date',
'min_trade_size')
tuples = ((attr, repr(getattr(self, attr, None)))
for attr in attrs)
strings = ('%s=%s' % (t[0], t[1]) for t in tuples)
@@ -170,7 +175,8 @@ cdef class Asset:
self.end_date,
self.first_traded,
self.auto_close_date,
self.exchange_full))
self.exchange_full,
self.min_trade_size))
cpdef to_dict(self):
"""
@@ -186,6 +192,7 @@ cdef class Asset:
'auto_close_date': self.auto_close_date,
'exchange': self.exchange,
'exchange_full': self.exchange_full,
'min_trade_size': self.min_trade_size
}
@classmethod
@@ -234,7 +241,7 @@ cdef class Equity(Asset):
def __repr__(self):
attrs = ('symbol', 'asset_name', 'exchange',
'start_date', 'end_date', 'first_traded', 'auto_close_date',
'exchange_full')
'exchange_full', 'min_trade_size')
tuples = ((attr, repr(getattr(self, attr, None)))
for attr in attrs)
strings = ('%s=%s' % (t[0], t[1]) for t in tuples)
catalyst/assets/asset_db_schema.py
+2 -1
View File
@@ -39,7 +39,8 @@ equities = sa.Table(
sa.Column('first_traded', sa.Integer),
sa.Column('auto_close_date', sa.Integer),
sa.Column('exchange', sa.Text),
sa.Column('exchange_full', sa.Text)
sa.Column('exchange_full', sa.Text),
sa.Column('min_trade_size', sa.Float)
)
equity_symbol_mappings = sa.Table(
catalyst/assets/asset_writer.py
+3
View File
@@ -73,6 +73,7 @@ _equities_defaults = {
'exchange': None,
# optional, something like "New York Stock Exchange"
'exchange_full': None,
'min_trade_size': 1
}
# Default values for the futures DataFrame
@@ -390,6 +391,8 @@ class AssetDBWriter(object):
The date on which to close any positions in this asset.
exchange : str
The exchange where this asset is traded.
min_trade_size: float, optional
The minimum denomination this asset can be traded.
The index of this dataframe should contain the sids.
futures : pd.DataFrame, optional
catalyst/curate/poloniex.py
+184 -71
View File
@@ -1,12 +1,10 @@
import json, time, csv
from datetime import datetime
import pandas as pd
import os
import time
import requests
import logbook
import os, time, shutil, requests, logbook
DT_START = time.mktime(datetime(2010, 1, 1, 0, 0).timetuple())
DT_START = int(time.mktime(datetime(2010, 1, 1, 0, 0).timetuple()))
DT_END = int(time.time())
CSV_OUT_FOLDER = '/var/tmp/catalyst/data/poloniex/'
CONN_RETRIES = 2
@@ -14,9 +12,9 @@ logbook.StderrHandler().push_application()
log = logbook.Logger(__name__)
class PoloniexCurator(object):
"""
'''
OHLCV data feed generator for crypto data. Based on Poloniex market data
"""
'''
_api_path = 'https://poloniex.com/public?'
currency_pairs = []
@@ -29,6 +27,9 @@ class PoloniexCurator(object):
log.error('Failed to create data folder: %s' % CSV_OUT_FOLDER)
log.exception(e)
'''
Retrieves and returns all currency pairs from the exchange
'''
def get_currency_pairs(self):
url = self._api_path + 'command=returnTicker'
@@ -47,98 +48,210 @@ class PoloniexCurator(object):
log.debug('Currency pairs retrieved successfully: %d' % (len(self.currency_pairs)))
def _get_start_date(self, csv_fn):
''' Function returns latest appended date, if the file has been previously written
the last line is an empty one, so we have to read the second to last line
'''
Helper function that reads tradeID and date fields from CSV readline
'''
def _retrieve_tradeID_date(self, row):
tId = int(row.split(',')[0])
d = pd.to_datetime( row.split(',')[1], infer_datetime_format=True).value // 10 ** 9
return tId, d
'''
Retrieves TradeHistory from exchange for a given currencyPair between start and end dates.
If no start date is provided, uses a system-wide one (beginning of time for cryptotrading)
If no end date is provided, 'now' is used
Stores results in CSV file on disk.
This function is called recursively to work around the limitations imposed by the provider API.
'''
def retrieve_trade_history(self, currencyPair, start=DT_START, end=DT_END, temp=None):
csv_fn = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv'
'''
Check what data we already have on disk, reading first and last lines from file.
Data is stored on file from NEWEST to OLDEST.
'''
try:
with open(csv_fn, 'ab+') as f:
f.seek(0, os.SEEK_END) # First check file is not zero size
if(f.tell() > 2):
f.seek(-2, os.SEEK_END) # Jump to the second last byte.
while f.read(1) != b"\n": # Until EOL is found...
f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more.
lastrow = f.readline()
return int(lastrow.split(',')[0]) + 300
f.seek(0, os.SEEK_END)
if(f.tell() > 2): # First check file is not zero size
f.seek(0) # Go to the beginning to read first line
last_tradeID, end_file = self._retrieve_tradeID_date(f.readline())
f.seek(-2, os.SEEK_END) # Jump to the second last byte.
while f.read(1) != b"\n": # Until EOL is found...
f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more.
first_tradeID, start_file = self._retrieve_tradeID_date(f.readline())
if( first_tradeID == 1 and end_file + 3600 > DT_END ):
return
except Exception as e:
log.error('Error opening file: %s' % csv_fn)
log.exception(e)
return DT_START
'''
Poloniex API limits querying TradeHistory to intervals smaller than 1 month,
so we make sure that start date is never more than 1 month apart from end date
'''
if( end - start > 2419200 ): # 60 s/min * 60 min/hr * 24 hr/day * 28 days
newstart = end - 2419200
else:
newstart = start
def get_data(self, currencyPair, start, end=9999999999, period=300):
url = self._api_path + 'command=returnChartData&currencyPair=' + currencyPair + '&start=' + str(start) + '&end=' + str(end) + '&period=' + str(period)
log.debug(currencyPair+': Retrieving from '+str(newstart)+' to '+str(end) +'\t '
+ time.ctime(newstart) + ' - '+ time.ctime(end))
url = self._api_path + 'command=returnTradeHistory&currencyPair=' + currencyPair + '&start=' + str(newstart) + '&end=' + str(end)
try:
response = requests.get(url)
except Exception as e:
log.error('Failed to retrieve candlestick chart data for %s' % currencyPair)
log.error('Failed to retrieve trade history data for %s' % currencyPair)
log.exception(e)
return None
else:
if isinstance(response.json(), dict) and response.json()['error']:
log.error('Failed to to retrieve trade history data for %s: %s' % (currencyPair,response.json()['error']))
exit(1)
'''
If we get to transactionId == 1, and we already have that on disk,
we got to the end of TradeHistory for this coin.
'''
if('first_tradeID' in locals() and response.json()[-1]['tradeID'] == first_tradeID):
return
'''
There are primarily two scenarios:
a) There is newer data available that we need to add at the beginning
of the file. We'll retrieve all what we need until we get to what
we already have, writing it to a temporary file; and we will write
that at the beginning of our existing file.
b) We are going back in time, appending at the end of our existing
TradeHistory until the first transaction for this currencyPair
'''
try:
if( 'end_file' in locals() and end_file + 3600 < end):
if (temp is None):
temp = os.tmpfile()
tempcsv = csv.writer(temp)
for item in response.json():
if( item['tradeID'] <= last_tradeID ):
continue
tempcsv.writerow([
item['tradeID'],
item['date'],
item['type'],
item['rate'],
item['amount'],
item['total'],
item['globalTradeID']
])
if( response.json()[-1]['tradeID'] > last_tradeID ):
end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9
self.retrieve_trade_history(currencyPair, start, end, temp=temp)
else:
with open(csv_fn,'rb+') as f:
shutil.copyfileobj(f,temp)
f.seek(0)
temp.seek(0)
shutil.copyfileobj(temp,f)
temp.close()
end = start_file
else:
with open(csv_fn, 'ab') as csvfile:
csvwriter = csv.writer(csvfile)
for item in response.json():
if( 'first_tradeID' in locals() and item['tradeID'] >= first_tradeID ):
continue
csvwriter.writerow([
item['tradeID'],
item['date'],
item['type'],
item['rate'],
item['amount'],
item['total'],
item['globalTradeID']
])
end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9
except Exception as e:
log.error('Error opening %s' % csv_fn)
log.exception(e)
'''
If we got here, we aren't done yet. Call recursively with 'end' times
that go sequentially back in time.
'''
self.retrieve_trade_history(currencyPair, start, end)
return response.json()
'''
Pulls latest data for a single pair
Generates OHLCV dataframe from a dataframe containing all TradeHistory
by resampling with 1-minute period
'''
def append_data_single_pair(self, currencyPair, repeat=0):
log.debug('Getting data for %s' % currencyPair)
csv_fn = CSV_OUT_FOLDER + 'crypto_prices-' + currencyPair + '.csv'
start = self._get_start_date(csv_fn)
# Only fetch data if more than 5min have passed since last fetch
if (time.time() > start):
data = self.get_data(currencyPair, start)
if data is not None:
try:
with open(csv_fn, 'ab') as csvfile:
csvwriter = csv.writer(csvfile)
for item in data:
if item['date'] == 0:
continue
csvwriter.writerow([
item['date'],
item['open'],
item['high'],
item['low'],
item['close'],
item['volume'],
])
except Exception as e:
log.error('Error opening %s' % csv_fn)
log.exception(e)
elif (repeat < CONN_RETRIES):
log.debug('Retrying: attemt %d' % (repeat+1) )
self.append_data_single_pair(currencyPair, repeat + 1)
def generate_ohlcv(self, df):
df.set_index('date', inplace=True) # Index by date
vol = df['total'].to_frame('volume') # Will deal with vol separately, as ohlc() messes it up
df.drop('total', axis=1, inplace=True) # Drop volume data from dataframe
ohlc = df.resample('T').ohlc() # Resample OHLC in 1min bins
ohlc.columns = ohlc.columns.map(lambda t: t[1]) # Raname columns by dropping 'rate'
closes = ohlc['close'].fillna(method='pad') # Pad forward missing 'close'
ohlc = ohlc.apply(lambda x: x.fillna(closes)) # Fill N/A with last close
vol = vol.resample('T').sum().fillna(0) # Add volumes by bin
ohlcv = pd.concat([ohlc,vol], axis=1) # Concatenate OHLC + Volume
return ohlcv
'''
Pulls latest data for all currency pairs
Generates OHLCV data file with 1minute bars from TradeHistory on disk
'''
def append_data(self):
for currencyPair in self.currency_pairs:
self.append_data_single_pair(currencyPair)
# Rate limit is 6 calls per second, sleep 1sec/6 to be safe
time.sleep(0.17)
def write_ohlcv_file(self, currencyPair):
csv_trades = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv'
csv_1min = CSV_OUT_FOLDER + 'crypto_1min-' + currencyPair + '.csv'
if( os.path.isfile(csv_1min) ):
log.debug(currencyPair+': 1min data already present. Delete the file if you want to rebuild it.')
else:
df = pd.read_csv(csv_trades, names=['tradeID','date','type','rate','amount','total','globalTradeID'],
dtype = {'tradeID': int, 'date': str, 'type': str, 'rate': float, 'amount': float, 'total': float, 'globalTradeID': int } )
df.drop(['tradeID','type','amount','globalTradeID'], axis=1, inplace=True)
df['date'] = pd.to_datetime(df['date'], infer_datetime_format=True)
ohlcv = self.generate_ohlcv(df)
try:
with open(csv_1min, 'ab') as csvfile:
csvwriter = csv.writer(csvfile)
for item in ohlcv.itertuples():
if item.Index == 0:
continue
csvwriter.writerow([
item.Index.value // 10 ** 9,
item.open,
item.high,
item.low,
item.close,
item.volume,
])
except Exception as e:
log.error('Error opening %s' % csv_fn)
log.exception(e)
log.debug(currencyPair+': Generated 1min OHLCV data.')
'''
Returns a data frame for all pairs, or for the requests currency pair.
Makes sure data is up to date
Returns a data frame for a given currencyPair from data on disk
'''
def to_dataframe(self, start, end, currencyPair=None):
csv_fn = CSV_OUT_FOLDER + 'crypto_prices-' + currencyPair + '.csv'
last_date = self._get_start_date(csv_fn)
if last_date + 300 < end or not os.path.exists(csv_fn):
# get latest data
self.append_data_single_pair(currencyPair)
# CSV holds the latest snapshot
df = pd.read_csv(csv_fn, names=['date', 'open', 'high', 'low', 'close', 'volume'])
df['date']=pd.to_datetime(df['date'],unit='s')
def onemin_to_dataframe(self, currencyPair, start, end):
csv_fn = CSV_OUT_FOLDER + 'crypto_1min-' + currencyPair + '.csv'
df = pd.read_csv(csv_fn, names=['date', 'open', 'high', 'low', 'close', 'volume'])
df['date'] = pd.to_datetime(df['date'],unit='s')
df.set_index('date', inplace=True)
return df[start : end]
return df[datetime.fromtimestamp(start):datetime.fromtimestamp(end-1)]
if __name__ == '__main__':
pc = PoloniexCurator()
pc.get_currency_pairs()
pc.append_data()
for currencyPair in pc.currency_pairs:
pc.retrieve_trade_history(currencyPair)
pc.write_ohlcv_file(currencyPair)
catalyst/data/_equities.pyx
+1 -1
View File
@@ -217,7 +217,7 @@ cpdef _read_bcolz_data(ctable_t table,
if column_name in ['open', 'high', 'low', 'close']:
where_nan = (outbuf == 0)
outbuf_as_float = outbuf.astype(float64) * .000001
outbuf_as_float = outbuf.astype(float64) * .000000001
outbuf_as_float[where_nan] = NAN
results.append(outbuf_as_float)
elif column_name != 'volume':
catalyst/data/bundles/base_pricing.py
+1
View File
@@ -24,6 +24,7 @@ class BasePricingBundle(BaseBundle):
('start_date', 'datetime64[ns]'),
('end_date', 'datetime64[ns]'),
('ac_date', 'datetime64[ns]'),
('min_trade_size', 'float'),
]
@lazyval
catalyst/data/bundles/poloniex.py
+31 -14
View File
@@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
from datetime import datetime
import pandas as pd
@@ -23,6 +25,8 @@ from catalyst.data.bundles.core import register_bundle
from catalyst.data.bundles.base_pricing import BaseCryptoPricingBundle
from catalyst.utils.memoize import lazyval
from catalyst.curate.poloniex import PoloniexCurator
class PoloniexBundle(BaseCryptoPricingBundle):
@lazyval
def name(self):
@@ -36,7 +40,7 @@ class PoloniexBundle(BaseCryptoPricingBundle):
def frequencies(self):
return set((
'daily',
#'5-minute',
'minute',
))
@lazyval
@@ -75,12 +79,14 @@ class PoloniexBundle(BaseCryptoPricingBundle):
start_date = sym_data.index[0]
end_date = sym_data.index[-1]
ac_date = end_date + pd.Timedelta(days=1)
min_trade_size = 0.00000001
return (
sym_md.symbol,
start_date,
end_date,
ac_date,
min_trade_size,
)
def fetch_raw_symbol_frame(self,
@@ -90,22 +96,28 @@ class PoloniexBundle(BaseCryptoPricingBundle):
start_date,
end_date,
frequency):
raw = pd.read_json(
self._format_data_url(
api_key,
symbol,
start_date,
end_date,
frequency,
),
orient='records',
)
raw.set_index('date', inplace=True)
if(frequency == 'minute'):
pc = PoloniexCurator()
raw = pc.onemin_to_dataframe(symbol, start_date, end_date)
else:
raw = pd.read_json(
self._format_data_url(
api_key,
symbol,
start_date,
end_date,
frequency,
),
orient='records',
)
raw.set_index('date', inplace=True)
# BcolzDailyBarReader introduces a 1/1000 factor in the way pricing is stored
# on disk, which we compensate here to get the right pricing amounts
# ref: data/us_equity_pricing.py
scale = 1000
scale = 1
raw.loc[:, 'open'] /= scale
raw.loc[:, 'high'] /= scale
raw.loc[:, 'low'] /= scale
@@ -166,4 +178,9 @@ register_bundle(PoloniexBundle, ['USDT_BTC',])
For a production environment make sure to use (to bundle all pairs):
register_bundle(PoloniexBundle)
'''
register_bundle(PoloniexBundle, create_writers=False)
if 'ingest' in sys.argv and '-c' in sys.argv:
register_bundle(PoloniexBundle)
else:
register_bundle(PoloniexBundle, create_writers=False)
catalyst/data/history_loader.py
+1 -1
View File
@@ -38,7 +38,7 @@ from catalyst.utils.numpy_utils import float64_dtype
from catalyst.utils.pandas_utils import find_in_sorted_index
# Default number of decimal places used for rounding asset prices.
DEFAULT_ASSET_PRICE_DECIMALS = 3
DEFAULT_ASSET_PRICE_DECIMALS = 9
class HistoryCompatibleUSEquityAdjustmentReader(object):
catalyst/data/us_equity_pricing.py
+7 -4
View File
@@ -11,6 +11,9 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import division # Python2 req to have division of ints yield float
from errno import ENOENT
from functools import partial
from os import remove
@@ -80,7 +83,6 @@ from catalyst.utils.cli import (
from ._equities import _compute_row_slices, _read_bcolz_data
from ._adjustments import load_adjustments_from_sqlite
logger = logbook.Logger('UsEquityPricing')
OHLC = frozenset(['open', 'high', 'low', 'close'])
@@ -116,6 +118,8 @@ SQLITE_STOCK_DIVIDEND_PAYOUT_COLUMN_DTYPES = {
UINT32_MAX = iinfo(uint32).max
UINT64_MAX = iinfo(uint64).max
PRICE_ADJUSTMENT_FACTOR = 1000000000 # Provides 9 decimals resolution. Also affects _equities.pyx L220
def check_uint32_safe(value, colname):
if value >= UINT32_MAX:
@@ -433,7 +437,7 @@ class BcolzDailyBarWriter(object):
return raw_data
winsorise_uint64(raw_data, invalid_data_behavior, 'volume', *OHLC)
processed = (raw_data[list(OHLC)] * 1000000).astype('uint64')
processed = (raw_data[list(OHLC)] * PRICE_ADJUSTMENT_FACTOR).astype('uint64')
dates = raw_data.index.values.astype('datetime64[s]')
check_uint32_safe(dates.max().view(np.int64), 'day')
processed['day'] = dates.astype('uint32')
@@ -519,7 +523,6 @@ class BcolzDailyBarReader(SessionBarReader):
# Need to test keeping the entire array in memory for the course of a
# process first.
self._spot_cols = {}
self.PRICE_ADJUSTMENT_FACTOR = 0.001
self._read_all_threshold = read_all_threshold
@lazyval
@@ -763,7 +766,7 @@ class BcolzDailyBarReader(SessionBarReader):
if price == 0:
return nan
else:
return price * 0.001
return price / PRICE_ADJUSTMENT_FACTOR
else:
return price
catalyst/exchange/algorithm_exchange.py
+91 -10
View File
@@ -19,6 +19,7 @@ from time import sleep
from os import listdir
from os.path import isfile, join
from collections import deque
import numpy as np
import logbook
import pandas as pd
@@ -28,14 +29,16 @@ from catalyst.algorithm import TradingAlgorithm
from catalyst.data.minute_bars import BcolzMinuteBarWriter, \
BcolzMinuteBarReader
from catalyst.errors import OrderInBeforeTradingStart
from catalyst.exchange.exchange_clock import ExchangeClock
from catalyst.exchange.simple_clock import SimpleClock
from catalyst.exchange.live_graph_clock import LiveGraphClock
from catalyst.exchange.exchange_errors import (
ExchangeRequestError,
ExchangePortfolioDataError,
ExchangeTransactionError
)
from catalyst.exchange.exchange_utils import get_exchange_minute_writer_root, \
save_algo_object, get_algo_object, get_algo_folder
save_algo_object, get_algo_object, get_algo_folder, get_algo_df, \
save_algo_df
from catalyst.exchange.stats_utils import get_pretty_stats
from catalyst.finance.performance.period import calc_period_stats
from catalyst.gens.tradesimulation import AlgorithmSimulator
@@ -56,8 +59,19 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
def __init__(self, *args, **kwargs):
self.exchange = kwargs.pop('exchange', None)
self.algo_namespace = kwargs.pop('algo_namespace', None)
self.orders = {}
self.live_graph = kwargs.pop('live_graph', None)
self._clock = None
self.minute_stats = deque(maxlen=60)
self.pnl_stats = get_algo_df(self.algo_namespace, 'pnl_stats')
self.custom_signals_stats = \
get_algo_df(self.algo_namespace, 'custom_signals_stats')
self.exposure_stats = \
get_algo_df(self.algo_namespace, 'exposure_stats')
self.is_running = True
self.retry_check_open_orders = 5
@@ -122,6 +136,13 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
sys.exit(0)
@property
def clock(self):
if self._clock is None:
return self._create_clock()
else:
return self._clock
def _create_clock(self):
# The calendar's execution times are the minutes over which we actually
@@ -137,10 +158,21 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
# This method is taken from TradingAlgorithm.
# The clock has been replaced to use RealtimeClock
# TODO: should we apply a time skew? not sure to understand the utility.
return ExchangeClock(
self.sim_params.sessions,
time_skew=self.exchange.time_skew
)
log.debug('creating clock')
if self.live_graph:
self._clock = LiveGraphClock(
self.sim_params.sessions,
time_skew=self.exchange.time_skew,
context=self
)
else:
self._clock = SimpleClock(
self.sim_params.sessions,
time_skew=self.exchange.time_skew
)
return self._clock
def _create_generator(self, sim_params):
if self.perf_tracker is None:
@@ -156,7 +188,7 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
self,
sim_params,
self.data_portal,
self._create_clock(),
self.clock,
self._create_benchmark_source(),
self.restrictions,
universe_func=self._calculate_universe
@@ -222,6 +254,49 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
error=e
)
def add_pnl_stats(self, period_stats):
starting = period_stats['starting_cash']
current = period_stats['portfolio_value']
appreciation = (current / starting) - 1
perc = (appreciation * 100) if current != 0 else 0
log.debug('adding pnl stats: {:6f}%'.format(perc))
df = pd.DataFrame(
data=[dict(performance=perc)],
index=[period_stats['period_close']]
)
self.pnl_stats = pd.concat([self.pnl_stats, df])
save_algo_df(self.algo_namespace, 'pnl_stats', self.pnl_stats)
def add_custom_signals_stats(self, period_stats):
log.debug('adding custom signals stats: {}'.format(self.recorded_vars))
df = pd.DataFrame(
data=[self.recorded_vars],
index=[period_stats['period_close']],
)
self.custom_signals_stats = pd.concat([self.custom_signals_stats, df])
save_algo_df(self.algo_namespace, 'custom_signals_stats',
self.custom_signals_stats)
def add_exposure_stats(self, period_stats):
data = dict(
long_exposure=period_stats['long_exposure'],
base_currency=period_stats['ending_cash']
)
log.debug('adding exposure stats: {}'.format(data))
df = pd.DataFrame(
data=[data],
index=[period_stats['period_close']],
)
self.exposure_stats = pd.concat([self.exposure_stats, df])
save_algo_df(self.algo_namespace, 'exposure_stats',
self.exposure_stats)
def prepare_period_stats(self, start_dt, end_dt):
"""
Creates a dictionary representing the state of the tracker.
@@ -314,9 +389,14 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
minute_stats = self.prepare_period_stats(
data.current_dt, data.current_dt + timedelta(minutes=1))
# Saving the last hour in memory
self.minute_stats.append(minute_stats)
self.add_pnl_stats(minute_stats)
self.add_custom_signals_stats(minute_stats)
self.add_exposure_stats(minute_stats)
print_df = pd.DataFrame(list(self.minute_stats))
log.debug(
'statistics for the last {stats_minutes} minutes:\n{stats}'.format(
@@ -401,8 +481,9 @@ class ExchangeTradingAlgorithm(TradingAlgorithm):
if order_id is not None:
order = self.portfolio.open_orders[order_id]
self.perf_tracker.process_order(order)
return order
return order
else:
return None
def round_order(self, amount):
"""
catalyst/exchange/bittrex/bittrex.py
+12 -1
View File
@@ -119,7 +119,18 @@ class Bittrex(Exchange):
)
return order
else:
raise CreateOrderError(exchange=self.name, error=order_status)
if order_status == 'INSUFFICIENT_FUNDS':
log.warn('not enough funds to create order')
return None
elif order_status == 'DUST_TRADE_DISALLOWED_MIN_VALUE_50K_SAT':
log.warn('Your order is too small, order at least 50K'
' Satoshi')
return None
else:
raise CreateOrderError(
exchange=self.name,
error=order_status
)
else:
raise InvalidOrderStyle(exchange=self.name,
style=style.__class__.__name__)
catalyst/exchange/exchange.py
+5 -4
View File
@@ -531,10 +531,11 @@ class Exchange:
)
)
order = self.create_order(asset, amount, is_buy, style)
self._portfolio.create_order(order)
return order.id
if order:
self._portfolio.create_order(order)
return order.id
else:
return None
@abstractmethod
def get_open_orders(self, asset):
catalyst/exchange/exchange_utils.py
+32
View File
@@ -3,6 +3,7 @@ import os
import pickle
import urllib
from datetime import date, datetime
import pandas as pd
from catalyst.exchange.exchange_errors import ExchangeAuthNotFound, \
ExchangeSymbolsNotFound
@@ -117,6 +118,37 @@ def append_algo_object(algo_name, key, obj, environ=None):
pickle.dump(obj, handle, protocol=pickle.HIGHEST_PROTOCOL)
def get_algo_df(algo_name, key, environ=None, rel_path=None):
folder = get_algo_folder(algo_name, environ)
if rel_path is not None:
folder = os.path.join(folder, rel_path)
filename = os.path.join(folder, key + '.csv')
if os.path.isfile(filename):
try:
with open(filename, 'rb') as handle:
return pd.read_csv(handle, index_col=0, parse_dates=True)
except IOError:
return pd.DataFrame()
else:
return pd.DataFrame()
def save_algo_df(algo_name, key, df, environ=None, rel_path=None):
folder = get_algo_folder(algo_name, environ)
if rel_path is not None:
folder = os.path.join(folder, rel_path)
ensure_directory(folder)
filename = os.path.join(folder, key + '.csv')
with open(filename, 'wb') as handle:
df.to_csv(handle)
def get_exchange_minute_writer_root(exchange_name, environ=None):
exchange_folder = get_exchange_folder(exchange_name, environ)
catalyst/exchange/live_graph_clock.py
+210
View File
@@ -0,0 +1,210 @@
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import timedelta
import matplotlib.dates as mdates
import pandas as pd
from catalyst.gens.sim_engine import (
BAR,
SESSION_START
)
from logbook import Logger
from matplotlib import pyplot as plt
from matplotlib import style
log = Logger('LiveGraphClock')
fmt = mdates.DateFormatter('%Y-%m-%d %H:%M')
class LiveGraphClock(object):
"""Realtime clock for live trading.
This class is a drop-in replacement for
:class:`zipline.gens.sim_engine.MinuteSimulationClock`.
This mixes the clock with a live graph.
Note
----
This seemingly awkward approach allows us to run the program using a single
thread. This is important because Matplotlib does not play nice with
multi-threaded environments. Zipline probably does not either.
Matplotlib has a pause() method which is a wrapper around time.sleep()
used in the SimpleClock. The key difference is that users
can still interact with the chart during the pause cycles. This is
what enables us to keep a single thread. This is also why we are not using
the 'animate' callback of Matplotlib. We need to direct access to the
__iter__ method in order to yield events to Zipline.
The :param:`time_skew` parameter represents the time difference between
the exchange and the live trading machine's clock. It's not used currently.
"""
def __init__(self, sessions, context, time_skew=pd.Timedelta('0s')):
self.sessions = sessions
self.time_skew = time_skew
self._last_emit = None
self._before_trading_start_bar_yielded = True
self.context = context
style.use('dark_background')
fig = plt.figure()
fig.canvas.set_window_title('Enigma Catalyst: {}'.format(
self.context.algo_namespace))
self.ax_pnl = fig.add_subplot(311)
self.ax_custom_signals = fig.add_subplot(312, sharex=self.ax_pnl)
self.ax_exposure = fig.add_subplot(313, sharex=self.ax_pnl)
if len(context.minute_stats) > 0:
self.draw_pnl()
self.draw_custom_signals()
self.draw_exposure()
# rotates and right aligns the x labels, and moves the bottom of the
# axes up to make room for them
fig.autofmt_xdate()
fig.subplots_adjust(hspace=0.5)
plt.tight_layout()
plt.ion()
plt.show()
def format_ax(self, ax):
"""
Trying to assign reasonable parameters to the time axis.
TODO: room for improvement
:param ax:
:return:
"""
ax.xaxis.set_major_locator(mdates.DayLocator(interval=1))
ax.xaxis.set_major_formatter(fmt)
locator = mdates.HourLocator(interval=4)
locator.MAXTICKS = 5000
ax.xaxis.set_minor_locator(locator)
datemin = pd.Timestamp.utcnow()
ax.set_xlim(datemin)
ax.grid(True)
def set_legend(self, ax):
ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1)
def draw_pnl(self):
ax = self.ax_pnl
df = self.context.pnl_stats
ax.clear()
ax.set_title('Performance')
ax.plot(df.index, df['performance'], '-',
color='green',
linewidth=1.0,
label='Performance'
)
def perc(val):
return '{:2f}'.format(val)
ax.format_ydata = perc
self.set_legend(ax)
self.format_ax(ax)
def draw_custom_signals(self):
ax = self.ax_custom_signals
df = self.context.custom_signals_stats
colors = ['blue', 'green', 'red', 'black', 'orange', 'yellow', 'pink']
ax.clear()
ax.set_title('Custom Signals')
for index, column in enumerate(df.columns.values.tolist()):
ax.plot(df.index, df[column], '-',
color=colors[index],
linewidth=1.0,
label=column
)
self.set_legend(ax)
self.format_ax(ax)
def draw_exposure(self):
ax = self.ax_exposure
context = self.context
df = context.exposure_stats
ax.clear()
ax.set_title('Exposure')
ax.plot(df.index, df['base_currency'], '-',
color='green',
linewidth=1.0,
label='Base Currency: {}'.format(
context.exchange.base_currency.upper()
)
)
positions = context.exchange.portfolio.positions
symbols = []
for position in positions:
symbols.append(position.symbol)
ax.plot(df.index, df['long_exposure'], '-',
color='blue',
linewidth=1.0,
label='Long Exposure: {}'.format(
', '.join(symbols).upper()
)
)
self.set_legend(ax)
self.format_ax(ax)
def __iter__(self):
yield pd.Timestamp.utcnow(), SESSION_START
while True:
current_time = pd.Timestamp.utcnow()
current_minute = current_time.floor('1 min')
if self._last_emit is None or current_minute > self._last_emit:
log.debug('emitting minutely bar: {}'.format(current_minute))
self._last_emit = current_minute
yield current_minute, BAR
try:
self.draw_pnl()
self.draw_custom_signals()
self.draw_exposure()
plt.draw()
except Exception as e:
log.warn('Unable to update the graph: {}'.format(e))
else:
# I can't use the "animate" reactive approach here because
# I need to yield from the main loop.
# Workaround: https://stackoverflow.com/a/33050617/814633
plt.pause(1)
catalyst/exchange/exchange_clock.py → catalyst/exchange/simple_clock.py
+1 -1
View File
@@ -25,7 +25,7 @@ from logbook import Logger
log = Logger('ExchangeClock')
class ExchangeClock(object):
class SimpleClock(object):
"""Realtime clock for live trading.
This class is a drop-in replacement for
catalyst/finance/slippage.py
+6 -3
View File
@@ -41,6 +41,7 @@ DEFAULT_EQUITY_VOLUME_SLIPPAGE_BAR_LIMIT = 0.025
DEFAULT_FUTURE_VOLUME_SLIPPAGE_BAR_LIMIT = 0.05
class LiquidityExceeded(Exception):
pass
@@ -205,20 +206,22 @@ class VolumeShareSlippage(SlippageModel):
def process_order(self, data, order):
volume = data.current(order.asset, "volume")
min_trade_size = order.asset.min_trade_size
max_volume = self.volume_limit * volume
# price impact accounts for the total volume of transactions
# created against the current minute bar
remaining_volume = max_volume - self.volume_for_bar
if remaining_volume < 1:
if remaining_volume < min_trade_size:
# we can't fill any more transactions
raise LiquidityExceeded()
# the current order amount will be the min of the
# volume available in the bar or the open amount.
cur_volume = int(min(remaining_volume, abs(order.open_amount)))
cur_volume = min(remaining_volume, abs(order.open_amount))
if cur_volume < 1:
if cur_volume < min_trade_size:
return None, None
# tally the current amount into our total amount ordered.
catalyst/finance/transaction.py
+1 -5
View File
@@ -65,14 +65,10 @@ def create_transaction(order, dt, price, amount):
# floor the amount to protect against non-whole number orders
# TODO: Investigate whether we can add a robust check in blotter
# and/or tradesimulation, as well.
amount_magnitude = int(abs(amount))
if amount_magnitude < 1:
raise Exception("Transaction magnitude must be at least 1.")
transaction = Transaction(
asset=order.asset,
amount=int(amount),
amount=amount,
dt=dt,
price=price,
order_id=order.id
catalyst/utils/math_utils.py
+2
View File
@@ -17,6 +17,8 @@ import math
from numpy import isnan
def round_nearest(x, a):
return round(round(x / a) * a, -int(math.floor(math.log10(a))))
def tolerant_equals(a, b, atol=10e-7, rtol=10e-7, equal_nan=False):
"""Check if a and b are equal with some tolerance.
catalyst/utils/run_algo.py
+9 -5
View File
@@ -95,7 +95,8 @@ def _run(handle_data,
live,
exchange,
algo_namespace,
base_currency):
base_currency,
live_graph):
"""Run a backtest for the given algorithm.
This is shared between the cli and :func:`catalyst.run_algo`.
@@ -277,7 +278,8 @@ def _run(handle_data,
)
env = TradingEnvironment(
load=partial(load_crypto_market_data, bundle=b, bundle_data=bundle_data, environ=environ),
load=partial(load_crypto_market_data, bundle=b,
bundle_data=bundle_data, environ=environ),
bm_symbol='USDT_BTC',
trading_calendar=open_calendar,
asset_db_path=connstr,
@@ -338,7 +340,7 @@ def _run(handle_data,
TradingAlgorithmClass = (
partial(ExchangeTradingAlgorithm, exchange=exchange,
algo_namespace=algo_namespace)
algo_namespace=algo_namespace, live_graph=live_graph)
if live and exchange else TradingAlgorithm)
perf = TradingAlgorithmClass(
@@ -439,7 +441,8 @@ def run_algorithm(initialize,
live=False,
exchange_name=None,
base_currency=None,
algo_namespace=None):
algo_namespace=None,
live_graph=False):
"""Run a trading algorithm.
Parameters
@@ -552,5 +555,6 @@ def run_algorithm(initialize,
live=live,
exchange=exchange_name,
algo_namespace=algo_namespace,
base_currency=base_currency
base_currency=base_currency,
live_graph=live_graph
)