mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 04:52:07 +00:00
remove old sdk files (#4498)
This commit is contained in:
Henrique Joaquim
@@ -1,164 +0,0 @@
|
||||
from typing import Any, List, Optional
|
||||
|
||||
import openbb_terminal.config_terminal as cfg
|
||||
from openbb_terminal.core.library.metadata import Metadata
|
||||
from openbb_terminal.core.library.operation import Operation
|
||||
from openbb_terminal.core.library.trail_map import TrailMap
|
||||
|
||||
# pylint: disable=import-outside-toplevel
|
||||
|
||||
|
||||
class MetadataBuilder:
|
||||
@staticmethod
|
||||
def get_option_list(trail: str, trail_map: TrailMap) -> List[str]:
|
||||
option_list = []
|
||||
for key in trail_map.map_dict:
|
||||
if trail == "":
|
||||
option = key.split(".")[0]
|
||||
elif key.startswith(trail) and key[len(trail)] == ".":
|
||||
option = key[len(trail) + 1 :].split(".")[0]
|
||||
else:
|
||||
option = None
|
||||
|
||||
if option:
|
||||
option_list.append(option)
|
||||
|
||||
return list(set(option_list))
|
||||
|
||||
@classmethod
|
||||
def build_dir_list(cls, trail: str, trail_map: TrailMap) -> List[str]:
|
||||
option_list = cls.get_option_list(trail=trail, trail_map=trail_map)
|
||||
|
||||
option_list_full = []
|
||||
for option in option_list:
|
||||
option_list_full.append(option)
|
||||
|
||||
option_view_trail = f"{trail}.{option}_chart"
|
||||
if trail_map.get_view_function(trail=option_view_trail):
|
||||
option_list_full.append(f"{option}_chart")
|
||||
|
||||
return option_list_full
|
||||
|
||||
@staticmethod
|
||||
def build_docstring(trail: str, dir_list: List[str]) -> str:
|
||||
if trail == "":
|
||||
docstring = """This is the OpenBB Terminal SDK.
|
||||
Use the SDK to get data directly into your jupyter notebook or directly use it in your application.
|
||||
For more information see the official documentation at: https://openbb-finance.github.io/OpenBBTerminal/SDK/
|
||||
"""
|
||||
else:
|
||||
docstring = (
|
||||
trail.rsplit(".")[-1].upper()
|
||||
+ " Menu\n\nThe SDK commands of the the menu:"
|
||||
)
|
||||
for command in dir_list:
|
||||
docstring += f"\n\t<openbb>.{trail}.{command}"
|
||||
|
||||
return docstring
|
||||
|
||||
@classmethod
|
||||
def build(cls, trail: str, trail_map: TrailMap) -> Metadata:
|
||||
dir_list = cls.build_dir_list(trail=trail, trail_map=trail_map)
|
||||
docstring = cls.build_docstring(trail=trail, dir_list=dir_list)
|
||||
metadata = Metadata(
|
||||
dir_list=dir_list,
|
||||
docstring=docstring,
|
||||
)
|
||||
return metadata
|
||||
|
||||
|
||||
class Breadcrumb:
|
||||
__version__ = cfg.VERSION
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
metadata: Optional[Metadata] = None,
|
||||
trail: str = "",
|
||||
trail_map: Optional[TrailMap] = None,
|
||||
) -> None:
|
||||
"""
|
||||
Generates a 'trail' that allows accessing OpenBB Terminal SDK methods.
|
||||
|
||||
Example:
|
||||
openbb.forex.get_currency_list()
|
||||
Breadcrumb(trail="").Breadcrumb(trail="forex").Operation(trail="forex.get_currency_list")()
|
||||
|
||||
Args:
|
||||
metadata (Optional[Metadata], optional):
|
||||
Object to generate Breadcrumb's metadata (__dir__, __doc__).
|
||||
Defaults to None.
|
||||
trail (str, optional):
|
||||
Current trail of the Breadcrumb.
|
||||
Defaults to "".
|
||||
trail_map (Optional[TrailMap], optional):
|
||||
Mapping with all the trails available and matching models and views.
|
||||
Defaults to None.
|
||||
"""
|
||||
trail_map = trail_map or TrailMap()
|
||||
metadata = metadata or MetadataBuilder.build(trail=trail, trail_map=trail_map)
|
||||
|
||||
self._metadata = metadata
|
||||
self._trail_map = trail_map
|
||||
self._trail = trail
|
||||
|
||||
self.__doc__ = metadata.docstring
|
||||
if trail == "":
|
||||
BreadcrumbLogger()
|
||||
|
||||
def __dir__(self):
|
||||
return self._metadata.dir_list
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
trail = self._trail
|
||||
trail_map = self._trail_map
|
||||
|
||||
trail_next = name if trail == "" else f"{trail}.{name}"
|
||||
|
||||
if trail_map.get_model_function(
|
||||
trail=trail_next
|
||||
) or trail_map.get_view_function(trail=trail_next):
|
||||
next_crumb: Any = Operation(
|
||||
trail=trail_next,
|
||||
trail_map=trail_map,
|
||||
)
|
||||
elif name in self._metadata.dir_list:
|
||||
next_crumb = Breadcrumb(
|
||||
metadata=MetadataBuilder.build(trail=trail_next, trail_map=trail_map),
|
||||
trail=trail_next,
|
||||
trail_map=trail_map,
|
||||
)
|
||||
else:
|
||||
raise AttributeError(
|
||||
f"Module or method '{trail}' has no attribute '{name}'."
|
||||
)
|
||||
|
||||
return next_crumb
|
||||
|
||||
def about(self):
|
||||
import webbrowser
|
||||
|
||||
trail = self._trail
|
||||
url = "https://docs.openbb.co/sdk/reference/"
|
||||
url += "/".join(trail.split("."))
|
||||
webbrowser.open(url)
|
||||
|
||||
|
||||
# pylint: disable=R0903
|
||||
class BreadcrumbLogger:
|
||||
def __init__(self) -> None:
|
||||
self.__check_initialize_logging()
|
||||
|
||||
def __check_initialize_logging(self):
|
||||
if not cfg.LOGGING_SUPPRESS:
|
||||
self.__initialize_logging()
|
||||
|
||||
@staticmethod
|
||||
def __initialize_logging() -> None:
|
||||
from openbb_terminal.core.log.generation.settings_logger import ( # pylint: disable=C0415
|
||||
log_all_settings,
|
||||
)
|
||||
from openbb_terminal.loggers import setup_logging # pylint: disable=C0415
|
||||
|
||||
cfg.LOGGING_SUB_APP = "sdk"
|
||||
setup_logging()
|
||||
log_all_settings()
|
||||
@@ -1,19 +0,0 @@
|
||||
from typing import List
|
||||
|
||||
|
||||
class Metadata:
|
||||
@property
|
||||
def dir_list(self) -> List[str]:
|
||||
return self.__dir_list
|
||||
|
||||
@property
|
||||
def docstring(self) -> str:
|
||||
return self.__docstring
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
dir_list: List[str],
|
||||
docstring: str = "",
|
||||
) -> None:
|
||||
self.__dir_list = dir_list
|
||||
self.__docstring = docstring
|
||||
@@ -1,291 +0,0 @@
|
||||
import json
|
||||
from importlib import import_module
|
||||
from inspect import signature
|
||||
from logging import Logger, getLogger
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
import openbb_terminal.config_terminal as cfg
|
||||
from openbb_terminal.core.library.metadata import Metadata
|
||||
from openbb_terminal.core.library.trail_map import TrailMap
|
||||
|
||||
# pylint: disable=import-outside-toplevel
|
||||
|
||||
|
||||
class MetadataBuilder:
|
||||
def __init__(
|
||||
self,
|
||||
method: Callable,
|
||||
trail: str,
|
||||
trail_map: TrailMap,
|
||||
) -> None:
|
||||
self.__method = method
|
||||
self.__trail = trail
|
||||
self.__trail_map = trail_map
|
||||
|
||||
@staticmethod
|
||||
def build_docstring(
|
||||
method: Callable,
|
||||
trail: str,
|
||||
trail_map: TrailMap,
|
||||
) -> str:
|
||||
doc = ""
|
||||
|
||||
view_trail = f"{trail}_chart"
|
||||
if trail_map.get_view_function(view_trail):
|
||||
doc += f"Use '{view_trail}' to access the view.\n"
|
||||
|
||||
if method.__doc__:
|
||||
doc += method.__doc__
|
||||
|
||||
return doc
|
||||
|
||||
def build(self) -> Metadata:
|
||||
method = self.__method
|
||||
trail = self.__trail
|
||||
trail_map = self.__trail_map
|
||||
|
||||
dir_list: List[str] = []
|
||||
docstring = self.build_docstring(
|
||||
method=method,
|
||||
trail=trail,
|
||||
trail_map=trail_map,
|
||||
)
|
||||
metadata = Metadata(dir_list=dir_list, docstring=docstring)
|
||||
|
||||
return metadata
|
||||
|
||||
|
||||
class Operation:
|
||||
@staticmethod
|
||||
def __get_method(method_path: str) -> Callable:
|
||||
module_path, function_name = method_path.rsplit(".", 1)
|
||||
module = import_module(module_path)
|
||||
func = getattr(module, function_name)
|
||||
|
||||
return func
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
trail: str,
|
||||
trail_map: Optional[TrailMap] = None,
|
||||
metadata: Optional[Metadata] = None,
|
||||
) -> None:
|
||||
trail_map = trail_map or TrailMap()
|
||||
|
||||
method_path = trail_map.get_model_or_view(trail=trail)
|
||||
method = self.__get_method(method_path=method_path)
|
||||
metadata = (
|
||||
metadata
|
||||
or MetadataBuilder(method=method, trail=trail, trail_map=trail_map).build()
|
||||
)
|
||||
|
||||
self._trail = trail
|
||||
self._trail_map = trail_map
|
||||
self._method = method
|
||||
self.__doc__ = metadata.docstring
|
||||
self.__signature__ = signature(method)
|
||||
|
||||
def __call__(self, *args: Any, **kwargs: Any) -> Any:
|
||||
method = self._method
|
||||
|
||||
operation_logger = OperationLogger(
|
||||
trail=self._trail,
|
||||
method_chosen=method,
|
||||
args=args,
|
||||
kwargs=kwargs,
|
||||
)
|
||||
|
||||
operation_logger.log_before_call()
|
||||
|
||||
method_result = method(*args, **kwargs)
|
||||
|
||||
operation_logger.log_after_call(method_result=method_result)
|
||||
|
||||
return method_result
|
||||
|
||||
def about(self):
|
||||
import webbrowser
|
||||
|
||||
trail = self._trail
|
||||
url = "https://docs.openbb.co/sdk/reference/"
|
||||
url += "/".join(trail.split("."))
|
||||
webbrowser.open(url)
|
||||
|
||||
|
||||
class OperationLogger:
|
||||
last_method: Dict[Any, Any] = {}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
trail: str,
|
||||
method_chosen: Callable,
|
||||
args: Any,
|
||||
kwargs: Any,
|
||||
logger: Optional[Logger] = None,
|
||||
) -> None:
|
||||
self.__trail = trail
|
||||
self.__method_chosen = method_chosen
|
||||
self.__logger = logger or getLogger(self.__method_chosen.__module__)
|
||||
self.__args = args
|
||||
self.__kwargs = kwargs
|
||||
|
||||
def log_before_call(
|
||||
self,
|
||||
):
|
||||
if self.__check_logging_conditions():
|
||||
logger = self.__logger
|
||||
self.__log_start(logger=logger, method_chosen=self.__method_chosen)
|
||||
self.__log_method_info(
|
||||
logger=logger,
|
||||
trail=self.__trail,
|
||||
method_chosen=self.__method_chosen,
|
||||
args=self.__args,
|
||||
kwargs=self.__kwargs,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def __log_start(logger: Logger, method_chosen: Callable):
|
||||
logger.info(
|
||||
"START",
|
||||
extra={"func_name_override": method_chosen.__name__},
|
||||
)
|
||||
|
||||
def __log_method_info(
|
||||
self,
|
||||
logger: Logger,
|
||||
trail: str,
|
||||
method_chosen: Callable,
|
||||
args: Any,
|
||||
kwargs: Any,
|
||||
):
|
||||
merged_args = self.__merge_function_args(method_chosen, args, kwargs)
|
||||
merged_args = self.__remove_key_and_log_state(
|
||||
method_chosen.__module__, merged_args
|
||||
)
|
||||
|
||||
logging_info: Dict[str, Any] = {}
|
||||
logging_info["INPUT"] = {
|
||||
key: str(value)[:100] for key, value in merged_args.items()
|
||||
}
|
||||
logging_info["VIRTUAL_PATH"] = trail
|
||||
logging_info["CHART"] = "chart" in kwargs and kwargs["chart"] is True
|
||||
|
||||
logger.info(
|
||||
f"{json.dumps(logging_info)}",
|
||||
extra={"func_name_override": method_chosen.__name__},
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def __merge_function_args(func: Callable, args: tuple, kwargs: dict) -> dict:
|
||||
"""
|
||||
Merge user input args and kwargs with signature defaults into a dictionary.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
|
||||
func : Callable
|
||||
Function to get the args from
|
||||
args : tuple
|
||||
Positional args
|
||||
kwargs : dict
|
||||
Keyword args
|
||||
|
||||
Returns
|
||||
----------
|
||||
dict
|
||||
Merged user args and signature defaults
|
||||
"""
|
||||
import inspect # pylint: disable=C0415
|
||||
|
||||
sig = inspect.signature(func)
|
||||
sig_args = {
|
||||
param.name: param.default
|
||||
for param in sig.parameters.values()
|
||||
if param.default is not inspect.Parameter.empty
|
||||
}
|
||||
# merge args with sig_args
|
||||
sig_args.update(dict(zip(sig.parameters, args)))
|
||||
# add kwargs elements to sig_args
|
||||
sig_args.update(kwargs)
|
||||
return sig_args
|
||||
|
||||
@staticmethod
|
||||
def __remove_key_and_log_state(func_module: str, function_args: dict) -> dict:
|
||||
"""
|
||||
Remove API key from the function args and log state of keys.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
func_module : str
|
||||
Module of the function
|
||||
function_args : dict
|
||||
Function args
|
||||
|
||||
Returns
|
||||
----------
|
||||
dict
|
||||
Function args with API key removed
|
||||
"""
|
||||
|
||||
if func_module == "openbb_terminal.keys_model":
|
||||
from openbb_terminal.core.log.generation.settings_logger import ( # pylint: disable=import-outside-toplevel
|
||||
log_keys,
|
||||
)
|
||||
|
||||
# remove key if defined
|
||||
function_args.pop("key", None)
|
||||
log_keys()
|
||||
return function_args
|
||||
|
||||
def log_after_call(
|
||||
self,
|
||||
method_result: Any,
|
||||
):
|
||||
if self.__check_logging_conditions():
|
||||
logger = self.__logger
|
||||
self.__log_exception_if_any(
|
||||
logger=logger,
|
||||
method_result=method_result,
|
||||
method_chosen=self.__method_chosen,
|
||||
)
|
||||
self.__log_end(
|
||||
logger=logger,
|
||||
method_chosen=self.__method_chosen,
|
||||
)
|
||||
OperationLogger.last_method = {
|
||||
f"{self.__method_chosen.__module__}.{self.__method_chosen.__name__}": {
|
||||
"args": str(self.__args)[:100],
|
||||
"kwargs": str(self.__kwargs)[:100],
|
||||
}
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def __log_exception_if_any(
|
||||
logger: Logger,
|
||||
method_chosen: Callable,
|
||||
method_result: Any,
|
||||
):
|
||||
if isinstance(method_result, Exception):
|
||||
logger.exception(
|
||||
f"Exception: {method_result}",
|
||||
extra={"func_name_override": method_chosen.__name__},
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def __log_end(logger: Logger, method_chosen: Callable):
|
||||
logger.info(
|
||||
"END",
|
||||
extra={"func_name_override": method_chosen.__name__},
|
||||
)
|
||||
|
||||
def __check_logging_conditions(self) -> bool:
|
||||
return not cfg.LOGGING_SUPPRESS and not self.__check_last_method()
|
||||
|
||||
def __check_last_method(self) -> bool:
|
||||
current_method = {
|
||||
f"{self.__method_chosen.__module__}.{self.__method_chosen.__name__}": {
|
||||
"args": str(self.__args)[:100],
|
||||
"kwargs": str(self.__kwargs)[:100],
|
||||
}
|
||||
}
|
||||
return OperationLogger.last_method == current_method
|
||||
@@ -1,141 +0,0 @@
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional
|
||||
|
||||
import dotenv
|
||||
import pandas as pd
|
||||
|
||||
from openbb_terminal.base_helpers import load_env_vars, strtobool
|
||||
from openbb_terminal.core.config.paths import MISCELLANEOUS_DIRECTORY, SETTINGS_ENV_FILE
|
||||
from openbb_terminal.rich_config import console
|
||||
|
||||
DISABLE_FORECASTING_WARNING = load_env_vars(
|
||||
"OPENBB_DISABLE_FORECASTING_WARNING", strtobool, False
|
||||
)
|
||||
try:
|
||||
import darts # pylint: disable=W0611 # noqa: F401
|
||||
from darts import utils # pylint: disable=W0611 # noqa: F401
|
||||
|
||||
FORECASTING_TOOLKIT_ENABLED = True
|
||||
except ModuleNotFoundError:
|
||||
FORECASTING_TOOLKIT_ENABLED = False
|
||||
if not DISABLE_FORECASTING_WARNING:
|
||||
dotenv.set_key(
|
||||
str(SETTINGS_ENV_FILE), "OPENBB_DISABLE_FORECASTING_WARNING", "True"
|
||||
)
|
||||
console.print(
|
||||
"[yellow]"
|
||||
"Forecasting Toolkit is disabled. "
|
||||
"To use the Forecasting features please install the toolkit following the "
|
||||
"instructions here: https://docs.openbb.co/sdk/quickstart/installation/"
|
||||
"\n"
|
||||
"[/yellow]"
|
||||
)
|
||||
|
||||
DISABLE_OPTIMIZATION_WARNING = load_env_vars(
|
||||
"OPENBB_DISABLE_OPTIMIZATION_WARNING", strtobool, False
|
||||
)
|
||||
try:
|
||||
# import riskfolio # pylint: disable=W0611 # noqa: F401
|
||||
|
||||
OPTIMIZATION_TOOLKIT_ENABLED = True
|
||||
except ModuleNotFoundError:
|
||||
OPTIMIZATION_TOOLKIT_ENABLED = False
|
||||
if not DISABLE_OPTIMIZATION_WARNING:
|
||||
dotenv.set_key(
|
||||
str(SETTINGS_ENV_FILE), "OPENBB_DISABLE_OPTIMIZATION_WARNING", "True"
|
||||
)
|
||||
console.print(
|
||||
"[yellow]"
|
||||
"Portfolio Optimization Toolkit is disabled. "
|
||||
"To use the Optimization features please install the toolkit following the "
|
||||
"instructions here: https://docs.openbb.co/sdk/quickstart/installation/"
|
||||
"\n"
|
||||
"[/yellow]"
|
||||
)
|
||||
|
||||
|
||||
class TrailMap:
|
||||
MAP_PATH = MISCELLANEOUS_DIRECTORY / "library" / "trail_map.csv"
|
||||
MAP_FORECASTING_PATH = (
|
||||
MISCELLANEOUS_DIRECTORY / "library" / "trail_map_forecasting.csv"
|
||||
)
|
||||
MAP_OPTIMIZATION_PATH = (
|
||||
MISCELLANEOUS_DIRECTORY / "library" / "trail_map_optimization.csv"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def load_csv(cls, path: Optional[Path] = None) -> Dict[str, Dict[str, str]]:
|
||||
path = path or cls.MAP_PATH
|
||||
df = pd.read_csv(path, keep_default_na=False)
|
||||
df.set_index("trail", inplace=True)
|
||||
return df.to_dict(orient="index")
|
||||
|
||||
@classmethod
|
||||
def save_csv(
|
||||
cls, map_dict: Dict[str, Dict[str, str]], path: Optional[Path] = None
|
||||
) -> None:
|
||||
path = path or cls.MAP_PATH
|
||||
df = pd.DataFrame.from_dict(data=map_dict, orient="index")
|
||||
df.index.name = "trail"
|
||||
df.to_csv(path_or_buf=path)
|
||||
|
||||
@property
|
||||
def map_dict(self) -> Dict[str, Dict[str, str]]:
|
||||
return self.__map_dict
|
||||
|
||||
def get_view_function(self, trail: str) -> Optional[str]:
|
||||
"""Retrieves the view function from the mapping.
|
||||
|
||||
Views ends with "_chart".
|
||||
|
||||
Args:
|
||||
trail (str): Trail like "futures.curves_chart"
|
||||
|
||||
Returns:
|
||||
Optional[str]: View function import path.
|
||||
"""
|
||||
|
||||
map_dict = self.__map_dict
|
||||
trail = trail[: -len("_chart")]
|
||||
|
||||
view = None
|
||||
if trail in map_dict and "view" in map_dict[trail] and map_dict[trail]["view"]:
|
||||
view = map_dict[trail]["view"]
|
||||
|
||||
return view
|
||||
|
||||
def get_model_function(self, trail: str) -> Optional[str]:
|
||||
map_dict = self.map_dict
|
||||
|
||||
model = None
|
||||
if trail in map_dict and "view" in map_dict[trail] and map_dict[trail]["model"]:
|
||||
model = map_dict[trail]["model"]
|
||||
|
||||
return model
|
||||
|
||||
def get_model_or_view(self, trail: str) -> str:
|
||||
model_path = self.get_model_function(trail=trail)
|
||||
view_path = self.get_view_function(trail=trail)
|
||||
|
||||
if model_path:
|
||||
method_path = model_path
|
||||
elif view_path:
|
||||
method_path = view_path
|
||||
else:
|
||||
raise AttributeError(f"Unknown method : {trail}")
|
||||
|
||||
return method_path
|
||||
|
||||
def load(self):
|
||||
map_dict = self.load_csv(path=self.MAP_PATH)
|
||||
if FORECASTING_TOOLKIT_ENABLED:
|
||||
map_forecasting_dict = self.load_csv(path=self.MAP_FORECASTING_PATH)
|
||||
map_dict = {**map_dict, **map_forecasting_dict}
|
||||
if OPTIMIZATION_TOOLKIT_ENABLED:
|
||||
map_optimization_dict = self.load_csv(path=self.MAP_OPTIMIZATION_PATH)
|
||||
map_dict = {**map_dict, **map_optimization_dict}
|
||||
|
||||
self.__map_dict = map_dict
|
||||
|
||||
def __init__(self):
|
||||
self.load()
|
||||
Reference in New Issue
Block a user