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 remote-tracking branch 'remotes/origin/develop' into new_exchange_config

Browse Source 
# Conflicts:
#	catalyst/exchange/exchange.py
#	catalyst/exchange/exchange_errors.py
This commit is contained in:
Frederic Fortier
2018-03-05 19:17:13 -05:00
parent 0d051a9496 cb4668f093
commit 6488ff5abe
15 changed files with 199 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/__main__.py
+17 -1
View File
@@ -767,12 +767,18 @@ def bundles():
@main.group()
@click.pass_context
def marketplace(ctx):
"""Access the Enigma Data Marketplace to:\n
- Register and Publish new datasets (seller-side)\n
- Subscribe and Ingest premium datasets (buyer-side)\n
"""
pass
@marketplace.command()
@click.pass_context
def ls(ctx):
"""List all available datasets.
"""
click.echo('Listing of available data sources on the marketplace:',
sys.stdout)
marketplace = Marketplace()
@@ -787,6 +793,8 @@ def ls(ctx):
)
@click.pass_context
def subscribe(ctx, dataset):
"""Subscribe to an exisiting dataset.
"""
if dataset is None:
ctx.fail("must specify a dataset to subscribe to with '--dataset'\n"
"List available dataset on the marketplace with "
@@ -825,6 +833,8 @@ def subscribe(ctx, dataset):
)
@click.pass_context
def ingest(ctx, dataset, data_frequency, start, end):
"""Ingest a dataset (requires subscription).
"""
if dataset is None:
ctx.fail("must specify a dataset to clean with '--dataset'\n"
"List available dataset on the marketplace with "
@@ -842,8 +852,10 @@ def ingest(ctx, dataset, data_frequency, start, end):
)
@click.pass_context
def clean(ctx, dataset):
"""Clean/Remove local data for a given dataset.
"""
if dataset is None:
ctx.fail("must specify a dataset to ingest with '--dataset'\n"
ctx.fail("must specify a dataset to clean up with '--dataset'\n"
"List available dataset on the marketplace with "
"'catalyst marketplace ls'")
click.echo('Cleaning data source: {}'.format(dataset), sys.stdout)
@@ -855,6 +867,8 @@ def clean(ctx, dataset):
@marketplace.command()
@click.pass_context
def register(ctx):
"""Register a new dataset.
"""
marketplace = Marketplace()
marketplace.register()
@@ -878,6 +892,8 @@ def register(ctx):
)
@click.pass_context
def publish(ctx, dataset, datadir, watch):
"""Publish data for a registered dataset.
"""
marketplace = Marketplace()
if dataset is None:
ctx.fail("must specify a dataset to publish data for "
catalyst/constants.py
+5 -5
View File
@@ -30,20 +30,20 @@ AUTH_SERVER = 'https://data.enigma.co'
# TODO: switch to mainnet
ETH_REMOTE_NODE = 'https://ropsten.infura.io/'
# TODO: move to MASTER branch on github
MARKETPLACE_CONTRACT = 'https://raw.githubusercontent.com/enigmampc/' \
'catalyst/develop/catalyst/marketplace/' \
'catalyst/master/catalyst/marketplace/' \
'contract_marketplace_address.txt'
MARKETPLACE_CONTRACT_ABI = 'https://raw.githubusercontent.com/enigmampc/' \
'catalyst/develop/catalyst/marketplace/' \
'catalyst/master/catalyst/marketplace/' \
'contract_marketplace_abi.json'
# TODO: switch to mainnet
ENIGMA_CONTRACT = 'https://raw.githubusercontent.com/enigmampc/catalyst/' \
'develop/catalyst/marketplace/' \
'master/catalyst/marketplace/' \
'contract_enigma_address.txt'
ENIGMA_CONTRACT_ABI = 'https://raw.githubusercontent.com/enigmampc/' \
'catalyst/develop/catalyst/marketplace/' \
'catalyst/master/catalyst/marketplace/' \
'contract_enigma_abi.json'
catalyst/examples/portfolio_optimization.py
+1 -1
View File
@@ -66,7 +66,7 @@ def handle_data(context, data):
# Define portfolio optimization parameters
n_portfolios = 50000
results_array = np.zeros((3 + context.nassets, n_portfolios))
for p in xrange(n_portfolios):
for p in range(n_portfolios):
weights = np.random.random(context.nassets)
weights /= np.sum(weights)
w = np.asmatrix(weights)
catalyst/exchange/exchange.py
+39 -34
View File
@@ -1,5 +1,4 @@
import abc
import pytz
from abc import ABCMeta, abstractmethod, abstractproperty
from datetime import timedelta
from time import sleep
@@ -14,7 +13,8 @@ from catalyst.exchange.exchange_bundle import ExchangeBundle
from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \
SymbolNotFoundOnExchange, \
PricingDataNotLoadedError, \
NoDataAvailableOnExchange, NoValueForField, LastCandleTooEarlyError, \
NoDataAvailableOnExchange, NoValueForField, \
NoCandlesReceivedFromExchange, \
TickerNotFoundError, NotEnoughCashError
from catalyst.exchange.utils.datetime_utils import get_delta, \
get_periods_range, \
@@ -257,7 +257,8 @@ class Exchange:
elif data_frequency is not None:
applies = (
(
data_frequency == 'minute' and a.end_minute is not None)
data_frequency == 'minute' and
a.end_minute is not None)
or (
data_frequency == 'daily' and a.end_daily is not None)
)
@@ -485,49 +486,52 @@ class Exchange:
freq, candle_size, unit, data_frequency = get_frequency(
frequency, data_frequency, supported_freqs=['T', 'D', 'H']
)
# we want to avoid receiving empty candles
# so we request more than needed
# TODO: consider defining a const per asset
# and/or some retry mechanism (in each iteration request more data)
requested_bar_count = bar_count + 30
# The get_history method supports multiple asset
candles = self.get_candles(
freq=freq,
assets=assets,
bar_count=bar_count,
bar_count=requested_bar_count,
end_dt=end_dt if not is_current else None,
)
series = dict()
# candles sanity check - verify no empty candles were received:
for asset in candles:
if candles[asset]:
first_candle = candles[asset][0]
asset_series = self.get_series_from_candles(
candles=candles[asset],
start_dt=first_candle['last_traded'],
if not candles[asset]:
raise NoCandlesReceivedFromExchange(
bar_count=requested_bar_count,
end_dt=end_dt,
data_frequency=frequency,
field=field,
)
asset=asset,
exchange=self.name)
delta_candle_size = candle_size * 60 if unit == 'H' else candle_size
# Checking to make sure that the dates match
delta = get_delta(delta_candle_size, data_frequency)
adj_end_dt = end_dt - delta
last_traded = asset_series.index[-1]
series = get_candles_df(candles=candles,
field=field,
freq=frequency,
bar_count=requested_bar_count,
end_dt=end_dt)
if last_traded < adj_end_dt:
raise LastCandleTooEarlyError(
last_traded=last_traded,
end_dt=adj_end_dt,
exchange=self.name,
)
else: # empty candle received
# because other assets are tz-aware, we need its tz to be set as well
asset_series = pd.Series([], index=pd.DatetimeIndex([], tz=pytz.utc))
series[asset] = asset_series
# TODO: consider how to approach this edge case
# delta_candle_size = candle_size * 60 if unit == 'H' else candle_size
# Checking to make sure that the dates match
# delta = get_delta(delta_candle_size, data_frequency)
# adj_end_dt = end_dt - delta
# last_traded = asset_series.index[-1]
# if last_traded < adj_end_dt:
# raise LastCandleTooEarlyError(
# last_traded=last_traded,
# end_dt=adj_end_dt,
# exchange=self.name,
# )
df = pd.DataFrame(series)
#df.dropna(inplace=True) # commented out due to issue 236
df.dropna(inplace=True)
return df
return df.tail(bar_count)
def get_history_window_with_bundle(self,
assets,
@@ -575,7 +579,8 @@ class Exchange:
A dataframe containing the requested data.
"""
# TODO: this function needs some work, we're currently using it just for benchmark data
# TODO: this function needs some work,
# we're currently using it just for benchmark data
freq, candle_size, unit, data_frequency = get_frequency(
frequency, data_frequency
)
@@ -601,7 +606,7 @@ class Exchange:
start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency)
trailing_dt = \
series[asset].index[-1] + get_delta(1, data_frequency) \
if asset in series else start_dt
if asset in series else start_dt
# The get_history method supports multiple asset
# Use the original frequency to let each api optimize
catalyst/exchange/exchange_algorithm.py
+19
View File
@@ -164,6 +164,25 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm):
style)
return amount, style
def _calculate_order_target_amount(self, asset, target):
"""
removes order amounts so we won't run into issues
when two orders are placed one after the other.
it then proceeds to removing positions amount at TradingAlgorithm
:param asset:
:param target:
:return: target
"""
if asset in self.blotter.open_orders:
for open_order in self.blotter.open_orders[asset]:
current_amount = open_order.amount
target -= current_amount
target = super(ExchangeTradingAlgorithmBase, self). \
_calculate_order_target_amount(asset, target)
return target
def round_order(self, amount, asset):
"""
We need fractions with cryptocurrencies
catalyst/exchange/exchange_blotter.py
+1 -1
View File
@@ -68,7 +68,7 @@ class TradingPairFeeSchedule(CommissionModel):
multiplier = maker \
if ((order.amount > 0 and order.limit < transaction.price)
or (order.amount < 0 and order.limit > transaction.price)) \
and order.limit_reached else taker
and order.limit_reached else taker
fee = cost * multiplier
return fee
catalyst/exchange/exchange_errors.py
+7
View File
@@ -324,6 +324,13 @@ class BalanceTooLowError(ZiplineError):
).strip()
class NoCandlesReceivedFromExchange(ZiplineError):
msg = (
'Although requesting {bar_count} candles until {end_dt} of asset {asset}, '
'an empty list of candles was received for {exchange}.'
).strip()
class MarketsNotFoundError(ZiplineError):
msg = (
'Exchange {exchange} contains no valid market so it is unusable in '
catalyst/exchange/utils/exchange_utils.py
+5 -15
View File
@@ -17,7 +17,6 @@ from catalyst.exchange.utils.serialization_utils import ExchangeJSONEncoder, \
ExchangeJSONDecoder, ConfigJSONEncoder
from catalyst.utils.paths import data_root, ensure_directory, \
last_modified_time
from catalyst.exchange.utils.datetime_utils import get_periods_range
def get_sid(symbol):
@@ -673,24 +672,15 @@ def transform_candles_to_df(candles):
return pd.DataFrame(candles).set_index('last_traded')
def get_candles_df(candles, field, freq, bar_count, end_dt=None):
def get_candles_df(candles, field, freq, bar_count, end_dt):
all_series = dict()
for asset in candles:
asset_df = transform_candles_to_df(candles[asset])
rounded_end_dt = end_dt.round(freq)
periods = get_periods_range(
start_dt=None, end_dt=rounded_end_dt,
freq=freq, periods=bar_count
)
if rounded_end_dt > end_dt:
periods = periods[:-1]
elif rounded_end_dt <= end_dt:
periods = periods[1:]
# periods = pd.date_range(end=end_dt, periods=bar_count, freq=freq)
rounded_end_dt = end_dt.floor(freq)
periods = pd.date_range(end=rounded_end_dt,
periods=bar_count,
freq=freq)
asset_df = forward_fill_df_if_needed(asset_df, periods)
all_series[asset] = pd.Series(asset_df[field])
catalyst/marketplace/marketplace.py
+24 -19
View File
@@ -23,7 +23,7 @@ from catalyst.exchange.utils.stats_utils import set_print_settings
from catalyst.marketplace.marketplace_errors import (
MarketplacePubAddressEmpty, MarketplaceDatasetNotFound,
MarketplaceNoAddressMatch, MarketplaceHTTPRequest,
MarketplaceNoCSVFiles)
MarketplaceNoCSVFiles, MarketplaceRequiresPython3)
from catalyst.marketplace.utils.auth_utils import get_key_secret, \
get_signed_headers
from catalyst.marketplace.utils.bundle_utils import merge_bundles
@@ -44,7 +44,10 @@ log = logbook.Logger('Marketplace', level=LOG_LEVEL)
class Marketplace:
def __init__(self):
global Web3
from web3 import Web3, HTTPProvider
try:
from web3 import Web3, HTTPProvider
except ImportError:
raise MarketplaceRequiresPython3()
self.addresses = get_user_pubaddr()
@@ -60,7 +63,8 @@ class Marketplace:
contract_url = urllib.urlopen(MARKETPLACE_CONTRACT)
self.mkt_contract_address = Web3.toChecksumAddress(
contract_url.readline().strip())
contract_url.readline().decode(
contract_url.info().get_content_charset()).strip())
abi_url = urllib.urlopen(MARKETPLACE_CONTRACT_ABI)
abi = json.load(abi_url)
@@ -73,7 +77,8 @@ class Marketplace:
contract_url = urllib.urlopen(ENIGMA_CONTRACT)
self.eng_contract_address = Web3.toChecksumAddress(
contract_url.readline().strip())
contract_url.readline().decode(
contract_url.info().get_content_charset()).strip())
abi_url = urllib.urlopen(ENIGMA_CONTRACT_ABI)
abi = json.load(abi_url)
@@ -148,13 +153,13 @@ class Marketplace:
'Gas Price:\t\t[Accept the default value]\n'
'Nonce:\t\t\t{nonce}\n'
'Data:\t\t\t{data}\n'.format(
_from=from_address,
to=tx['to'],
value=tx['value'],
gas=tx['gas'],
nonce=tx['nonce'],
data=tx['data'], )
)
_from=from_address,
to=tx['to'],
value=tx['value'],
gas=tx['gas'],
nonce=tx['nonce'],
data=tx['data'], )
)
signed_tx = input('Copy and Paste the "Signed Transaction" '
'field here:\n')
@@ -259,14 +264,14 @@ class Marketplace:
'buy: {} ENG. Get enough ENG to cover the costs of the '
'monthly\nsubscription for what you are trying to buy, '
'and try again.'.format(
address, from_grains(balance), price))
address, from_grains(balance), price))
return
while True:
agree_pay = input('Please confirm that you agree to pay {} ENG '
'for a monthly subscription to the dataset "{}" '
'starting today. [default: Y] '.format(
price, dataset)) or 'y'
price, dataset)) or 'y'
if agree_pay.lower() not in ('y', 'n'):
print("Please answer Y or N.")
else:
@@ -369,7 +374,7 @@ class Marketplace:
'You can now ingest this dataset anytime during the '
'next month by running the following command:\n'
'catalyst marketplace ingest --dataset={}'.format(
dataset, address, dataset))
dataset, address, dataset))
def process_temp_bundle(self, ds_name, path):
"""
@@ -426,10 +431,10 @@ class Marketplace:
print('Your subscription to dataset "{}" expired on {} UTC.'
'Please renew your subscription by running:\n'
'catalyst marketplace subscribe --dataset={}'.format(
ds_name,
pd.to_datetime(check_sub[4], unit='s', utc=True),
ds_name)
)
ds_name,
pd.to_datetime(check_sub[4], unit='s', utc=True),
ds_name)
)
if 'key' in self.addresses[address_i]:
key = self.addresses[address_i]['key']
@@ -621,7 +626,7 @@ class Marketplace:
)
except Exception as e:
print('Unable to subscribe to data source: {}'.format(e))
print('Unable to register the requested dataset: {}'.format(e))
return
self.check_transaction(tx_hash)
catalyst/marketplace/marketplace_errors.py
+10 -1
View File
@@ -9,7 +9,8 @@ def silent_except_hook(exctype, excvalue, exctraceback):
MarketplaceNoAddressMatch, MarketplaceHTTPRequest,
MarketplaceNoCSVFiles, MarketplaceContractDataNoMatch,
MarketplaceSubscriptionExpired, MarketplaceJSONError,
MarketplaceWalletNotSupported, MarketplaceEmptySignature]:
MarketplaceWalletNotSupported, MarketplaceEmptySignature,
MarketplaceRequiresPython3]:
fn = traceback.extract_tb(exctraceback)[-1][0]
ln = traceback.extract_tb(exctraceback)[-1][1]
print("Error traceback: {1} (line {2})\n"
@@ -86,3 +87,11 @@ class MarketplaceJSONError(ZiplineError):
'The configuration file {file} is malformed. Please correct '
'the following error:\n{error}'
)
class MarketplaceRequiresPython3(ZiplineError):
msg = (
'\nCatalyst requires Python3 to access the Enigma Data Marketplace.\n'
'If you want to use the Data Marketplace, you need to reinstall '
'Catalyst\nwith Python3. See the documentation website for additional '
'information.')
etc/python2.7-environment.yml
+3 -1
View File
@@ -23,7 +23,9 @@ dependencies:
- bottleneck==1.2.1
- chardet==3.0.4
- ccxt==1.10.1094
- web3==4.0.0b7
# The Enigma Data Marketplace requires Python3 because it depends on
# web3, which requires Python3, as building its dependencies breaks in Python2
# - web3==4.0.0b7
- requests-toolbelt==0.8.0
- click==6.7
- contextlib2==0.5.5
etc/requirements.txt
+1 -1
View File
@@ -84,5 +84,5 @@ tables==3.3.0
ccxt==1.10.1094
boto3==1.4.8
redo==1.6
web3==4.0.0b7
web3==4.0.0b11; python_version > '3.4'
requests-toolbelt==0.8.0
etc/requirements_marketplace.txt
-2
View File
@@ -1,2 +0,0 @@
web3==4.0.0b7
requests-toolbelt==0.8.0
tests/exchange/test_bundle.py
+6 -4
View File
@@ -51,8 +51,10 @@ class TestExchangeBundle:
exchange.get_asset('eng_eth')
]
start = pd.to_datetime('2018-02-01', utc=True)
end = pd.to_datetime('2018-02-10', utc=True)
# start = pd.to_datetime('2018-02-01', utc=True)
# end = pd.to_datetime('2018-02-28', utc=True)
start = None
end = None
log.info('ingesting exchange bundle {}'.format(exchange_name))
exchange_bundle.ingest(
@@ -60,8 +62,8 @@ class TestExchangeBundle:
include_symbols=','.join([asset.symbol for asset in assets]),
# include_symbols=None,
exclude_symbols=None,
start=start,
end=end,
start=None,
end=None,
show_progress=False,
show_breakdown=False
)
tests/exchange/test_exchange_utils.py
+61 -48
View File
@@ -2,6 +2,7 @@ from catalyst.exchange.utils.exchange_utils import transform_candles_to_df, \
forward_fill_df_if_needed, get_candles_df
from catalyst.testing.fixtures import WithLogger, ZiplineTestCase
from datetime import timedelta
from pandas import Timestamp, DataFrame, concat
import numpy as np
@@ -15,9 +16,59 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase):
new_df.index.name = None
return new_df
@classmethod
def verify_forward_fill_df_if_needed(cls, candles, periods, expected_df):
observed_df = forward_fill_df_if_needed(
transform_candles_to_df(candles),
periods)
assert (expected_df.equals(observed_df))
@classmethod
def verify_get_candles_df(cls, assets, candles, end_fixed_dt,
expected_df, check_next_candle=False):
# run on all the fields
for field in ['volume', 'open', 'close', 'high', 'low']:
field_dt = cls.get_specific_field_from_df(expected_df,
field,
assets[0])
# run on several timestamps
for delta in range(5):
end_dt = end_fixed_dt + timedelta(minutes=delta)
assert (field_dt.equals(get_candles_df({assets[0]: candles},
field, '5T', 3,
end_dt=end_dt)))
field_dt_a1 = cls.get_specific_field_from_df(expected_df,
field,
assets[0])
field_dt_a2 = cls.get_specific_field_from_df(expected_df,
field,
assets[1])
observed_df = get_candles_df({assets[0]: candles,
assets[1]: candles},
field, '5T', 3,
end_dt=end_dt)
assert (observed_df.equals(concat([field_dt_a1, field_dt_a2],
axis=1)))
if check_next_candle:
# one candle forward
end_dt = end_fixed_dt + timedelta(minutes=6)
observed_df = get_candles_df({assets[0]: candles,
assets[1]: candles},
field, '5T', 3,
end_dt=end_dt)
assert (not observed_df.equals(concat([field_dt_a1,
field_dt_a2],
axis=1)))
assert (concat([field_dt_a1, field_dt_a2],
axis=1)[1:].equals(observed_df[:-1]))
def test_get_candles_df(self):
asset = 'btc_usdt'
asset2 = 'eth_usdt'
assets = ['btc_usdt', 'eth_usdt']
# test forward fill in the end
candles = [{'high': 595, 'volume': 10, 'low': 594,
@@ -51,20 +102,12 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase):
Timestamp('2018-03-01 09:50:00+0000', tz='UTC'),
Timestamp('2018-03-01 09:55:00+0000', tz='UTC')]
observed_df = forward_fill_df_if_needed(
transform_candles_to_df(candles),
periods)
expected_df = transform_candles_to_df(expected)
assert (expected_df.equals(observed_df))
for field in ['volume', 'open', 'close', 'high', 'low']:
field_dt = self.get_specific_field_from_df(expected_df,
field,
asset)
assert (field_dt.equals(get_candles_df({asset: candles},
field, '5T', 3,
end_dt=periods[2])))
self.verify_forward_fill_df_if_needed(candles, periods,
expected_df)
self.verify_get_candles_df(assets, candles, periods[2],
expected_df, True)
# test forward fill in the middle
candles = [{'high': 595, 'volume': 10, 'low': 594,
@@ -94,28 +137,9 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase):
tz='UTC')
}]
df = transform_candles_to_df(candles)
observed_df = forward_fill_df_if_needed(df, periods)
expected_df = transform_candles_to_df(expected)
assert (expected_df.equals(observed_df))
for field in ['volume', 'open', 'close', 'high', 'low']:
# test several assets as well
observed_df = get_candles_df({asset: candles,
asset2: candles},
field, '5T', 3,
end_dt=periods[2])
field_dt_a1 = self.get_specific_field_from_df(expected_df,
field,
asset)
field_dt_a2 = self.get_specific_field_from_df(expected_df,
field,
asset2)
assert(observed_df.equals(concat([field_dt_a1, field_dt_a2],
axis=1)))
self.verify_forward_fill_df_if_needed(candles, periods, expected_df)
self.verify_get_candles_df(assets, candles, periods[2], expected_df)
# test "forward fill" at the beginning
candles = [{'high': 595, 'volume': 10, 'low': 594,
@@ -145,18 +169,7 @@ class TestExchangeUtils(WithLogger, ZiplineTestCase):
tz='UTC')
}]
df = transform_candles_to_df(candles)
observed_df = forward_fill_df_if_needed(df, periods)
expected_df = transform_candles_to_df(expected)
assert (expected_df.equals(observed_df))
self.verify_forward_fill_df_if_needed(candles, periods, expected_df)
# Not the same due to dropna - commenting out for now
"""
for field in ['volume', 'open', 'close', 'high', 'low']:
field_dt = self.get_specific_field_from_df(observed_df,
field,
asset)
assert(field_dt.equals(get_candles_df({asset:candles},
field, '5T', 3,
end_dt=periods[2])))
"""
# self.verify_get_candles_df(assets, candles, periods[2], expected_df)