mirror of
https://github.com/wassname/catalyst.git
synced 2026-09-12 12:12:04 +08:00
MAINT: PEP8 compliance
This commit is contained in:
@@ -1,6 +1,5 @@
|
||||
# These imports are necessary to force module-scope register calls to happen.
|
||||
from . import quandl # noqa
|
||||
from . import poloniex
|
||||
from .core import (
|
||||
UnknownBundle,
|
||||
bundles,
|
||||
|
||||
@@ -13,10 +13,9 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
|
||||
from itertools import count
|
||||
import tarfile
|
||||
from time import time, sleep
|
||||
from time import sleep
|
||||
|
||||
from abc import abstractmethod, abstractproperty
|
||||
import logbook
|
||||
@@ -37,6 +36,7 @@ log = logbook.Logger(__name__, level=LOG_LEVEL)
|
||||
|
||||
DEFAULT_RETRIES = 5
|
||||
|
||||
|
||||
class BaseBundle(object):
|
||||
def __init__(self, asset_filter=[]):
|
||||
self._asset_filter = asset_filter
|
||||
@@ -104,11 +104,11 @@ class BaseBundle(object):
|
||||
|
||||
def post_process_symbol_metadata(self, metadata, data):
|
||||
return metadata
|
||||
|
||||
|
||||
@abstractmethod
|
||||
def fetch_raw_symbol_frame(self, api_key, symbol, start_date, end_date):
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def ingest(self,
|
||||
environ,
|
||||
asset_db_writer,
|
||||
@@ -128,7 +128,7 @@ class BaseBundle(object):
|
||||
retries = environ.get('CATALYST_DOWNLOAD_ATTEMPTS', 5)
|
||||
|
||||
if is_compile:
|
||||
# User has instructed local compilation and ingestion of bundle.
|
||||
# User has instructed local compilation & ingestion of bundle.
|
||||
# Fetch raw metadata for all symbols.
|
||||
raw_metadata = self._fetch_metadata_frame(
|
||||
api_key,
|
||||
@@ -157,9 +157,9 @@ class BaseBundle(object):
|
||||
show_progress=show_progress,
|
||||
)
|
||||
|
||||
# Post-process metadata using cached symbol frames, and write to
|
||||
# disk. This metadata must be written before any attempt to write
|
||||
# minute data.
|
||||
# Post-process metadata using cached symbol frames, and write
|
||||
# to disk. This metadata must be written before any attempt
|
||||
# to write minute data.
|
||||
metadata = self._post_process_metadata(
|
||||
raw_metadata,
|
||||
cache,
|
||||
@@ -184,10 +184,11 @@ class BaseBundle(object):
|
||||
show_progress=show_progress,
|
||||
)
|
||||
|
||||
# For legacy purposes, this call is required to ensure the database
|
||||
# contains an appropriately initialized file structure. We don't
|
||||
# forsee a usecase for adjustments at this time, but may later
|
||||
# choose to expose this functionality in the future.
|
||||
# For legacy purposes, this call is required to ensure the
|
||||
# database contains an appropriately initialized file
|
||||
# structure. We don't forsee a usecase for adjustments at
|
||||
# this time, but may later choose to expose this functionality
|
||||
# in the future.
|
||||
adjustment_writer.write(
|
||||
splits=(
|
||||
pd.concat(self.splits, ignore_index=True)
|
||||
@@ -232,12 +233,12 @@ class BaseBundle(object):
|
||||
tar.extractall(output_dir)
|
||||
|
||||
def _fetch_metadata_frame(self,
|
||||
api_key,
|
||||
cache,
|
||||
retries=DEFAULT_RETRIES,
|
||||
environ=None,
|
||||
show_progress=False):
|
||||
|
||||
api_key,
|
||||
cache,
|
||||
retries=DEFAULT_RETRIES,
|
||||
environ=None,
|
||||
show_progress=False):
|
||||
|
||||
# Setup raw metadata iterator to fetch pages if necessary.
|
||||
raw_iter = self._fetch_metadata_iter(api_key, cache, retries, environ)
|
||||
|
||||
@@ -251,7 +252,7 @@ class BaseBundle(object):
|
||||
show_percent=False,
|
||||
) as blocks:
|
||||
metadata = pd.concat(blocks, ignore_index=True)
|
||||
|
||||
|
||||
return metadata
|
||||
|
||||
def _fetch_metadata_iter(self, api_key, cache, retries, environ):
|
||||
@@ -269,21 +270,20 @@ class BaseBundle(object):
|
||||
page_number,
|
||||
)
|
||||
break
|
||||
except ValueError as e:
|
||||
except ValueError:
|
||||
raw = pd.DataFrame([])
|
||||
break
|
||||
except Exception as e:
|
||||
except Exception:
|
||||
log.exception(
|
||||
'Failed to load metadata from {}. '
|
||||
'Retrying.'.format(self.name)
|
||||
)
|
||||
)
|
||||
else:
|
||||
raise ValueError(
|
||||
'Failed to download metadata page {} after {} '
|
||||
'attempts.'.format(page_number, retries)
|
||||
)
|
||||
|
||||
|
||||
if raw.empty:
|
||||
# Empty DataFrame signals completion.
|
||||
break
|
||||
@@ -305,7 +305,7 @@ class BaseBundle(object):
|
||||
columns=self.md_column_names,
|
||||
index=metadata.index,
|
||||
)
|
||||
|
||||
|
||||
# Iterate over the available symbols, loading the asset's raw symbol
|
||||
# data from the cache. The final metadata is computed and recorded in
|
||||
# the appropriate row depending on the asset's id.
|
||||
@@ -318,22 +318,22 @@ class BaseBundle(object):
|
||||
show_percent=False,
|
||||
) as symbols_map:
|
||||
for asset_id, symbol in symbols_map:
|
||||
# Attempt to load data from disk, the cache should have an entry
|
||||
# for each symbol at this point of the execution. If one does
|
||||
# not exist, we should fail.
|
||||
# Attempt to load data from disk, the cache should have an
|
||||
# entry for each symbol at this point of the execution. If one
|
||||
# does not exist, we should fail.
|
||||
key = '{sym}.daily.frame'.format(sym=symbol)
|
||||
try:
|
||||
raw_data = cache[key]
|
||||
except KeyError:
|
||||
raise ValueError(
|
||||
'Unable to find cached data for symbol: {0}'.format(symbol)
|
||||
)
|
||||
'Unable to find cached data for symbol:'
|
||||
' {0}'.format(symbol))
|
||||
|
||||
# Perform and require post-processing of metadata.
|
||||
final_symbol_metadata = self.post_process_symbol_metadata(
|
||||
asset_id,
|
||||
metadata.iloc[asset_id],
|
||||
raw_data,
|
||||
raw_data,
|
||||
)
|
||||
|
||||
# Record symbol's final metadata.
|
||||
@@ -363,8 +363,8 @@ class BaseBundle(object):
|
||||
# returns the cached data unaltered. The `should_sleep` flag
|
||||
# indicates that an API call was attempted, and that we should be
|
||||
# ensure aren't exceeding our rate limit before proceeding to the
|
||||
# next symbol. If the raw_data is updated, it is cached before being
|
||||
# returned.
|
||||
# next symbol. If the raw_data is updated, it is cached before
|
||||
# being returned.
|
||||
raw_data, should_sleep = self._maybe_update_symbol_frame(
|
||||
start_time,
|
||||
api_key,
|
||||
@@ -414,7 +414,7 @@ class BaseBundle(object):
|
||||
last = start_session
|
||||
if raw_data is not None and len(raw_data) > 0:
|
||||
last = raw_data.index[-1].tz_localize('UTC')
|
||||
|
||||
|
||||
should_sleep = False
|
||||
|
||||
# Determine time at which cached data will be considered stale.
|
||||
@@ -455,7 +455,7 @@ class BaseBundle(object):
|
||||
retries=DEFAULT_RETRIES):
|
||||
|
||||
# Data for symbol is old enough to attempt an update or is not
|
||||
# present in the cache. Fetch raw data for a single symbol
|
||||
# present in the cache. Fetch raw data for a single symbol
|
||||
# with requested intervals and frequency. Retry as necessary.
|
||||
for _ in range(retries):
|
||||
try:
|
||||
@@ -468,7 +468,6 @@ class BaseBundle(object):
|
||||
data_frequency,
|
||||
)
|
||||
raw_data.index = pd.to_datetime(raw_data.index, utc=True)
|
||||
#raw_data.index = raw_data.index.tz_localize('UTC')
|
||||
|
||||
# Filter incoming data to fit start and end sessions.
|
||||
raw_data = raw_data[
|
||||
@@ -482,7 +481,7 @@ class BaseBundle(object):
|
||||
|
||||
return raw_data
|
||||
|
||||
except Exception as e:
|
||||
except Exception:
|
||||
log.exception(
|
||||
'Exception raised fetching {name} data. Retrying.'
|
||||
.format(name=self.name)
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
from catalyst.data.bundles.base import BaseBundle
|
||||
from catalyst.utils.memoize import lazyval
|
||||
|
||||
|
||||
class BasePricingBundle(BaseBundle):
|
||||
@lazyval
|
||||
def md_dtypes(self):
|
||||
@@ -38,6 +39,7 @@ class BasePricingBundle(BaseBundle):
|
||||
('volume', 'float64'),
|
||||
]
|
||||
|
||||
|
||||
class BaseCryptoPricingBundle(BasePricingBundle):
|
||||
@lazyval
|
||||
def calendar_name(self):
|
||||
@@ -55,6 +57,7 @@ class BaseCryptoPricingBundle(BasePricingBundle):
|
||||
def dividends(self):
|
||||
return []
|
||||
|
||||
|
||||
class BaseEquityPricingBundle(BasePricingBundle):
|
||||
@lazyval
|
||||
def calendar_name(self):
|
||||
|
||||
@@ -37,6 +37,7 @@ from catalyst.utils.cli import maybe_show_progress
|
||||
|
||||
ONE_MEGABYTE = 1024 * 1024
|
||||
|
||||
|
||||
def asset_db_path(bundle_name, timestr, environ=None, db_version=None):
|
||||
return pth.data_path(
|
||||
asset_db_relative(bundle_name, timestr, environ, db_version),
|
||||
@@ -135,6 +136,7 @@ def ingestions_for_bundle(bundle, environ=None):
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
|
||||
def download_with_progress(url, chunk_size, **progress_kwargs):
|
||||
"""
|
||||
Download streaming data from a URL, printing progress information to the
|
||||
@@ -705,4 +707,5 @@ def _make_bundle_core():
|
||||
)
|
||||
|
||||
|
||||
bundles, register_bundle, register, unregister, ingest, load, clean = _make_bundle_core()
|
||||
bundles, register_bundle, register, unregister, ingest, load, clean = \
|
||||
_make_bundle_core()
|
||||
|
||||
@@ -14,19 +14,17 @@
|
||||
# limitations under the License.
|
||||
|
||||
import sys
|
||||
|
||||
from datetime import datetime
|
||||
from six.moves.urllib.parse import urlencode
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from six.moves.urllib.parse import urlencode
|
||||
|
||||
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):
|
||||
@@ -46,7 +44,8 @@ class PoloniexBundle(BaseCryptoPricingBundle):
|
||||
@lazyval
|
||||
def tar_url(self):
|
||||
return (
|
||||
'https://s3.amazonaws.com/enigmaco/catalyst-bundles/poloniex/poloniex-bundle.tar.gz'
|
||||
'https://s3.amazonaws.com/enigmaco/catalyst-bundles/'
|
||||
'poloniex/poloniex-bundle.tar.gz'
|
||||
)
|
||||
|
||||
@lazyval
|
||||
@@ -67,12 +66,11 @@ class PoloniexBundle(BaseCryptoPricingBundle):
|
||||
|
||||
raw = raw.sort_index().reset_index()
|
||||
raw.rename(
|
||||
columns={'index':'symbol'},
|
||||
columns={'index': 'symbol'},
|
||||
inplace=True,
|
||||
)
|
||||
|
||||
raw = raw[raw['isFrozen'] == 0]
|
||||
|
||||
return raw
|
||||
|
||||
def post_process_symbol_metadata(self, asset_id, sym_md, sym_data):
|
||||
@@ -98,7 +96,8 @@ class PoloniexBundle(BaseCryptoPricingBundle):
|
||||
frequency):
|
||||
|
||||
# TODO: replace this with direct exchange call
|
||||
# The end date and frequency should be used to calculate the number of bars
|
||||
# The end date and frequency should be used to
|
||||
# calculate the number of bars
|
||||
if(frequency == 'minute'):
|
||||
pc = PoloniexCurator()
|
||||
raw = pc.onemin_to_dataframe(symbol, start_date, end_date)
|
||||
@@ -116,8 +115,9 @@ class PoloniexBundle(BaseCryptoPricingBundle):
|
||||
)
|
||||
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
|
||||
# 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 = 1
|
||||
raw.loc[:, 'open'] /= scale
|
||||
@@ -139,7 +139,6 @@ class PoloniexBundle(BaseCryptoPricingBundle):
|
||||
|
||||
return self._format_polo_query(query_params)
|
||||
|
||||
|
||||
def _format_data_url(self,
|
||||
api_key,
|
||||
symbol,
|
||||
@@ -162,27 +161,26 @@ class PoloniexBundle(BaseCryptoPricingBundle):
|
||||
('end', end_date.value / 10**9),
|
||||
('period', period),
|
||||
]
|
||||
|
||||
|
||||
return self._format_polo_query(query_params)
|
||||
|
||||
|
||||
def _format_polo_query(self, query_params):
|
||||
# TODO: got against the exchange object
|
||||
return 'https://poloniex.com/public?{query}'.format(
|
||||
query=urlencode(query_params),
|
||||
)
|
||||
|
||||
'''
|
||||
As a second parameter, you can pass an array of currency pairs
|
||||
that will be processed as an asset_filter to only process that
|
||||
|
||||
'''
|
||||
As a second parameter, you can pass an array of currency pairs
|
||||
that will be processed as an asset_filter to only process that
|
||||
subset of assets in the bundle, such as:
|
||||
register_bundle(PoloniexBundle, ['USDT_BTC',])
|
||||
|
||||
For a production environment make sure to use (to bundle all pairs):
|
||||
register_bundle(PoloniexBundle)
|
||||
'''
|
||||
|
||||
if 'ingest' in sys.argv and '-c' in sys.argv:
|
||||
register_bundle(PoloniexBundle)
|
||||
else:
|
||||
register_bundle(PoloniexBundle, create_writers=False)
|
||||
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
from datetime import datetime
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from six.moves.urllib.parse import urlencode
|
||||
|
||||
from catalyst.data.bundles.core import register_bundle
|
||||
@@ -26,25 +25,16 @@ from catalyst.utils.memoize import lazyval
|
||||
"""
|
||||
Module for building a complete daily dataset from Quandl's WIKI dataset.
|
||||
"""
|
||||
from itertools import count
|
||||
import tarfile
|
||||
from time import time, sleep
|
||||
from datetime import datetime
|
||||
|
||||
from logbook import Logger
|
||||
import pandas as pd
|
||||
from six.moves.urllib.parse import urlencode
|
||||
|
||||
from catalyst.utils.calendars import register_calendar_alias
|
||||
from catalyst.utils.cli import maybe_show_progress
|
||||
|
||||
from . import core as bundles
|
||||
|
||||
from catalyst.constants import LOG_LEVEL
|
||||
from catalyst.utils.calendars import register_calendar_alias
|
||||
|
||||
|
||||
log = Logger(__name__, level=LOG_LEVEL)
|
||||
seconds_per_call = (pd.Timedelta('10 minutes') / 2000).total_seconds()
|
||||
|
||||
|
||||
class QuandlBundle(BaseEquityPricingBundle):
|
||||
@lazyval
|
||||
def name(self):
|
||||
@@ -109,8 +99,8 @@ class QuandlBundle(BaseEquityPricingBundle):
|
||||
# Filter out invalid symbols
|
||||
raw = raw[~raw.symbol.isin(self._excluded_symbols)]
|
||||
|
||||
# cut out all the other stuff in the name column
|
||||
# we need to escape the paren because it is actually splitting on a regex
|
||||
# cut out all the other stuff in the name column. We need to
|
||||
# escape the paren because it is actually splitting on a regex
|
||||
raw.asset_name = raw.asset_name.str.split(r' \(', 1).str.get(0)
|
||||
|
||||
return raw
|
||||
@@ -175,7 +165,6 @@ class QuandlBundle(BaseEquityPricingBundle):
|
||||
df['sid'] = asset_id
|
||||
self.splits.append(df)
|
||||
|
||||
|
||||
def _update_dividends(self, asset_id, raw_data):
|
||||
divs = raw_data.ex_dividend
|
||||
df = pd.DataFrame({'amount': divs[divs != 0]})
|
||||
@@ -186,7 +175,6 @@ class QuandlBundle(BaseEquityPricingBundle):
|
||||
df['record_date'] = df['declared_date'] = df['pay_date'] = pd.NaT
|
||||
self.dividends.append(df)
|
||||
|
||||
|
||||
def _format_metadata_url(self, api_key, page_number):
|
||||
"""Build the query RL for the quandl WIKI metadata.
|
||||
"""
|
||||
@@ -200,10 +188,10 @@ class QuandlBundle(BaseEquityPricingBundle):
|
||||
query_params = [('api_key', api_key)] + query_params
|
||||
|
||||
return (
|
||||
'https://www.quandl.com/api/v3/datasets.csv?' + urlencode(query_params)
|
||||
'https://www.quandl.com/api/v3/datasets.csv?'
|
||||
+ urlencode(query_params)
|
||||
)
|
||||
|
||||
|
||||
def _format_wiki_url(self,
|
||||
api_key,
|
||||
symbol,
|
||||
@@ -229,5 +217,6 @@ class QuandlBundle(BaseEquityPricingBundle):
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
register_calendar_alias('QUANDL', 'NYSE')
|
||||
register_bundle(QuandlBundle)
|
||||
|
||||
@@ -656,11 +656,11 @@ class DataPortal(object):
|
||||
return spot_value
|
||||
|
||||
def _get_minutely_spot_value(self,
|
||||
asset,
|
||||
column,
|
||||
dt,
|
||||
data_frequency,
|
||||
ffill=False):
|
||||
asset,
|
||||
column,
|
||||
dt,
|
||||
data_frequency,
|
||||
ffill=False):
|
||||
|
||||
reader = self._get_pricing_reader(data_frequency)
|
||||
|
||||
@@ -706,7 +706,7 @@ class DataPortal(object):
|
||||
asset,
|
||||
column,
|
||||
dt,
|
||||
ffill,
|
||||
ffill,
|
||||
'minute',
|
||||
)
|
||||
|
||||
|
||||
@@ -133,11 +133,13 @@ class AssetDispatchBarReader(with_metaclass(ABCMeta)):
|
||||
|
||||
return results
|
||||
|
||||
|
||||
class AssetDispatchMinuteBarReader(AssetDispatchBarReader):
|
||||
|
||||
def _dt_window_size(self, start_dt, end_dt):
|
||||
return len(self.trading_calendar.minutes_in_range(start_dt, end_dt))
|
||||
|
||||
|
||||
class AssetDispatchSessionBarReader(AssetDispatchBarReader):
|
||||
|
||||
def _dt_window_size(self, start_dt, end_dt):
|
||||
|
||||
+19
-78
@@ -12,7 +12,6 @@
|
||||
# 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.
|
||||
import datetime
|
||||
import os
|
||||
from collections import OrderedDict
|
||||
|
||||
@@ -129,11 +128,13 @@ def load_crypto_market_data(trading_day=None, trading_days=None,
|
||||
# before this date.
|
||||
'''
|
||||
if(bundle_data):
|
||||
# If we are using the bundle to retrieve the cryptobenchmark, find the last
|
||||
# date for which there is trading data in the bundle
|
||||
asset = bundle_data.asset_finder.lookup_symbol(symbol=bm_symbol,as_of_date=None)
|
||||
# If we are using the bundle to retrieve the cryptobenchmark, find
|
||||
# the last date for which there is trading data in the bundle
|
||||
asset = bundle_data.asset_finder.lookup_symbol(
|
||||
symbol=bm_symbol,as_of_date=None)
|
||||
ix = bundle_data.daily_bar_reader._last_rows[asset.sid]
|
||||
last_date = pd.to_datetime(bundle_data.daily_bar_reader._spot_col('day')[ix],unit='s')
|
||||
last_date = pd.to_datetime(
|
||||
bundle_data.daily_bar_reader._spot_col('day')[ix],unit='s')
|
||||
else:
|
||||
last_date = trading_days[trading_days.get_loc(now, method='ffill') - 2]
|
||||
'''
|
||||
@@ -164,8 +165,8 @@ def load_crypto_market_data(trading_day=None, trading_days=None,
|
||||
br.loc[start_dt] = 0
|
||||
br = br.sort_index()
|
||||
|
||||
# Override first_date for treasury data since we have it for many more years
|
||||
# and is independent of crypto data
|
||||
# Override first_date for treasury data since we have it for many more
|
||||
# years and is independent of crypto data
|
||||
first_date_treasury = pd.Timestamp('1990-01-02', tz='UTC')
|
||||
tc = ensure_treasury_data(
|
||||
bm_symbol,
|
||||
@@ -301,14 +302,14 @@ def ensure_crypto_benchmark_data(symbol,
|
||||
|
||||
if (bundle == 'poloniex'):
|
||||
'''
|
||||
If we're using the Poloniex bundle, we'll get the benchmark from the bundle
|
||||
instead of downloading it from Poloniex every time we need it.
|
||||
Poloniex has a captcha for API queries originating from outside the US that
|
||||
prevents users abroad from getting Catalyst to work
|
||||
If we're using the Poloniex bundle, we'll get the benchmark from the
|
||||
bundle instead of downloading it from Poloniex every time we need it.
|
||||
Poloniex has a captcha for API queries originating from outside the US
|
||||
that prevents users abroad from getting Catalyst to work
|
||||
'''
|
||||
logger.info(
|
||||
(
|
||||
'Retrieving benchmark data from bundle for {symbol!r} from {first_date} to {last_date}'),
|
||||
('Retrieving benchmark data from bundle for {symbol!r}'
|
||||
' from {first_date} to {last_date}'),
|
||||
symbol=symbol, first_date=first_date, last_date=last_date)
|
||||
|
||||
asset = bundle_data.asset_finder.lookup_symbol(symbol=symbol,
|
||||
@@ -330,11 +331,12 @@ def ensure_crypto_benchmark_data(symbol,
|
||||
last_date)]
|
||||
|
||||
else:
|
||||
# This is how it used to be: downloading the benchmark everytime.
|
||||
# Leaving this code here to be repurposed in the future for other bundles.
|
||||
# This is how it used to be: downloading the benchmark everytime.
|
||||
# Leaving this code here to be repurposed in the future for
|
||||
# other bundles.
|
||||
logger.info(
|
||||
(
|
||||
'Downloading benchmark data for {symbol!r} from {first_date} to {last_date}'),
|
||||
('Downloading benchmark data for {symbol!r}'
|
||||
' from {first_date} to {last_date}'),
|
||||
symbol=symbol, first_date=first_date, last_date=last_date)
|
||||
|
||||
raise DeprecationWarning('poloniex bundle deprecated')
|
||||
@@ -431,67 +433,6 @@ def ensure_benchmark_data(symbol, first_date, last_date, now, trading_day,
|
||||
return data
|
||||
|
||||
|
||||
def ensure_benchmark_data(symbol, first_date, last_date, now, trading_day,
|
||||
environ=None):
|
||||
"""
|
||||
Ensure we have benchmark data for `symbol` from `first_date` to `last_date`
|
||||
|
||||
Parameters
|
||||
----------
|
||||
symbol : str
|
||||
The symbol for the benchmark to load.
|
||||
first_date : pd.Timestamp
|
||||
First required date for the cache.
|
||||
last_date : pd.Timestamp
|
||||
Last required date for the cache.
|
||||
now : pd.Timestamp
|
||||
The current time. This is used to prevent repeated attempts to
|
||||
re-download data that isn't available due to scheduling quirks or other
|
||||
failures.
|
||||
trading_day : pd.CustomBusinessDay
|
||||
A trading day delta. Used to find the day before first_date so we can
|
||||
get the close of the day prior to first_date.
|
||||
|
||||
We attempt to download data unless we already have data stored at the data
|
||||
cache for `symbol` whose first entry is before or on `first_date` and whose
|
||||
last entry is on or after `last_date`.
|
||||
|
||||
If we perform a download and the cache criteria are not satisfied, we wait
|
||||
at least one hour before attempting a redownload. This is determined by
|
||||
comparing the current time to the result of os.path.getmtime on the cache
|
||||
path.
|
||||
"""
|
||||
filename = get_benchmark_filename(symbol)
|
||||
data = _load_cached_data(filename, first_date, last_date, now, 'benchmark',
|
||||
environ)
|
||||
if data is not None:
|
||||
return data
|
||||
|
||||
# If no cached data was found or it was missing any dates then download the
|
||||
# necessary data.
|
||||
logger.info(
|
||||
('Downloading benchmark data for {symbol!r} '
|
||||
'from {first_date} to {last_date}'),
|
||||
symbol=symbol,
|
||||
first_date=first_date - trading_day,
|
||||
last_date=last_date
|
||||
)
|
||||
|
||||
try:
|
||||
data = get_benchmark_returns(
|
||||
symbol,
|
||||
first_date - trading_day,
|
||||
last_date,
|
||||
)
|
||||
data.to_csv(get_data_filepath(filename, environ))
|
||||
except (OSError, IOError, HTTPError):
|
||||
logger.exception('Failed to cache the new benchmark returns')
|
||||
raise
|
||||
if not has_data_for_dates(data, first_date, last_date):
|
||||
logger.warn("Still don't have expected data after redownload!")
|
||||
return data
|
||||
|
||||
|
||||
def ensure_treasury_data(symbol, first_date, last_date, now, environ=None):
|
||||
"""
|
||||
Ensure we have treasury data from treasury module associated with
|
||||
|
||||
@@ -341,12 +341,10 @@ class BcolzMinuteBarMetadata(object):
|
||||
'end_session': str(self.end_session.date()),
|
||||
# Write these values for backwards compatibility
|
||||
'first_trading_day': str(self.start_session.date()),
|
||||
'market_opens': (
|
||||
market_opens.values.astype('datetime64[m]').
|
||||
astype(np.int64).tolist()),
|
||||
'market_closes': (
|
||||
market_closes.values.astype('datetime64[m]').
|
||||
astype(np.int64).tolist()),
|
||||
'market_opens': (market_opens.values.astype('datetime64[m]').
|
||||
astype(np.int64).tolist()),
|
||||
'market_closes': (market_closes.values.astype('datetime64[m]').
|
||||
astype(np.int64).tolist()),
|
||||
}
|
||||
with open(self.metadata_path(rootdir), 'w+') as fp:
|
||||
json.dump(metadata, fp)
|
||||
@@ -1256,8 +1254,8 @@ class BcolzMinuteBarReader(MinuteBarReader):
|
||||
values = carray[start_idx:end_idx + 1]
|
||||
if indices_to_exclude is not None:
|
||||
for excl_start, excl_stop in indices_to_exclude[::-1]:
|
||||
excl_slice = np.s_[
|
||||
excl_start - start_idx:excl_stop - start_idx + 1]
|
||||
excl_slice = np.s_[excl_start - start_idx:excl_stop
|
||||
- start_idx + 1]
|
||||
values = np.delete(values, excl_slice)
|
||||
|
||||
where = values != 0
|
||||
@@ -1320,9 +1318,8 @@ class H5MinuteBarUpdateWriter(object):
|
||||
|
||||
def __init__(self, path, complevel=None, complib=None):
|
||||
self._complevel = complevel if complevel \
|
||||
is not None else self._COMPLEVEL
|
||||
self._complib = complib if complib \
|
||||
is not None else self._COMPLIB
|
||||
is not None else self._COMPLEVEL
|
||||
self._complib = complib if complib is not None else self._COMPLIB
|
||||
self._path = path
|
||||
|
||||
def write(self, frames):
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
# 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 __future__ import division # Python2 req for division of ints yield float
|
||||
|
||||
from errno import ENOENT
|
||||
from functools import partial
|
||||
@@ -120,7 +120,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
|
||||
# Provides 9 decimals resolution. Also affects _equities.pyx L220
|
||||
PRICE_ADJUSTMENT_FACTOR = 1000000000
|
||||
|
||||
|
||||
def check_uint32_safe(value, colname):
|
||||
@@ -130,6 +131,7 @@ def check_uint32_safe(value, colname):
|
||||
"for uint32" % (value, colname)
|
||||
)
|
||||
|
||||
|
||||
def check_uint64_safe(value, colname):
|
||||
if value >= UINT64_MAX:
|
||||
raise ValueError(
|
||||
@@ -322,8 +324,8 @@ class BcolzDailyBarWriter(object):
|
||||
# Maps column name -> output carray.
|
||||
columns = {
|
||||
k: carray(array([], dtype=uint64))
|
||||
if k in OHLCV
|
||||
else carray(array([], dtype=uint32))
|
||||
if k in OHLCV
|
||||
else carray(array([], dtype=uint32))
|
||||
for k in US_EQUITY_PRICING_BCOLZ_COLUMNS
|
||||
}
|
||||
|
||||
@@ -439,11 +441,13 @@ class BcolzDailyBarWriter(object):
|
||||
return raw_data
|
||||
|
||||
winsorise_uint64(raw_data, invalid_data_behavior, 'volume', *OHLC)
|
||||
processed = (raw_data[list(OHLC)] * PRICE_ADJUSTMENT_FACTOR).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')
|
||||
processed['volume'] = (raw_data.volume * PRICE_ADJUSTMENT_FACTOR).astype('uint64')
|
||||
processed['volume'] = (raw_data.volume
|
||||
* PRICE_ADJUSTMENT_FACTOR).astype('uint64')
|
||||
return ctable.fromdataframe(processed)
|
||||
|
||||
|
||||
@@ -496,7 +500,7 @@ class BcolzDailyBarReader(SessionBarReader):
|
||||
|
||||
The data in these columns is interpreted as follows:
|
||||
|
||||
- Price columns ('open', 'high', 'low', 'close') and Volume are interpreted
|
||||
- Price columns ('open', 'high', 'low', 'close') and Volume are interpreted
|
||||
as 10^9 * as-traded dollar value.
|
||||
- Day is interpreted as seconds since midnight UTC, Jan 1, 1970.
|
||||
- Id is the asset id of the row.
|
||||
|
||||
Reference in New Issue
Block a user