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

Fixed some issues and optimized data.history() in live mode

Browse Source
This commit is contained in:
fredfortier
2017-10-19 05:19:01 -04:00
parent 6097128d5c
commit 51172759d3
6 changed files with 172 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/examples/simple_loop.py
+15 -5
View File
@@ -27,15 +27,25 @@ def handle_data(context, data):
pass
# run_algorithm(
# capital_base=250,
# start=pd.to_datetime('2017-08-01', utc=True),
# end=pd.to_datetime('2017-9-30', utc=True),
# data_frequency='daily',
# initialize=initialize,
# handle_data=handle_data,
# analyze=None,
# exchange_name='poloniex',
# algo_namespace='simple_loop',
# base_currency='eth'
# )
run_algorithm(
capital_base=250,
start=pd.to_datetime('2015-2-19', utc=True),
end=pd.to_datetime('2017-9-30', utc=True),
data_frequency='daily',
initialize=initialize,
handle_data=handle_data,
analyze=None,
exchange_name='poloniex',
live=True,
algo_namespace='simple_loop',
base_currency='eth'
base_currency='eth',
live_graph=False
)
catalyst/exchange/bundle_utils.py
+17 -4
View File
@@ -135,11 +135,11 @@ def get_adj_dates(start, end, assets, data_frequency):
end = last_entry
if end is None:
raise NoDataAvailableOnExchange(
raise NoDataAvailableOnExchange(
exchange=asset.exchange.title(),
symbol=[asset.symbol.encode('utf-8')],
data_frequency=data_frequency,
)
)
if end is None or start >= end:
raise PricingDataBeforeTradingError(
@@ -184,8 +184,21 @@ def get_year_start_end(dt):
return year_start, year_end
def get_ffill_candles(candles, bar_count, end_dt, data_frequency,
previous_candle=None):
def get_df_from_arrays(arrays, periods):
ohlcv = dict()
for index, field in enumerate(
['open', 'high', 'low', 'close', 'volume']):
ohlcv[field] = arrays[index].flatten()
df = pd.DataFrame(
data=ohlcv,
index=periods
)
return df
def get_df_from_candles(candles, bar_count, end_dt, data_frequency,
previous_candle=None):
"""
Create candles for each period of the specified range, forward-filling
missing candles with the previous value.
catalyst/exchange/data_portal_exchange.py
+14
View File
@@ -12,6 +12,7 @@
# limitations under the License.
import abc
from datetime import timedelta
from time import sleep
import pandas as pd
@@ -19,6 +20,7 @@ from catalyst.assets._assets import TradingPair
from logbook import Logger
from catalyst.data.data_portal import DataPortal
from catalyst.errors import HistoryWindowStartsBeforeData
from catalyst.exchange.exchange_bundle import ExchangeBundle
from catalyst.exchange.exchange_errors import (
ExchangeRequestError,
@@ -293,6 +295,18 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
session = self.trading_calendar.minute_to_session_label(end_dt)
dts = self._get_days_for_window(session, bar_count)
if len(dts) == 0:
symbols = [asset.symbol for asset in assets]
raise PricingDataNotLoadedError(
field=field,
symbols=symbols,
exchange=exchange.name,
first_trading_day= \
min([asset.start_date for asset in assets]),
data_frequency=data_frequency,
symbol_list=','.join(symbols)
)
self.ensure_after_first_day(dts[0], assets)
else:
catalyst/exchange/exchange.py
+114 -105
View File
@@ -1,5 +1,4 @@
import abc
import random
import re
from abc import ABCMeta, abstractmethod, abstractproperty
from datetime import timedelta
@@ -12,7 +11,8 @@ from logbook import Logger
from catalyst.data.data_portal import BASE_FIELDS
from catalyst.exchange.bundle_utils import get_start_dt, \
get_delta, get_trailing_candles_dt, get_periods, get_adj_dates
get_delta, get_trailing_candles_dt, get_periods, get_adj_dates, \
get_df_from_candles
from catalyst.exchange.exchange_bundle import ExchangeBundle
from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \
InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \
@@ -100,12 +100,6 @@ class Exchange:
delta = now - cpt_date
sleep_period = 60 - delta.total_seconds()
# log.debug(
# 'max requests {} reached, sleeping for {} seconds'.format(
# self.max_requests_per_minute,
# sleep_period
# ))
sleep(sleep_period)
now = pd.Timestamp.utcnow()
@@ -174,7 +168,8 @@ class Exchange:
asset = self.assets[key]
if not asset:
supported_symbols = [pair.symbol.encode('utf-8') for pair in self.assets.values()]
supported_symbols = [pair.symbol.encode('utf-8') for pair in
self.assets.values()]
raise SymbolNotFoundOnExchange(
symbol=symbol,
exchange=self.name.title(),
@@ -367,35 +362,76 @@ class Exchange:
)
)
# Don't use a timezone here
dt = pd.Timestamp.utcnow().floor('1 min')
ohlc = self.get_candles(data_frequency, asset)
if field not in ohlc:
raise KeyError('Invalid column: %s' % field)
if self.minute_writer is not None:
df = pd.DataFrame(
[ohlc],
index=pd.DatetimeIndex([dt]),
columns=['open', 'high', 'low', 'close', 'volume']
)
try:
# TODO: use victor's modified branch using int64
self.minute_writer.write_sid(
sid=asset.sid,
df=df
)
log.debug('wrote minute data: {}'.format(dt))
except Exception as e:
log.warn(
'unable to write minute data: {} {}'.format(dt, e))
value = ohlc[field]
log.debug('got spot value: {}'.format(value))
value = ohlc[field]
log.debug('got spot value: {}'.format(value))
return value
def get_series_from_bundle(self, assets, start_dt, end_dt, data_frequency,
field):
"""
:return:
"""
reader = self.bundle.get_reader(data_frequency)
if reader is None:
raise BundleNotFoundError(
exchange=self.name.title(),
data_frequency=data_frequency
)
series = dict()
try:
arrays = reader.load_raw_arrays(
sids=[asset.sid for asset in assets],
fields=[field],
start_dt=start_dt,
end_dt=end_dt
)
periods = self.bundle.get_calendar_periods_range(
start_dt, end_dt, data_frequency
)
for asset_index, asset in enumerate(assets):
asset_values = arrays[asset_index]
value_series = pd.Series(asset_values[0], index=periods)
series[asset] = value_series
except Exception as e:
log.debug('unable to retreive from bundle: {}'.format(e))
return series
def get_series_from_candles(self, candles, start_dt, end_dt,
field, previous_value=None):
"""
Get a series of field data for the specified candles.
:param candles:
:param start_dt:
:param end_dt:
:param field:
:param previous_value:
:return:
"""
dates = [candle['last_traded'] for candle in candles]
values = [candle[field] for candle in candles]
periods = pd.date_range(start_dt, end_dt)
series = pd.Series(values, index=dates)
series.reindex(periods, method='ffill', fill_value=previous_value)
return series
def get_history_window(self,
assets,
end_dt,
@@ -448,11 +484,8 @@ class Exchange:
raise InvalidHistoryFrequencyError(frequency)
if unit.lower() == 'd':
if data_frequency != 'daily':
raise MismatchingFrequencyError(
frequency=frequency,
data_frequency=data_frequency
)
if data_frequency == 'minute':
data_frequency = 'daily'
elif unit.lower() == 'm':
if data_frequency != 'minute':
@@ -467,94 +500,70 @@ class Exchange:
adj_bar_count = candle_size * bar_count
start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency)
start_dt, end_dt = get_adj_dates(start_dt, end_dt, assets,
data_frequency)
adj_start_dt, adj_end_dt = get_adj_dates(
start_dt, end_dt, assets, data_frequency
)
missing_assets = self.bundle.filter_existing_assets(
assets=assets,
start_dt=start_dt,
end_dt=end_dt,
start_dt=adj_start_dt,
end_dt=adj_end_dt,
data_frequency=data_frequency
)
if missing_assets:
self.bundle.ingest_assets(
assets=assets,
start_dt=start_dt,
end_dt=end_dt,
start_dt=adj_start_dt,
end_dt=adj_end_dt,
data_frequency=data_frequency
)
# We check again for data which may be too recent for the consolidated
# exchanges service
trailing_assets = self.bundle.filter_existing_assets(
series = self.get_series_from_bundle(
assets=assets,
start_dt=start_dt,
end_dt=end_dt,
data_frequency=data_frequency
start_dt=adj_start_dt,
end_dt=adj_end_dt,
data_frequency=data_frequency,
field=field
)
if trailing_assets:
# Adding bars too recent to be contained in the consolidated
# exchanges bundles. We go directly against the exchange
# to retrieve the candles.
for asset in trailing_assets:
trailing_candles_dt = get_trailing_candles_dt(
asset=asset,
start_dt=start_dt,
end_dt=end_dt,
data_frequency=data_frequency
for asset in assets:
if asset not in series or series[asset].index[-1] < end_dt:
# Adding bars too recent to be contained in the consolidated
# exchanges bundles. We go directly against the exchange
# to retrieve the candles.
trailing_dt = \
series[asset].index[-1] + get_delta(1, data_frequency) \
if asset in series else start_dt
trailing_bar_count = \
get_periods(trailing_dt, end_dt, data_frequency)
# The get_history method supports multiple asset
candles = self.get_candles(
data_frequency=data_frequency,
assets=asset,
bar_count=trailing_bar_count,
end_dt=end_dt
)
if trailing_candles_dt is not None:
trailing_bar_count = \
get_periods(start_dt, end_dt, data_frequency)
last_value = series[asset].iloc(0) if asset in series \
else np.nan
# The get_history method supports multiple asset
candles = self.get_candles(
data_frequency=data_frequency,
assets=[asset],
bar_count=trailing_bar_count,
end_dt=end_dt
)
candle_series = self.get_series_from_candles(
candles=candles,
start_dt=trailing_dt,
end_dt=end_dt,
field=field,
previous_value=last_value
)
# TODO: Do I need the previous_candle?
self.bundle.ingest_candles(
candles=candles,
bar_count=trailing_bar_count,
start_dt=start_dt,
end_dt=end_dt,
data_frequency=data_frequency
)
if asset in series:
series[asset].append(candle_series)
reader = self.bundle.get_reader(data_frequency)
if reader is None:
raise BundleNotFoundError(
exchange=self.name.title(),
data_frequency=data_frequency
)
values = reader.load_raw_arrays(
sids=[asset.sid for asset in assets],
fields=[field],
start_dt=start_dt,
end_dt=end_dt
)
series = dict()
for asset_index, asset in enumerate(assets):
all_dates = []
asset_values = []
# TODO: use numpy to avoid the loop
date = start_dt
for value in values[0]:
all_dates.append(date)
asset_values.append(value[asset_index])
date += get_delta(1, data_frequency)
value_series = pd.Series(asset_values, index=all_dates)
series[asset] = value_series
else:
series[asset] = candle_series
df = pd.DataFrame(series)
catalyst/exchange/exchange_bcolz.py
+6 -4
View File
@@ -61,12 +61,13 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader):
num_days = len(periods)
shape = num_days, len(sids)
if len(fields) == 1 and fields[0] == 'volume':
fields.insert(0, 'close')
all_fields = fields[:]
if len(all_fields) == 1 and all_fields[0] == 'volume':
all_fields.insert(0, 'close')
mask = None
data = []
for field in fields:
for field in all_fields:
if field != 'volume':
out = np.full(shape, np.nan)
else:
@@ -83,6 +84,7 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader):
a[mask] * self._ohlc_ratio_inverse_for_sid(sid)
)
data.append(out)
if field in fields:
data.append(out)
return data
catalyst/exchange/exchange_bundle.py
+6 -88
View File
@@ -2,16 +2,15 @@ import os
import shutil
from datetime import timedelta
import numpy as np
import pandas as pd
from logbook import Logger, INFO
from catalyst import get_calendar
from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
BcolzMinuteBarMetadata
from catalyst.exchange.bundle_utils import get_ffill_candles, range_in_bundle, \
from catalyst.exchange.bundle_utils import range_in_bundle, \
get_bcolz_chunk, get_delta, get_adj_dates, get_month_start_end, \
get_year_start_end, get_periods_range
get_year_start_end, get_periods_range, get_df_from_arrays
from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
BcolzExchangeBarWriter
from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
@@ -192,79 +191,6 @@ class ExchangeBundle:
invalid_data_behavior='raise'
)
def ingest_candles(self, candles, bar_count, start_dt, end_dt,
data_frequency,
previous_candle=dict()):
"""
Ingest candles obtained via the get_candles API of an exchange.
Since exchange APIs generally only do not return candles when there
are no transactions in the period, we ffill values using the
previous candle to ensure that each period has a candle.
:param bar_count:
:param end_dt:
:param data_frequency:
:param asset:
:param writer:
:param previous_candle
:return:
"""
writer = self.get_writer(start_dt, end_dt, data_frequency)
num_candles = 0
data = []
for asset in candles:
asset_candles = candles[asset]
if not asset_candles:
log.debug(
'no data: {symbols} on {exchange}, date {end}'.format(
symbols=asset,
exchange=self.exchange.name,
end=end_dt
)
)
continue
previous = previous_candle[asset] \
if asset in previous_candle else None
all_dates, all_candles = get_ffill_candles(
candles=asset_candles,
bar_count=bar_count,
end_dt=end_dt,
data_frequency=data_frequency,
previous_candle=previous
)
previous_candle[asset] = all_candles[-1]
df = pd.DataFrame(
data=all_candles,
index=all_dates,
columns=['open', 'high', 'low', 'close', 'volume']
)
if not df.empty:
df.sort_index(inplace=True)
sid = asset.sid
num_candles += len(df.values)
data.append((sid, df))
log.debug(
'writing {num_candles} candles for {bar_count} bars'
'ending {end}'.format(
num_candles=num_candles,
bar_count=bar_count,
end=end_dt
)
)
self._write(data, writer, data_frequency)
return data
def get_calendar_periods_range(self, start_dt, end_dt, data_frequency):
return self.calendar.minutes_in_range(start_dt, end_dt) \
if data_frequency == 'minute' \
@@ -295,9 +221,6 @@ class ExchangeBundle:
period=period
)
periods = self.get_calendar_periods_range(
start_dt, end_dt, data_frequency
)
reader = self.get_reader(data_frequency, path=path)
if reader is None:
raise TempBundleNotFoundError(path=path)
@@ -312,16 +235,12 @@ class ExchangeBundle:
if not arrays:
return path
ohlcv = dict()
for index, field in enumerate(
['open', 'high', 'low', 'close', 'volume']):
ohlcv[field] = arrays[index].flatten()
df = pd.DataFrame(
data=ohlcv,
index=periods
periods = self.get_calendar_periods_range(
start_dt, end_dt, data_frequency
)
df = get_df_from_arrays(arrays, periods)
if empty_rows_behavior is not 'ignore':
nan_rows = df[df.isnull().T.any().T].index
@@ -369,7 +288,6 @@ class ExchangeBundle:
if not df.empty:
df.sort_index(inplace=True)
data.append((asset.sid, df))
self._write(data, writer, data_frequency)
if cleanup: