Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_quandl_exceptions_no_retries(self, request_method):
ApiConfig.use_retries = False
quandl_errors = [('QELx04', 429, LimitExceededError),
('QEMx01', 500, InternalServerError),
('QEAx01', 400, AuthenticationError),
('QEPx02', 403, ForbiddenError),
('QESx03', 422, InvalidRequestError),
('QECx05', 404, NotFoundError),
('QEXx01', 503, ServiceUnavailableError),
('QEZx02', 400, QuandlError)]
httpretty.register_uri(getattr(httpretty, request_method),
"https://www.quandl.com/api/v3/databases",
responses=[httpretty.Response(body=json.dumps(
{'quandl_error':
{'code': x[0], 'message': 'something went wrong'}}),
status=x[1]) for x in quandl_errors]
)
for expected_error in quandl_errors:
self.assertRaises(
expected_error[2], lambda: Connection.request(request_method, 'databases'))
end_date = arrow.now().format("YYYY-MM-DD")
start_date = arrow.now()
start_date = start_date.replace(days=(num_days_back*-1)).format("YYYY-MM-DD")
quandl_api_key = "DqEaArDZQP8SfgHTd_Ko"
quandl.ApiConfig.api_key = quandl_api_key
source = "WIKI/" + stock_ticker
print(" Retrieving data from quandl API...")
data = quandl.get(source, start_date=str(start_date), end_date=str(end_date))
#Quandl will not error if it cannot get the amount of tweets specified. It will get however many exist.
if len(data) < minimum_days:
raise quandl.errors.quandl_error.NotFoundError
print(" Retrieving twitter data and performing sentiment analysis...")
#get sentiment from the dates
dates = list(data.index)
for i in range(0, len(dates)):
dates[i] = str(dates[i]).split(" ")[0]
sentiments = dates_to_sentiment(dates, ticker, max_tweets)
data = data[["Open", "High", "Low", "Volume", "Close"]].as_matrix()
data = np.insert(data, 4, sentiments, axis=1)
return data
data_keys = ["trX", "trY", "testX", "testY", "price", "X", "Y"]
if self.mode == "test":
data_keys = [e for e in data_keys if e != "Y"]
if not self.test:
data_keys = [e for e in data_keys if e not in ("trX", "trY", "testX", "testY")]
for key in data_keys:
output[stock_idx][key] = _data[key]
else:
# print("No pickle found, getting data for", sec)
try:
# print("Getting data for", stock_code)
df = quandl.get(stock_code, start_date=start_date, end_date=end_date)
except quandl.errors.quandl_error.NotFoundError:
# Log as faulty code
self._log_faulty_code(stock_code)
output[stock_idx] = None
continue
# Trim column names to standard O-H-L-C-V
df = _rename_columns(df)
df.reset_index(drop=True, inplace=True)
# Actually build all the technical indicators
df = self._build_indicators(df)
price = df["close"].values
# If target Y data is requested, build it
if self.mode == "train":
def get_price_quandl(self,ticker ):
try:
if Cache.data_request_delay != None:
time.sleep(Cache.data_request_delay)
if Cache.quandl_key != None:
df = quandl.get("WIKI/"+ticker, authtoken=Cache.quandl_key)
else:
df = quandl.get("WIKI/"+ticker)
except quandl.errors.quandl_error.NotFoundError as e:
print("WIKI/"+ticker)
raise e
df = df[['Adj. Open', 'Adj. High', 'Adj. Low', 'Adj. Close', 'Adj. Volume']]
df.rename(columns={'Adj. Open':'Open','Adj. High':'High', 'Adj. Low':'Low', 'Adj. Close':'Close', 'Adj. Volume':'Volume'},inplace = True)
df.dropna(inplace=True)
return df
def data(self, **options):
# handle_not_found_error if set to True will add an empty DataFrame
# for a non-existent dataset instead of raising an error
handle_not_found_error = options.pop('handle_not_found_error', False)
handle_column_not_found = options.pop('handle_column_not_found', False)
# default order to ascending, and respect whatever user passes in
params = {
'database_code': self.database_code,
'dataset_code': self.dataset_code,
'order': 'asc'
}
updated_options = Util.merge_options('params', params, **options)
try:
return Data.all(**updated_options)
except NotFoundError:
if handle_not_found_error:
return DataList(Data, [], {'column_names': [six.u('None'), six.u('Not Found')]})
raise
except ColumnNotFound:
if handle_column_not_found:
return DataList(Data, [], {'column_names': [six.u('None'), six.u('Not Found')]})
raise
-------
Dictionary containing a Pandas dataframe and the labels, one for train data, \
one for test.
Dictionary keys: train_x, train_y, test_x, test_y
i.e. `[{'train_x': train1_data, 'train_y': train1_labels,
'test_x': test1_data, 'test_y': test1_labels}]`
"""
print("Building", stock_code)
output = {}
data = _get_cached_data(stock_code, build_labels, start_date, end_date) if cache else None
need_to_cache = True
if data is None:
try:
data = quandl.get(stock_code, start_date=start_date, end_date=end_date)
except quandl.errors.quandl_error.NotFoundError:
# invalid stock code
print("Invalid code: {}".format(stock_code))
return None
data = _rename_columns(data)
data.reset_index(inplace=True, drop=True)
data = build_indicators(data)
if build_labels:
data["Y"] = compute_labels(data["Close"])
data = data.replace([np.inf, -np.inf], np.nan)
data.dropna(inplace=True)
# If target Y data is requested, build it
if build_labels:
labels = data.pop("Y")
pickle_name += "_notrand"
if start_date and end_date:
pickle_name += start_date + "to" + end_date
elif start_date:
pickle_name += start_date
elif end_date:
pickle_name += "to" + end_date
if not os.path.isfile("./stock_data/" + pickle_name + "_data.pickle"):
# print("No pickle found, getting data for", sec)
try:
# print("Getting data for", stock_code)
df = quandl.get(stock_code, start_date=start_date, end_date=end_date)
except quandl.errors.quandl_error.NotFoundError:
invalid_stock_codes += [stock_code]
f.write(stock_code + "\n")
stock_code = yield None
continue
if "Adj. Close" in df.columns:
df = df[["Adj. Open", "Adj. High", "Adj. Low", "Adj. Close", "Adj. Volume"]]
df.rename(
columns=lambda x: x[5:].lower(), inplace=True
) # Remove the "Adj. " and make lowercase
elif "Close" in df.columns:
df = df[["Open", "High", "Low", "Close", "Volume"]]
df.rename(columns=lambda x: x.lower(), inplace=True) # make lowercase
price = df["close"].values
minIdxs = argrelextrema(price, np.less)
calendardate={'gte': START_DATE, 'lte': END_DATE},
ticker=ticker,
qopts={'columns': ['dimension', 'datekey'] + fields})
df = df.rename(columns={'datekey': 'Date'}).set_index('Date')
# loop over the fields and dimensions
series = []
for i, field in enumerate(fields):
s = df[df.dimension == dimensions[i]][field]
series.append(s)
df = pd.concat(series, axis=1)
print(df)
# write raw file: raw/
df.to_csv(os.path.join(raw_path, "{}.csv".format(sid)))
except quandl.errors.quandl_error.NotFoundError:
print("error with ticker: {}".format(ticker))
def run(self):
"""
Loads data, thus enabling get_similarity_matrix and
get_covariance_matrix methods in the base class.
"""
self._check_provider_valid()
if self._token:
quandl.ApiConfig.api_key = self._token
quandl.ApiConfig.api_version = '2015-04-09'
self._data = []
for _, __s in enumerate(self._tickers):
try:
__d = quandl.get("WIKI/" + __s,
start_date=self._start,
end_date=self._end)
except NotFoundError as ex:
raise QiskitFinanceError(
"Cannot retrieve Wikipedia data due to an invalid token."
) from ex
# The exception will be urllib3 NewConnectionError, but it can get dressed by quandl
except Exception as ex: # pylint: disable=broad-except
raise QiskitFinanceError(
"Cannot retrieve Wikipedia data.") from ex
try:
self._data.append(__d["Adj. Close"])
except KeyError as ex:
raise QiskitFinanceError("Cannot parse quandl output.") from ex
end_date = arrow.now().format("YYYY-MM-DD")
start_date = arrow.now()
start_date = start_date.replace(days=(num_days_back*-1)).format("YYYY-MM-DD")
quandl_api_key = "DqEaArDZQP8SfgHTd_Ko"
quandl.ApiConfig.api_key = quandl_api_key
source = "WIKI/" + stock_ticker
print(" Retrieving data from quandl API...")
data = quandl.get(source, start_date=str(start_date), end_date=str(end_date))
data = data[["Open", "High", "Low", "Volume", "Close"]].as_matrix()
if len(data) < minimum_days:
raise quandl.errors.quandl_error.NotFoundError
return data