#!/usr/bin/env python # -*- coding: utf-8 -*- # # yfinance - market data downloader # https://github.com/ranaroussi/yfinance # # Copyright 2017-2019 Ran Aroussi # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # 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. # from __future__ import print_function import requests as _requests import re as _re import pandas as _pd import numpy as _np import sys as _sys try: import ujson as _json except ImportError: import json as _json user_agent_headers = { 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36'} def is_isin(string): return bool(_re.match("^([A-Z]{2})([A-Z0-9]{9})([0-9]{1})$", string)) def get_all_by_isin(isin, proxy=None, session=None): if not(is_isin(isin)): raise ValueError("Invalid ISIN number") from .base import _BASE_URL_ session = session or _requests url = "{}/v1/finance/search?q={}".format(_BASE_URL_, isin) data = session.get(url=url, proxies=proxy, headers=user_agent_headers) try: data = data.json() ticker = data.get('quotes', [{}])[0] return { 'ticker': { 'symbol': ticker['symbol'], 'shortname': ticker['shortname'], 'longname': ticker['longname'], 'type': ticker['quoteType'], 'exchange': ticker['exchDisp'], }, 'news': data.get('news', []) } except Exception: return {} def get_ticker_by_isin(isin, proxy=None, session=None): data = get_all_by_isin(isin, proxy, session) return data.get('ticker', {}).get('symbol', '') def get_info_by_isin(isin, proxy=None, session=None): data = get_all_by_isin(isin, proxy, session) return data.get('ticker', {}) def get_news_by_isin(isin, proxy=None, session=None): data = get_all_by_isin(isin, proxy, session) return data.get('news', {}) def empty_df(index=[]): empty = _pd.DataFrame(index=index, data={ 'Open': _np.nan, 'High': _np.nan, 'Low': _np.nan, 'Close': _np.nan, 'Adj Close': _np.nan, 'Volume': _np.nan}) empty.index.name = 'Date' return empty def empty_earnings_dates_df(): empty = _pd.DataFrame( columns=["Symbol", "Company", "Earnings Date", "EPS Estimate", "Reported EPS", "Surprise(%)"]) return empty def get_html(url, proxy=None, session=None): session = session or _requests html = session.get(url=url, proxies=proxy, headers=user_agent_headers).text return html def get_json(url, proxy=None, session=None): session = session or _requests html = session.get(url=url, proxies=proxy, headers=user_agent_headers).text if "QuoteSummaryStore" not in html: html = session.get(url=url, proxies=proxy).text if "QuoteSummaryStore" not in html: return {} json_str = html.split('root.App.main =')[1].split( '(this)')[0].split(';\n}')[0].strip() data = _json.loads(json_str)[ 'context']['dispatcher']['stores']['QuoteSummaryStore'] # add data about Shares Outstanding for companies' tickers if they are available try: data['annualBasicAverageShares'] = _json.loads( json_str)['context']['dispatcher']['stores'][ 'QuoteTimeSeriesStore']['timeSeries']['annualBasicAverageShares'] except Exception: pass # return data new_data = _json.dumps(data).replace('{}', 'null') new_data = _re.sub( r'\{[\'|\"]raw[\'|\"]:(.*?),(.*?)\}', r'\1', new_data) return _json.loads(new_data) def camel2title(o): return [_re.sub("([a-z])([A-Z])", r"\g<1> \g<2>", i).title() for i in o] def auto_adjust(data): df = data.copy() ratio = df["Close"] / df["Adj Close"] df["Adj Open"] = df["Open"] / ratio df["Adj High"] = df["High"] / ratio df["Adj Low"] = df["Low"] / ratio df.drop( ["Open", "High", "Low", "Close"], axis=1, inplace=True) df.rename(columns={ "Adj Open": "Open", "Adj High": "High", "Adj Low": "Low", "Adj Close": "Close" }, inplace=True) df = df[["Open", "High", "Low", "Close", "Volume"]] return df[["Open", "High", "Low", "Close", "Volume"]] def back_adjust(data): """ back-adjusted data to mimic true historical prices """ df = data.copy() ratio = df["Adj Close"] / df["Close"] df["Adj Open"] = df["Open"] * ratio df["Adj High"] = df["High"] * ratio df["Adj Low"] = df["Low"] * ratio df.drop( ["Open", "High", "Low", "Adj Close"], axis=1, inplace=True) df.rename(columns={ "Adj Open": "Open", "Adj High": "High", "Adj Low": "Low" }, inplace=True) return df[["Open", "High", "Low", "Close", "Volume"]] def parse_quotes(data, tz=None): timestamps = data["timestamp"] ohlc = data["indicators"]["quote"][0] volumes = ohlc["volume"] opens = ohlc["open"] closes = ohlc["close"] lows = ohlc["low"] highs = ohlc["high"] adjclose = closes if "adjclose" in data["indicators"]: adjclose = data["indicators"]["adjclose"][0]["adjclose"] quotes = _pd.DataFrame({"Open": opens, "High": highs, "Low": lows, "Close": closes, "Adj Close": adjclose, "Volume": volumes}) quotes.index = _pd.to_datetime(timestamps, unit="s") quotes.sort_index(inplace=True) if tz is not None: quotes.index = quotes.index.tz_localize(tz) return quotes def parse_actions(data, tz=None): dividends = _pd.DataFrame( columns=["Dividends"], index=_pd.DatetimeIndex([])) splits = _pd.DataFrame( columns=["Stock Splits"], index=_pd.DatetimeIndex([])) if "events" in data: if "dividends" in data["events"]: dividends = _pd.DataFrame( data=list(data["events"]["dividends"].values())) dividends.set_index("date", inplace=True) dividends.index = _pd.to_datetime(dividends.index, unit="s") dividends.sort_index(inplace=True) if tz is not None: dividends.index = dividends.index.tz_localize(tz) dividends.columns = ["Dividends"] if "splits" in data["events"]: splits = _pd.DataFrame( data=list(data["events"]["splits"].values())) splits.set_index("date", inplace=True) splits.index = _pd.to_datetime(splits.index, unit="s") splits.sort_index(inplace=True) if tz is not None: splits.index = splits.index.tz_localize(tz) splits["Stock Splits"] = splits["numerator"] / \ splits["denominator"] splits = splits["Stock Splits"] return dividends, splits class ProgressBar: def __init__(self, iterations, text='completed'): self.text = text self.iterations = iterations self.prog_bar = '[]' self.fill_char = '*' self.width = 50 self.__update_amount(0) self.elapsed = 1 def completed(self): if self.elapsed > self.iterations: self.elapsed = self.iterations self.update_iteration(1) print('\r' + str(self), end='') _sys.stdout.flush() print() def animate(self, iteration=None): if iteration is None: self.elapsed += 1 iteration = self.elapsed else: self.elapsed += iteration print('\r' + str(self), end='') _sys.stdout.flush() self.update_iteration() def update_iteration(self, val=None): val = val if val is not None else self.elapsed / float(self.iterations) self.__update_amount(val * 100.0) self.prog_bar += ' %s of %s %s' % ( self.elapsed, self.iterations, self.text) def __update_amount(self, new_amount): percent_done = int(round((new_amount / 100.0) * 100.0)) all_full = self.width - 2 num_hashes = int(round((percent_done / 100.0) * all_full)) self.prog_bar = '[' + self.fill_char * \ num_hashes + ' ' * (all_full - num_hashes) + ']' pct_place = (len(self.prog_bar) // 2) - len(str(percent_done)) pct_string = '%d%%' % percent_done self.prog_bar = self.prog_bar[0:pct_place] + \ (pct_string + self.prog_bar[pct_place + len(pct_string):]) def __str__(self): return str(self.prog_bar)