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

Added exchange get_history method which merge historical bars from the Catalyst and exchange APIs

Browse Source
This commit is contained in:
fredfortier
2017-10-09 16:23:02 -04:00
parent 8811aa669a
commit 83af12c52c
3 changed files with 162 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/exchange/bundle_utils.py
+41 -12
View File
@@ -6,6 +6,7 @@ import pandas as pd
import pytz
from catalyst.data.bundles import from_bundle_ingest_dirname
from catalyst.utils.deprecate import deprecated
from catalyst.utils.paths import data_path
log = Logger('test_exchange_bundle')
@@ -106,6 +107,46 @@ def get_history(exchange_name, data_frequency, symbol, start=None, end=None):
return data
def get_ffill_candles(candles, start_dt, end_dt, data_frequency,
previous_candle=None):
"""
Create candles for each period of the specified range, forward-filling
missing candles with the previous value.
:param candles:
:param start_dt:
:param end_dt:
:param data_frequency:
:param previous_candle:
:return:
"""
all_dates = []
all_candles = []
date = start_dt
while date <= end_dt:
candle = next((
candle for candle in candles if candle['last_traded'] == date
), previous_candle)
if candle is not None:
all_dates.append(date)
all_candles.append(candle)
previous_candle = candle
if data_frequency == 'minute':
date += datetime.timedelta(minutes=1)
elif data_frequency == 'daily':
date += datetime.timedelta(days=1)
else:
raise ValueError('invalid data frequency')
return all_dates, all_candles
@deprecated
def get_history_mock(exchange_name, data_frequency, symbol, start_ms, end_ms,
exchanges):
"""
@@ -174,18 +215,6 @@ def get_history_mock(exchange_name, data_frequency, symbol, start_ms, end_ms,
return ohlcv
def fetch_candles_chunk(exchange, assets, data_frequency, end_dt, bar_count):
calc_start_dt = end_dt - datetime.timedelta(minutes=bar_count)
candles = exchange.get_candles(
data_frequency=data_frequency,
assets=assets,
bar_count=bar_count,
start_dt=calc_start_dt,
end_dt=end_dt
)
return candles
def find_most_recent_time(bundle_name):
"""
Find most recent "time folder" for a given bundle.
catalyst/exchange/exchange.py
+94 -3
View File
@@ -10,6 +10,7 @@ from catalyst.assets._assets import TradingPair
from logbook import Logger
from catalyst.data.data_portal import BASE_FIELDS
from catalyst.exchange import bundle_utils
from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \
InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange
from catalyst.exchange.exchange_execution import ExchangeStopLimitOrder, \
@@ -412,6 +413,95 @@ class Exchange:
return value
def get_history(self, assets, end_dt, bar_count, data_frequency):
"""
Retrieve OHLCV bars from the Catalyst and/or exchange API.
If Catalyst does not have the full data set, retrieve the missing
portion from the exchange API if the exchanges supports
historical data.
:param assets: list[TradingPair]
The TradingPair asset.
:param data_frequency: str
The bar frequency: daily or minute
:param bar_count: int
The number of bars desired.
:param end: datetime
The last trading date of the last bar.
:return:
"""
candles = dict()
for asset in assets:
candles[asset] = self.get_asset_history(
asset=asset,
end=end_dt,
bar_count=bar_count,
data_frequency=data_frequency
)
return candles
def get_asset_history(self, asset, end, bar_count, data_frequency):
"""
Retrieve the OHLVC bars of a single asset.
:param asset: TradingPair
The TradingPair asset.
:param data_frequency: str
The bar frequency: daily or minute
:param bar_count: int
The number of bars desired.
:param end: datetime
The last trading date of the last bar.
:return:
"""
start = end - timedelta(minutes=bar_count)
exchange_start = None
catalyst_end = None
if start < asset.end_minute:
catalyst_start = start
if end <= asset.end_minute:
catalyst_end = end
else:
catalyst_end = asset.end_minute
delta = timedelta(minutes=1) \
if data_frequency == 'minute' else timedelta(days=1)
exchange_start = catalyst_end + delta
exchange_end = end
else:
exchange_start = start
exchange_end = end
data = []
if catalyst_end is not None:
# TODO: support multiple assets in the Catalyst API.
candles = bundle_utils.get_history(
exchange_name=self.name,
data_frequency=data_frequency,
symbol=asset.exchange_symbol, # TODO: use Catalyst symbol
start=catalyst_start,
end=catalyst_end
)
data += candles
if exchange_start is not None:
candles = self.get_candles(
data_frequency=data_frequency,
assets=[asset],
bar_count=bar_count,
start_dt=exchange_start,
end_dt=exchange_end
)
data += candles[asset]
return data
def get_history_window(self,
assets,
end_dt,
@@ -455,11 +545,12 @@ class Exchange:
A dataframe containing the requested data.
"""
candles = self.get_candles(
data_frequency=frequency,
# TODO: try to read from bundle first
candles = self.get_history(
assets=assets,
end_dt=end_dt,
bar_count=bar_count,
end_dt=end_dt
data_frequency=data_frequency
)
series = dict()
catalyst/exchange/exchange_bundle.py
+27 -50
View File
@@ -10,8 +10,7 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
BcolzMinuteBarWriter, BcolzMinuteBarReader
from catalyst.data.us_equity_pricing import BcolzDailyBarWriter, \
BcolzDailyBarReader
from catalyst.exchange.bundle_utils import fetch_candles_chunk, get_history, \
get_seconds_from_date
from catalyst.exchange.bundle_utils import get_ffill_candles
from catalyst.exchange.exchange_utils import get_exchange_folder
from catalyst.exchange.init_utils import get_exchange
from catalyst.utils.cli import maybe_show_progress
@@ -196,6 +195,16 @@ class ExchangeBundle:
def ingest_chunk(self, chunk, previous_candle, data_frequency, assets,
writer):
"""
Retrieve the specified OHLCV chunk and write it to the bundle
:param chunk:
:param previous_candle:
:param data_frequency:
:param assets:
:param writer:
:return:
"""
chunk_end = chunk['end']
chunk_start = chunk_end - timedelta(minutes=chunk['bar_count'])
@@ -215,36 +224,12 @@ class ExchangeBundle:
log.debug('the data chunk already exists')
return
candles = dict()
for asset in missing_assets:
if chunk_start < asset.end_minute:
# TODO: fetch delta candles from exchanges
history_end = chunk_end \
if chunk_end <= asset.end_minute else asset.end_minute
# TODO: switch to Catalyst symbol convention
candles[asset] = get_history(
exchange_name=self.exchange.name,
data_frequency=data_frequency,
symbol=asset.exchange_symbol,
start=chunk_start,
end=history_end
)
else:
log.debug(
'no data in Catalyst api for chunk '
'{} to {}'.format(chunk_start, chunk_end)
)
# if data_frequency == 'minute':
# # TODO: ensure correct behavior for assets starting in the chunk
# candles = fetch_candles_chunk(
# exchange=self.exchange,
# assets=missing_assets,
# data_frequency=data_frequency,
# end_dt=chunk_end,
# bar_count=chunk['bar_count']
# )
# else:
candles = self.exchange.get_history(
assets=missing_assets,
end_dt=chunk_end,
bar_count=chunk['bar_count'],
data_frequency=data_frequency
)
num_candles = 0
data = []
@@ -260,25 +245,17 @@ class ExchangeBundle:
)
continue
all_dates = []
all_candles = []
date = chunk_start
while date <= chunk_end:
previous = previous_candle[asset] \
if asset in previous_candle else None
previous = previous_candle[asset] \
if asset in previous_candle else None
candle = next((candle for candle in asset_candles \
if candle['last_traded'] == date),
previous)
if candle is not None:
all_dates.append(date)
all_candles.append(candle)
previous_candle[asset] = candle
date += timedelta(minutes=1)
all_dates, all_candles = get_ffill_candles(
candles=asset_candles,
start_dt=chunk_start,
end_dt=chunk_end,
data_frequency=data_frequency,
previous_candle=previous
)
previous_candle[asset] = all_candles[-1]
df = pd.DataFrame(all_candles, index=all_dates)
if not df.empty: