mbk-dev / okama / 290

pending completion

Build # 290

Build Type

push

travis-ci-com

Committed by

chilango74

Commit Message

docs: update EfficientFrontier.plot_transition_map example in README.md

Run Details

1947 of 2222 relevant lines covered (87.62%)

2.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.83

/okama/api/api_methods.py

from typing import Union

import pandas as pd
import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

from okama import settings


class API:
    """
    Set of methods to get data from API.
    """

    # TODO: introduce 'from' & 'to' for dates.

    api_url = "http://api.okama.io:5000"
    default_timeout = 10  # seconds

    endpoint_ror = "/api/ts/ror/"
    endpoint_symbol = "/api/symbol/"
    endpoint_search = "/api/search/"
    endpoint_live_price = "/api/live_price/"
    endpoint_adjusted_close = "/api/ts/adjusted_close/"
    endpoint_close = "/api/ts/close/"
    endpoint_dividends = "/api/ts/dividends/"
    endpoint_nav = "/api/ts/nav/"
    endpoint_macro = "/api/ts/macro/"
    # namespaces endpoints
    endpoint_namespaces = "/api/namespaces/"
    endpoint_assets_namespaces = "/api/assets_namespaces/"
    endpoint_macro_namespaces = "/api/macro_namespaces/"
    endpoint_no_dividends_namespaces = "/api/no_dividends_namespaces/"

    @classmethod
    def connect(
        cls,
        endpoint: str = endpoint_ror,
        symbol: str = settings.default_ticker,
        first_date: Union[str, pd.Timestamp, None] = None,
        last_date: Union[str, pd.Timestamp, None] = None,
        period: str = "d",
    ) -> str:
        session = requests.session()
        retry_strategy = Retry(total=3, backoff_factor=0.1, status_forcelist=[429, 500, 502, 503, 504])
        adapter = HTTPAdapter(max_retries=retry_strategy)
        request_url = cls.api_url + endpoint + symbol
        params = {"first_date": first_date, "last_date": last_date, "period": period}
        session.mount("https://", adapter)
        session.mount("http://", adapter)
        try:
            r = session.get(request_url, params=params, verify=False, timeout=cls.default_timeout)
            r.raise_for_status()
        except requests.exceptions.HTTPError as errh:
            if r.status_code == 404:
                raise requests.exceptions.HTTPError(f"{symbol} is not found in the database.", 404) from errh
            raise requests.exceptions.HTTPError(
                f"HTTP error fetching data for {symbol}:",
                r.status_code,
                r.reason,
                request_url,
            ) from errh
        return r.text

    @classmethod
    def get_ror(
        cls,
        symbol: str = settings.default_ticker,
        first_date: Union[str, pd.Timestamp, None] = None,
        last_date: Union[str, pd.Timestamp, None] = None,
        period: str = "m",
    ):
        return cls.connect(
            endpoint=cls.endpoint_ror,
            symbol=symbol,
            first_date=first_date,
            last_date=last_date,
            period=period,
        )

    @classmethod
    def get_adjusted_close(
        cls,
        symbol: str = settings.default_ticker,
        first_date: Union[str, pd.Timestamp, None] = None,
        last_date: Union[str, pd.Timestamp, None] = None,
        period: str = "m",
    ):
        return cls.connect(
            endpoint=cls.endpoint_adjusted_close,
            symbol=symbol,
            first_date=first_date,
            last_date=last_date,
            period=period,
        )

    @classmethod
    def get_close(
        cls,
        symbol: str = settings.default_ticker,
        first_date: Union[str, pd.Timestamp, None] = None,
        last_date: Union[str, pd.Timestamp, None] = None,
        period: str = "m",
    ):
        return cls.connect(
            endpoint=cls.endpoint_close,
            symbol=symbol,
            first_date=first_date,
            last_date=last_date,
            period=period,
        )

    @classmethod
    def get_dividends(
        cls,
        symbol: str = settings.default_ticker,
        first_date: Union[str, pd.Timestamp, None] = None,
        last_date: Union[str, pd.Timestamp, None] = None,
    ):
        return cls.connect(
            endpoint=cls.endpoint_dividends,
            symbol=symbol,
            first_date=first_date,
            last_date=last_date,
        )

    @classmethod
    def get_nav(
        cls,
        symbol: str = settings.default_ticker,
        first_date: Union[str, pd.Timestamp, None] = None,
        last_date: Union[str, pd.Timestamp, None] = None,
        period: str = "m",
    ):
        return cls.connect(
            endpoint=cls.endpoint_nav,
            symbol=symbol,
            first_date=first_date,
            last_date=last_date,
            period=period,
        )

    @classmethod
    def get_macro(
        cls,
        symbol: str = settings.default_ticker,
        first_date: Union[str, pd.Timestamp, None] = None,
        last_date: Union[str, pd.Timestamp, None] = None,
        period: str = "m",
    ):
        """
        Get macro time series (monthly).
        """
        return cls.connect(
            endpoint=cls.endpoint_macro,
            symbol=symbol,
            first_date=first_date,
            last_date=last_date,
            period=period,
        )

    @classmethod
    def get_namespaces(cls):
        return cls.connect(endpoint=cls.endpoint_namespaces, symbol="")

    @classmethod
    def get_symbols_in_namespace(cls, namespace: str = settings.default_namespace):
        return cls.connect(endpoint=cls.endpoint_namespaces, symbol=namespace)

    @classmethod
    def get_assets_namespaces(cls):
        return cls.connect(endpoint=cls.endpoint_assets_namespaces, symbol="")

    @classmethod
    def get_macro_namespaces(cls):
        return cls.connect(endpoint=cls.endpoint_macro_namespaces, symbol="")

    @classmethod
    def get_no_dividends_namespaces(cls):
        return cls.connect(endpoint=cls.endpoint_no_dividends_namespaces, symbol="")

    @classmethod
    def get_symbol_info(cls, symbol: str):
        return cls.connect(endpoint=cls.endpoint_symbol, symbol=symbol)

    @classmethod
    def search(cls, search_string: str):
        return cls.connect(endpoint=cls.endpoint_search, symbol=search_string)

    @classmethod
    def get_live_price(cls, symbol: str):
        return cls.connect(endpoint=cls.endpoint_live_price, symbol=symbol)

1	from typing import Union	3✔
2
3	import pandas as pd	3✔
4	import requests	3✔
5	from urllib3.util.retry import Retry	3✔
6	from requests.adapters import HTTPAdapter	3✔
7
8	from okama import settings	3✔
9
10
11	class API:	3✔
12	"""
13	Set of methods to get data from API.
14	"""
15
16	# TODO: introduce 'from' & 'to' for dates.
17
18	api_url = "http://api.okama.io:5000"	3✔
19	default_timeout = 10 # seconds	3✔
20
21	endpoint_ror = "/api/ts/ror/"	3✔
22	endpoint_symbol = "/api/symbol/"	3✔
23	endpoint_search = "/api/search/"	3✔
24	endpoint_live_price = "/api/live_price/"	3✔
25	endpoint_adjusted_close = "/api/ts/adjusted_close/"	3✔
26	endpoint_close = "/api/ts/close/"	3✔
27	endpoint_dividends = "/api/ts/dividends/"	3✔
28	endpoint_nav = "/api/ts/nav/"	3✔
29	endpoint_macro = "/api/ts/macro/"	3✔
30	# namespaces endpoints
31	endpoint_namespaces = "/api/namespaces/"	3✔
32	endpoint_assets_namespaces = "/api/assets_namespaces/"	3✔
33	endpoint_macro_namespaces = "/api/macro_namespaces/"	3✔
34	endpoint_no_dividends_namespaces = "/api/no_dividends_namespaces/"	3✔
35
36	@classmethod	3✔
37	def connect(	3✔
38	cls,
39	endpoint: str = endpoint_ror,
40	symbol: str = settings.default_ticker,
41	first_date: Union[str, pd.Timestamp, None] = None,
42	last_date: Union[str, pd.Timestamp, None] = None,
43	period: str = "d",
44	) -> str:
45	session = requests.session()	3✔
46	retry_strategy = Retry(total=3, backoff_factor=0.1, status_forcelist=[429, 500, 502, 503, 504])	3✔
47	adapter = HTTPAdapter(max_retries=retry_strategy)	3✔
48	request_url = cls.api_url + endpoint + symbol	3✔
49	params = {"first_date": first_date, "last_date": last_date, "period": period}	3✔
50	session.mount("https://", adapter)	3✔
51	session.mount("http://", adapter)	3✔
52	try:	3✔
53	r = session.get(request_url, params=params, verify=False, timeout=cls.default_timeout)	3✔
54	r.raise_for_status()	3✔
55	except requests.exceptions.HTTPError as errh:	×
56	if r.status_code == 404:	×
57	raise requests.exceptions.HTTPError(f"{symbol} is not found in the database.", 404) from errh	×
58	raise requests.exceptions.HTTPError(	×
59	f"HTTP error fetching data for {symbol}:",
60	r.status_code,
61	r.reason,
62	request_url,
63	) from errh
64	return r.text	3✔
65
66	@classmethod	3✔
67	def get_ror(	3✔
68	cls,
69	symbol: str = settings.default_ticker,
70	first_date: Union[str, pd.Timestamp, None] = None,
71	last_date: Union[str, pd.Timestamp, None] = None,
72	period: str = "m",
73	):
74	return cls.connect(	3✔
75	endpoint=cls.endpoint_ror,
76	symbol=symbol,
77	first_date=first_date,
78	last_date=last_date,
79	period=period,
80	)
81
82	@classmethod	3✔
83	def get_adjusted_close(	3✔
84	cls,
85	symbol: str = settings.default_ticker,
86	first_date: Union[str, pd.Timestamp, None] = None,
87	last_date: Union[str, pd.Timestamp, None] = None,
88	period: str = "m",
89	):
90	return cls.connect(	3✔
91	endpoint=cls.endpoint_adjusted_close,
92	symbol=symbol,
93	first_date=first_date,
94	last_date=last_date,
95	period=period,
96	)
97
98	@classmethod	3✔
99	def get_close(	3✔
100	cls,
101	symbol: str = settings.default_ticker,
102	first_date: Union[str, pd.Timestamp, None] = None,
103	last_date: Union[str, pd.Timestamp, None] = None,
104	period: str = "m",
105	):
106	return cls.connect(	3✔
107	endpoint=cls.endpoint_close,
108	symbol=symbol,
109	first_date=first_date,
110	last_date=last_date,
111	period=period,
112	)
113
114	@classmethod	3✔
115	def get_dividends(	3✔
116	cls,
117	symbol: str = settings.default_ticker,
118	first_date: Union[str, pd.Timestamp, None] = None,
119	last_date: Union[str, pd.Timestamp, None] = None,
120	):
121	return cls.connect(	3✔
122	endpoint=cls.endpoint_dividends,
123	symbol=symbol,
124	first_date=first_date,
125	last_date=last_date,
126	)
127
128	@classmethod	3✔
129	def get_nav(	3✔
130	cls,
131	symbol: str = settings.default_ticker,
132	first_date: Union[str, pd.Timestamp, None] = None,
133	last_date: Union[str, pd.Timestamp, None] = None,
134	period: str = "m",
135	):
136	return cls.connect(	3✔
137	endpoint=cls.endpoint_nav,
138	symbol=symbol,
139	first_date=first_date,
140	last_date=last_date,
141	period=period,
142	)
143
144	@classmethod	3✔
145	def get_macro(	3✔
146	cls,
147	symbol: str = settings.default_ticker,
148	first_date: Union[str, pd.Timestamp, None] = None,
149	last_date: Union[str, pd.Timestamp, None] = None,
150	period: str = "m",
151	):
152	"""
153	Get macro time series (monthly).
154	"""
155	return cls.connect(	3✔
156	endpoint=cls.endpoint_macro,
157	symbol=symbol,
158	first_date=first_date,
159	last_date=last_date,
160	period=period,
161	)
162
163	@classmethod	3✔
164	def get_namespaces(cls):	3✔
165	return cls.connect(endpoint=cls.endpoint_namespaces, symbol="")	×
166
167	@classmethod	3✔
168	def get_symbols_in_namespace(cls, namespace: str = settings.default_namespace):	3✔
169	return cls.connect(endpoint=cls.endpoint_namespaces, symbol=namespace)	3✔
170
171	@classmethod	3✔
172	def get_assets_namespaces(cls):	3✔
173	return cls.connect(endpoint=cls.endpoint_assets_namespaces, symbol="")	3✔
174
175	@classmethod	3✔
176	def get_macro_namespaces(cls):	3✔
177	return cls.connect(endpoint=cls.endpoint_macro_namespaces, symbol="")	3✔
178
179	@classmethod	3✔
180	def get_no_dividends_namespaces(cls):	3✔
181	return cls.connect(endpoint=cls.endpoint_no_dividends_namespaces, symbol="")	3✔
182
183	@classmethod	3✔
184	def get_symbol_info(cls, symbol: str):	3✔
185	return cls.connect(endpoint=cls.endpoint_symbol, symbol=symbol)	3✔
186
187	@classmethod	3✔
188	def search(cls, search_string: str):	3✔
189	return cls.connect(endpoint=cls.endpoint_search, symbol=search_string)	3✔
190
191	@classmethod	3✔
192	def get_live_price(cls, symbol: str):	3✔
193	return cls.connect(endpoint=cls.endpoint_live_price, symbol=symbol)	3✔

mbk-dev / okama / 290

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous