Compare commits

..
20 Commits
Author SHA1 Message Date
cmallwitzandGitHub 1ec1908948 Merge pull request #161 from Michael-C-Strobel/master
Swap to endpoint for fetching crumb
2026-03-28 16:18:16 +00:00
Michael Strobel 03770b4a2e Swap to endpoint for fetching crumb. 2026-03-07 15:26:53 -08:00
cmallwitz fef8d3e442 Package additional Python modules 2025-05-18 15:27:55 +01:00
cmallwitz af69d7f9fd Bundle Python module "requests" 2025-05-13 18:21:09 +01:00
cmallwitz b5b1ca5d09 README.md update 2025-05-12 22:06:32 +01:00
cmallwitz e50f084886 README.md update 2025-05-12 19:35:48 +01:00
cmallwitz d3e2937286 Workaround for Yahoo HTTPS fingerprinting 2025-05-12 18:11:13 +01:00
cmallwitz a0f723cfba Fix historic price rounding and missing summaryDetail 2025-04-21 16:12:10 +01:00
cmallwitz 376e0999c9 Fixed Yahoo consent form handling 2025-02-03 21:23:34 +00:00
cmallwitz 7b451bf872 [Issue 115] GETHISTORIC not working 2024-09-29 14:41:52 +01:00
cmallwitz 28ae9efa74 Fix for FT redirect to unencoded URLs 2024-09-02 20:37:37 +01:00
cmallwitz 090e604d1d Improve Yahoo crumb handling 2024-09-01 17:35:26 +01:00
cmallwitz a4432a85e6 Fix Yahoo crumb handling 2024-05-17 15:56:43 +01:00
cmallwitz a593336b08 Handle Yahoo site redesign 2024-05-01 18:35:41 +01:00
cmallwitz 06d9b91793 Update version and some cleanups 2024-03-13 12:11:42 +00:00
cmallwitzandGitHub 60d5d7b764 Merge pull request #99 from beckjake/handle-data-value
Handle the case where there is a data-value but no value
2024-03-12 17:18:17 +00:00
Jacob Beck 345f57420c Handle the case where there is a data-value but no value 2024-03-08 21:51:21 -07:00
cmallwitz 88b039d471 Yahoo: Don't re-fetch non-existing statistic/profile data 2023-12-29 22:22:28 +00:00
cmallwitz 7e751928de Rework Yahoo cookies (always fetch from scratch) 2023-12-29 18:52:19 +00:00
cmallwitz 0c07c4c3ba Update build for LibeOffice 7.4+ 2023-12-24 19:49:56 +00:00
17 changed files with 526 additions and 1248 deletions
+73 -17
View File
@@ -1,22 +1,75 @@
# Financials-Extension
Version 3.3.0 includes improved cookie handling and somewhat improved logic to deal with network issues.
## Overview
This is a Python based extension for LibreOffice Calc to make market data available in Calc
spreadsheets - currently supporting Yahoo's (FX, crypto, equities, indices, futures, options) and Financial Times'
(FX, equities, indices, futures) websites using old-fashioned web scraping.
Starting with version 3.1.0, we received a contribution to get crypto data directly from Coinbase
## Latest version vs Yahoo HTTPS fingerprinting
### Feedback requested:
Latest version 3.8.0 was created to bypass Yahoo's recently adding crazy HTTPS fingerprinting
to their website. In a step back to before or rather a return to times long gone some Python
modules need to be installed such that LibreOffice can find them - otherwise Yahoo will not work.
Please provide feedback about using the extension [here](https://github.com/cmallwitz/Financials-Extension/issues/10)
Update for version: 3.8.2 - this bundles the Python module 'requests' and dependencies so users only
using 'FT' as source should not require anything else.
### Usage:
Everyone else using 'Yahoo' as source needs to install module 'curl_cffi'.
Under 'Releases' on GitHub [there](https://github.com/cmallwitz/Financials-Extension/releases) is a downloadable **Financials-Extension.oxt** file - load it into Calc
### Ubuntu / Linux Mint / etc.
Install Python curl_cffi module as root (such that LibroOffice can find it)
- Optionally, if you don't have pip3 installed: ```sudo apt install python3-pip```
- Then ```sudo pip3 install curl_cffi --upgrade```
Note: For a normal Python script just installing curl_cffi is enough to bypass Yahoo's HTTPS fingerprinting.
Because LibreOffice on Linux is loading the stock curl library long before executing the extension
code directly, a second step are required.
The second bit requires a download of [curl-impersonate](https://github.com/lwthiker/curl-impersonate/releases) e.g.
(currently) libcurl-impersonate-v0.6.1.x86_64-linux-gnu.tar.gz - unpack it somewhere
Then I used the below (adjust your location of libcurl-impersonate-chrome.so) to run LibreOffice Calc
directly from command line - alternatively you could define/export LD_PRELOAD and CURL_IMPERSONATE
e.g. in /etc/environment or ~/.bashrc - but make sure the variables are really set when you run LibreOffice.
Note: the setting chrome101 is just that - a setting on what browser to "impersonate". You don't
need to use or install Chrome for this.
```
LD_PRELOAD=/tmp/curl-impersonate/libcurl-impersonate-chrome.so CURL_IMPERSONATE=chrome101 /usr/lib/libreoffice/program/soffice.bin --calc
```
In LibreOffice Calc this I can see something like the below in the output
from `=GETREALTIME("SUPPORT")` and the examples.ods file from this repo can load data
for Yahoo again.
```
...
requests=curl_cffi_0.10.0
LD_PRELOAD=/tmp/curl-impersonate/libcurl-impersonate-chrome.so
CURL_IMPERSONATE=chrome101
curl_version="libcurl/8.1.1 BoringSSL zlib/1.2.11 brotli/1.0.9 nghttp2/1.56.0"
```
### Windows
- Download the script https://bootstrap.pypa.io/get-pip.py to your computer
- Start a Command Prompt (CMD) as Administrator on the command prompt run (change path as required)
```"c:\Program Files\LibreOffice\program\python.exe" c:\temp\get-pip.py``` and then
```"c:\Program Files\LibreOffice\program\python.exe" -m pip install curl_cffi --upgrade```
- Add a new user environment variable CURL_IMPERSONATE, setting it to value chrome101, the Linux
LD_PRELOAD is not needed for the LibreOffice 7.1 I tested this with.
## Usage of extension:
Under 'Releases' on GitHub is a [downloadable](https://github.com/cmallwitz/Financials-Extension/releases) **Financials-Extension.oxt** file - load it into Calc
under menu item: Tools, Extension Manager...
Please make sure, not to rename the OXT file when downloading and before installing: LO will mess up the installation otherwise and the extension won't work.
@@ -31,7 +84,8 @@ Getting data should be as simple as having this in a cell:
Codes 21 and 90 stand for "last price" and "close" (see below), respectively.
Only Yahoo has historic data available.
There is a file **examples.ods** there too with usage examples and possible arguments to functions.
There is a file **examples.ods** in the same Release area with usage examples
and possible arguments to functions.
You have to check the respective websites to work out what symbol is the right one for you. Make sure today or the date
requested is a trading day (exchange is not closed). If a website doesn't have
@@ -121,26 +175,28 @@ refresh things.
### Build:
You will need the LibreOffice SDK installed.
I only ever tried building on a Linux box.
On my system (Ubuntu) I installed packages: libreoffice-dev libreoffice-java-common libreoffice-script-provider-python
You need to install LibreOffice SDK packages: libreoffice-dev libreoffice-java-common libreoffice-script-provider-python
Since the Yahoo HTTPS fingerprinting issue, additionally curl_cffi needs to be installed (see beginning of README)
\# depending on your location...
cd ~/tech/IdeaProjects/Financials-Extension/
cd ~/tech/Financials-Extension/
python3 -m unittest discover src
\# Assuming curl-cffi is installed, LD_PRELOAD is not required here
CURL_IMPERSONATE=chrome101 python3 -m unittest discover src
\# This builds file **Financials-Extension.oxt**
./compile.sh
### Tested with:
- Windows 10 / LibreOffice Calc 7.1.2.2 / Python 3.8.8
- Ubuntu 22.04.1 / LibreOffice Calc 7.3.7.2 / Python 3.10.6
- MacOS 10.15.7 / LibreOffice Calc 7.2.0.4 / Python 3.8.10
- Ubuntu 22.04.5 / LibreOffice Calc 7.3.7.2 / Python 3.10.12
(Previous versions)
(Previously)
- Windows 10 / LibreOffice Calc 7.1.2.2 / Python 3.8.8
- MacOS 10.15.7 / LibreOffice Calc 7.2.0.4 / Python 3.8.10
- Debian 10.3 / LibreOffice Calc 6.1.5.2 / Python 3.7.3
- Ubuntu 20.04.5 / LibreOffice Calc 6.4.7.2 / Python 3.8.10
- Ubuntu 18.04.5 / LibreOffice Calc 6 / Python 3.6.9
+45 -12
View File
@@ -29,13 +29,23 @@ mkdir "${PWD}"/build/META-INF/
# Compile the binaries
echo "Calling idlc..."
idlc -w -verbose "${PWD}"/idl/XFinancials.idl
if command -v idlc &> /dev/null
then
echo "Calling idlc..."
idlc -w -verbose "${PWD}"/idl/XFinancials.idl
echo "Calling regmerge..."
regmerge -v "${PWD}"/build/XFinancials.rdb UCR "${PWD}"/idl/XFinancials.urd
echo "Calling regmerge..."
regmerge -v "${PWD}"/build/XFinancials.rdb UCR "${PWD}"/idl/XFinancials.urd
rm "${PWD}"/idl/XFinancials.urd
rm "${PWD}"/idl/XFinancials.urd
else
# LibreOffice 7.4+
export unoTypes=/usr/lib/libreoffice/program/types.rdb
export offTypes=/usr/lib/libreoffice/program/types/offapi.rdb
echo "Calling unoidl-write..."
unoidl-write $unoTypes $offTypes "${PWD}"/idl/XFinancials.idl "${PWD}"/build/XFinancials.rdb
fi
echo "Generating meta files..."
python3 "${PWD}"/src/generate_metainfo.py
@@ -43,35 +53,58 @@ python3 "${PWD}"/src/generate_metainfo.py
cp -f "${PWD}"/src/financials.py "${PWD}"/build/
cp -f "${PWD}"/src/datacode.py "${PWD}"/build/
cp -f "${PWD}"/src/baseclient.py "${PWD}"/build/
cp -f "${PWD}"/src/jsonParser.py "${PWD}"/build/
cp -f "${PWD}"/src/naivehtmlparser.py "${PWD}"/build/
cp -f "${PWD}"/src/tz.py "${PWD}"/build/
cp -f "${PWD}"/src/financials_ft.py "${PWD}"/build/
cp -f "${PWD}"/src/financials_google.py "${PWD}"/build/
cp -f "${PWD}"/src/financials_yahoo.py "${PWD}"/build/
cp -f "${PWD}"/src/financials_coinbase.py "${PWD}"/build/
# this copies python modules dateutil, pytz, pyparsing to extension so it doesn't have to be installed by user
# this copies python some modules to extension so they doesn't have to be installed by user
TMPFILE=`mktemp`
wget "https://files.pythonhosted.org/packages/36/7a/87837f39d0296e723bb9b62bbb257d0355c7f6128853c78955f57342a56d/python_dateutil-2.8.2-py2.py3-none-any.whl" -O $TMPFILE
# https://pypi.org/project/python-dateutil/
wget "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE dateutil/\* -d "${PWD}"/build/
rm $TMPFILE
wget "https://files.pythonhosted.org/packages/7f/99/ad6bd37e748257dd70d6f85d916cafe79c0b0f5e2e95b11f7fbc82bf3110/pytz-2023.3-py2.py3-none-any.whl" -O $TMPFILE
# https://pypi.org/project/pytz/
wget "https://files.pythonhosted.org/packages/81/c4/34e93fe5f5429d7570ec1fa436f1986fb1f00c3e0f43a589fe2bbcd22c3f/pytz-2025.2-py2.py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE pytz/\* -d "${PWD}"/build/
rm $TMPFILE
# https://pypi.org/project/pyparsing/
# lastest version of "single-file" pyparsing 2.x - used by Ubuntu 22.04 as python3-pyparsing
wget "https://files.pythonhosted.org/packages/8a/bb/488841f56197b13700afd5658fc279a2025a39e22449b7cf29864669b15d/pyparsing-2.4.7-py2.py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE pyparsing.py -d "${PWD}"/build/
rm $TMPFILE
# Windows LibreOffice 7.1 Python is missing this...
wget "https://files.pythonhosted.org/packages/d9/5a/e7c31adbe875f2abbb91bd84cf2dc52d792b5a01506781dbcf25c91daf11/six-1.16.0-py2.py3-none-any.whl" -O $TMPFILE
# https://pypi.org/project/six/
wget "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE six.py -d "${PWD}"/build/
rm $TMPFILE
# https://pypi.org/project/requests/
wget "https://files.pythonhosted.org/packages/f9/9b/335f9764261e915ed497fcdeb11df5dfd6f7bf257d4a6a2a686d80da4d54/requests-2.32.3-py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE requests/\* -d "${PWD}"/build/
rm $TMPFILE
# https://pypi.org/project/urllib3/
# urllib3-2.2.3 is last version supporting Python 3.8 used by LibreOffice 7.1
wget "https://files.pythonhosted.org/packages/ce/d9/5f4c13cecde62396b0d3fe530a50ccea91e7dfc1ccf0e09c228841bb5ba8/urllib3-2.2.3-py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE urllib3/\* -d "${PWD}"/build/
rm $TMPFILE
# https://pypi.org/project/certifi/
wget "https://files.pythonhosted.org/packages/4a/7e/3db2bd1b1f9e95f7cddca6d6e75e2f2bd9f51b1246e546d88addca0106bd/certifi-2025.4.26-py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE certifi/\* -d "${PWD}"/build/
rm $TMPFILE
# https://pypi.org/project/idna/
wget "https://files.pythonhosted.org/packages/76/c6/c88e154df9c4e1a2a66ccf0005a88dfb2650c1dffb6f5ce603dfbd452ce3/idna-3.10-py3-none-any.whl" -O $TMPFILE
unzip $TMPFILE idna/\* -d "${PWD}"/build/
rm $TMPFILE
echo "Package into oxt file..."
pushd "${PWD}"/build/
zip -r "${PWD}"/Financials-Extension.zip ./*
BIN
View File
Binary file not shown.
+72 -134
View File
@@ -8,17 +8,11 @@
# version 3 of the License, or (at your option) any later version.
import codecs
import gzip
import logging
import os
import pathlib
import random
import select
import urllib.request
from http import cookiejar
from http.client import HTTPConnection, HTTPSConnection, HTTPException
from importlib import util
from datacode import Datacode
logger = logging.getLogger(__name__)
@@ -27,154 +21,92 @@ logger = logging.getLogger(__name__)
# logger.setLevel(logging.DEBUG)
class RedirectException(HTTPException):
def __init__(self, location):
self.location = location
curl_cffi_present = not util.find_spec("curl_cffi") is None
requests_present = not util.find_spec("requests") is None
if curl_cffi_present:
logger.debug("Importing curl_cffi...")
from curl_cffi import requests, __version__ as requests_version, __name__ as requests_name
elif requests_present:
logger.debug("Importing requests...")
import requests
requests_version = requests.__version__
requests_name = requests.__name__
else:
raise Exception("Neither curl_cffi nor requests found.")
# import requests
class HttpException(HTTPException):
def __init__(self, url, status):
class HttpException(Exception):
def __init__(self, url, response):
self.url = url
self.status = status
self.response = response
def __str__(self):
if self.response is None:
return f"url='{self.url}'"
if type(self.response) is str:
return f"url='{self.url}' status='{self.response}'"
if self.response.headers:
h = '\n'.join(sorted(self.response.headers.__str__().splitlines(), key=lambda l: l.lower()))
return f"url='{self.url}' status={self.response.status_code} reason='{self.response.reason}' headers={h}\n"
else:
return f"url='{self.url}' status={self.response.status_code} reason='{self.response.reason}'"
class BaseClient:
def __init__(self):
self.connections = {}
self.cookies = cookiejar.CookieJar()
self.last_url = None
self.redirect_count = 0 # will be set later
self.redirect_count = 0
self.basedir = os.path.join(str(pathlib.Path.home()), '.financials-extension')
os.makedirs(self.basedir, exist_ok=True)
user_agents = [
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:100.0) Gecko/20100101 Firefox/100.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:101.0) Gecko/20100101 Firefox/101.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:102.0) Gecko/20100101 Firefox/102.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:103.0) Gecko/20100101 Firefox/103.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:104.0) Gecko/20100101 Firefox/104.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:105.0) Gecko/20100101 Firefox/105.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:106.0) Gecko/20100101 Firefox/106.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:107.0) Gecko/20100101 Firefox/107.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:108.0) Gecko/20100101 Firefox/108.0'
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:133.0) Gecko/20100101 Firefox/133.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:134.0) Gecko/20100101 Firefox/134.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:135.0) Gecko/20100101 Firefox/135.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:136.0) Gecko/20100101 Firefox/136.0',
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:137.0) Gecko/20100101 Firefox/137.0',
]
self.default_headers = {
'User-Agent': random.sample(user_agents, 1)[0],
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Encoding': 'gzip, deflate',
'Accept-Language': 'en-US,en;q=0.5',
'Connection': 'keep-alive',
'Cache-Control': 'max-age=0'
}
if curl_cffi_present:
self.session = requests.Session()
if logger.isEnabledFor(logging.DEBUG) and self.session.curl:
self.session.curl.debug()
else:
self.session = requests.Session()
self.session.headers.update({'User-Agent': random.sample(user_agents, 1)[0],
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Encoding': 'gzip, deflate',
'Accept-Language': 'en-US,en;q=0.5',
'Connection': 'keep-alive',
'Cache-Control': 'max-age=0',
})
self.response = None
self.session.max_redirects = 5
def request(self, method: str, url: str, data=None, headers={}, cookies=[], **kwargs):
_headers = self.default_headers.copy()
if headers:
for key, value in headers.items():
_headers[key] = value
if cookies:
for c in cookies:
self.cookies.set_cookie(c)
connection = None
scheme, _, host, path = url.split('/', 3)
if (scheme, host) in self.connections:
connection = self.connections.get((scheme, host))
if connection and select.select([connection.sock], [], [], 0)[0]:
connection.close()
connection = None
if not connection:
logger.debug('Creating connection --------------------------------------------------')
connection = HTTPConnection(host, **kwargs) if scheme == 'http:' else HTTPSConnection(host, **kwargs)
logger.debug('Creating request -----------------------------------------------------')
logger.debug("url='%s'", url)
self.last_url = url
# generate and add cookie headers
request = urllib.request.Request(url)
self.cookies.add_cookie_header(request)
if request.get_header('Cookie'):
_headers['Cookie'] = request.get_header('Cookie')
for key, value in _headers.items():
logger.debug('Header: %s=%s', key, value)
# request
connection.request(method, '/' + path, data, _headers)
response = connection.getresponse()
logger.debug('Processing response --------------------------------------------------')
logger.debug('response.status=%s', response.status)
for key, value in response.getheaders():
logger.debug('Header: %s=%s', key, value)
self.cookies.extract_cookies(response, request)
self.connections[(scheme, host)] = connection
return response
def urlopen(self, url, redirect=True, data=None, headers={}, cookies=[], **kwargs):
def urlopen(self, url, data=None):
self.last_url = None
self.response = self.request('POST' if data else 'GET', url, data, headers, cookies, **kwargs)
text = self.response.read()
resp = self.session.request('POST' if data else 'GET', url, data=data)
# Allow redirects - used by Yahoo for some cookie based consent
self.redirect_count = 5
if 400 <= resp.status_code < 500:
if resp.headers.get('X-Cache') == 'Error from cloudfront':
resp = self.session.request('POST' if data else 'GET', url, data=data)
# (for Yahoo) AWS CloudFront occasionally returns an incorrect, cached error responses
# try mitigating by re-requesting straight away
if 400 <= self.response.status < 500:
if self.response.getheader('X-Cache') == 'Error from cloudfront':
self.response = self.request('POST' if data else 'GET', url, data, headers, cookies, **kwargs)
text = self.response.read()
if resp.status_code >= 400:
logger.warning("url='%s' status=%s reason='%s' headers=%s", resp.url,
resp.status_code, resp.reason,
'\n'.join(sorted(resp.headers.__str__().splitlines(), key=lambda l: l.lower())))
raise HttpException(url, resp)
while 300 <= self.response.status < 400 and self.redirect_count >= 0:
self.redirect_count = len(resp.history)
self.last_url = resp.url
self.redirect_count -= 1
location = self.response.getheader('Location')
if location and redirect:
if location.startswith('/'):
scheme, _, host, path = url.split('/', 3)
location = '{}//{}{}'.format(scheme, host, location)
self.response = self.request('POST' if data else 'GET', location, data, headers, cookies, **kwargs)
text = self.response.read()
else:
raise RedirectException(location)
if self.response.status >= 400:
logger.warning("last_url='%s' status=%s headers=%s", self.last_url, self.response.status,
'\n'.join(sorted(self.response.headers.__str__().splitlines(), key=lambda l: l.lower())))
raise HttpException(url, self.response.status)
if self.response.getheader('Content-Encoding') == 'gzip':
text = gzip.decompress(text)
content_type = self.response.headers.get_content_charset()
if content_type is None:
content_type = 'utf-8'
text = codecs.decode(text, encoding=content_type, errors='ignore')
return text
return resp.text
def get_ticker(self):
@@ -218,9 +150,6 @@ class BaseClient:
tick[Datacode.TIMEZONE] = None
tick[Datacode.VOLUME] = None
tick[Datacode.YAHOO_SUMMARY_RECEIVED] = False
tick[Datacode.YAHOO_STATISTIC_RECEIVED] = False
tick[Datacode.YAHOO_PROFILE_RECEIVED] = False
tick[Datacode.TIMESTAMP] = None
return tick
@@ -385,3 +314,12 @@ class BaseClient:
pass
return None
def version(self):
return requests_name + "_" + requests_version
def curl(self):
return curl_version
def close(self):
self.session.close()
-3
View File
@@ -60,9 +60,6 @@ class Datacode(Enum):
NAME = 104
TIMEZONE = 105
YAHOO_SUMMARY_RECEIVED = 996
YAHOO_STATISTIC_RECEIVED = 997
YAHOO_PROFILE_RECEIVED = 998
TIMESTAMP = 999
@classmethod
+19 -6
View File
@@ -54,7 +54,6 @@ import pyparsing
import six
from datacode import Datacode
import financials_google as google
import financials_yahoo as yahoo
import financials_coinbase as coinbase
import financials_ft as ft
@@ -95,7 +94,6 @@ class FinancialsImpl(unohelper.Base, Financials):
def __init__(self, ctx):
self.ctx = ctx
self.google = google.createInstance(ctx)
self.yahoo = yahoo.createInstance(ctx)
self.coinbase = coinbase.createInstance(ctx)
self.ft = ft.createInstance(ctx)
@@ -136,9 +134,6 @@ class FinancialsImpl(unohelper.Base, Financials):
ticker = str(ticker).strip()
source = str(source).upper()
# if source == 'GOOGLE':
# s = self.google.getRealtime(ticker, datacode)
# el
if source == 'YAHOO':
s = self.yahoo.getRealtime(ticker, datacode)
elif source == 'FT':
@@ -245,7 +240,8 @@ class FinancialsImpl(unohelper.Base, Financials):
if e.tag.endswith('version'):
version = e.attrib['value']
s = 'ctx={}\nid(self)={}\nversion={}\nfile={}\ncwd={}\nhome={}\nuname={}\npid={}\nsys.executable={}\nsys.version={}\nsys.path={}\nlocale={}\ndefaultlocale={}\ndateutil={}\npytz={}\npyparsing={}\nsix={}'.format(
s = ('ctx={}\nid(self)={}\nversion={}\nfile={}\ncwd={}\nhome={}\nuname={}\npid={}\nsys.executable={}\nsys.version={}\nsys.path={}\n' +
'locale={}\ndefaultlocale={}\ndateutil={}\npytz={}\npyparsing={}\nsix={}\nrequests={}').format(
self.ctx,
id(self),
version,
@@ -263,8 +259,20 @@ class FinancialsImpl(unohelper.Base, Financials):
pytz.__version__,
pyparsing.__version__,
six.__version__,
self.ft.version()
)
ld_preload = os.environ.get('LD_PRELOAD')
if ld_preload:
s += f"\nLD_PRELOAD={ld_preload}"
curl_impersonate = os.environ.get('CURL_IMPERSONATE')
if curl_impersonate:
s += f"\nCURL_IMPERSONATE={curl_impersonate}"
if 'curl_cffi' in self.ft.version():
s += f"\ncurl_version=\"{self.ft.session.curl.version().decode()}\""
if datacode:
s = '{}\ntype(datacode)={}\nstr(datacode)={}'.format(
s,
@@ -273,6 +281,11 @@ class FinancialsImpl(unohelper.Base, Financials):
return s
def close(self):
if self.yahoo: self.yahoo.close()
if self.coinbase: self.coinbase.close()
if self.ft: self.ft.close()
def createInstance(ctx):
return FinancialsImpl(ctx)
+2 -4
View File
@@ -20,7 +20,6 @@ import json
import dateutil.parser
import pytz
import jsonParser
from baseclient import BaseClient, HttpException
from datacode import Datacode
@@ -35,7 +34,6 @@ class Coinbase(BaseClient):
self.crumb = None
self.realtime = {}
self.js = jsonParser.jsonObject
def getRealtime(self, ticker, datacode):
@@ -61,7 +59,7 @@ class Coinbase(BaseClient):
url = 'https://api.exchange.coinbase.com/products/{}/stats'.format(ticker)
try:
text = self.urlopen(url, redirect=True, data=None, headers=None)
text = self.urlopen(url)
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s last_url=%s redirect_count=%s", ticker, datacode, self.last_url, self.redirect_count)
del self.realtime[ticker]
@@ -79,7 +77,7 @@ class Coinbase(BaseClient):
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Coinbase.getRealtime({}, {}) - crumb: {}'.format(ticker, datacode, e)
return 'Coinbase.getRealtime({}, {}) - exception: {}'.format(ticker, datacode, e)
try:
price = results['last']
+1 -3
View File
@@ -16,7 +16,6 @@ import urllib.parse
import dateutil.parser
import jsonParser
from baseclient import BaseClient
from datacode import Datacode
from tz import whois_timezone_info
@@ -48,7 +47,6 @@ class FT(BaseClient):
self.crumb = None
self.realtime = {}
self.historicdata = {}
self.js = jsonParser.jsonObject
def getRealtime(self, ticker: str, datacode: int):
@@ -78,7 +76,7 @@ class FT(BaseClient):
url = f'https://markets.ft.com/data/{asset_class}/tearsheet/summary?s={urllib.parse.quote_plus(ticker)}'
try:
text = self.urlopen(url, redirect=True, data=None, headers=None)
text = self.urlopen(url)
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s last_url=%s redirect_count=%s", ticker, datacode, self.last_url, self.redirect_count)
del self.realtime[ticker]
-248
View File
@@ -1,248 +0,0 @@
# financials_google.py
#
# license: GNU LGPL
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 3 of the License, or (at your option) any later version.
import html
import logging
import os
import re
import time
import xml.etree.ElementTree as ET
import dateutil
from baseclient import BaseClient, RedirectException
from datacode import Datacode
from naivehtmlparser import NaiveHTMLParser
from tz import whois_timezone_info
logger = logging.getLogger(__name__)
# logger.setLevel(logging.DEBUG)
def handle_abbreviations(s):
s = str(s).strip()
if s.endswith('M'):
return float(s[:-1]) * 1000000
elif s.endswith('B'):
return float(s[:-1]) * 1000000000
elif s.endswith('T'):
return float(s[:-1]) * 1000000000000
return float(s)
def un_span(s):
return re.sub(r'<span [^>]*>', '', s).replace('</span>', '')
class Google(BaseClient):
def __init__(self, ctx):
super().__init__()
self.realtime = {}
self.location = None
def getRealtime(self, ticker: str, datacode: int):
"""
Retrieve realtime data for ticker from Google Finance and cache it for further lookups
:param ticker: the ticker symbol e.g. LON:VOD
:param datacode: the requested datacode
:return:
"""
# remove white space
ticker = "".join(ticker.split())
# use cached value for up to 60 seconds
if ticker in self.realtime:
tick = self.realtime[ticker]
if Datacode.TIMESTAMP in tick and type(tick[Datacode.TIMESTAMP]) == float and time.time() - 60 < tick[Datacode.TIMESTAMP]:
return self._return_value(tick, datacode)
else:
del self.realtime[ticker]
q_param = 'q=' + ticker
if not self.location:
url = 'https://www.google.com/search?hl=en&tbm=fin&' + q_param
try:
self.urlopen(url, redirect=False)
except RedirectException as e:
self.location = e.location.replace('&' + q_param, '')
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
return 'Google.getRealtime(\'{}\', {}) - location: {}'.format(ticker, datacode, e)
if not self.location:
url = 'https://www.google.com/search?tbm=fin&' + q_param
else:
url = self.location + '&' + q_param
try:
text = self.urlopen(url)
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
return 'Google.getRealtime(\'{}\', {}) - urlopen: {} {}'.format(ticker, datacode, e, url)
try:
temp = ticker.replace(':', '_') # Windows can't have ':' in file names
with open(os.path.join(self.basedir, 'google-{}.html'.format(temp)), "w", encoding="utf-8") as text_file:
print(f"<!-- '{url}' -->\r\n\r\n{text}", file=text_file)
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
if ticker not in self.realtime:
self.realtime[ticker] = self.get_ticker()
tick = self.realtime[ticker]
tick[Datacode.TIMESTAMP] = time.time()
try:
r = '<span[^>]+role="heading"[^>]+>(.*?)</span>'
pattern = re.compile(r)
match = pattern.search(text)
if not match:
return None
start = match.span(0)[1]
tick[Datacode.NAME] = self.save_wrapper(
lambda: html.unescape(un_span(match.group(1)).strip()))
# next div is TICKER
r = '<div [^>]*><div [^>]*>(.*?)</div></div>'
pattern = re.compile(r)
match = pattern.search(text, start)
if not match:
return 'Google.getRealtime({}, {}) - no match'.format(ticker, datacode)
ticker = self.save_wrapper(
lambda: html.unescape(match.group(1)).replace(' ', ''))
tick[Datacode.EXCHANGE] = self.save_wrapper(lambda: ticker.split(':')[0])
tick[Datacode.TICKER] = self.save_wrapper(lambda: ticker.split(':')[1])
except BaseException as e:
return 'Google.getRealtime({}, {}) - process: {}'.format(ticker, datacode, e)
try:
r = '<sticky-header [^>]*>(.*?)</sticky-header>'
pattern = re.compile(r, flags=re.DOTALL)
match = re.search(pattern, text)
if match:
text = match.group(1)
else:
return 'Data for \'{}\' not found'.format(ticker)
parser = NaiveHTMLParser()
root = parser.feed(text)
parser.close()
cards = root.findall('.//g-card-section')
if len(cards) < 4:
return 'Data for \'{}\' not found'.format(ticker)
header = cards[1]
tick[Datacode.LAST_PRICE] = self.save_wrapper(
lambda: float(
html.unescape(header.find('./span[1]/span[1]/span[1]').text).replace(',', '').strip()))
tick[Datacode.CURRENCY] = self.save_wrapper(
lambda: html.unescape(header.find('./span[1]/span[1]/span[2]').text).strip())
tick[Datacode.CHANGE] = self.save_wrapper(
lambda: float(
html.unescape(header.find('./span[2]/span[1]').text).replace('', '-').replace(',', '').strip()))
# percentage is always wrapped in (...) and always positive even if there is a price drop
tick[Datacode.CHANGE_IN_PERCENT] = self.save_wrapper(
lambda: float(
html.unescape(header.find('./span[2]/span[2]/span[1]').text).strip()
.replace('(', '').replace(')', '').replace('%', '')))
try:
value = html.unescape(header.find('./div[1]/span[1]/span[2]').text).replace('·', '').strip()
logger.debug(value)
dt = dateutil.parser.parse(value, yearfirst=True, dayfirst=False, tzinfos=whois_timezone_info)
tick[Datacode.LAST_PRICE_DATE] = dt.date()
tick[Datacode.LAST_PRICE_TIME] = dt.time()
time_bits = value.split(' ')
if len(time_bits) >= 4:
tick[Datacode.TIMEZONE] = time_bits[-1]
except BaseException as e:
pass
footer = cards[3]
logger.debug(ET.tostring(footer))
# parse 'footer' for remaining fields
table = footer.find('./div[1]/div[1]/div[1]/table[1]')
tick[Datacode.OPEN] = self.save_wrapper(
lambda: float(
html.unescape(table.find('./tr[1]/td[2]').text).replace(',', '').strip()))
tick[Datacode.HIGH] = self.save_wrapper(
lambda: float(
html.unescape(table.find('./tr[2]/td[2]').text).replace(',', '').strip()))
tick[Datacode.LOW] = self.save_wrapper(
lambda: float(
html.unescape(table.find('./tr[3]/td[2]').text).replace(',', '').strip()))
tick[Datacode.MARKET_CAP] = self.save_wrapper(
lambda: handle_abbreviations(
html.unescape(table.find('./tr[4]/td[2]').text).replace(',', '').replace('-', '').strip()))
table = footer.find('./div[1]/div[1]/div[2]/table[1]')
# for indices: first item on right side is LOW
if html.unescape(table.find('./tr[1]/td[1]').text).strip() == 'Low':
tick[Datacode.LOW] = self.save_wrapper(
lambda: float(
html.unescape(table.find('./tr[1]/td[2]').text).replace(',', '').strip()))
tick[Datacode.PREV_CLOSE] = self.save_wrapper(
lambda: float(
html.unescape(table.find('./tr[2]/td[2]').text).replace(',', '').strip()))
tick[Datacode.HIGH_52_WEEK] = self.save_wrapper(
lambda: float(
html.unescape(table.find('./tr[3]/td[2]').text).replace(',', '').strip()))
tick[Datacode.LOW_52_WEEK] = self.save_wrapper(
lambda: float(
html.unescape(table.find('./tr[4]/td[2]').text).replace(',', '').strip()))
logger.info(tick)
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
return 'Google.getRealtime({}, {}) - process: {}'.format(ticker, datacode, e)
return self._return_value(self.realtime[ticker], datacode)
def getHistoric(self, ticker, datacode, date):
return 'Google.getHistoric: Historic Data not implemented.'
def createInstance(ctx):
return Google(ctx)
+214 -381
View File
@@ -8,18 +8,17 @@
# version 3 of the License, or (at your option) any later version.
import csv
import html
import datetime
import json
import logging
import os
import pytz
import re
import time
import urllib.parse
from http import cookiejar
import dateutil.parser
import jsonParser
from baseclient import BaseClient, HttpException
from datacode import Datacode
from naivehtmlparser import NaiveHTMLParser
@@ -57,25 +56,6 @@ def handle_abbreviations(s):
return float(s)
def cookie(name, value):
return cookiejar.Cookie(version=0, name=name, value=value,
port=None, port_specified=False, domain=".yahoo.com", domain_specified=True,
domain_initial_dot=True, path="/", path_specified=True, secure=True, expires=None,
discard=False, comment=None, comment_url=None, rest=dict())
def get_cookies():
return [
cookie("A1", "d=AQABBE6aomQCEJUamSIzqCl9UJ-spkNvMfkFEgABCAHqo2TKZPbPb2UBAiAAAAcIS5qiZF3wy-8&S=AQAAAnSEZV80kjd9J2RUh8TD5cY"),
cookie("A1S", "d=AQABBE6aomQCEJUamSIzqCl9UJ-spkNvMfkFEgABCAHqo2TKZPbPb2UBAiAAAAcIS5qiZF3wy-8&S=AQAAAnSEZV80kjd9J2RUh8TD5cY&j=GDPR"),
cookie("A3", "d=AQABBE6aomQCEJUamSIzqCl9UJ-spkNvMfkFEgABCAHqo2TKZPbPb2UBAiAAAAcIS5qiZF3wy-8&S=AQAAAnSEZV80kjd9J2RUh8TD5cY"),
cookie("EuConsent", "CPuVJsAPuVJsAAOACKENDICgAAAAAAAAACiQAAAAAABhoAMAAQSrEQAYAAglWKgAwABBKsA"),
cookie("GUC", "AQ81G6Da"),
cookie("maex", "{\"v2\":{}}"),
cookie("thamba", "1")
]
class Yahoo(BaseClient):
def __init__(self, ctx):
super().__init__()
@@ -83,91 +63,125 @@ class Yahoo(BaseClient):
self.crumb = None
self.realtime = {}
self.historicdata = {}
self.js = jsonParser.jsonObject
self.create_cookies()
def _read_ticker_json_file(self, ticker):
def create_cookies(self):
cookiejar_path = os.path.join(self.basedir, 'yahoo.cookiejar')
cookiejar_exists = os.path.isfile(cookiejar_path) and os.stat(cookiejar_path).st_size >= 0
if cookiejar_exists:
try:
self.cookies = cookiejar.LWPCookieJar()
self.cookies.load(cookiejar_path, ignore_discard=True)
required_cookie_names = ["A1", "A1S", "A3", "GUC", "maex", "thamba"]
for c in self.cookies:
if c.name in required_cookie_names:
required_cookie_names.remove(c.name)
if len(required_cookie_names) > 0:
cookiejar_exists = False
logger.info("Overriding cookiejar '%s'", cookiejar_path)
except BaseException:
cookiejar_exists = False
logger.exception("BaseException initial loading cookiejar_path=%s", cookiejar_path)
if not cookiejar_exists:
try:
lwp_cookiejar = cookiejar.LWPCookieJar()
for c in get_cookies():
lwp_cookiejar.set_cookie(c)
lwp_cookiejar.save(cookiejar_path, ignore_discard=True)
logger.info("Created cookiejar '%s'", cookiejar_path)
except BaseException:
logger.exception("BaseException creating cookiejar_path=%s", cookiejar_path)
try:
self.cookies = cookiejar.LWPCookieJar()
self.cookies.load(cookiejar_path, ignore_discard=True)
logger.info("Loaded cookiejar '%s'", cookiejar_path)
for c in self.cookies:
logger.info("Cookie name'%s' value='%s' path='%s'", c.name, c.value, c.path)
except BaseException:
logger.exception("BaseException loading cookiejar_path=%s", cookiejar_path)
def save_cookies(self):
cookiejar_path = os.path.join(self.basedir, 'yahoo.cookiejar')
try:
self.cookies.save(cookiejar_path, ignore_discard=True)
logger.debug("Saved cookiejar '%s'", cookiejar_path)
except BaseException:
logger.exception("BaseException saving cookiejar_path=%s", cookiejar_path)
def _read_ticker_csv_file(self, ticker):
fn = os.path.join(self.basedir, 'yahoo-{}.csv'.format(ticker))
fn = os.path.join(self.basedir, 'yahoo-hist-{}.json'.format(ticker))
if not os.path.isfile(fn):
return
with open(fn, newline='', encoding="utf-8") as csvfile:
reader = csv.DictReader(csvfile)
with open(fn, newline='', encoding="utf-8") as jsonfile:
js = jsonfile.read()
ticks = {}
parsed = json.loads(js)
parsed = parsed['chart']['result'][0]
price_hint = 2
if 'priceHint' in parsed['meta']:
price_hint = str(parsed['meta']['priceHint'])
if price_hint and price_hint.isnumeric():
price_hint = int(price_hint)
else:
price_hint = 2
tz = datetime.timezone(datetime.timedelta(seconds=parsed['meta']['gmtoffset']), parsed['meta']['exchangeTimezoneName'])
rows = list(
zip((datetime.datetime.fromtimestamp(ts, tz).date() for ts in parsed['timestamp']),
parsed['indicators']['quote'][0]['open'],
parsed['indicators']['quote'][0]['low'],
parsed['indicators']['quote'][0]['high'],
parsed['indicators']['quote'][0]['volume'],
parsed['indicators']['quote'][0]['close'],
parsed['indicators']['adjclose'][0]['adjclose']))
ticks = {}
for row in rows:
tick = self.get_ticker()
try:
tick[Datacode.OPEN] = round(float(row[1]), price_hint)
tick[Datacode.LOW] = round(float(row[2]), price_hint)
tick[Datacode.HIGH] = round(float(row[3]), price_hint)
tick[Datacode.VOLUME] = round(float(row[4]), price_hint)
tick[Datacode.CLOSE] = round(float(row[5]), price_hint)
tick[Datacode.ADJ_CLOSE] = round(float(row[6]), price_hint)
except:
pass
if len(tick) > 0:
ticks[str(row[0])] = tick # Date
self.historicdata[ticker] = ticks
def handleCookiesAndConsent(self, url, ticker, datacode, html_file):
try:
text = self.urlopen(url)
except BaseException as e:
logger.exception("BaseException (1) ticker=%s datacode=%s last_url=%s redirect_count=%s %s",
ticker, datacode, self.last_url, self.redirect_count, e)
return None
try:
with open(os.path.join(self.basedir, html_file), "w", encoding="utf-8") as text_file:
print(f"<!-- '{self.last_url}' -->\r\n\r\n{text}", file=text_file)
except BaseException as e:
logger.exception("BaseException (2) ticker=%s datacode=%s %s", ticker, datacode, e)
if not text:
return None
try:
parser = NaiveHTMLParser()
root = parser.feed(text)
parser.close()
except BaseException as e:
logger.exception("BaseException (3) ticker=%s datacode=%s - HTML parsing - %s", ticker, datacode, e)
return None
form = root.find(f".//form[@class='consent-form']")
if form:
inputs = form.findall(f".//input")
if inputs:
data = {'reject': 'reject'}
for d in inputs:
if 'name' in d.attrib and 'value' in d.attrib:
data[d.attrib['name']] = d.attrib['value']
for row in reader:
tick = self.get_ticker()
try:
tick[Datacode.OPEN] = float(row['Open'])
tick[Datacode.LOW] = float(row['Low'])
tick[Datacode.HIGH] = float(row['High'])
tick[Datacode.VOLUME] = float(row['Volume'])
tick[Datacode.CLOSE] = float(row['Close'])
tick[Datacode.ADJ_CLOSE] = float(row['Adj Close'])
except:
pass
text = self.urlopen(self.last_url, data=data)
except BaseException as e:
logger.exception("BaseException (4) ticker=%s datacode=%s last_url=%s redirect_count=%s %s",
ticker, datacode, self.last_url, self.redirect_count, e)
if len(tick) > 0:
ticks[row['Date']] = tick
try:
with open(os.path.join(self.basedir, html_file), "w", encoding="utf-8") as text_file:
print(f"<!-- '{self.last_url}' (after consent handling) -->\r\n\r\n{text}", file=text_file)
except BaseException as e:
logger.exception("BaseException (5) ticker=%s datacode=%s %s", ticker, datacode, e)
self.historicdata[ticker] = ticks
return text
def fetch_crumb(self):
"""Refreshes the crumb using the current session."""
try:
# Grab cookies
self.session.get("https://finance.yahoo.com/quote/SPY", timeout=10)
# Request crumb endpoint
response = self.session.get("https://query1.finance.yahoo.com/v1/test/getcrumb", timeout=10)
if response.status_code == 200:
return response.text.strip()
except Exception as e:
logger.error(f"Failed to fetch crumb: {e}")
return None
def getRealtime(self, ticker, datacode):
@@ -182,338 +196,158 @@ class Yahoo(BaseClient):
# remove white space
ticker = "".join(ticker.split())
needStatistics = datacode in [Datacode.SHARES_OUT.value, Datacode.FREE_FLOAT.value, Datacode.PAYOUT_RATIO.value]
needProfile = datacode in [Datacode.SECTOR.value, Datacode.INDUSTRY.value]
# use cached value for up to 60 seconds
if ticker in self.realtime:
tick = self.realtime[ticker]
if Datacode.TIMESTAMP in tick and type(tick[Datacode.TIMESTAMP]) == float and time.time() - 60 < tick[Datacode.TIMESTAMP]:
if (tick[Datacode.YAHOO_STATISTIC_RECEIVED] or not needStatistics) and (
tick[Datacode.YAHOO_PROFILE_RECEIVED] or not needProfile) and (
tick[Datacode.YAHOO_SUMMARY_RECEIVED]):
return self._return_value(tick, datacode)
return self._return_value(tick, datacode)
else:
del self.realtime[ticker]
if ticker not in self.realtime:
self.realtime[ticker] = self.get_ticker()
if needStatistics:
return self.getRealtimeStatistics(ticker, datacode)
if needProfile:
return self.getRealtimeProfile(ticker, datacode)
return self.getRealtimeSummary(ticker, datacode)
def getRealtimeSummary(self, ticker, datacode):
"""
Retrieve realtime data from Yahoo Finance - Summary tab
"""
tick = self.realtime[ticker]
url = 'https://finance.yahoo.com/quote/{}?p={}'.format(ticker, ticker)
if not self.crumb:
new_crumb = self.fetch_crumb()
if new_crumb:
self.crumb = new_crumb
logger.debug(f"Crumb successfully set: {self.crumb}")
else:
del self.realtime[ticker]
return f'Yahoo.getRealtime({ticker}, {datacode}) - failed to fetch crumb'
if not self.crumb:
return 'Yahoo.getRealtime({}, {}) - crumb missing'.format(ticker, datacode)
try:
text = self.urlopen(url, redirect=True)
self.save_cookies()
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s last_url=%s redirect_count=%s", ticker, datacode, self.last_url, self.redirect_count)
url = 'https://query1.finance.yahoo.com/v10/finance/quoteSummary/{}?formatted=true&' \
'modules=summaryProfile,financialData,quoteType,recommendationTrend,earnings,equityPerformance,summaryDetail,defaultKeyStatistics,calendarEvents,esgScores,price,pageViews,financialsTemplate&' \
'lang=en-US&region=US&crumb={}' \
.format(ticker, urllib.parse.quote_plus(self.crumb))
js = self.urlopen(url)
except HttpException as e:
logger.exception("HttpException querying ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Yahoo.getRealtimeSummary({}, {}) - urlopen: {}'.format(ticker, datacode, e)
return None
try:
with open(os.path.join(self.basedir, 'yahoo-{}.html'.format(ticker)), "w", encoding="utf-8") as text_file:
print(f"<!-- '{self.last_url}' -->\r\n\r\n{text}", file=text_file)
except BaseException:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
try:
r = '"crumb":"([^"]{11})"'
pattern = re.compile(r)
match = pattern.search(text)
if match:
self.crumb = match.group(1)
with open(os.path.join(self.basedir, 'yahoo-{}.json'.format(ticker)), "w", encoding="utf-8") as json_file:
print(f"<!-- '{self.last_url}' -->\r\n\r\n{js}", file=json_file)
parsed = json.loads(js)
parsed = parsed['quoteSummary']['result'][0]
summaryDetail = dict()
if 'summaryDetail' in parsed:
summaryDetail = dict(sorted(parsed['summaryDetail'].items()))
price = dict(sorted(parsed['price'].items()))
if 'defaultKeyStatistics' in parsed:
defaultKeyStatistics = dict(sorted(parsed['defaultKeyStatistics'].items()))
else:
defaultKeyStatistics = {}
if 'summaryProfile' in parsed:
summaryProfile = dict(sorted(parsed['summaryProfile'].items()))
else:
summaryProfile = {}
quoteType = dict(sorted(parsed['quoteType'].items()))
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
logger.exception("BaseException parsing ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Yahoo.getRealtimeSummary({}, {}) - crumb: {}'.format(ticker, datacode, e)
return 'Yahoo.getRealtimeSummary({}, {}) - exception: {}'.format(ticker, datacode, e)
try:
parser = NaiveHTMLParser()
root = parser.feed(text)
parser.close()
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
return 'Yahoo.getRealtimeSummary({}, {}) - HTML parsing: {}'.format(ticker, datacode, e)
try:
if not root:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
return 'Yahoo.getRealtimeSummary({}, {}) - root missing'.format(ticker, datacode)
tick[Datacode.TICKER] = ticker
tick[Datacode.TIMESTAMP] = time.time()
tick[Datacode.YAHOO_SUMMARY_RECEIVED] = True
parsed = {}
found = root.findall(f".//fin-streamer[@data-symbol='{ticker}']")
for d in found:
if hasattr(d, 'attrib') and 'data-field' in d.attrib:
parsed[d.attrib['data-field']] = default(d.attrib, 'value').replace('', '-').replace(',', '').strip()
# for futures "regularMarketVolume" is from actual future ticker (potentially different to requested one)
if 'regularMarketVolume' not in parsed:
found = root.findall(f".//fin-streamer[@data-field='regularMarketVolume']")
for d in found:
if hasattr(d, 'attrib') and 'data-field' in d.attrib and 'data-symbol' in d.attrib:
parsed[d.attrib['data-field']] = default(d.attrib, 'value').replace('', '-').replace(',', '').strip()
tick[Datacode.TICKER] = default(d.attrib, 'data-symbol').strip()
found = root.findall(f".//td[@data-test]")
for d in found:
if d:
span = d.find('./span')
if hasattr(d, 'attrib') and hasattr(span, 'text'):
parsed[d.attrib['data-test']] = default(span, 'text').replace('', '-').replace(',', '').strip()
else:
if hasattr(d, 'attrib') and hasattr(d, 'text'):
parsed[d.attrib['data-test']] = default(d, 'text').replace('', '-').replace(',', '').strip()
if 'regularMarketPrice' not in parsed:
if 'regularMarketPrice' not in price:
return None
tick[Datacode.PREV_CLOSE] = self.save_wrapper(lambda: float(parsed['PREV_CLOSE-value']))
tick[Datacode.OPEN] = self.save_wrapper(lambda: float(parsed['OPEN-value']))
tick[Datacode.CHANGE] = self.save_wrapper(lambda: float(parsed['regularMarketChange']))
tick[Datacode.CHANGE_IN_PERCENT] = self.save_wrapper(lambda: float(parsed['regularMarketChangePercent']))
tick[Datacode.PREV_CLOSE] = self.save_wrapper(lambda: float(price['regularMarketPreviousClose']['raw']))
tick[Datacode.OPEN] = self.save_wrapper(lambda: float(price['regularMarketOpen']['raw']))
tick[Datacode.CHANGE] = self.save_wrapper(lambda: float(price['regularMarketChange']['raw']))
tick[Datacode.CHANGE_IN_PERCENT] = self.save_wrapper(lambda: float(price['regularMarketChangePercent']['raw']))
t = default(parsed, 'DAYS_RANGE-value').split(' - ')
tick[Datacode.LOW] = self.save_wrapper(lambda: float(t[0]))
tick[Datacode.HIGH] = self.save_wrapper(lambda: float(t[1]))
tick[Datacode.LOW] = self.save_wrapper(lambda: float(price['regularMarketDayLow']['raw']))
tick[Datacode.HIGH] = self.save_wrapper(lambda: float(price['regularMarketDayHigh']['raw']))
tick[Datacode.LAST_PRICE] = self.save_wrapper(lambda: float(parsed['regularMarketPrice']))
tick[Datacode.VOLUME] = self.save_wrapper(lambda: float(parsed['regularMarketVolume']))
tick[Datacode.AVG_DAILY_VOL_3MONTH] = self.save_wrapper(lambda: float(parsed['AVERAGE_VOLUME_3MONTH-value']))
tick[Datacode.BETA] = self.save_wrapper(lambda: float(parsed['BETA_5Y-value']))
tick[Datacode.EPS] = self.save_wrapper(lambda: float(parsed['EPS_RATIO-value']))
tick[Datacode.PE_RATIO] = self.save_wrapper(lambda: float(parsed['PE_RATIO-value']))
t = default(parsed, 'DIVIDEND_AND_YIELD-value').replace('(', '').replace(')', '').replace('%', '').strip().split(' ')
tick[Datacode.DIV] = self.save_wrapper(lambda: float(t[0]))
tick[Datacode.DIV_YIELD] = self.save_wrapper(lambda: float(t[1])/100.0)
tick[Datacode.LAST_PRICE] = self.save_wrapper(lambda: float(price['regularMarketPrice']['raw']))
tick[Datacode.VOLUME] = self.save_wrapper(lambda: float(price['regularMarketVolume']['raw']))
tick[Datacode.AVG_DAILY_VOL_3MONTH] = self.save_wrapper(lambda: float(price['averageDailyVolume3Month']['raw']))
tick[Datacode.BETA] = self.save_wrapper(lambda: float(defaultKeyStatistics['beta']['raw']))
tick[Datacode.EPS] = self.save_wrapper(lambda: float(defaultKeyStatistics['trailingEps']['raw']))
tick[Datacode.PE_RATIO] = self.save_wrapper(lambda: float(summaryDetail['trailingPE']['raw']))
tick[Datacode.EX_DIV_DATE] = self.save_wrapper(
lambda: dateutil.parser.parse(parsed['EX_DIVIDEND_DATE-value'], yearfirst=True, dayfirst=False).date())
lambda: dateutil.parser.parse(summaryDetail['exDividendDate']['fmt'], yearfirst=True, dayfirst=False).date())
t = default(parsed, 'FIFTY_TWO_WK_RANGE-value').split(' - ')
tick[Datacode.LOW_52_WEEK] = self.save_wrapper(lambda: float(t[0]))
tick[Datacode.HIGH_52_WEEK] = self.save_wrapper(lambda: float(t[1]))
tick[Datacode.LOW_52_WEEK] = self.save_wrapper(lambda: float(summaryDetail['fiftyTwoWeekLow']['raw']))
tick[Datacode.HIGH_52_WEEK] = self.save_wrapper(lambda: float(summaryDetail['fiftyTwoWeekHigh']['raw']))
tick[Datacode.MARKET_CAP] = self.save_wrapper(lambda: float(handle_abbreviations(parsed['MARKET_CAP-value'])))
tick[Datacode.MARKET_CAP] = self.save_wrapper(lambda: float(price['marketCap']['raw']))
t = default(parsed, 'BID-value').split(' x ')
tick[Datacode.BID] = self.save_wrapper(lambda: float(t[0]))
tick[Datacode.BIDSIZE] = self.save_wrapper(lambda: float(t[1]))
tick[Datacode.BID] = self.save_wrapper(lambda: float(summaryDetail['bid']['raw']))
tick[Datacode.BIDSIZE] = self.save_wrapper(lambda: float(summaryDetail['bidSize']['raw']))
t = default(parsed, 'ASK-value').split(' x ')
tick[Datacode.ASK] = self.save_wrapper(lambda: float(t[0]))
tick[Datacode.ASKSIZE] = self.save_wrapper(lambda: float(t[1]))
tick[Datacode.ASK] = self.save_wrapper(lambda: float(summaryDetail['ask']['raw']))
tick[Datacode.ASKSIZE] = self.save_wrapper(lambda: float(summaryDetail['askSize']['raw']))
tick[Datacode.EXPIRY_DATE] = self.save_wrapper(
lambda: dateutil.parser.parse(parsed['EXPIRE_DATE-value'], yearfirst=True, dayfirst=False).date())
if quoteType:
t = int(price['regularMarketTime'])
tz = pytz.timezone(quoteType['timeZoneFullName'])
tick[Datacode.SETTLEMENT_DATE] = self.save_wrapper(
lambda: dateutil.parser.parse(parsed['SETTLEMENT_DATE-value'], yearfirst=True, dayfirst=False).date())
tick[Datacode.TIMEZONE] = tz
dt = datetime.datetime.fromtimestamp(t, tz)
r = '<div id="quote-market-notice"[^>]*><span>([^>]*?)(. Market open.)?</span></div>'
match = re.compile(r, flags=re.DOTALL).search(text)
if match:
t = html.unescape(match.group(1)).strip().split(' ')
tick[Datacode.TIMEZONE] = self.save_wrapper(lambda: t[-1])
tick[Datacode.LAST_PRICE_DATE] = dt.date()
tick[Datacode.LAST_PRICE_TIME] = dt.time()
# if quoteType:
# t = int(price['regularMarketTime'])
# tz = pytz.timezone(quoteType['exchangeTimezoneName'])
#
# tick[Datacode.TIMEZONE] = tz
# dt = datetime.datetime.fromtimestamp(t, tz)
#
# tick[Datacode.LAST_PRICE_DATE] = dt.date()
# tick[Datacode.LAST_PRICE_TIME] = dt.time()
tick[Datacode.EXCHANGE] = self.save_wrapper(lambda: price['exchangeName'])
tick[Datacode.CURRENCY] = self.save_wrapper(lambda: price['currency'])
r = '<span>([ \\w]+?) - [^>]*Currency in ([\\w]+)[^>]*</span>'
match = re.compile(r, flags=re.DOTALL).search(text)
if match:
tick[Datacode.EXCHANGE] = self.save_wrapper(lambda: html.unescape(match.group(1)).strip())
tick[Datacode.CURRENCY] = self.save_wrapper(lambda: html.unescape(match.group(2)).strip())
tick[Datacode.DIV] = self.save_wrapper(lambda: float(summaryDetail['dividendRate']['raw']))
tick[Datacode.DIV_YIELD] = self.save_wrapper(lambda: float(summaryDetail['dividendYield']['raw']))
# fallback for dividend/yield on mutual funds and ETFs
# fallback to last dividend on mutual funds and ETFs
if not tick[Datacode.DIV]:
tick[Datacode.DIV] = self.save_wrapper(lambda: float(parsed['LAST_DIVIDEND-value']))
if not tick[Datacode.DIV_YIELD]:
tick[Datacode.DIV_YIELD] = self.save_wrapper(lambda: float(parsed['TD_YIELD-value'].replace('%', '').strip())/100.0)
tick[Datacode.DIV] = self.save_wrapper(lambda: float(defaultKeyStatistics['lastDividendValue']['raw']))
tick[Datacode.NAME] = self.save_wrapper(
lambda: html.unescape(root.find('.//h1').text).strip())
if default(price, 'quoteType') == 'FUTURE':
tick[Datacode.TICKER] = self.save_wrapper(lambda: price['underlyingSymbol'])
tick[Datacode.NAME] = self.save_wrapper(lambda: price['shortName'])
tick[Datacode.SETTLEMENT_DATE] = self.save_wrapper(
lambda: dateutil.parser.parse(summaryDetail['expireDate']['fmt'], yearfirst=True, dayfirst=False).date())
else:
tick[Datacode.NAME] = self.save_wrapper(lambda: price['longName'])
tick[Datacode.EXPIRY_DATE] = self.save_wrapper(
lambda: dateutil.parser.parse(summaryDetail['expireDate']['fmt'], yearfirst=True, dayfirst=False).date())
tick[Datacode.SETTLEMENT_DATE] = None
if not tick[Datacode.NAME]:
tick[Datacode.NAME] = tick[Datacode.TICKER]
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Yahoo.getRealtimeSummary({}, {}) - process: {}'.format(ticker, datacode, e)
tick[Datacode.SECTOR] = self.save_wrapper(lambda: summaryProfile['sector'])
tick[Datacode.INDUSTRY] = self.save_wrapper(lambda: summaryProfile['industry'])
return self._return_value(self.realtime[ticker], datacode)
def getRealtimeStatistics(self, ticker, datacode):
"""
Retrieve realtime data from Yahoo Finance - Statistics tab
"""
tick = self.realtime[ticker]
url = 'https://finance.yahoo.com/quote/{}/key-statistics?p={}'.format(ticker, ticker)
try:
text = self.urlopen(url, redirect=True)
self.save_cookies()
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s last_url=%s redirect_count=%s", ticker, datacode, self.last_url, self.redirect_count)
del self.realtime[ticker]
return 'Yahoo.getRealtimeStatistics({}, {}) - urlopen: {}'.format(ticker, datacode, e)
try:
with open(os.path.join(self.basedir, 'yahoo-{}-statistics.html'.format(ticker)), "w", encoding="utf-8") as text_file:
print(f"<!-- '{url}' -->\r\n\r\n{text}", file=text_file)
except BaseException:
logger.exception("BaseException open/write ticker=%s datacode=%s", ticker, datacode)
try:
parser = NaiveHTMLParser()
root = parser.feed(text)
parser.close()
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Yahoo.getRealtimeStatistics({}, {}) - HTML parsing: {}'.format(ticker, datacode, e)
try:
parsed = {}
statistics = root.find(".//section[@data-test='qsp-statistics']")
if statistics is None:
return None
tick[Datacode.TICKER] = ticker
tick[Datacode.TIMESTAMP] = time.time()
tick[Datacode.YAHOO_STATISTIC_RECEIVED] = True
# Valuation Measures
found = statistics.find('./div[2]/div[1]//table')
if found:
for d in found.findall('.//tr'):
key = d.find('./td[1]/span').text
if key is not None:
parsed[key] = d.find('./td[2]').text
# Stock Price History
found = statistics.find('./div[2]/div[2]/div[1]/div[1]//table')
if found:
for d in found.findall('.//tr'):
key = d.find('./td[1]/span').text
if key is not None:
parsed[key] = d.find('./td[2]').text
# Share Statistics
found = statistics.find('./div[2]/div[2]/div[1]/div[2]//table')
if found:
for d in found.findall('.//tr'):
key = d.find('./td[1]/span').text
if key is not None:
parsed[key] = d.find('./td[2]').text
# Dividends & Splits
found = statistics.find('./div[2]/div[2]/div[1]/div[3]//table')
if found:
for d in found.findall('.//tr'):
key = d.find('./td[1]/span').text
if key is not None:
parsed[key] = d.find('./td[2]').text
tick[Datacode.SHARES_OUT] = self.save_wrapper(
lambda: float(handle_abbreviations(parsed['Shares Outstanding'])))
tick[Datacode.FREE_FLOAT] = self.save_wrapper(
lambda: float(handle_abbreviations(parsed['Float'])))
tick[Datacode.PAYOUT_RATIO] = self.save_wrapper(
lambda: float(handle_abbreviations(parsed['Payout Ratio'].replace('%', '').strip()))/100.0)
tick[Datacode.SHARES_OUT] = self.save_wrapper(lambda: float(defaultKeyStatistics['sharesOutstanding']['raw']))
tick[Datacode.FREE_FLOAT] = self.save_wrapper(lambda: float(defaultKeyStatistics['floatShares']['raw']))
tick[Datacode.PAYOUT_RATIO] = self.save_wrapper(lambda: float(summaryDetail['payoutRatio']['raw']))
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Yahoo.getRealtimeStatistics({}, {}) - process: {}'.format(ticker, datacode, e)
return self._return_value(self.realtime[ticker], datacode)
def getRealtimeProfile(self, ticker, datacode):
"""
Retrieve realtime data from Yahoo Finance - Profile tab
"""
tick = self.realtime[ticker]
url = 'https://finance.yahoo.com/quote/{}/profile?p={}'.format(ticker, ticker)
try:
text = self.urlopen(url, redirect=True)
self.save_cookies()
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s last_url=%s redirect_count=%s", ticker, datacode, self.last_url, self.redirect_count)
del self.realtime[ticker]
return 'Yahoo.getRealtimeProfile({}, {}) - urlopen: {}'.format(ticker, datacode, e)
try:
with open(os.path.join(self.basedir, 'yahoo-{}-profile.html'.format(ticker)), "w", encoding="utf-8") as text_file:
print(f"<!-- '{url}' -->\r\n\r\n{text}", file=text_file)
except BaseException:
logger.exception("BaseException open/write ticker=%s datacode=%s", ticker, datacode)
try:
parser = NaiveHTMLParser()
root = parser.feed(text)
parser.close()
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Yahoo.getRealtimeProfile({}, {}) - HTML parsing: {}'.format(ticker, datacode, e)
try:
p = root.find(".//*[span='Sector(s)']")
if p is None:
return None
tick[Datacode.TICKER] = ticker
tick[Datacode.TIMESTAMP] = time.time()
tick[Datacode.YAHOO_PROFILE_RECEIVED] = True
tick[Datacode.SECTOR] = self.save_wrapper(lambda: p.find("./span[2]").text)
tick[Datacode.INDUSTRY] = self.save_wrapper(lambda: p.find("./span[4]").text)
except BaseException as e:
logger.exception("BaseException ticker=%s datacode=%s", ticker, datacode)
del self.realtime[ticker]
return 'Yahoo.getRealtimeProfile({}, {}) - process: {}'.format(ticker, datacode, e)
return 'Yahoo.getRealtime({}, {}) - process: {}'.format(ticker, datacode, e)
return self._return_value(self.realtime[ticker], datacode)
@@ -536,7 +370,7 @@ class Yahoo(BaseClient):
# the moment we are asked for ADJ_CLOSE we ignore the ticker cache to refresh
if Datacode.ADJ_CLOSE != datacode and ticker not in self.historicdata:
self._read_ticker_csv_file(ticker)
self._read_ticker_json_file(ticker)
try:
date_as_dt = dateutil.parser.parse(date, yearfirst=True, dayfirst=False)
@@ -590,16 +424,16 @@ class Yahoo(BaseClient):
try:
url = 'https://query1.finance.yahoo.com/v7/finance/download/{}' \
url = 'https://query1.finance.yahoo.com/v8/finance/chart/{}' \
'?period1={}&period2={}&interval=1d&events=history&crumb={}' \
.format(ticker, t1, t2, urllib.parse.quote_plus(self.crumb))
text = self.urlopen(url)
with open(os.path.join(self.basedir, 'yahoo-{}.csv'.format(ticker)), "w", encoding="utf-8") as csv_file:
with open(os.path.join(self.basedir, 'yahoo-hist-{}.json'.format(ticker)), "w", encoding="utf-8") as csv_file:
print(text, file=csv_file)
self._read_ticker_csv_file(ticker)
self._read_ticker_json_file(ticker)
except HttpException:
logger.exception("HttpException ticker=%s datacode=%s date=%s", ticker, datacode, date)
@@ -629,6 +463,5 @@ class Yahoo(BaseClient):
return None
def createInstance(ctx):
return Yahoo(ctx)
+1 -1
View File
@@ -14,7 +14,7 @@ import os
cur_dir = os.getcwd()
addin_id = "com.financials.getinfo"
addin_version = "3.3.2"
addin_version = "3.8.2"
addin_displayname = "Financial Market Extension"
addin_publisher_link = "https://github.com/cmallwitz/Financials-Extension"
addin_publisher_name = "The Publisher"
-115
View File
@@ -1,115 +0,0 @@
# jsonParser.py
#
# Implementation of a simple JSON parser, returning a hierarchical
# ParseResults object support both list- and dict-style data access.
#
# Copyright 2006, by Paul McGuire
#
# Updated 8 Jan 2007 - fixed dict grouping bug, and made elements and
# members optional in array and object collections
#
# Updated 9 Aug 2016 - use more current pyparsing constructs/idioms
#
# https://github.com/pyparsing/pyparsing/blob/master/examples/jsonParser.py - revision 53d1b4a on 1 Nov 2019
json_bnf = """
object
{ members }
{}
members
string : value
members , string : value
array
[ elements ]
[]
elements
value
elements , value
value
string
number
object
array
true
false
null
"""
import pyparsing as pp
from pyparsing import pyparsing_common as ppc
def make_keyword(kwd_str, kwd_value):
return pp.Keyword(kwd_str).setParseAction(pp.replaceWith(kwd_value))
TRUE = make_keyword("true", True)
FALSE = make_keyword("false", False)
NULL = make_keyword("null", None)
LBRACK, RBRACK, LBRACE, RBRACE, COLON = map(pp.Suppress, "[]{}:")
jsonString = pp.dblQuotedString().setParseAction(pp.removeQuotes)
jsonNumber = ppc.number()
jsonObject = pp.Forward()
jsonValue = pp.Forward()
jsonElements = pp.delimitedList(jsonValue)
jsonArray = pp.Group(LBRACK + pp.Optional(jsonElements, []) + RBRACK)
jsonValue << (
jsonString | jsonNumber | pp.Group(jsonObject) | jsonArray | TRUE | FALSE | NULL
)
memberDef = pp.Group(jsonString + COLON + jsonValue)
jsonMembers = pp.delimitedList(memberDef)
jsonObject << pp.Dict(LBRACE + pp.Optional(jsonMembers) + RBRACE)
jsonComment = pp.cppStyleComment
jsonObject.ignore(jsonComment)
if __name__ == "__main__":
testdata = """
{
"glossary": {
"title": "example glossary",
"GlossDiv": {
"title": "S",
"GlossList":
{
"ID": "SGML",
"SortAs": "SGML",
"GlossTerm": "Standard Generalized Markup Language",
"TrueValue": true,
"FalseValue": false,
"Gravity": -9.8,
"LargestPrimeLessThan100": 97,
"AvogadroNumber": 6.02E23,
"EvenPrimesGreaterThan2": null,
"PrimesLessThan10" : [2,3,5,7],
"Acronym": "SGML",
"Abbrev": "ISO 8879:1986",
"GlossDef": "A meta-markup language, used to create markup languages such as DocBook.",
"GlossSeeAlso": ["GML", "XML", "markup"],
"EmptyDict" : {},
"EmptyList" : []
}
}
}
}
"""
results = jsonObject.parseString(testdata)
results.pprint()
print()
def testPrint(x):
print(type(x), repr(x))
print(list(results.glossary.GlossDiv.GlossList.keys()))
testPrint(results.glossary.title)
testPrint(results.glossary.GlossDiv.GlossList.ID)
testPrint(results.glossary.GlossDiv.GlossList.FalseValue)
testPrint(results.glossary.GlossDiv.GlossList.Acronym)
testPrint(results.glossary.GlossDiv.GlossList.EvenPrimesGreaterThan2)
testPrint(results.glossary.GlossDiv.GlossList.PrimesLessThan10)
+5
View File
@@ -25,6 +25,11 @@ financials = financials.createInstance(None)
class Test(unittest.TestCase):
@classmethod
def tearDownClass(cls):
# this avoids "ResourceWarning: unclosed..." on cached socket connections
financials.close()
def test_currency(self):
s = financials.getRealtime('ETH-EUR', Datacode.LAST_PRICE.value, 'COINBASE')
self.assertEqual(float, type(s), 'test_currency LAST_PRICE')
+25 -11
View File
@@ -23,6 +23,11 @@ financials = financials.createInstance(None)
class Test(unittest.TestCase):
@classmethod
def tearDownClass(cls):
# this avoids "ResourceWarning: unclosed..." on cached socket connections
financials.close()
def test_currency(self):
s = financials.getRealtime('EURGBP', 'LAST_PRICE', 'FT')
self.assertEqual(float, type(s), 'test_currency LAST_PRICE')
@@ -137,29 +142,31 @@ class Test(unittest.TestCase):
def test_US_futures(self):
s = financials.getRealtime('ESU3:IOM', Datacode.NAME.value, 'FT')
self.assertEqual(str, type(s), 'test_realtime_US_futures NAME {}'.format(s))
self.assertEqual('EMINI S&P SEP3', s, 'test_US_futures NAME {}'.format(s))
# https://markets.ft.com/data/commodities/tearsheet/summary?s=823439664 ESH26:IOM - EMINI S&P MAR26
s = financials.getRealtime('ESU3:IOM', Datacode.LAST_PRICE.value, 'FT')
s = financials.getRealtime('823439664', Datacode.NAME.value, 'FT')
self.assertEqual(str, type(s), 'test_realtime_US_futures NAME {}'.format(s))
self.assertEqual('EMINI S&P MAR26', s, 'test_US_futures NAME {}'.format(s))
s = financials.getRealtime('823439664', Datacode.LAST_PRICE.value, 'FT')
self.assertEqual(float, type(s), 'test_US_futures LAST_PRICE {}'.format(s))
# s = financials.getRealtime('ESH3:IOM', Datacode.OPEN.value, 'FT')
# s = financials.getRealtime('775326843', Datacode.OPEN.value, 'FT')
# self.assertEqual(float, type(s), 'test_US_futures OPEN {}'.format(s))
s = financials.getRealtime('ESU3:IOM', Datacode.VOLUME.value, 'FT')
s = financials.getRealtime('823439664', Datacode.VOLUME.value, 'FT')
self.assertEqual(float, type(s), 'test_US_futures VOLUME {}'.format(s))
s = financials.getRealtime('ESU3:IOM', Datacode.LOW_52_WEEK.value, 'FT')
s = financials.getRealtime('823439664', Datacode.LOW_52_WEEK.value, 'FT')
self.assertEqual(float, type(s), 'test_US_futures LOW_52_WEEK {}'.format(s))
s = financials.getRealtime('ESU3:IOM', Datacode.HIGH_52_WEEK.value, 'FT')
s = financials.getRealtime('823439664', Datacode.HIGH_52_WEEK.value, 'FT')
self.assertEqual(float, type(s), 'test_US_futures HIGH_52_WEEK {}'.format(s))
s = financials.getRealtime('ESU3:IOM', Datacode.CHANGE.value, 'FT')
s = financials.getRealtime('823439664', Datacode.CHANGE.value, 'FT')
self.assertEqual(float, type(s), 'test_US_futures CHANGE {}'.format(s))
s = financials.getRealtime('ESU3:IOM', Datacode.CHANGE_IN_PERCENT.value, 'FT')
s = financials.getRealtime('823439664', Datacode.CHANGE_IN_PERCENT.value, 'FT')
self.assertEqual(float, type(s), 'test_US_futures CHANGE_IN_PERCENT {}'.format(s))
def test_UK_ETF(self):
@@ -268,7 +275,7 @@ class Test(unittest.TestCase):
self.assertTrue(testutils.is_date(s), 'test_DE_equity EX_DIV_DATE {}'.format(s))
s = financials.getRealtime('ISHAX:GER', 'NAME', 'FT')
self.assertEqual('INTERSHOP Communications AG', s, 'test_DE_equity NAME {}'.format(s))
self.assertEqual('Intershop Communications AG', s, 'test_DE_equity NAME {}'.format(s))
s = financials.getRealtime('ISHAX:GER', 'BETA', 'FT')
self.assertTrue(testutils.is_positive_float(s), 'test_DE_equity BETA {}'.format(s))
@@ -302,6 +309,13 @@ class Test(unittest.TestCase):
self.assertEqual(str, type(s), 'test_DK_equity INDUSTRY {}'.format(s))
self.assertEqual('Pharmaceuticals and Biotechnology', s, 'test_DK_equity INDUSTRY {}'.format(s))
def test_SE_equity(self):
s = financials.getRealtime('ACRI A:STO', 'name', 'FT')
self.assertEqual('Acrinova AB (publ)', s, 'test_SE_equity NAME {}'.format(s))
s = financials.getRealtime('SE0015660014', 'name', 'FT')
self.assertEqual('Acrinova AB (publ)', s, 'test_SE_equity NAME {}'.format(s))
def test_TY_equity(self):
s = financials.getRealtime('6503:TYO', 'OPEN', 'FT')
self.assertEqual(float, type(s), 'test_TY_equity OPEN {}'.format(s))
-259
View File
@@ -1,259 +0,0 @@
# test_google.py
#
# license: GNU LGPL
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 3 of the License, or (at your option) any later version.
import argparse
import logging
import sys
import unittest
logging.basicConfig(level=logging.ERROR, format="%(asctime)s %(name)s %(levelname)s %(message)s")
import financials
from datacode import Datacode
financials = financials.createInstance(None)
# class Test(unittest.TestCase):
class Test:
def test_currency(self):
s = financials.getRealtime('EURGBP', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertIsNone(s, 'test_currency LAST_PRICE')
def test_UK_equity(self):
s = financials.getRealtime('LON:VOD', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_UK_equity LAST_PRICE {}'.format(s))
s = financials.getRealtime('LON:VOD', Datacode.TICKER.value, 'GOOGLE')
self.assertEqual(s, 'VOD', 'test_UK_equity TICKER')
s = financials.getRealtime('LON:VOD', Datacode.NAME.value, 'GOOGLE')
self.assertEqual(type(s), str, 'test_UK_equity NAME')
s = financials.getRealtime('LON:VOD', Datacode.EXCHANGE.value, 'GOOGLE')
self.assertEqual(s, 'LON', 'test_UK_equity EXCHANGE')
s = financials.getRealtime('LON:VOD', Datacode.PREV_CLOSE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_UK_equity PREV_CLOSE {}'.format(s))
# MARKET_CAP missing for UK stock but available for German stock - weekend issue (FX) ?
s = financials.getRealtime('LON:VOD', Datacode.MARKET_CAP.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_UK_equity MARKET_CAP {}'.format(s))
def test_UK_ETF(self):
s = financials.getRealtime('LON:CSP1', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_UK_ETF LAST_PRICE {}'.format(s))
s = financials.getRealtime('LON:CSP1', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual(s, 'GBX', 'test_UK_ETF CURRENCY')
s = financials.getRealtime('LON:FTAL', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_UK_ETF LAST_PRICE {}'.format(s))
s = financials.getRealtime('LON:FTAL', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual(s, 'GBP', 'test_UK_ETF CURRENCY')
s = financials.getRealtime('LON:FTAL', Datacode.NAME.value, 'GOOGLE')
self.assertEqual(type(s), str, 'test_UK_ETF NAME')
def test_DE_equity(self):
s = financials.getRealtime('FRA:SAP', 7, 'GOOGLE')
self.assertEqual(float, type(s), 'test_DE_equity 7')
s = financials.getRealtime('FRA:SAP', '7', 'GOOGLE')
self.assertEqual(float, type(s), 'test_DE_equity \'7\'')
s = financials.getRealtime('FRA:SAP', 8, 'GOOGLE')
self.assertEqual(type(s), str, 'test_DE_equity 8')
s = financials.getRealtime('FRA:SAP', 8.1, 'GOOGLE')
self.assertEqual(type(s), str, 'test_DE_equity 8.1')
s = financials.getRealtime('FRA:SAP', '8.1', 'GOOGLE')
self.assertEqual(type(s), str, 'test_DE_equity \'8.1\'')
s = financials.getRealtime('FRA:SAP', 10, 'GOOGLE')
self.assertEqual(type(s), str, 'test_DE_equity 10')
s = financials.getRealtime('FRA:SAP', '11', 'GOOGLE')
self.assertEqual(float, type(s), 'test_DE_equity \'11\'')
s = financials.getRealtime('FRA:SAP', '21', 'GOOGLE')
self.assertEqual(float, type(s), 'test_DE_equity \'21\'')
s = financials.getRealtime('FRA:SAP', Datacode.TIMEZONE.value, 'GOOGLE')
# self.assertEqual(s, 'Europe/Berlin', 'test_DE_equity TIMEZONE')
self.assertTrue(s == 'CET' or s == 'CEST', 'test_DE_equity TIMEZONE: {}'.format(s))
def test_DE_ETF(self):
s = financials.getRealtime('FRA:C060', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_DE_ETF LAST_PRICE {}'.format(s))
s = financials.getRealtime('FRA:C060', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual('EUR', s, 'test_DE_ETF CURRENCY')
s = financials.getRealtime('FRA:C060', Datacode.TICKER.value, 'GOOGLE')
self.assertEqual('C060', s, 'test_DE_ETF TICKER')
s = financials.getRealtime('FRA:C060', Datacode.EXCHANGE.value, 'GOOGLE')
self.assertEqual('FRA', s, 'test_DE_ETF EXCHANGE')
s = financials.getRealtime('FRA:C060', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual('EUR', s, 'test_DE_ETF CURRENCY')
s = financials.getRealtime('FRA:C060', Datacode.MARKET_CAP.value, 'GOOGLE')
self.assertIsNone(s, 'test_DE_ETF MARKET_CAP {}'.format(s))
def test_TY_equity(self):
s = financials.getRealtime('TYO:6503', Datacode.OPEN.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_TY_equity OPEN {}'.format(s))
s = financials.getRealtime('TYO:6503', Datacode.LOW.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_TY_equity LOW {}'.format(s))
s = financials.getRealtime('TYO:6503', Datacode.HIGH.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_TY_equity HIGH {}'.format(s))
s = financials.getRealtime('TYO:6503', Datacode.LOW_52_WEEK.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_TY_equity LOW_52_WEEK {}'.format(s))
s = financials.getRealtime('TYO:6503', Datacode.HIGH_52_WEEK.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_TY_equity HIGH_52_WEEK {}'.format(s))
s = financials.getRealtime('TYO:6503', Datacode.MARKET_CAP.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_TY_equity MARKET_CAP {}'.format(s))
# s = financials.getRealtime('TYO:6503', Datacode.VOLUME.value, 'GOOGLE')
# self.assertEqual(float, type(s), 'test_TY_equity VOLUME {}'.format(s))
s = financials.getRealtime('TYO:6503', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual(s, 'JPY', 'test_TY_equity CURRENCY')
def test_US_equity(self):
s = financials.getRealtime(' NASDAQ : AAPL ', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_equity LAST_PRICE {}'.format(s))
s = financials.getRealtime(' NASDAQ : AAPL ', Datacode.TICKER.value, 'GOOGLE')
self.assertEqual(s, 'AAPL', 'test_US_equity TICKER')
s = financials.getRealtime(' NASDAQ : AAPL ', Datacode.EXCHANGE.value, 'GOOGLE')
self.assertEqual(s, 'NASDAQ', 'test_US_equity EXCHANGE')
s = financials.getRealtime(' NASDAQ : AAPL ', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual(s, 'USD', 'test_US_equity CURRENCY')
s = financials.getRealtime('NYSE:IBM', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_equity LAST_PRICE {}'.format(s))
s = financials.getRealtime('NYSE:IBM', Datacode.TICKER.value, 'GOOGLE')
self.assertEqual(s, 'IBM', 'test_US_equity TICKER')
s = financials.getRealtime('NYSE:IBM', Datacode.EXCHANGE.value, 'GOOGLE')
self.assertEqual(s, 'NYSE', 'test_US_equity EXCHANGE')
s = financials.getRealtime('NYSE:IBM', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual(s, 'USD', 'test_US_equity CURRENCY')
s = financials.getRealtime('NYSE:IBM', Datacode.NAME.value, 'GOOGLE')
self.assertEqual(type(s), str, 'test_US_equity NAME')
self.assertEqual(s, 'IBM Common Stock', 'test_US_equity NAME')
s = financials.getRealtime('NYSE:IBM', Datacode.LOW.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_equity LOW {}'.format(s))
s = financials.getRealtime('NYSE:IBM', Datacode.HIGH.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_equity HIGH {}'.format(s))
s = financials.getRealtime('NYSE:IBM', Datacode.LOW_52_WEEK.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_equity LOW_52_WEEK {}'.format(s))
s = financials.getRealtime('NYSE:IBM', Datacode.HIGH_52_WEEK.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_equity HIGH_52_WEEK {}'.format(s))
s = financials.getRealtime('NYSE:IBM', Datacode.MARKET_CAP.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_equity MARKET_CAP {}'.format(s))
# s = financials.getRealtime('NYSE:IBM', Datacode.VOLUME.value, 'GOOGLE')
# self.assertEqual(float, type(s), 'test_US_equity VOLUME {}'.format(s))
s = financials.getRealtime('NYSE:IBM', Datacode.TIMESTAMP.value, 'GOOGLE')
self.assertEqual(s, 'Data doesn\'t exist - 999', 'test_US_equity TIMESTAMP')
s = financials.getRealtime('NYSE:IBM', Datacode.TIMEZONE.value, 'GOOGLE')
# self.assertEqual('America/New_York', s, 'test_US_equity TIMEZONE')
# self.assertEqual('GMT-4', s, 'test_US_equity TIMEZONE')
self.assertEqual('GMT-5', s, 'test_US_equity TIMEZONE')
def test_US_mutuals(self):
s = financials.getRealtime('MUTF:VFIAX', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_US_mutuals LAST_PRICE - {}'.format(s))
s = financials.getRealtime('MUTF:VFIAX', Datacode.CURRENCY.value, 'GOOGLE')
self.assertEqual(s, 'USD', 'test_US_mutuals CURRENCY')
s = financials.getRealtime('MUTF:VFIAX', Datacode.TIMEZONE.value, 'GOOGLE')
self.assertIsNone(s, 'test_US_mutuals TIMEZONE')
def test_index(self):
s = financials.getRealtime('INDEXDB:DAX', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_index LAST_PRICE {}'.format(s))
s = financials.getRealtime('INDEXDB:DAX', Datacode.CHANGE_IN_PERCENT.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_index CHANGE_IN_PERCENT')
s = financials.getRealtime('INDEXDB:DAX', Datacode.CHANGE.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_index CHANGE')
s = financials.getRealtime('INDEXDB:DAX', Datacode.OPEN.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_index OPEN')
s = financials.getRealtime('INDEXDB:DAX', Datacode.LOW.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_index LOW')
s = financials.getRealtime('INDEXDB:DAX', Datacode.HIGH.value, 'GOOGLE')
self.assertEqual(float, type(s), 'test_index HIGH')
def test_errors(self):
s = financials.getRealtime(None, Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(s, 'Ticker is empty', 'test_errors')
s = financials.getRealtime('NYS:IBM', None, 'GOOGLE')
self.assertEqual(s, 'Datacode is empty', 'test_errors')
s = financials.getRealtime('DOES_NOT_EXISTS', Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertIsNone(s, 'test_errors')
s = financials.getRealtime('NYS:IBM', 'Foo', 'GOOGLE')
self.assertEqual(s, 'Datacode is invalid', 'test_errors')
# Historic data not supported on GOOGLE
s = financials.getHistoric('NYS:IBM', Datacode.LAST_PRICE.value, '2017-01-01', 'GOOGLE')
self.assertEqual(s, 'Source \'GOOGLE\' not supported', 'test_errors')
def test_errors_cell_range_passed(self):
cell_range = ((1, 2), ('3', '4'), (5.0, 6.0))
s = financials.getRealtime(cell_range, Datacode.LAST_PRICE.value, 'GOOGLE')
self.assertEqual(s, 'Cell range not allowed for ticker', 'test_errors')
s = financials.getRealtime('NYS:IBM', cell_range, 'GOOGLE')
self.assertEqual(s, 'Cell range not allowed for datacode', 'test_errors')
s = financials.getRealtime('NYS:IBM', Datacode.LAST_PRICE.value, cell_range)
self.assertEqual(s, 'Cell range not allowed for source', 'test_errors')
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('unittest_args', nargs='*')
args = parser.parse_args()
unit_argv = [sys.argv[0]] + args.unittest_args
unittest.main(argv=unit_argv)
+1 -1
View File
@@ -1,4 +1,4 @@
# test_google.py
# test_support.py
#
# license: GNU LGPL
#
+68 -53
View File
@@ -24,12 +24,17 @@ import testutils
financials = financials.createInstance(None)
def urlopen_fail(self, url, redirect=True, data=None, headers={}, cookies=[], **kwargs):
raise baseclient.HttpException(url, 'simulated urlopen() failed')
def urlopen_fail(self, url, data=None):
raise baseclient.HttpException(url, 'ERROR: simulated urlopen() failed')
class Test(unittest.TestCase):
@classmethod
def tearDownClass(cls):
# this avoids "ResourceWarning: unclosed..." on cached socket connections
financials.close()
def test_recovery_from_urlopen_error_issue(self):
financials.yahoo.last_url = 'blank'
@@ -100,19 +105,19 @@ class Test(unittest.TestCase):
s = financials.getRealtime('IBM', Datacode.NAME.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_equity NAME {}'.format(s))
self.assertEqual(s, 'International Business Machines Corporation (IBM)',
self.assertEqual(s, 'International Business Machines Corporation',
'test_realtime_US_equity NAME {}'.format(s))
s = financials.getRealtime('IBM', Datacode.SECTOR.value, 'YAHOO')
s = financials.getRealtime('IBM', 'SECTOR', 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_equity SECTOR {}'.format(s))
self.assertEqual(s, 'Technology', 'test_realtime_US_equity SECTOR {}'.format(s))
s = financials.getRealtime('IBM', Datacode.INDUSTRY.value, 'YAHOO')
s = financials.getRealtime('IBM', 'INDUSTRY', 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_equity INDUSTRY {}'.format(s))
self.assertEqual(s, 'Information Technology Services', 'test_realtime_US_equity INDUSTRY {}'.format(s))
s = financials.getRealtime('IBM', Datacode.TIMEZONE.value, 'YAHOO')
self.assertEqual(s, 'EDT', 'test_realtime_US_equity TIMEZONE {}'.format(s))
self.assertTrue(s == 'America/New_York', 'test_realtime_US_equity TIMEZONE: {}'.format(s))
s = financials.getRealtime('IBM', Datacode.BETA.value, 'YAHOO')
self.assertTrue(testutils.is_positive_float(s), 'test_realtime_US_equity BETA {}'.format(s))
@@ -133,14 +138,17 @@ class Test(unittest.TestCase):
self.assertEqual(str, type(s), 'test_realtime_US_equity EX_DIV_DATE {}'.format(s))
self.assertTrue(testutils.is_date(s), 'test_realtime_US_equity EX_DIV_DATE {}'.format(s))
s = financials.getRealtime('IBM', Datacode.PAYOUT_RATIO.value, 'YAHOO')
s = financials.getRealtime('IBM', 'PAYOUT_RATIO', 'YAHOO')
self.assertTrue(testutils.is_positive_float(s), 'test_realtime_US_equity PAYOUT_RATIO {}'.format(s))
# self.assertEqual(None, s, 'test_realtime_US_equity PAYOUT_RATIO {}'.format(s))
s = financials.getRealtime('IBM', Datacode.SHARES_OUT.value, 'YAHOO')
s = financials.getRealtime('IBM', 'SHARES_OUT', 'YAHOO')
self.assertTrue(testutils.is_positive_float(s), 'test_realtime_US_equity SHARES_OUT {}'.format(s))
# self.assertEqual(None, s, 'test_realtime_US_equity SHARES_OUT {}'.format(s))
s = financials.getRealtime('IBM', Datacode.FREE_FLOAT.value, 'YAHOO')
s = financials.getRealtime('IBM', 'FREE_FLOAT', 'YAHOO')
self.assertTrue(testutils.is_positive_float(s), 'test_realtime_US_equity FREE_FLOAT {}'.format(s))
# self.assertEqual(None, s, 'test_realtime_US_equity FREE_FLOAT {}'.format(s))
s = financials.getRealtime('IBM', Datacode.EXCHANGE.value, 'YAHOO')
self.assertEqual(s, 'NYSE', 'test_realtime_US_equity EXCHANGE')
@@ -157,74 +165,71 @@ class Test(unittest.TestCase):
self.assertEqual(float, type(s), 'test_realtime_US_mutuals DIV {}'.format(s))
s = financials.getRealtime('VFIAX', Datacode.DIV_YIELD.value, 'YAHOO')
# self.assertIsNone(s, 'test_realtime_US_mutuals DIV_YIELD {}'.format(s)) # no yield
self.assertTrue(testutils.is_positive_float(s), 'test_realtime_US_mutuals DIV_YIELD {}'.format(s))
self.assertIsNone(s, 'test_realtime_US_mutuals DIV_YIELD {}'.format(s)) # no yield
s = financials.getRealtime('SHRAX', Datacode.DIV.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_mutuals DIV {}'.format(s))
s = financials.getRealtime('SHRAX', Datacode.DIV_YIELD.value, 'YAHOO')
# self.assertIsNone(s, 'test_realtime_US_mutuals DIV_YIELD {}'.format(s)) # no yield
self.assertEqual(float, type(s), 'test_realtime_US_mutuals DIV_YIELD {}'.format(s))
self.assertIsNone(s, 'test_realtime_US_mutuals DIV_YIELD {}'.format(s)) # no yield
# s = financials.getRealtime('VERX.L', Datacode.DIV.value, 'YAHOO')
# self.assertIsNone(s, 'test_realtime_US_mutuals DIV {}'.format(s)) # no dividend
s = financials.getRealtime('VERX.L', Datacode.DIV.value, 'YAHOO')
self.assertIsNone(s, 'test_realtime_US_mutuals DIV {}'.format(s)) # no dividend
s = financials.getRealtime('VERX.L', Datacode.DIV_YIELD.value, 'YAHOO')
self.assertIsNone(s, 'test_realtime_US_mutuals DIV_YIELD {}'.format(s)) # no yield
# self.assertEqual(float, type(s), 'test_realtime_US_mutuals DIV_YIELD {}'.format(s))
def test_realtime_US_options(self):
# symbol from https://finance.yahoo.com/quote/IBM/options?p=IBM
s = financials.getRealtime('IBM240119C00120000', Datacode.PREV_CLOSE.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.PREV_CLOSE.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_options PREV_CLOSE {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.NAME.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.NAME.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_options NAME {}'.format(s))
self.assertEqual('IBM Jan 2024 120.000 call', s, 'test_realtime_US_options NAME {}'.format(s))
self.assertEqual('IBM Jan 2026 230.000 call', s, 'test_realtime_US_options NAME {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.EXPIRY_DATE.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.EXPIRY_DATE.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_options EXPIRY_DATE {}'.format(s))
self.assertTrue(testutils.is_date(s), 'test_realtime_US_options EXPIRY_DATE {}'.format(s))
self.assertEqual("2024-01-19", s, 'test_realtime_US_options EXPIRY_DATE {}'.format(s))
self.assertEqual("2026-01-16", s, 'test_realtime_US_options EXPIRY_DATE {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.LAST_PRICE.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.LAST_PRICE.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_options LAST_PRICE {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.OPEN.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.OPEN.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_options OPEN {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.VOLUME.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.VOLUME.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_options VOLUME {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.BID.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.BID.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_options BID {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.ASK.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.ASK.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_options ASK {}'.format(s))
s = financials.getRealtime('IBM240119C00120000', Datacode.PAYOUT_RATIO.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.PAYOUT_RATIO.value, 'YAHOO')
self.assertIsNone(s, 'test_realtime_US_options PAYOUT_RATIO {}'.format(s))
s = financials.getRealtime('IBM230120C00130000', Datacode.SECTOR.value, 'YAHOO')
s = financials.getRealtime('IBM260116C00230000', Datacode.SECTOR.value, 'YAHOO')
self.assertIsNone(s, 'test_realtime_US_options SECTOR {}'.format(s))
def test_realtime_US_futures(self):
s = financials.getRealtime('ES=F', Datacode.NAME.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_futures NAME {}'.format(s))
self.assertEqual('E-Mini S&P 500 Sep 23 (ES=F)', s, 'test_realtime_US_futures NAME {}'.format(s))
self.assertEqual('E-Mini S&P 500 Jun 25', s, 'test_realtime_US_futures NAME {}'.format(s))
s = financials.getRealtime('ES=F', Datacode.TICKER.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_futures TICKER {}'.format(s))
self.assertEqual('ESU23.CME', s, 'test_realtime_US_futures TICKER {}'.format(s))
self.assertEqual('ESM25.CME', s, 'test_realtime_US_futures TICKER {}'.format(s))
s = financials.getRealtime('ES=F', Datacode.SETTLEMENT_DATE.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_US_futures SETTLEMENT_DATE {}'.format(s))
self.assertTrue(testutils.is_date(s), 'test_realtime_US_futures SETTLEMENT_DATE {}'.format(s))
self.assertEqual("2023-09-15", s, 'test_realtime_US_futures SETTLEMENT_DATE {}'.format(s))
self.assertEqual("2025-06-20", s, 'test_realtime_US_futures SETTLEMENT_DATE {}'.format(s))
s = financials.getRealtime('ES=F', Datacode.LAST_PRICE.value, 'YAHOO')
self.assertEqual(float, type(s), 'test_realtime_US_futures LAST_PRICE {}'.format(s))
@@ -259,12 +264,23 @@ class Test(unittest.TestCase):
self.assertEqual(float, type(s), 'test_realtime_UK_ETF LAST_PRICE {}'.format(s))
s = financials.getRealtime('VERX.L', Datacode.TIMEZONE.value, 'YAHOO')
# self.assertEqual('GMT', s, 'test_realtime_UK_ETF TIMEZONE {}'.format(s))
self.assertEqual('BST', s, 'test_realtime_UK_ETF TIMEZONE {}'.format(s))
self.assertTrue(s == 'Europe/London', 'test_realtime_UK_ETF TIMEZONE: {}'.format(s))
s = financials.getRealtime('CSP1.L', Datacode.NAME.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_UK_ETF NAME {}'.format(s))
self.assertEqual('iShares VII PLC - iShares Core S&P 500 UCITS ETF (CSP1.L)', s, 'test_realtime_UK_ETF NAME {}'.format(s))
self.assertEqual('iShares Core S&P 500 UCITS ETF USD (Acc)', s, 'test_realtime_UK_ETF NAME {}'.format(s))
s = financials.getRealtime('VERX.L', 'SECTOR', 'YAHOO')
self.assertIsNone(s, 'test_realtime_UK_ETF SECTOR {}'.format(s))
s = financials.getRealtime('VERX.L', 'INDUSTRY', 'YAHOO')
self.assertIsNone(s, 'test_realtime_UK_ETF INDUSTRY {}'.format(s))
s = financials.getRealtime('VERX.L', 'PAYOUT_RATIO', 'YAHOO')
self.assertIsNone(s, 'test_realtime_UK_ETF PAYOUT_RATIO {}'.format(s))
s = financials.getRealtime('VERX.L', 'SHARES_OUT', 'YAHOO')
self.assertIsNone(s, 'test_realtime_UK_ETF SHARES_OUT {}'.format(s))
def test_realtime_DE_equity(self):
@@ -272,8 +288,7 @@ class Test(unittest.TestCase):
self.assertEqual(float, type(s), 'test_realtime_DE_equity LAST_PRICE {}'.format(s))
s = financials.getRealtime('SAP.DE', Datacode.TIMEZONE.value, 'YAHOO')
# self.assertEqual('CET', s, 'test_realtime_DE_equity TIMEZONE {}'.format(s))
self.assertEqual('CEST', s, 'test_realtime_DE_equity TIMEZONE {}'.format(s))
self.assertTrue(s == 'Europe/Berlin', 'test_realtime_DE_equity TIMEZONE: {}'.format(s))
s = financials.getRealtime('SAP.DE', Datacode.SECTOR.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_DE_equity SECTOR {}'.format(s))
@@ -281,7 +296,7 @@ class Test(unittest.TestCase):
s = financials.getRealtime('SAP.DE', Datacode.INDUSTRY.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_DE_equity INDUSTRY {}'.format(s))
self.assertEqual(s, 'SoftwareApplication', 'test_realtime_DE_equity INDUSTRY {}'.format(s))
self.assertEqual(s, 'Software - Application', 'test_realtime_DE_equity INDUSTRY {}'.format(s))
s = financials.getRealtime('LYY8.DE', Datacode.NAME.value, 'YAHOO')
self.assertEqual(str, type(s), 'test_realtime_DE_equity NAME {}'.format(s))
@@ -324,21 +339,21 @@ class Test(unittest.TestCase):
s = financials.getRealtime('LUMI.TA', 'INDUSTRY', 'YAHOO')
self.assertEqual(str, type(s), 'test_TLV_equity INDUSTRY {}'.format(s))
self.assertEqual('BanksRegional', s, 'test_TA_equity INDUSTRY {}'.format(s))
self.assertEqual('Banks - Regional', s, 'test_TA_equity INDUSTRY {}'.format(s))
def test_DK_equity(self):
s = financials.getRealtime('NOVO-B.CO', 'last_price', 'YAHOO')
self.assertEqual(float, type(s), 'test_DK_equity LAST_PRICE {}'.format(s))
s = financials.getRealtime('NOVO-B.CO', 'name', 'YAHOO')
self.assertEqual('Novo Nordisk A/S (NOVO-B.CO)', s, 'test_DK_equity NAME {}'.format(s))
self.assertEqual('Novo Nordisk A/S', s, 'test_DK_equity NAME {}'.format(s))
s = financials.getRealtime('NOVO-B.CO', 'currency', 'YAHOO')
self.assertEqual('DKK', s, 'test_DK_equity CURRENCY {}'.format(s))
s = financials.getRealtime('NOVO-B.CO', 'industry', 'YAHOO')
self.assertEqual(str, type(s), 'test_DK_equity INDUSTRY {}'.format(s))
self.assertEqual('Biotechnology', s, 'test_DK_equity INDUSTRY {}'.format(s))
self.assertEqual('Drug Manufacturers - General', s, 'test_DK_equity INDUSTRY {}'.format(s))
s = financials.getRealtime('MAERSK-B.CO', 'currency', 'YAHOO')
self.assertEqual('DKK', s, 'test_DK_equity CURRENCY {}'.format(s))
@@ -377,7 +392,7 @@ class Test(unittest.TestCase):
self.assertEqual(s, 'JPY', 'test_TY_equity CURRENCY')
s = financials.getRealtime('6503.T', Datacode.TIMEZONE.value, 'YAHOO')
self.assertEqual(s, 'JST', 'test_TY_equity TIMEZONE')
self.assertEqual(s, 'Asia/Tokyo', 'test_TY_equity TIMEZONE')
def test_historic_US_equity(self):
@@ -391,15 +406,15 @@ class Test(unittest.TestCase):
self.assertIsNone(s, 'test_historic_US_equity LAST_PRICE {}'.format(s))
s = financials.getHistoric('IBM', Datacode.CLOSE.value, '2017-01-03', 'YAHOO')
self.assertEqual(159.837479, s, 'test_historic_US_equity CLOSE {}'.format(s))
self.assertAlmostEqual(159.84, s, 2, 'test_historic_US_equity CLOSE {}'.format(s))
financials.yahoo.historicdata = {}
s = financials.getHistoric('IBM', Datacode.CLOSE.value, '2017-01-03', 'YAHOO')
self.assertEqual(159.837479, s, 'test_historic_US_equity CLOSE {}'.format(s))
self.assertAlmostEqual(159.84, s, 2, 'test_historic_US_equity CLOSE {}'.format(s))
directory = os.path.join(str(pathlib.Path.home()), '.financials-extension')
ibm = os.path.join(directory, 'yahoo-IBM.csv')
ibm = os.path.join(directory, 'yahoo-hist-IBM.json')
try:
os.unlink(ibm)
except:
@@ -408,7 +423,7 @@ class Test(unittest.TestCase):
financials.yahoo.historicdata = {}
s = financials.getHistoric('IBM', Datacode.CLOSE.value, '2017-01-03', 'YAHOO')
self.assertEqual(159.837479, s, 'test_historic_US_equity CLOSE {}'.format(s))
self.assertAlmostEqual(159.84, s, 2, 'test_historic_US_equity CLOSE {}'.format(s))
s = financials.getHistoric('IBM', Datacode.ADJ_CLOSE.value, '2017-01-03', 'YAHOO')
self.assertEqual(float, type(s), 'test_historic_US_equity ADJ_CLOSE {}'.format(s))
@@ -416,7 +431,7 @@ class Test(unittest.TestCase):
def test_historic_UK_ETF(self):
directory = os.path.join(str(pathlib.Path.home()), '.financials-extension')
verx = os.path.join(directory, 'yahoo-VERX.L.csv')
verx = os.path.join(directory, 'yahoo-hist-VERX.L.json')
try:
os.unlink(verx)
except:
@@ -428,10 +443,10 @@ class Test(unittest.TestCase):
self.assertEqual(s, 'Not a trading day \'2017-01-01\'', 'test_historic_UK_ETF LAST_PRICE {}'.format(s))
s = financials.getHistoric('VERX.L', Datacode.CLOSE.value, '2017-01-03', 'YAHOO')
self.assertEqual(s, 23.24, 'test_historic_UK_ETF CLOSE {}'.format(s))
self.assertAlmostEqual(s, 23.24, 2, 'test_historic_UK_ETF CLOSE {}'.format(s))
s = financials.getHistoric('VERX.L', Datacode.CLOSE.value, '2016-10-03', 'YAHOO')
self.assertEqual(s, 22.26, 'test_historic_UK_ETF CLOSE {}'.format(s))
self.assertAlmostEqual(s, 22.26, 2, 'test_historic_UK_ETF CLOSE {}'.format(s))
# Inception Date 2014-09-30
s = financials.getHistoric('VERX.L', Datacode.CLOSE.value, '2018-04-02', 'YAHOO')
@@ -442,13 +457,13 @@ class Test(unittest.TestCase):
self.assertEqual(s, 'Not a trading day \'2015-01-01\'', 'test_historic_UK_ETF CLOSE {}'.format(s))
s = financials.getHistoric('VERX.L', Datacode.CLOSE.value, 42738, 'YAHOO') # 2017-01-03
self.assertEqual(s, 23.24, 'test_historic_UK_ETF CLOSE {}'.format(s))
self.assertAlmostEqual(s, 23.24, 2, 'test_historic_UK_ETF CLOSE {}'.format(s))
s = financials.getHistoric('VERX.L', Datacode.CLOSE.value, 42738.0, 'YAHOO') # 2017-01-03
self.assertEqual(s, 23.24, 'test_historic_UK_ETF CLOSE {}'.format(s))
self.assertAlmostEqual(s, 23.24, 2, 'test_historic_UK_ETF CLOSE {}'.format(s))
s = financials.getHistoric('VERX.L', Datacode.CLOSE.value, 42646.0, 'YAHOO') # 2016-10-03
self.assertEqual(s, 22.26, 'test_historic_UK_ETF CLOSE {}'.format(s))
self.assertAlmostEqual(s, 22.26, 2, 'test_historic_UK_ETF CLOSE {}'.format(s))
def test_historic_DE_equity(self):
@@ -456,10 +471,10 @@ class Test(unittest.TestCase):
self.assertEqual(s, 'Not a trading day \'2017-01-01\'', 'test_historic_DE_equity LAST_PRICE {}'.format(s))
s = financials.getHistoric('SAP.DE', Datacode.CLOSE.value, '2017-01-03', 'YAHOO')
self.assertEqual(s, 82.889999, 'test_historic_DE_equity CLOSE {}'.format(s))
self.assertAlmostEqual(s, 82.89, 2, 'test_historic_DE_equity CLOSE {}'.format(s))
s = financials.getHistoric('LYY8.DE', Datacode.CLOSE.value, '2017-01-03', 'YAHOO')
self.assertEqual(s, 96.010002, 'test_historic_DE_equity CLOSE {}'.format(s))
self.assertAlmostEqual(s, 96.01, 2, 'test_historic_DE_equity CLOSE {}'.format(s))
def test_realtime_errors(self):