Upload 355 files

openbb_platform/core/README.md

@@ -0,0 +1,57 @@
+# OpenBB Platform - Core
+
+## Overview
+
+The Core extension serves as the foundational component of the OpenBB Platform. It encapsulates essential functionalities and serves as an infrastructural base for other extensions. This extension is vital for maintaining the integrity and standardization of the platform.
+
+## Key Features
+
+- **Standardized Data Model** (`Data` Class): A flexible and dynamic Pydantic model capable of handling various data structures.
+- **Standardized Query Params** (`QueryParams` Class): A Pydantic model for handling querying to different providers.
+- **Dynamic Field Support**: Enables handling of undefined fields, providing versatility in data processing.
+- **Robust Data Validation**: Utilizes Pydantic's validation features to ensure data integrity.
+- **API Routing Mechanism** (`Router` Class): Simplifies the process of defining API routes and endpoints - out of the box Python and Web endpoints.
+
+## Getting Started
+
+### Prerequisites
+
+- Python 3.9 or higher.
+- Familiarity with FastAPI and Pydantic.
+
+### Installation
+
+Installing through pip:
+
+```bash
+pip install openbb-core
+```
+
+> Note that, the openbb-core is an infrastructural component of the OpenBB Platform. It is not intended to be used as a standalone package.
+
+### Usage
+
+The Core extension is used as the basis for building and integrating new data sources, providers, and extensions into the OpenBB Platform. It provides the necessary classes and structures for standardizing and handling data.
+
+### Contributing
+
+We welcome contributions! If you're looking to contribute, please:
+
+- Follow the existing coding standards and conventions.
+- Write clear, documented code.
+- Ensure your code does not negatively impact performance.
+- Test your contributions thoroughly.
+
+Please refer to our [Contributing Guidelines](https://docs.openbb.co/platform/developer_guide/contributing).
+
+### Collaboration
+
+Engage with the development team and the community. Be open to feedback and collaborative discussions.
+
+### Support
+
+For support, questions, or more information, please visit [OpenBB Platform Documentation](https://docs.openbb.co/platform).
+
+### License
+
+This project is licensed under the MIT License - see the [LICENSE.md](https://github.com/OpenBB-finance/OpenBB/blob/main/LICENSE) file for details.

openbb_platform/core/__init__.py

@@ -0,0 +1 @@
+"""OpenBB Core Module."""

openbb_platform/core/integration/test_obbject.py

@@ -0,0 +1,96 @@
+"""Test the OBBject."""
+
+import contextlib
+import sys
+
+import pytest
+
+with contextlib.suppress(ImportError):
+    import polars as pl
+
+with contextlib.suppress(ImportError):
+    import pandas as pd
+
+with contextlib.suppress(ImportError):
+    import numpy as np
+
+with contextlib.suppress(ImportError):
+    from openbb_charting.core.openbb_figure import OpenBBFigure
+
+
+# pylint: disable=inconsistent-return-statements
+@pytest.fixture(scope="session")
+def obb(pytestconfig):
+    """Fixture to setup obb."""
+
+    if pytestconfig.getoption("markexpr") != "not integration":
+        import openbb  # pylint: disable=import-outside-toplevel
+
+        return openbb.obb
+
+
+# pylint: disable=redefined-outer-name
+
+
+@pytest.mark.skipif("pandas" not in sys.modules, reason="pandas not installed")
+@pytest.mark.integration
+def test_to_dataframe(obb):
+    """Test obbject to dataframe."""
+
+    stocks_df = obb.equity.price.historical("AAPL", provider="fmp").to_dataframe()
+    assert isinstance(stocks_df, pd.DataFrame)
+
+
+@pytest.mark.skipif(
+    "polars" not in sys.modules or "polars-lts-cpu" not in sys.modules,
+    reason="polars not installed",
+)
+@pytest.mark.integration
+def test_to_polars(obb):
+    """Test obbject to polars."""
+
+    crypto_pl = obb.crypto.price.historical("BTC-USD", provider="fmp").to_polars()
+    assert isinstance(crypto_pl, pl.DataFrame)
+
+
+@pytest.mark.skipif("numpy" not in sys.modules, reason="numpy not installed")
+@pytest.mark.integration
+def test_to_numpy(obb):
+    """Test obbject to numpy array."""
+
+    cpi_np = obb.economy.cpi(
+        country=["portugal", "spain", "switzerland"], frequency="annual"
+    ).to_numpy()
+    assert isinstance(cpi_np, np.ndarray)
+
+
+@pytest.mark.integration
+def test_to_dict(obb):
+    """Test obbject to dict."""
+
+    fed_dict = obb.fixedincome.rate.ameribor(start_date="2020-01-01").to_dict()
+    assert isinstance(fed_dict, dict)
+
+
+@pytest.mark.skipif(
+    "openbb_charting" not in sys.modules, reason="openbb_charting not installed"
+)
+@pytest.mark.integration
+def test_to_chart(obb):
+    """Test obbject to chart."""
+
+    res = obb.equity.price.historical("AAPL", provider="fmp")
+    res.charting.to_chart(render=False)
+    assert isinstance(res.chart.fig, OpenBBFigure)
+
+
+@pytest.mark.skipif(
+    "openbb_charting" not in sys.modules, reason="openbb_charting not installed"
+)
+@pytest.mark.integration
+def test_show(obb):
+    """Test obbject to chart."""
+
+    stocks_data = obb.equity.price.historical("AAPL", provider="fmp", chart=True)
+    assert isinstance(stocks_data.chart.fig, OpenBBFigure)
+    assert stocks_data.chart.fig.show() is None

openbb_platform/core/openbb_core/__init__.py

@@ -0,0 +1 @@
+"""OpenBB Core."""

openbb_platform/core/openbb_core/api/app_loader.py

@@ -0,0 +1,45 @@
+"""App loader module."""
+
+from typing import List, Optional
+
+from fastapi import APIRouter, FastAPI
+from fastapi.exceptions import ResponseValidationError
+from openbb_core.api.exception_handlers import ExceptionHandlers
+from openbb_core.app.model.abstract.error import OpenBBError
+from openbb_core.app.router import RouterLoader
+from openbb_core.provider.utils.errors import EmptyDataError, UnauthorizedError
+from pydantic import ValidationError
+
+
+class AppLoader:
+    """App loader."""
+
+    @staticmethod
+    def add_routers(app: FastAPI, routers: List[Optional[APIRouter]], prefix: str):
+        """Add routers."""
+        for router in routers:
+            if router:
+                app.include_router(router=router, prefix=prefix)
+
+    @staticmethod
+    def add_openapi_tags(app: FastAPI):
+        """Add openapi tags."""
+        main_router = RouterLoader.from_extensions()
+        # Add tag data for each router in the main router
+        app.openapi_tags = [
+            {
+                "name": r,
+                "description": main_router.get_attr(r, "description"),
+            }
+            for r in main_router.routers
+        ]
+
+    @staticmethod
+    def add_exception_handlers(app: FastAPI):
+        """Add exception handlers."""
+        app.exception_handlers[Exception] = ExceptionHandlers.exception
+        app.exception_handlers[ValidationError] = ExceptionHandlers.validation
+        app.exception_handlers[ResponseValidationError] = ExceptionHandlers.validation
+        app.exception_handlers[OpenBBError] = ExceptionHandlers.openbb
+        app.exception_handlers[EmptyDataError] = ExceptionHandlers.empty_data
+        app.exception_handlers[UnauthorizedError] = ExceptionHandlers.unauthorized

openbb_platform/core/openbb_core/api/auth/user.py

@@ -0,0 +1,57 @@
+"""User authentication."""
+
+import secrets
+from typing import Optional
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+from openbb_core.app.model.user_settings import UserSettings
+from openbb_core.app.service.user_service import UserService
+from openbb_core.env import Env
+from typing_extensions import Annotated
+
+security = HTTPBasic() if Env().API_AUTH else lambda: None
+
+
+async def authenticate_user(
+    credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)],
+):
+    """Authenticate the user."""
+    if credentials:
+        username = Env().API_USERNAME
+        password = Env().API_PASSWORD
+
+        is_correct_username = False
+        is_correct_password = False
+
+        if username is not None and password is not None:
+            current_username_bytes = credentials.username.encode("utf8")
+            correct_username_bytes = username.encode("utf8")
+            is_correct_username = secrets.compare_digest(
+                current_username_bytes, correct_username_bytes
+            )
+            current_password_bytes = credentials.password.encode("utf8")
+            correct_password_bytes = password.encode("utf8")
+            is_correct_password = secrets.compare_digest(
+                current_password_bytes, correct_password_bytes
+            )
+
+        if not (is_correct_username and is_correct_password):
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Incorrect email or password",
+                headers={"WWW-Authenticate": "Basic"},
+            )
+
+
+async def get_user_service() -> UserService:
+    """Get user service."""
+    return UserService()
+
+
+async def get_user_settings(
+    _: Annotated[None, Depends(authenticate_user)],
+    user_service: Annotated[UserService, Depends(get_user_service)],
+) -> UserSettings:
+    """Get user settings."""
+    return user_service.read_from_file()

openbb_platform/core/openbb_core/api/dependency/__init__.py

@@ -0,0 +1 @@
+"""OpenBB Core API Dependency."""

openbb_platform/core/openbb_core/api/dependency/coverage.py

@@ -0,0 +1,21 @@
+"""Coverage dependency."""
+
+from fastapi import Depends
+from openbb_core.app.provider_interface import ProviderInterface
+from openbb_core.app.router import CommandMap
+from openbb_core.app.service.auth_service import AuthService
+from typing_extensions import Annotated
+
+
+async def get_command_map(
+    _: Annotated[None, Depends(AuthService().auth_hook)]
+) -> CommandMap:
+    """Get command map."""
+    return CommandMap()
+
+
+async def get_provider_interface(
+    _: Annotated[None, Depends(AuthService().auth_hook)]
+) -> ProviderInterface:
+    """Get provider interface."""
+    return ProviderInterface()

openbb_platform/core/openbb_core/api/dependency/system.py

@@ -0,0 +1,20 @@
+"""System dependency."""
+
+from fastapi import Depends
+from openbb_core.app.model.system_settings import SystemSettings
+from openbb_core.app.service.auth_service import AuthService
+from openbb_core.app.service.system_service import SystemService
+from typing_extensions import Annotated
+
+
+async def get_system_service() -> SystemService:
+    """Get system service."""
+    return SystemService()
+
+
+async def get_system_settings(
+    _: Annotated[None, Depends(AuthService().auth_hook)],
+    system_service: Annotated[SystemService, Depends(get_system_service)],
+) -> SystemSettings:
+    """Get system settings."""
+    return system_service.system_settings

openbb_platform/core/openbb_core/api/exception_handlers.py

@@ -0,0 +1,134 @@
+"""Exception handlers module."""
+
+# pylint: disable=unused-argument
+
+import logging
+from collections.abc import Iterable
+from typing import Any, Union
+
+from fastapi import Request
+from fastapi.exceptions import ResponseValidationError
+from fastapi.responses import JSONResponse, Response
+from openbb_core.app.model.abstract.error import OpenBBError
+from openbb_core.env import Env
+from openbb_core.provider.utils.errors import EmptyDataError, UnauthorizedError
+from pydantic import ValidationError
+
+logger = logging.getLogger("uvicorn.error")
+
+
+class ExceptionHandlers:
+    """Exception handlers."""
+
+    @staticmethod
+    async def _handle(exception: Exception, status_code: int, detail: Any):
+        """Exception handler."""
+        if Env().DEBUG_MODE:
+            raise exception
+        logger.error(exception)
+        return JSONResponse(
+            status_code=status_code,
+            content={
+                "detail": detail,
+            },
+        )
+
+    @staticmethod
+    async def exception(_: Request, error: Exception) -> JSONResponse:
+        """Exception handler for Base Exception."""
+        errors = error.errors if hasattr(error, "errors") else error
+
+        if errors:
+            if isinstance(errors, ValueError):
+                return await ExceptionHandlers._handle(
+                    exception=errors,
+                    status_code=422,
+                    detail=errors.args,
+                )
+            # Required parameters are missing and is not handled by ValidationError.
+            if isinstance(errors, Iterable):
+                for err in errors:
+                    if err.get("type") == "missing":
+                        return await ExceptionHandlers._handle(
+                            exception=error,
+                            status_code=422,
+                            detail={**err},
+                        )
+        return await ExceptionHandlers._handle(
+            exception=error,
+            status_code=500,
+            detail=f"Unexpected Error -> {error.__class__.__name__} -> {error}",
+        )
+
+    @staticmethod
+    async def validation(
+        request: Request, error: Union[ValidationError, ResponseValidationError]
+    ):
+        """Exception handler for ValidationError."""
+        # Some validation is performed at Fetcher level.
+        # So we check if the validation error comes from a QueryParams class.
+        # And that it is in the request query params.
+        # If yes, we update the error location with query.
+        # If not, we handle it as a base Exception error.
+        query_params = dict(request.query_params)
+        if isinstance(error, ResponseValidationError):
+            detail = [
+                {
+                    **{k: v for k, v in err.items() if k != "ctx"},
+                    "loc": ("query",) + err.get("loc", ()),
+                }
+                for err in error.errors()
+            ]
+            return await ExceptionHandlers._handle(
+                exception=error,
+                status_code=422,
+                detail=detail,
+            )
+        try:
+            errors = (
+                error.errors(include_url=False)
+                if hasattr(error, "errors")
+                else error.errors
+            )
+        except Exception:
+            errors = error.errors if hasattr(error, "errors") else error
+        all_in_query = all(
+            loc in query_params for err in errors for loc in err.get("loc", ())
+        )
+        if "QueryParams" in error.title and all_in_query:
+            detail = [
+                {
+                    **{k: v for k, v in err.items() if k != "ctx"},
+                    "loc": ("query",) + err.get("loc", ()),
+                }
+                for err in errors
+            ]
+            return await ExceptionHandlers._handle(
+                exception=error,
+                status_code=422,
+                detail=detail,
+            )
+        return await ExceptionHandlers.exception(request, error)
+
+    @staticmethod
+    async def openbb(_: Request, error: OpenBBError):
+        """Exception handler for OpenBBError."""
+        return await ExceptionHandlers._handle(
+            exception=error,
+            status_code=400,
+            detail=str(error.original),
+        )
+
+    @staticmethod
+    async def empty_data(_: Request, error: EmptyDataError):
+        """Exception handler for EmptyDataError."""
+        return Response(status_code=204)
+
+    @staticmethod
+    async def unauthorized(_: Request, error: UnauthorizedError):
+        """Exception handler for OpenBBError."""
+        return await ExceptionHandlers._handle(
+            exception=error,
+            status_code=502,
+            detail=str(error.original),
+        )

openbb_platform/core/openbb_core/api/rest_api.py

@@ -0,0 +1,105 @@
+"""REST API for the OpenBB Platform."""
+
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from openbb_core.api.app_loader import AppLoader
+from openbb_core.api.router.commands import router as router_commands
+from openbb_core.api.router.coverage import router as router_coverage
+from openbb_core.api.router.system import router as router_system
+from openbb_core.app.service.auth_service import AuthService
+from openbb_core.app.service.system_service import SystemService
+from openbb_core.env import Env
+
+logger = logging.getLogger("uvicorn.error")
+
+system = SystemService().system_settings
+
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    """Startup event."""
+    auth = "ENABLED" if Env().API_AUTH else "DISABLED"
+    banner = rf"""
+
+                   ███╗
+  █████████████████╔══█████████████████╗       OpenBB Platform v{system.version}
+  ███╔══════════███║  ███╔══════════███║
+  █████████████████║  █████████████████║       Authentication: {auth}
+  ╚═════════════███║  ███╔═════════════╝
+     ██████████████║  ██████████████╗
+     ███╔═══════███║  ███╔═══════███║
+     ██████████████║  ██████████████║
+     ╚═════════════╝  ╚═════════════╝
+Investment research for everyone, anywhere.
+
+    https://my.openbb.co/app/platform
+
+"""
+    logger.info(banner)
+    yield
+
+
+app = FastAPI(
+    title=system.api_settings.title,
+    description=system.api_settings.description,
+    version=system.api_settings.version,
+    terms_of_service=system.api_settings.terms_of_service,
+    contact={
+        "name": system.api_settings.contact_name,
+        "url": system.api_settings.contact_url,
+        "email": system.api_settings.contact_email,
+    },
+    license_info={
+        "name": system.api_settings.license_name,
+        "url": system.api_settings.license_url,
+    },
+    servers=[
+        {
+            "url": s.url,
+            "description": s.description,
+        }
+        for s in system.api_settings.servers
+    ],
+    lifespan=lifespan,
+)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=system.api_settings.cors.allow_origins,
+    allow_methods=system.api_settings.cors.allow_methods,
+    allow_headers=system.api_settings.cors.allow_headers,
+)
+AppLoader.add_routers(
+    app=app,
+    routers=(
+        [AuthService().router, router_system, router_coverage, router_commands]
+        if Env().DEV_MODE
+        else (
+            [router_commands, router_coverage]
+            if hasattr(router_commands, "routes") and router_commands.routes
+            else [router_commands]
+        )
+    ),
+    prefix=system.api_settings.prefix,
+)
+AppLoader.add_openapi_tags(app)
+AppLoader.add_exception_handlers(app)
+
+
+if __name__ == "__main__":
+    # pylint: disable=import-outside-toplevel
+    import uvicorn
+
+    # This initializes the OpenBB environment variables so they can be read before uvicorn is run.
+    Env()
+    uvicorn_kwargs = system.python_settings.model_dump().get("uvicorn", {})
+    uvicorn_reload = uvicorn_kwargs.pop("reload", None)
+
+    if uvicorn_reload is None or uvicorn_reload:
+        uvicorn_kwargs["reload"] = True
+
+    uvicorn_app = uvicorn_kwargs.pop("app", "openbb_core.api.rest_api:app")
+
+    uvicorn.run(uvicorn_app, **uvicorn_kwargs)

openbb_platform/core/openbb_core/api/router/__init__.py

@@ -0,0 +1 @@
+"""OpenBB Core API Router."""

openbb_platform/core/openbb_core/api/router/commands.py

@@ -0,0 +1,266 @@
+"""Commands: generates the command map."""
+
+import inspect
+from functools import partial, wraps
+from inspect import Parameter, Signature, signature
+from typing import Any, Callable, Dict, Optional, Tuple, TypeVar
+
+from fastapi import APIRouter, Depends, Header
+from fastapi.routing import APIRoute
+from openbb_core.app.command_runner import CommandRunner
+from openbb_core.app.model.command_context import CommandContext
+from openbb_core.app.model.obbject import OBBject
+from openbb_core.app.model.user_settings import UserSettings
+from openbb_core.app.router import RouterLoader
+from openbb_core.app.service.auth_service import AuthService
+from openbb_core.app.service.system_service import SystemService
+from openbb_core.app.service.user_service import UserService
+from openbb_core.env import Env
+from pydantic import BaseModel
+from typing_extensions import Annotated, ParamSpec
+
+try:
+    from openbb_charting import Charting
+
+    CHARTING_INSTALLED = True
+except ImportError:
+    CHARTING_INSTALLED = False
+
+T = TypeVar("T")
+P = ParamSpec("P")
+
+router = APIRouter(prefix="")
+
+
+def build_new_annotation_map(sig: Signature) -> Dict[str, Any]:
+    """Build new annotation map."""
+    annotation_map = {}
+    parameter_list = sig.parameters.values()
+
+    for parameter in parameter_list:
+        annotation_map[parameter.name] = parameter.annotation
+
+    annotation_map["return"] = sig.return_annotation
+
+    return annotation_map
+
+
+def build_new_signature(path: str, func: Callable) -> Signature:
+    """Build new function signature."""
+    sig = signature(func)
+    parameter_list = sig.parameters.values()
+    return_annotation = sig.return_annotation
+    new_parameter_list = []
+    var_kw_pos = len(parameter_list)
+    for pos, parameter in enumerate(parameter_list):
+        if parameter.name == "cc" and parameter.annotation == CommandContext:
+            continue
+
+        if parameter.kind == Parameter.VAR_KEYWORD:
+            # We track VAR_KEYWORD parameter to insert the any additional
+            # parameters we need to add before it and avoid a SyntaxError
+            var_kw_pos = pos
+
+        new_parameter_list.append(
+            Parameter(
+                parameter.name,
+                kind=parameter.kind,
+                default=parameter.default,
+                annotation=parameter.annotation,
+            )
+        )
+
+    if CHARTING_INSTALLED and path.replace("/", "_")[1:] in Charting.functions():
+        new_parameter_list.insert(
+            var_kw_pos,
+            Parameter(
+                "chart",
+                kind=Parameter.POSITIONAL_OR_KEYWORD,
+                default=False,
+                annotation=bool,
+            ),
+        )
+        var_kw_pos += 1
+
+    if custom_headers := SystemService().system_settings.api_settings.custom_headers:
+        for name, default in custom_headers.items():
+            new_parameter_list.insert(
+                var_kw_pos,
+                Parameter(
+                    name.replace("-", "_"),
+                    kind=Parameter.POSITIONAL_OR_KEYWORD,
+                    default=default,
+                    annotation=Annotated[
+                        Optional[str], Header(include_in_schema=False)
+                    ],
+                ),
+            )
+            var_kw_pos += 1
+
+    if Env().API_AUTH:
+        new_parameter_list.insert(
+            var_kw_pos,
+            Parameter(
+                "__authenticated_user_settings",
+                kind=Parameter.POSITIONAL_OR_KEYWORD,
+                default=UserSettings(),
+                annotation=Annotated[
+                    UserSettings, Depends(AuthService().user_settings_hook)
+                ],
+            ),
+        )
+        var_kw_pos += 1
+
+    return Signature(
+        parameters=new_parameter_list,
+        return_annotation=return_annotation,
+    )
+
+
+def validate_output(c_out: OBBject) -> OBBject:
+    """
+    Validate OBBject object.
+
+    Checks against the OBBject schema and removes fields that contain the
+    `exclude_from_api` extra `pydantic.Field` kwarg.
+    Note that the modification to the `OBBject` object is done in-place.
+
+    Parameters
+    ----------
+    c_out : OBBject
+        OBBject object to validate.
+
+    Returns
+    -------
+    Dict
+        Serialized OBBject.
+    """
+
+    def is_model(type_):
+        return inspect.isclass(type_) and issubclass(type_, BaseModel)
+
+    def exclude_fields_from_api(key: str, value: Any):
+        type_ = type(value)
+        field = c_out.model_fields.get(key, None)
+        json_schema_extra = field.json_schema_extra if field else None
+
+        # case where 1st layer field needs to be excluded
+        if (
+            json_schema_extra
+            and isinstance(json_schema_extra, dict)
+            and json_schema_extra.get("exclude_from_api", None)
+        ):
+            delattr(c_out, key)
+
+        # if it's a model with nested fields
+        elif is_model(type_):
+            for field_name, field in type_.model_fields.items():
+                extra = getattr(field, "json_schema_extra", None)
+                if (
+                    extra
+                    and isinstance(extra, dict)
+                    and extra.get("exclude_from_api", None)
+                ):
+                    delattr(value, field_name)
+
+                # if it's a yet a nested model we need to go deeper in the recursion
+                elif is_model(getattr(field, "annotation", None)):
+                    exclude_fields_from_api(field_name, getattr(value, field_name))
+
+    # Let a non-OBBject object pass through without validation
+    if not isinstance(c_out, OBBject):
+        return c_out
+
+    for k, v in c_out.model_copy():
+        exclude_fields_from_api(k, v)
+
+    return c_out
+
+
+def build_api_wrapper(
+    command_runner: CommandRunner,
+    route: APIRoute,
+) -> Callable:
+    """Build API wrapper for a command."""
+    func: Callable = route.endpoint  # type: ignore
+    path: str = route.path  # type: ignore
+
+    no_validate = (
+        openapi_extra.get("no_validate")
+        if (openapi_extra := getattr(route, "openapi_extra", None))
+        else None
+    )
+    new_signature = build_new_signature(path=path, func=func)
+    new_annotations_map = build_new_annotation_map(sig=new_signature)
+    func.__signature__ = new_signature  # type: ignore
+    func.__annotations__ = new_annotations_map
+
+    if no_validate is True:
+        route.response_model = None
+
+    @wraps(wrapped=func)
+    async def wrapper(*args: Tuple[Any], **kwargs: Dict[str, Any]) -> OBBject:
+        user_settings: UserSettings = UserSettings.model_validate(
+            kwargs.pop(
+                "__authenticated_user_settings",
+                UserService.read_from_file(),
+            )
+        )
+        p = path.strip("/").replace("/", ".")
+        defaults = (
+            getattr(user_settings.defaults, "__dict__", {})
+            .get("commands", {})
+            .get(p, {})
+        )
+
+        if defaults:
+            _provider = defaults.pop("provider", None)
+            standard_params = getattr(
+                kwargs.pop("standard_params", None), "__dict__", {}
+            )
+            extra_params = getattr(kwargs.pop("extra_params", None), "__dict__", {})
+
+            if "chart" in defaults:
+                kwargs["chart"] = defaults.pop("chart", False)
+
+            if "chart_params" in defaults:
+                extra_params["chart_params"] = defaults.pop("chart_params", {})
+
+            for k, v in defaults.items():
+                if k in standard_params and standard_params[k] is None:
+                    standard_params[k] = v
+                elif (k in standard_params and standard_params[k] is not None) or (
+                    k in extra_params and extra_params[k] is not None
+                ):
+                    continue
+                elif k not in extra_params or (
+                    k in extra_params and extra_params[k] is None
+                ):
+                    extra_params[k] = v
+
+            kwargs["standard_params"] = standard_params
+            kwargs["extra_params"] = extra_params
+
+        execute = partial(command_runner.run, path, user_settings)
+        output = await execute(*args, **kwargs)
+
+        if isinstance(output, OBBject) and not no_validate:
+            return validate_output(output)
+
+        return output
+
+    return wrapper
+
+
+def add_command_map(command_runner: CommandRunner, api_router: APIRouter) -> None:
+    """Add command map to the API router."""
+    plugins_router = RouterLoader.from_extensions()
+
+    for route in plugins_router.api_router.routes:
+        route.endpoint = build_api_wrapper(command_runner=command_runner, route=route)  # type: ignore # noqa
+    api_router.include_router(router=plugins_router.api_router)
+
+
+system_settings = SystemService(logging_sub_app="api").system_settings
+command_runner_instance = CommandRunner(system_settings=system_settings)
+add_command_map(command_runner=command_runner_instance, api_router=router)

openbb_platform/core/openbb_core/api/router/coverage.py

@@ -0,0 +1,105 @@
+"""Coverage API router."""
+
+import json
+
+from fastapi import APIRouter, Depends
+from openbb_core.api.dependency.coverage import get_command_map, get_provider_interface
+from openbb_core.app.provider_interface import ProviderInterface
+from openbb_core.app.router import CommandMap
+from typing_extensions import Annotated
+
+router = APIRouter(prefix="/coverage", tags=["Coverage"])
+
+
+@router.get("/command_model", openapi_extra={"widget_config": {"exclude": True}})
+async def get_commands_model_map(
+    command_map: Annotated[CommandMap, Depends(get_command_map)],
+    provider_interface: Annotated[ProviderInterface, Depends(get_provider_interface)],
+):
+    """Get the command to provider model mapping."""
+
+    commands_map: dict = {}
+
+    for command in command_map.commands_model:
+        model = command_map.commands_model[command]
+        pi_command = provider_interface.map[model]
+        schema = provider_interface.return_annotations[model]
+        providers = list(pi_command)
+        new_command: dict = {}
+        new_command["response_schema_name"] = schema.__name__ if schema else None
+        for provider in providers:
+            new_command[provider] = {
+                "QueryParams": {"docstring": "", "fields": {}},
+                "Data": {"docstring": "", "fields": {}},
+            }
+            p = pi_command[provider]
+            query = p.get("QueryParams", {})
+            query_fields = query.get("fields", {})
+            data = p.get("Data", {})
+            data_fields = data.get("fields", {})
+
+            for field, field_info in query_fields.items():
+                attributes = (
+                    field_info._attributes_set  # pylint: disable=protected-access
+                )
+                if attributes.get("annotation"):
+                    _annotation = str(attributes.get("annotation"))
+                    attributes["annotation"] = _annotation
+
+                new_command[provider]["QueryParams"]["fields"][field] = attributes
+
+            new_command[provider]["QueryParams"]["docstring"] = query.get("docstring")
+
+            for field, field_info in data_fields.items():
+                attributes = (
+                    field_info._attributes_set  # pylint: disable=protected-access
+                )
+                if attributes.get("annotation"):
+                    _annotation = str(attributes.get("annotation"))
+                    attributes["annotation"] = _annotation
+                new_command[provider]["Data"]["fields"][field] = attributes
+
+            new_command[provider]["Data"]["docstring"] = data.get("docstring")
+
+            if openbb_info := new_command.get("openbb", {}):
+                for key in list(new_command):
+                    if key == "response_schema_name":
+                        continue
+
+                    if obb_params := openbb_info.get("QueryParams", {}).get(
+                        "fields", {}
+                    ):
+                        old_fields = new_command[key]["QueryParams"].get("fields", {})
+                        new_command[key]["QueryParams"]["fields"] = {
+                            **obb_params,
+                            **old_fields,
+                        }
+                    if obb_data := openbb_info.get("Data", {}).get("fields", {}):
+                        old_fields = new_command[key]["Data"].get("fields", {})
+                        new_command[key]["Data"]["fields"] = {**obb_data, **old_fields}
+        _ = new_command.pop("openbb")
+        commands_map[command] = new_command
+
+    def serializer(obj):
+        """Serialize the object."""
+        if isinstance(obj, type):
+            return str(obj)
+        return obj
+
+    return json.loads(json.dumps(commands_map, default=serializer, indent=4))
+
+
+@router.get("/providers", openapi_extra={"widget_config": {"exclude": True}})
+async def get_provider_coverage(
+    command_map: Annotated[CommandMap, Depends(get_command_map)]
+):
+    """Get command coverage by provider."""
+    return command_map.provider_coverage
+
+
+@router.get("/commands", openapi_extra={"widget_config": {"exclude": True}})
+async def get_command_coverage(
+    command_map: Annotated[CommandMap, Depends(get_command_map)]
+):
+    """Get provider coverage by command."""
+    return command_map.command_coverage

openbb_platform/core/openbb_core/api/router/helpers/__init__.py

@@ -0,0 +1 @@
+"""The init of the coverage helpers."""

openbb_platform/core/openbb_core/api/router/helpers/coverage_helpers.py

@@ -0,0 +1,120 @@
+"""Coverage API router helper functions."""
+
+from inspect import _empty, signature
+from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, Type
+
+from openbb_core.app.provider_interface import ProviderInterface
+from pydantic import BaseModel, Field, create_model
+
+if TYPE_CHECKING:
+    from openbb_core.app.static.app_factory import BaseApp
+
+provider_interface = ProviderInterface()
+
+
+def get_route_callable(app: "BaseApp", route: str) -> Callable:
+    """Get the callable for a route."""
+    # TODO: Add return typing Optional[Callable] to this function. First need to
+    # figure how to do that starting from "BaseApp" and account for the possibility
+    # of a route not existing. Then remove the type: ignore from the function.
+
+    split_route = route.replace(".", "/").split("/")[1:]
+
+    return_callable = app
+
+    for route_path in split_route:
+        return_callable = getattr(return_callable, route_path)
+
+    return return_callable  # type: ignore
+
+
+def signature_to_fields(app: "BaseApp", route: str) -> Dict[str, Tuple[Any, Field]]:  # type: ignore
+    """Convert a command signature to pydantic fields."""
+    return_callable = get_route_callable(app, route)
+    sig = signature(return_callable)
+
+    fields = {}
+    for name, param in sig.parameters.items():
+        if name not in ["kwargs", "args"]:
+            type_annotation = (
+                param.annotation if param.annotation is not _empty else Any
+            )
+            description = (
+                param.annotation.__metadata__[0].description
+                if hasattr(param.annotation, "__metadata__")
+                else None
+            )
+            fields[name] = (
+                type_annotation,
+                Field(..., title="openbb", description=description),
+            )
+
+    return fields
+
+
+def dataclass_to_fields(model_name: str) -> Dict[str, Tuple[Any, Field]]:  # type: ignore
+    """Convert a dataclass to pydantic fields."""
+    dataclass = provider_interface.params[model_name]["extra"]
+    fields = {}
+    for name, field in dataclass.__dataclass_fields__.items():
+        type_annotation = field.default.annotation if field.default is not None else Any  # type: ignore
+        description = field.default.description if field.default is not None else None  # type: ignore
+        title = field.default.title if field.default is not None else None  # type: ignore
+        fields[name] = (
+            type_annotation,
+            Field(..., title=title, description=description),
+        )
+
+    return fields
+
+
+def create_combined_model(
+    model_name: str,
+    *field_sets: Dict[str, Tuple[Any, Field]],  # type: ignore
+    filter_by_provider: Optional[str] = None,
+) -> Type[BaseModel]:
+    """Create a combined pydantic model."""
+    combined_fields = {}
+    for fields in field_sets:
+        for name, (type_annotation, field) in fields.items():
+            if (
+                filter_by_provider is None
+                or "openbb" in field.title  # type: ignore
+                or (filter_by_provider in field.title)  # type: ignore
+            ):
+                combined_fields[name] = (type_annotation, field)
+
+    model = create_model(model_name, **combined_fields)  # type: ignore
+
+    # # Clean up the metadata
+    for field in model.model_fields.values():
+        if hasattr(field, "metadata"):
+            field.metadata = None  # type: ignore
+
+    return model
+
+
+def get_route_schema_map(
+    app: "BaseApp",
+    command_model_map: Dict[str, str],
+    filter_by_provider: Optional[str] = None,
+) -> Dict[str, Dict[str, Any]]:
+    """Get the route schema map."""
+    route_schema_map = {}
+    for route, model in command_model_map.items():
+        input_model = create_combined_model(
+            route,
+            signature_to_fields(app, route),
+            dataclass_to_fields(model),
+            filter_by_provider=filter_by_provider,
+        )
+        output_model = provider_interface.return_schema[model]
+        return_callable = get_route_callable(app, route)
+
+        route_schema_map[route] = {
+            "input": input_model,
+            "output": output_model,
+            "callable": return_callable,
+        }
+
+    return route_schema_map

openbb_platform/core/openbb_core/api/router/system.py

@@ -0,0 +1,16 @@
+"""System router."""
+
+from fastapi import APIRouter, Depends
+from openbb_core.api.dependency.system import get_system_settings
+from openbb_core.app.model.system_settings import SystemSettings
+from typing_extensions import Annotated
+
+router = APIRouter(prefix="/system", tags=["System"])
+
+
+@router.get("")
+async def get_system_model(
+    system_settings: Annotated[SystemSettings, Depends(get_system_settings)],
+):
+    """Get system model."""
+    return system_settings

openbb_platform/core/openbb_core/api/router/user.py

@@ -0,0 +1,18 @@
+"""OpenBB Platform API Account Router."""
+
+from fastapi import APIRouter, Depends
+from openbb_core.api.auth.user import authenticate_user, get_user_settings
+from openbb_core.app.model.user_settings import UserSettings
+from typing_extensions import Annotated
+
+router = APIRouter(prefix="/user", tags=["User"])
+auth_hook = authenticate_user
+user_settings_hook = get_user_settings
+
+
+@router.get("/me")
+async def read_user_settings(
+    user_settings: Annotated[UserSettings, Depends(get_user_settings)]
+):
+    """Read current user settings."""
+    return user_settings

openbb_platform/core/openbb_core/app/__init__.py

@@ -0,0 +1 @@
+"""OpenBB Core App Module."""

openbb_platform/core/openbb_core/app/command_runner.py

@@ -0,0 +1,512 @@
+"""Command runner module."""
+
+# pylint: disable=R0903
+
+from copy import deepcopy
+from dataclasses import asdict, is_dataclass
+from datetime import datetime
+from inspect import Parameter, signature
+from sys import exc_info
+from time import perf_counter_ns
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Type
+from warnings import catch_warnings, showwarning, warn
+
+from openbb_core.app.model.abstract.error import OpenBBError
+from openbb_core.app.model.abstract.warning import OpenBBWarning, cast_warning
+from openbb_core.app.model.metadata import Metadata
+from openbb_core.app.model.obbject import OBBject
+from openbb_core.app.provider_interface import ExtraParams
+from openbb_core.app.static.package_builder import PathHandler
+from openbb_core.env import Env
+from openbb_core.provider.utils.helpers import maybe_coroutine, run_async
+from pydantic import BaseModel, ConfigDict, create_model
+
+if TYPE_CHECKING:
+    from fastapi.routing import APIRoute
+    from openbb_core.app.model.system_settings import SystemSettings
+    from openbb_core.app.model.user_settings import UserSettings
+    from openbb_core.app.router import CommandMap
+
+
+class ExecutionContext:
+    """Execution context."""
+
+    # For checking if the command specifies no validation in the API Route
+    _route_map = PathHandler.build_route_map()
+
+    def __init__(
+        self,
+        command_map: "CommandMap",
+        route: str,
+        system_settings: "SystemSettings",
+        user_settings: "UserSettings",
+    ) -> None:
+        """Initialize the execution context."""
+        self.command_map = command_map
+        self.route = route
+        self.system_settings = system_settings
+        self.user_settings = user_settings
+
+    @property
+    def api_route(self) -> "APIRoute":
+        """API route."""
+        return self._route_map[self.route]
+
+
+class ParametersBuilder:
+    """Build parameters for a function."""
+
+    @staticmethod
+    def get_polished_parameter_list(func: Callable) -> List[Parameter]:
+        """Get the signature parameters values as a list."""
+        sig = signature(func)
+        parameter_list = list(sig.parameters.values())
+
+        return parameter_list
+
+    @staticmethod
+    def get_polished_func(func: Callable) -> Callable:
+        """Remove __authenticated_user_settings from the function signature and annotations."""
+        func = deepcopy(func)
+        sig = signature(func)
+        parameter_map = dict(sig.parameters)
+
+        if "__authenticated_user_settings" in parameter_map:
+            parameter_map.pop("__authenticated_user_settings")
+
+        parameter_list = list(parameter_map.values())
+        new_signature = signature(func).replace(parameters=parameter_list)
+
+        func.__signature__ = new_signature  # type: ignore
+        func.__annotations__ = parameter_map
+
+        return func
+
+    @classmethod
+    def merge_args_and_kwargs(
+        cls,
+        func: Callable,
+        args: Tuple[Any, ...],
+        kwargs: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Merge args and kwargs into a single dict."""
+        args = deepcopy(args)
+        kwargs = deepcopy(kwargs)
+        parameter_list = cls.get_polished_parameter_list(func=func)
+        parameter_map = {}
+
+        for index, parameter in enumerate(parameter_list):
+            if index < len(args):
+                parameter_map[parameter.name] = args[index]
+            elif parameter.name in kwargs:
+                parameter_map[parameter.name] = kwargs[parameter.name]
+            elif parameter.default is not parameter.empty:
+                parameter_map[parameter.name] = parameter.default
+            else:
+                parameter_map[parameter.name] = None
+
+        return parameter_map
+
+    @staticmethod
+    def update_command_context(
+        func: Callable,
+        kwargs: Dict[str, Any],
+        system_settings: "SystemSettings",
+        user_settings: "UserSettings",
+    ) -> Dict[str, Any]:
+        """Update the command context with the available user and system settings."""
+        # pylint: disable=import-outside-toplevel
+        from openbb_core.app.model.command_context import CommandContext
+
+        argcount = func.__code__.co_argcount
+        if "cc" in func.__code__.co_varnames[:argcount]:
+            kwargs["cc"] = CommandContext(
+                user_settings=user_settings,
+                system_settings=system_settings,
+            )
+
+        return kwargs
+
+    @staticmethod
+    def _warn_kwargs(
+        extra_params: Dict[str, Any],
+        model: Type[BaseModel],
+    ) -> None:
+        """Warn if kwargs received and ignored by the validation model."""
+        # We only check the extra_params annotation because ignored fields
+        # will always be there
+        annotation = getattr(
+            model.model_fields.get("extra_params", None), "annotation", None
+        )
+        if is_dataclass(annotation) and any(
+            t is ExtraParams for t in getattr(annotation, "__bases__", [])
+        ):
+            valid = asdict(annotation())  # type: ignore
+            for p in extra_params:
+                if "chart_params" in p:
+                    continue
+                if p not in valid:
+                    warn(
+                        message=f"Parameter '{p}' not found.",
+                        category=OpenBBWarning,
+                    )
+
+    @staticmethod
+    def _as_dict(obj: Any) -> Dict[str, Any]:
+        """Safely convert an object to a dict."""
+        try:
+            if isinstance(obj, dict):
+                return obj
+            return asdict(obj) if is_dataclass(obj) else dict(obj)  # type: ignore
+        except Exception:
+            return {}
+
+    @staticmethod
+    def validate_kwargs(
+        func: Callable,
+        kwargs: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Validate kwargs and if possible coerce to the correct type."""
+        sig = signature(func)
+        fields = {
+            n: (
+                Any if p.annotation is Parameter.empty else p.annotation,
+                ... if p.default is Parameter.empty else p.default,
+            )
+            for n, p in sig.parameters.items()
+        }
+        # We allow extra fields to return with model with 'cc: CommandContext'
+        config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
+        # pylint: disable=C0103
+        ValidationModel = create_model(func.__name__, __config__=config, **fields)  # type: ignore
+        # Validate and coerce
+        model = ValidationModel(**kwargs)
+        ParametersBuilder._warn_kwargs(
+            ParametersBuilder._as_dict(kwargs.get("extra_params", {})),
+            ValidationModel,
+        )
+        return dict(model)
+
+    # pylint: disable=R0913
+    @classmethod
+    def build(
+        cls,
+        args: Tuple[Any, ...],
+        execution_context: ExecutionContext,
+        func: Callable,
+        kwargs: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Build the parameters for a function."""
+        func = cls.get_polished_func(func=func)
+        system_settings = execution_context.system_settings
+        user_settings = execution_context.user_settings
+
+        kwargs = cls.merge_args_and_kwargs(
+            func=func,
+            args=args,
+            kwargs=kwargs,
+        )
+        kwargs = cls.update_command_context(
+            func=func,
+            kwargs=kwargs,
+            system_settings=system_settings,
+            user_settings=user_settings,
+        )
+        kwargs = cls.validate_kwargs(
+            func=func,
+            kwargs=kwargs,
+        )
+        return kwargs
+
+
+# pylint: disable=too-few-public-methods
+class StaticCommandRunner:
+    """Static Command Runner."""
+
+    @classmethod
+    async def _command(
+        cls,
+        func: Callable,
+        kwargs: Dict[str, Any],
+        show_warnings: bool = True,  # pylint: disable=unused-argument   # type: ignore
+    ) -> OBBject:
+        """Run a command and return the output."""
+        obbject = await maybe_coroutine(func, **kwargs)
+        if isinstance(obbject, OBBject):
+            obbject.provider = getattr(
+                kwargs.get("provider_choices"),
+                "provider",
+                getattr(obbject, "provider", None),
+            )
+        return obbject
+
+    @classmethod
+    def _chart(
+        cls,
+        obbject: OBBject,
+        **kwargs,
+    ) -> None:
+        """Create a chart from the command output."""
+        try:
+            if "charting" not in obbject.accessors:
+                raise OpenBBError(
+                    "Charting is not installed. Please install `openbb-charting`."
+                )
+            # Here we will pop the chart_params kwargs and flatten them into the kwargs.
+            chart_params = {}
+            extra_params = getattr(obbject, "_extra_params", {})
+
+            if extra_params and "chart_params" in extra_params:
+                chart_params = extra_params.get("chart_params", {})
+
+            if kwargs.get("chart_params"):
+                chart_params.update(kwargs.pop("chart_params", {}))
+            # Verify that kwargs is not nested as kwargs so we don't miss any chart params.
+            if (
+                "kwargs" in kwargs
+                and "chart_params" in kwargs["kwargs"]
+                and kwargs["kwargs"].get("chart_params")
+            ):
+                chart_params.update(kwargs.pop("kwargs", {}).get("chart_params", {}))
+
+            if chart_params:
+                kwargs.update(chart_params)
+            obbject.charting.show(render=False, **kwargs)  # type: ignore[attr-defined]
+        except Exception as e:  # pylint: disable=broad-exception-caught
+            if Env().DEBUG_MODE:
+                raise OpenBBError(e) from e
+            warn(str(e), OpenBBWarning)
+
+    @classmethod
+    def _extract_params(cls, kwargs, key) -> Dict:
+        """Extract params models from kwargs and convert to a dictionary."""
+        params = kwargs.get(key, {})
+        if hasattr(params, "__dict__"):
+            return params.__dict__
+        return params
+
+    # pylint: disable=R0913, R0914
+    @classmethod
+    async def _execute_func(  # pylint: disable=too-many-positional-arguments
+        cls,
+        route: str,
+        args: Tuple[Any, ...],
+        execution_context: ExecutionContext,
+        func: Callable,
+        kwargs: Dict[str, Any],
+    ) -> OBBject:
+        """Execute a function and return the output."""
+        # pylint: disable=import-outside-toplevel
+        from openbb_core.app.logs.logging_service import LoggingService
+
+        user_settings = execution_context.user_settings
+        system_settings = execution_context.system_settings
+        raised_warnings: list = []
+        custom_headers: Optional[dict[str, Any]] = None
+
+        try:
+            with catch_warnings(record=True) as warning_list:
+                # If we're on Jupyter we need to pop here because we will lose "chart" after
+                # ParametersBuilder.build. This needs to be fixed in a way that chart is
+                # added to the function signature and shared for jupyter and api
+                # We can check in the router decorator if the given function has a chart
+                # in the charting extension then we add it there. This way we can remove
+                # the chart parameter from the commands.py and package_builder, it will be
+                # added to the function signature in the router decorator
+                chart = kwargs.pop("chart", False)
+
+                kwargs = ParametersBuilder.build(
+                    args=args,
+                    execution_context=execution_context,
+                    func=func,
+                    kwargs=kwargs,
+                )
+                kwargs = kwargs if kwargs is not None else {}
+                # If we're on the api we need to remove "chart" here because the parameter is added on
+                # commands.py and the function signature does not expect "chart"
+                kwargs.pop("chart", None)
+                # We also pop custom headers
+                model_headers = system_settings.api_settings.custom_headers or {}
+                custom_headers = {
+                    name: kwargs.pop(name.replace("-", "_"), default)
+                    for name, default in model_headers.items() or {}
+                } or None
+
+                obbject = await cls._command(func, kwargs)
+                # The output might be from a router command with 'no_validate=True'
+                # It might be of a different type than OBBject.
+                # In this case, we avoid accessing those attributes.
+                if isinstance(obbject, OBBject):
+                    # This section prepares the obbject to pass to the charting service.
+                    obbject._route = route  # pylint: disable=protected-access
+                    std_params = cls._extract_params(kwargs, "standard_params") or (
+                        kwargs if "data" in kwargs else {}
+                    )
+                    extra_params = cls._extract_params(kwargs, "extra_params")
+                    obbject._standard_params = (  # pylint: disable=protected-access
+                        std_params
+                    )
+                    obbject._extra_params = (  # pylint: disable=protected-access
+                        extra_params
+                    )
+                    if chart and obbject.results:
+                        cls._chart(obbject, **kwargs)
+
+                raised_warnings = warning_list if warning_list else []
+        finally:
+            if raised_warnings:
+                if isinstance(obbject, OBBject):
+                    obbject.warnings = []
+                for w in raised_warnings:
+                    if isinstance(obbject, OBBject):
+                        obbject.warnings.append(cast_warning(w))
+                    if user_settings.preferences.show_warnings:
+                        showwarning(
+                            message=w.message,
+                            category=w.category,
+                            filename=w.filename,
+                            lineno=w.lineno,
+                            file=w.file,
+                            line=w.line,
+                        )
+            ls = LoggingService(system_settings, user_settings)
+            ls.log(
+                user_settings=user_settings,
+                system_settings=system_settings,
+                route=route,
+                func=func,
+                kwargs=kwargs,
+                exec_info=exc_info(),
+                custom_headers=custom_headers,
+            )
+
+        return obbject
+
+    # pylint: disable=W0718
+    @classmethod
+    async def run(
+        cls,
+        execution_context: ExecutionContext,
+        /,
+        *args,
+        **kwargs,
+    ) -> OBBject:
+        """Run a command and return the OBBject as output."""
+        timestamp = datetime.now()
+        start_ns = perf_counter_ns()
+
+        command_map = execution_context.command_map
+        route = execution_context.route
+
+        if func := command_map.get_command(route=route):
+            obbject = await cls._execute_func(
+                route=route,
+                args=args,  # type: ignore
+                execution_context=execution_context,
+                func=func,
+                kwargs=kwargs,
+            )
+        else:
+            raise AttributeError(f"Invalid command : route={route}")
+
+        duration = perf_counter_ns() - start_ns
+
+        if execution_context.user_settings.preferences.metadata and isinstance(
+            obbject, OBBject
+        ):
+            try:
+                obbject.extra["metadata"] = Metadata(
+                    arguments=kwargs,
+                    duration=duration,
+                    route=route,
+                    timestamp=timestamp,
+                )
+            except Exception as e:
+                if Env().DEBUG_MODE:
+                    raise OpenBBError(e) from e
+                warn(str(e), OpenBBWarning)
+
+        return obbject
+
+
+class CommandRunner:
+    """Command runner."""
+
+    def __init__(
+        self,
+        command_map: Optional["CommandMap"] = None,
+        system_settings: Optional["SystemSettings"] = None,
+        user_settings: Optional["UserSettings"] = None,
+    ) -> None:
+        """Initialize the command runner."""
+        # pylint: disable=import-outside-toplevel
+        from openbb_core.app.router import CommandMap
+        from openbb_core.app.service.system_service import SystemService
+        from openbb_core.app.service.user_service import UserService
+
+        self._command_map = command_map or CommandMap()
+        self._system_settings = system_settings or SystemService().system_settings
+        self._user_settings = user_settings or UserService.read_from_file()
+
+    def init_logging_service(self) -> None:
+        """Initialize the logging service."""
+        # pylint: disable=import-outside-toplevel
+        from openbb_core.app.logs.logging_service import LoggingService
+
+        _ = LoggingService(
+            system_settings=self._system_settings, user_settings=self._user_settings
+        )
+
+    @property
+    def command_map(self) -> "CommandMap":
+        """Command map."""
+        return self._command_map
+
+    @property
+    def system_settings(self) -> "SystemSettings":
+        """System settings."""
+        return self._system_settings
+
+    @property
+    def user_settings(self) -> "UserSettings":
+        """User settings."""
+        return self._user_settings
+
+    @user_settings.setter
+    def user_settings(self, user_settings: "UserSettings") -> None:
+        self._user_settings = user_settings
+
+    # pylint: disable=W1113
+    async def run(
+        self,
+        route: str,
+        user_settings: Optional["UserSettings"] = None,
+        /,
+        *args,
+        **kwargs,
+    ) -> OBBject:
+        """Run a command and return the OBBject as output."""
+        # pylint: disable=import-outside-toplevel
+
+        self._user_settings = user_settings or self._user_settings
+
+        execution_context = ExecutionContext(
+            command_map=self._command_map,
+            route=route,
+            system_settings=self._system_settings,
+            user_settings=self._user_settings,
+        )
+
+        return await StaticCommandRunner.run(execution_context, *args, **kwargs)
+
+    # pylint: disable=W1113
+    def sync_run(
+        self,
+        route: str,
+        user_settings: Optional["UserSettings"] = None,
+        /,
+        *args,
+        **kwargs,
+    ) -> OBBject:
+        """Run a command and return the OBBject as output."""
+        return run_async(self.run, route, user_settings, *args, **kwargs)

openbb_platform/core/openbb_core/app/constants.py

@@ -0,0 +1,8 @@
+"""Constants for the OpenBB Platform."""
+
+from pathlib import Path
+
+HOME_DIRECTORY = Path.home()
+OPENBB_DIRECTORY = Path(HOME_DIRECTORY, ".openbb_platform")
+USER_SETTINGS_PATH = Path(OPENBB_DIRECTORY, "user_settings.json")
+SYSTEM_SETTINGS_PATH = Path(OPENBB_DIRECTORY, "system_settings.json")

openbb_platform/core/openbb_core/app/deprecation.py

@@ -0,0 +1,65 @@
+"""
+OpenBB-specific deprecation warnings.
+
+This implementation was inspired from Pydantic's specific warnings and modified to suit OpenBB's needs.
+"""
+
+from typing import Optional, Tuple
+
+from openbb_core.app.version import VERSION, get_major_minor
+
+
+class DeprecationSummary(str):
+    """A string subclass that can be used to store deprecation metadata."""
+
+    def __new__(cls, value: str, metadata: DeprecationWarning):
+        """Create a new instance of the class."""
+        obj = str.__new__(cls, value)
+        setattr(obj, "metadata", metadata)
+        return obj
+
+
+class OpenBBDeprecationWarning(DeprecationWarning):
+    """
+    A OpenBB specific deprecation warning.
+
+    This warning is raised when using deprecated functionality in OpenBB. It provides information on when the
+    deprecation was introduced and the expected version in which the corresponding functionality will be removed.
+
+    Attributes
+    ----------
+        message: Description of the warning.
+        since: Version in what the deprecation was introduced.
+        expected_removal: Version in what the corresponding functionality expected to be removed.
+    """
+
+    # The choice to use class variables is based on the potential for extending the class in future developments.
+    # Example: launching Platform V5 and decide to create a subclimagine we areass named OpenBBDeprecatedSinceV4,
+    # which inherits from OpenBBDeprecationWarning. In this subclass, we would set since=4.X and expected_removal=5.0.
+    # It's important for these values to be defined at the class level, rather than just at the instance level,
+    # to ensure consistency and clarity in our deprecation warnings across the platform.
+
+    message: str
+    since: Tuple[int, int]
+    expected_removal: Tuple[int, int]
+
+    def __init__(
+        self,
+        message: str,
+        *args: object,
+        since: Optional[Tuple[int, int]] = None,
+        expected_removal: Optional[Tuple[int, int]] = None,
+    ) -> None:
+        """Initialize the warning."""
+        super().__init__(message, *args)
+        self.message = message.rstrip(".")
+        self.since = since or get_major_minor(VERSION)
+        self.expected_removal = expected_removal or (self.since[0] + 1, 0)
+        self.long_message = (
+            f"{self.message}. Deprecated in OpenBB Platform V{self.since[0]}.{self.since[1]}"
+            f" to be removed in V{self.expected_removal[0]}.{self.expected_removal[1]}."
+        )
+
+    def __str__(self) -> str:
+        """Return the warning message."""
+        return self.long_message

openbb_platform/core/openbb_core/app/extension_loader.py

@@ -0,0 +1,177 @@
+"""Extension Loader."""
+
+from enum import Enum
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+from importlib_metadata import EntryPoint, EntryPoints, entry_points
+from openbb_core.app.model.abstract.singleton import SingletonMeta
+from openbb_core.app.model.extension import Extension
+
+if TYPE_CHECKING:
+    from openbb_core.app.router import Router
+    from openbb_core.provider.abstract.provider import Provider
+
+
+class OpenBBGroups(Enum):
+    """OpenBB Extension Groups."""
+
+    core = "openbb_core_extension"
+    provider = "openbb_provider_extension"
+    obbject = "openbb_obbject_extension"
+
+    @staticmethod
+    def groups() -> List[str]:
+        """Return the OpenBBGroups."""
+        return [
+            OpenBBGroups.core.value,
+            OpenBBGroups.provider.value,
+            OpenBBGroups.obbject.value,
+        ]
+
+
+class ExtensionLoader(metaclass=SingletonMeta):
+    """Extension loader class."""
+
+    def __init__(
+        self,
+    ) -> None:
+        """Initialize the extension loader."""
+        self._obbject_entry_points: EntryPoints = self._sorted_entry_points(
+            group=OpenBBGroups.obbject.value
+        )
+        self._core_entry_points: EntryPoints = self._sorted_entry_points(
+            group=OpenBBGroups.core.value
+        )
+        self._provider_entry_points: EntryPoints = self._sorted_entry_points(
+            group=OpenBBGroups.provider.value
+        )
+        self._obbject_objects: Dict[str, Extension] = {}
+        self._core_objects: Dict[str, Router] = {}
+        self._provider_objects: Dict[str, Provider] = {}
+
+    @property
+    def obbject_entry_points(self) -> EntryPoints:
+        """Return the obbject entry points."""
+        return self._obbject_entry_points
+
+    @property
+    def core_entry_points(self) -> EntryPoints:
+        """Return the core entry points."""
+        return self._core_entry_points
+
+    @property
+    def provider_entry_points(self) -> EntryPoints:
+        """Return the provider entry points."""
+        return self._provider_entry_points
+
+    @property
+    def entry_points(self) -> List[EntryPoints]:
+        """Return the entry points."""
+        return [
+            self._core_entry_points,
+            self._provider_entry_points,
+            self._obbject_entry_points,
+        ]
+
+    @staticmethod
+    def _get_entry_point(
+        entry_points_: EntryPoints, ext_name: str
+    ) -> Optional[EntryPoint]:
+        """Given an extension name and a list of entry points, return the corresponding entry point."""
+        return next((ep for ep in entry_points_ if ep.name == ext_name), None)
+
+    def get_obbject_entry_point(self, ext_name: str) -> Optional[EntryPoint]:
+        """Given an extension name, return the corresponding entry point."""
+        return self._get_entry_point(self._obbject_entry_points, ext_name)
+
+    def get_core_entry_point(self, ext_name: str) -> Optional[EntryPoint]:
+        """Given an extension name, return the corresponding entry point."""
+        return self._get_entry_point(self._core_entry_points, ext_name)
+
+    def get_provider_entry_point(self, ext_name: str) -> Optional[EntryPoint]:
+        """Given an extension name, return the corresponding entry point."""
+        return self._get_entry_point(self._provider_entry_points, ext_name)
+
+    @property
+    @lru_cache
+    def obbject_objects(self) -> Dict[str, Extension]:
+        """Return a dict of obbject extension objects."""
+        self._obbject_objects = self._load_entry_points(
+            self._obbject_entry_points, OpenBBGroups.obbject
+        )
+        return self._obbject_objects
+
+    @property
+    @lru_cache
+    def core_objects(self) -> Dict[str, "Router"]:
+        """Return a dict of core extension objects."""
+        self._core_objects = self._load_entry_points(
+            self._core_entry_points, OpenBBGroups.core
+        )
+        return self._core_objects
+
+    @property
+    @lru_cache
+    def provider_objects(self) -> Dict[str, "Provider"]:
+        """Return a dict of provider extension objects."""
+        self._provider_objects = self._load_entry_points(
+            self._provider_entry_points, OpenBBGroups.provider
+        )
+        return self._provider_objects
+
+    @staticmethod
+    def _sorted_entry_points(group: str) -> EntryPoints:
+        """Return a sorted dictionary of entry points."""
+        return sorted(entry_points(group=group))  # type: ignore
+
+    def _load_entry_points(
+        self, entry_points_: EntryPoints, group: OpenBBGroups
+    ) -> Dict[str, Any]:
+        """Return a dict of objects matching the entry points."""
+
+        def load_obbject(eps: EntryPoints) -> Dict[str, Extension]:
+            """
+            Return a dictionary of obbject objects.
+
+            Keys are entry point names and values are instances of the Extension class.
+            """
+            return {
+                ep.name: entry
+                for ep in eps
+                if isinstance((entry := ep.load()), Extension)
+            }
+
+        def load_core(eps: EntryPoints) -> Dict[str, "Router"]:
+            """Return a dictionary of core objects."""
+            # pylint: disable=import-outside-toplevel
+            from openbb_core.app.router import Router
+
+            return {
+                ep.name: entry for ep in eps if isinstance((entry := ep.load()), Router)
+            }
+
+        def load_provider(eps: EntryPoints) -> Dict[str, "Provider"]:
+            """
+            Return a dictionary of provider objects.
+
+            Keys are entry point names and values are instances of the Provider class.
+            """
+            # pylint: disable=import-outside-toplevel
+            from openbb_core.provider.abstract.provider import Provider
+
+            entries: dict = {}
+            for ep in eps:
+                try:
+                    if isinstance((entry := ep.load()), Provider):
+                        entries[ep.name] = entry
+                except ModuleNotFoundError:
+                    continue
+            return entries
+
+        func = {
+            OpenBBGroups.obbject: load_obbject,
+            OpenBBGroups.core: load_core,
+            OpenBBGroups.provider: load_provider,
+        }
+        return func[group](entry_points_)  # type: ignore

openbb_platform/core/openbb_core/app/logs/formatters/formatter_with_exceptions.py

@@ -0,0 +1,208 @@
+"""Logging Formatter that includes formatting of Exceptions."""
+
+import logging
+
+from openbb_core.app.logs.models.logging_settings import LoggingSettings
+
+
+class FormatterWithExceptions(logging.Formatter):
+    """Logging Formatter that includes formatting of Exceptions."""
+
+    DATEFORMAT = "%Y-%m-%dT%H:%M:%S%z"
+    LOGFORMAT = "%(asctime)s|%(name)s|%(funcName)s|%(lineno)s|%(message)s"
+    LOGPREFIXFORMAT = (
+        "%(levelname)s|%(appName)s|%(commitHash)s|%(appId)s|%(sessionId)s|%(userId)s|"
+    )
+
+    @staticmethod
+    def calculate_level_name(record: logging.LogRecord) -> str:
+        """Calculate the level name of the log record."""
+        if record.exc_text:
+            level_name = "X"
+        elif record.levelname:
+            level_name = record.levelname[0]
+        else:
+            level_name = "U"
+
+        return level_name
+
+    @staticmethod
+    def extract_log_extra(record: logging.LogRecord):
+        """Extract extra log information from the record."""
+        log_extra = dict()
+
+        if hasattr(record, "func_name_override"):
+            record.funcName = record.func_name_override  # type: ignore
+            record.lineno = 0
+
+        if hasattr(record, "session_id"):
+            log_extra["sessionId"] = record.session_id  # type: ignore
+
+        return log_extra
+
+    @staticmethod
+    def mock_ipv4(text: str) -> str:
+        """Mock IPv4 addresses in the text."""
+        # pylint: disable=import-outside-toplevel
+        import re
+
+        pattern = r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}"
+        replacement = " FILTERED_IP "
+        text_mocked = re.sub(pattern, replacement, text)
+
+        return text_mocked
+
+    @staticmethod
+    def mock_email(text: str) -> str:
+        """Mock email addresses in the text."""
+        # pylint: disable=import-outside-toplevel
+        import re
+
+        pattern = r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"
+        replacement = " FILTERED_EMAIL "
+        text_mocked = re.sub(pattern, replacement, text)
+
+        return text_mocked
+
+    @staticmethod
+    def mock_password(text: str) -> str:
+        """Mock passwords in the text."""
+        # pylint: disable=import-outside-toplevel
+        import re
+
+        pattern = r'("password": ")[^"]+'
+        replacement = r"\1 FILTERED_PASSWORD "
+        text_mocked = re.sub(pattern, replacement, text)
+        return text_mocked
+
+    @staticmethod
+    def mock_flair(text: str) -> str:
+        """Mock flair in the text."""
+        # pylint: disable=import-outside-toplevel
+        import re
+
+        pattern = r'("FLAIR": "\[)(.*?)\]'
+        replacement = r"\1 FILTERED_FLAIR ]"
+        text_mocked = re.sub(pattern, replacement, text)
+
+        return text_mocked
+
+    @staticmethod
+    def mock_home_directory(text: str) -> str:
+        """Mock home directory in the text."""
+        # pylint: disable=import-outside-toplevel
+        from pathlib import Path
+
+        user_home_directory = str(Path.home().as_posix())
+        text_mocked = text.replace("\\", "/").replace(
+            user_home_directory, "MOCKING_USER_PATH"
+        )
+
+        return text_mocked
+
+    @staticmethod
+    def filter_special_tags(text: str) -> str:
+        """Filter special tags in the text."""
+        text_filtered = text.replace("\n", " MOCKING_BREAKLINE ")
+        text_filtered = text_filtered.replace("'Traceback", "Traceback")
+
+        return text_filtered
+
+    @classmethod
+    def filter_piis(cls, text: str) -> str:
+        """Filter Personally Identifiable Information in the text."""
+        text_filtered = cls.mock_ipv4(text=text)
+        text_filtered = cls.mock_email(text=text_filtered)
+        text_filtered = cls.mock_password(text=text_filtered)
+        text_filtered = cls.mock_home_directory(text=text_filtered)
+        text_filtered = cls.mock_flair(text=text_filtered)
+
+        return text_filtered
+
+    @classmethod
+    def filter_log_line(cls, text: str):
+        """Filter log line."""
+        text_filtered = cls.filter_special_tags(text=text)
+        text_filtered = cls.filter_piis(text=text_filtered)
+
+        return text_filtered
+
+    # OVERRIDE
+    def __init__(
+        self,
+        settings: LoggingSettings,
+        style="%",
+        validate=True,
+    ) -> None:
+        """Initialize the FormatterWithExceptions."""
+        super().__init__(
+            fmt=self.LOGFORMAT,
+            datefmt=self.DATEFORMAT,
+            style=style,
+            validate=validate,
+        )
+        self.settings = settings
+
+    @property
+    def settings(self) -> LoggingSettings:
+        """Get the settings."""
+        # pylint: disable=import-outside-toplevel
+        from copy import deepcopy
+
+        return deepcopy(self.__settings)
+
+    @settings.setter
+    def settings(self, settings: LoggingSettings) -> None:
+        """Set the settings."""
+        self.__settings = settings
+
+    # OVERRIDE
+    def formatException(self, ei) -> str:
+        """Define the Exception formatting handler.
+
+        Parameters
+        ----------
+        ei : logging._SysExcInfoType
+            Exception to be logged
+        Returns
+        ----------
+        str
+            Formatted exception
+        """
+        result = super().formatException(ei)
+        return repr(result)
+
+    # OVERRIDE
+    def format(self, record: logging.LogRecord) -> str:
+        """Define the Log formatter.
+
+        Parameters
+        ----------
+        record : logging.LogRecord
+            Logging record
+        Returns
+        ----------
+        str
+            Formatted_log message
+        """
+        level_name = self.calculate_level_name(record=record)
+        log_prefix_content = {
+            "appName": self.settings.app_name,
+            "levelname": level_name,
+            "appId": self.settings.app_id,
+            "sessionId": self.settings.session_id,
+            "commitHash": "unknown-commit",
+            "userId": self.settings.user_id,
+        }
+
+        log_extra = self.extract_log_extra(record=record)
+        log_prefix_content = {**log_prefix_content, **log_extra}
+        log_prefix = self.LOGPREFIXFORMAT % log_prefix_content
+
+        record.msg = record.msg.replace("|", "-MOCK_PIPE-")
+
+        log_line = super().format(record)
+        log_line = self.filter_log_line(text=log_line)
+        log_line_full = log_prefix + log_line
+
+        return log_line_full

openbb_platform/core/openbb_core/app/logs/handlers/path_tracking_file_handler.py

@@ -0,0 +1,86 @@
+"""Path Tracking File Handler."""
+
+# IMPORTATION STANDARD
+from copy import deepcopy
+from logging.handlers import TimedRotatingFileHandler
+from pathlib import Path
+
+# IMPORTATION THIRD PARTY
+# IMPORTATION INTERNAL
+from openbb_core.app.logs.models.logging_settings import LoggingSettings
+from openbb_core.app.logs.utils.expired_files import (
+    get_expired_file_list,
+    get_timestamp_from_x_days,
+    remove_file_list,
+)
+
+ARCHIVES_FOLDER_NAME = "archives"
+TMP_FOLDER_NAME = "tmp"
+
+
+class PathTrackingFileHandler(TimedRotatingFileHandler):
+    """Path Tracking File Handler."""
+
+    @staticmethod
+    def build_log_file_path(settings: LoggingSettings) -> Path:
+        """Build the log file path."""
+        app_name = settings.app_name
+        directory = settings.user_logs_directory
+        session_id = settings.session_id
+
+        path = directory.absolute().joinpath(f"{app_name}_{session_id}")
+        return path
+
+    def clean_expired_files(self, before_timestamp: float):
+        """Remove expired files from logs directory."""
+        logs_dir = self.settings.user_logs_directory
+        archives_directory = logs_dir / ARCHIVES_FOLDER_NAME
+        tmp_directory = logs_dir / TMP_FOLDER_NAME
+
+        expired_logs_file_list = get_expired_file_list(
+            directory=logs_dir,
+            before_timestamp=before_timestamp,
+        )
+        expired_archives_file_list = get_expired_file_list(
+            directory=archives_directory,
+            before_timestamp=before_timestamp,
+        )
+        expired_tmp_file_list = get_expired_file_list(
+            directory=tmp_directory,
+            before_timestamp=before_timestamp,
+        )
+        remove_file_list(file_list=expired_logs_file_list)
+        remove_file_list(file_list=expired_archives_file_list)
+        remove_file_list(file_list=expired_tmp_file_list)
+
+    @property
+    def settings(self) -> LoggingSettings:
+        """Get the settings."""
+        return deepcopy(self.__settings)
+
+    @settings.setter
+    def settings(self, settings: LoggingSettings) -> None:
+        """Set the settings."""
+        self.__settings = settings
+
+    # OVERRIDE
+    def __init__(
+        self,
+        settings: LoggingSettings,
+        *args,
+        **kwargs,
+    ) -> None:
+        """Initialize the PathTrackingFileHandler."""
+        # SETUP PARENT CLASS
+        filename = str(self.build_log_file_path(settings=settings))
+        frequency = settings.frequency
+        kwargs["when"] = frequency
+
+        super().__init__(filename, *args, **kwargs)
+
+        self.suffix += ".log"
+
+        # SETUP CURRENT CLASS
+        self.__settings = settings
+
+        self.clean_expired_files(before_timestamp=get_timestamp_from_x_days(x=5))

openbb_platform/core/openbb_core/app/logs/handlers/posthog_handler.py

@@ -0,0 +1,151 @@
+"""Posthog Handler."""
+
+import json
+import logging
+from enum import Enum
+from typing import Any, Dict
+
+import posthog
+from openbb_core.app.logs.formatters.formatter_with_exceptions import (
+    FormatterWithExceptions,
+)
+from openbb_core.app.logs.models.logging_settings import LoggingSettings
+from openbb_core.env import Env
+
+
+class PosthogHandler(logging.Handler):
+    """Posthog Handler."""
+
+    class AllowedEvents(Enum):
+        """Allowed Posthog Events."""
+
+        log_startup = "log_startup"
+        log_cmd = "log_cmd"
+        log_error = "log_error"
+        log_warning = "log_warning"
+
+    def __init__(self, settings: LoggingSettings):
+        """Initialize Posthog Handler."""
+        # pylint: disable=import-outside-toplevel
+        from openbb_core.provider.utils.helpers import get_requests_session
+
+        super().__init__()
+        self._settings = settings
+        self.logged_in = False
+        posthog.api_key = "phc_6FXLqu4uW9yxfyN8DpPdgzCdlYXOmIWdMGh6GnBgJLX"  # pragma: allowlist secret  # noqa
+        posthog.host = "https://app.posthog.com"  # noqa
+        posthog.request._session = (  # pylint: disable=protected-access
+            get_requests_session()
+        )
+
+    @property
+    def settings(self) -> LoggingSettings:
+        """Get logging settings."""
+        # pylint: disable=import-outside-toplevel
+        from copy import deepcopy
+
+        return deepcopy(self._settings)
+
+    @settings.setter
+    def settings(self, settings: LoggingSettings) -> None:
+        """Set logging settings."""
+        self._settings = settings
+
+    def emit(self, record: logging.LogRecord):
+        """Emit log record."""
+        try:
+            self.send(record=record)
+        except Exception as e:
+            self.handleError(record)
+            if Env().DEBUG_MODE:
+                raise e
+
+    def distinct_id(self) -> str:
+        """Get distinct id."""
+        return self._settings.user_id or self._settings.app_id
+
+    def identify(self) -> None:
+        """Identify user."""
+        if self.logged_in or not self._settings.user_id:
+            return
+
+        posthog.identify(
+            self._settings.user_id,
+            {
+                "email": self._settings.user_email,
+                "primaryUsage": self._settings.user_primary_usage,
+            },
+        )
+
+        if self._settings.sub_app_name == "pro":
+            return
+
+        self.logged_in = True
+        posthog.alias(self._settings.user_id, self._settings.app_id)
+
+    def log_to_dict(self, log_info: str) -> dict:
+        """Log to dict."""
+        # pylint: disable=import-outside-toplevel
+        import re
+
+        log_regex = r"(STARTUP|CMD|ERROR): (.*)"
+        log_dict: Dict[str, Any] = {}
+
+        for log in re.findall(log_regex, log_info):
+            log_dict[log[0]] = json.loads(log[1])
+
+        return log_dict
+
+    def send(self, record: logging.LogRecord):
+        """Send log record to Posthog."""
+        # pylint: disable=import-outside-toplevel
+        import re
+
+        level_name = logging.getLevelName(record.levelno)
+        log_line = FormatterWithExceptions.filter_log_line(text=record.getMessage())
+
+        log_extra = self.extract_log_extra(record=record)
+        log_extra.update(dict(level=level_name, message=log_line))
+        event_name = f"log_{level_name.lower()}"
+
+        if log_dict := self.log_to_dict(log_info=log_line):
+            event_name = f"log_{list(log_dict.keys())[0].lower()}"
+            log_dict = log_dict.get("STARTUP", log_dict)
+
+            log_extra = {**log_extra, **log_dict}
+            log_extra.pop("message", None)
+
+        if re.match(r"^(QUEUE|START|END|INPUT:)", log_line) and not log_dict:
+            return
+
+        if event_name not in [e.value for e in self.AllowedEvents]:
+            return
+
+        self.identify()
+        posthog.capture(self.distinct_id(), event_name, properties=log_extra)
+
+    def extract_log_extra(self, record: logging.LogRecord) -> Dict[str, Any]:
+        """Extract log extra from record."""
+        log_extra: Dict[str, Any] = {
+            "appName": self._settings.app_name,
+            "subAppName": self._settings.sub_app_name,
+            "appId": self._settings.app_id,
+            "sessionId": self._settings.session_id,
+            "platform": self._settings.platform,
+            "pythonVersion": self._settings.python_version,
+            "obbPlatformVersion": self._settings.platform_version,
+        }
+
+        if self._settings.user_id:
+            log_extra["userId"] = self._settings.user_id
+
+        if hasattr(record, "extra"):
+            log_extra = {**log_extra, **record.extra}  # type: ignore
+
+        if record.exc_info:
+            log_extra["exception"] = {
+                "type": str(record.exc_info[0]),
+                "value": str(record.exc_info[1]),
+            }
+
+        return log_extra

openbb_platform/core/openbb_core/app/logs/handlers_manager.py

@@ -0,0 +1,89 @@
+"""Handlers Manager."""
+
+import logging
+import sys
+
+from openbb_core.app.logs.formatters.formatter_with_exceptions import (
+    FormatterWithExceptions,
+)
+from openbb_core.app.logs.handlers.path_tracking_file_handler import (
+    PathTrackingFileHandler,
+)
+from openbb_core.app.logs.handlers.posthog_handler import PosthogHandler
+from openbb_core.app.logs.models.logging_settings import LoggingSettings
+
+
+class HandlersManager:
+    """Handlers Manager."""
+
+    def __init__(self, logger: logging.Logger, settings: LoggingSettings):
+        """Initialize the HandlersManager."""
+        self._logger = logger
+        self._handlers = settings.handler_list
+        self._settings = settings
+
+    def setup(self):
+        """Set the logger handlers and settings."""
+        # Disable propagation to root logger to avoid duplicate logs
+        self._logger.propagate = False
+        self._logger.setLevel(self._settings.verbosity)
+
+        for handler_type in self._handlers:
+            if handler_type == "stdout":
+                self._add_stdout_handler()
+            elif handler_type == "stderr":
+                self._add_stderr_handler()
+            elif handler_type == "noop":
+                self._add_noop_handler()
+            elif handler_type == "file" and not self._settings.logging_suppress:
+                self._add_file_handler()
+            elif handler_type == "posthog" and not self._settings.logging_suppress:
+                self._add_posthog_handler()
+            else:
+                self._logger.debug("Unknown log handler.")
+
+    def _add_posthog_handler(self):
+        """Add a Posthog handler."""
+        handler = PosthogHandler(settings=self._settings)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        self._logger.addHandler(handler)
+
+    def _add_stdout_handler(self):
+        """Add a stdout handler."""
+        handler = logging.StreamHandler(sys.stdout)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        self._logger.addHandler(handler)
+
+    def _add_stderr_handler(self):
+        """Add a stderr handler."""
+        handler = logging.StreamHandler(sys.stderr)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        self._logger.addHandler(handler)
+
+    def _add_noop_handler(self):
+        """Add a null handler."""
+        handler = logging.NullHandler()
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        self._logger.addHandler(handler)
+
+    def _add_file_handler(self):
+        """Add a file handler."""
+        handler = PathTrackingFileHandler(settings=self._settings)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        self._logger.addHandler(handler)
+
+    def update_handlers(self, settings: LoggingSettings):
+        """Update the handlers with new settings."""
+        logger = self._logger
+        for hdlr in logger.handlers:
+            if (
+                isinstance(hdlr, (PathTrackingFileHandler, PosthogHandler))
+                and not settings.logging_suppress
+            ):
+                hdlr.settings = settings
+                hdlr.formatter.settings = settings  # type: ignore

openbb_platform/core/openbb_core/app/logs/logging_service.py

@@ -0,0 +1,260 @@
+"""Logging Service Module."""
+
+import json
+import logging
+from enum import Enum
+from types import TracebackType
+from typing import Any, Callable, Dict, Optional, Tuple, Type, Union
+
+from openbb_core.app.logs.formatters.formatter_with_exceptions import (
+    FormatterWithExceptions,
+)
+from openbb_core.app.logs.handlers_manager import HandlersManager
+from openbb_core.app.logs.models.logging_settings import LoggingSettings
+from openbb_core.app.model.abstract.singleton import SingletonMeta
+from openbb_core.app.model.system_settings import SystemSettings
+from openbb_core.app.model.user_settings import UserSettings
+from pydantic import BaseModel
+from pydantic_core import to_jsonable_python
+
+
+class DummyProvider(BaseModel):
+    """Dummy Provider for error handling with logs."""
+
+    provider: str = "not_passed_to_kwargs"
+
+
+class LoggingService(metaclass=SingletonMeta):
+    """Logging Service class responsible for managing logging settings and handling logs.
+
+    Attributes
+    ----------
+    _user_settings : Optional[UserSettings]
+        User Settings object.
+    _system_settings : Optional[SystemSettings]
+        System Settings object.
+    _logging_settings : LoggingSettings
+        LoggingSettings object containing the current logging settings.
+    _handlers_manager : HandlersManager
+        HandlersManager object managing logging handlers.
+
+    Methods
+    -------
+    __init__(system_settings, user_settings)
+        Logging Manager Constructor.
+
+    log(user_settings, system_settings, obbject, route, func, kwargs)
+        Log command output and relevant information.
+
+    logging_settings
+        Property to access the current logging settings.
+
+    logging_settings.setter
+        Setter method to update the logging settings.
+
+    _setup_handlers()
+        Setup Logging Handlers.
+
+    _log_startup()
+        Log startup information.
+    """
+
+    _logger = logging.getLogger("openbb.logging_service")
+
+    def __init__(
+        self,
+        system_settings: SystemSettings,
+        user_settings: UserSettings,
+    ) -> None:
+        """Define the Logging Service Constructor.
+
+        Sets up the logging settings and handlers and then logs the startup information.
+
+        Parameters
+        ----------
+        system_settings : SystemSettings
+            System Settings, by default None
+        user_settings : UserSettings
+            User Settings, by default None
+        """
+        self._user_settings = user_settings
+        self._system_settings = system_settings
+        self._logging_settings = LoggingSettings(
+            user_settings=self._user_settings,
+            system_settings=self._system_settings,
+        )
+        self._handlers_manager = self._setup_handlers()
+        self._log_startup()
+
+    @property
+    def logging_settings(self) -> LoggingSettings:
+        """Define the Current logging settings.
+
+        Returns
+        -------
+        LoggingSettings
+            LoggingSettings object containing the current logging settings.
+        """
+        return self._logging_settings
+
+    @logging_settings.setter
+    def logging_settings(self, value: Tuple[SystemSettings, UserSettings]):
+        """Define the Setter for updating the logging settings.
+
+        Parameters
+        ----------
+        value : Tuple[SystemSettings, UserSettings]
+            Tuple containing updated SystemSettings and UserSettings.
+        """
+        system_settings, user_settings = value
+        self._logging_settings = LoggingSettings(
+            user_settings=user_settings,
+            system_settings=system_settings,
+        )
+
+    def _setup_handlers(self) -> HandlersManager:
+        """Set up Logging Handlers.
+
+        Returns
+        -------
+        HandlersManager
+            Handlers Manager object.
+        """
+        handlers_manager = HandlersManager(
+            self._logger, settings=self._logging_settings
+        )
+        handlers_manager.setup()
+
+        self._logger.info("Logging configuration finished")
+        self._logger.info("Logging set to %s", self._logging_settings.handler_list)
+        self._logger.info("Verbosity set to %s", self._logging_settings.verbosity)
+        self._logger.info(
+            "LOGFORMAT: %s%s",
+            FormatterWithExceptions.LOGPREFIXFORMAT.replace("|", "-"),
+            FormatterWithExceptions.LOGFORMAT.replace("|", "-"),
+        )
+
+        return handlers_manager
+
+    def _log_startup(
+        self,
+        route: Optional[str] = None,
+        custom_headers: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log startup information."""
+
+        def check_credentials_defined(credentials: Dict[str, Any]):
+            class CredentialsDefinition(Enum):
+                defined = "defined"
+                undefined = "undefined"
+
+            return {
+                c: (
+                    CredentialsDefinition.defined.value
+                    if credentials[c]
+                    else CredentialsDefinition.undefined.value
+                )
+                for c in credentials
+            }
+
+        self._logger.info(
+            "STARTUP: %s ",
+            json.dumps(
+                {
+                    "route": route,
+                    "PREFERENCES": self._user_settings.preferences,
+                    "KEYS": check_credentials_defined(
+                        self._user_settings.credentials.model_dump()
+                        if self._user_settings.credentials
+                        else {}
+                    ),
+                    "SYSTEM": self._system_settings,
+                    "custom_headers": custom_headers,
+                },
+                default=to_jsonable_python,
+            ),
+        )
+
+    # pylint: disable=R0917
+    def log(
+        self,
+        user_settings: UserSettings,
+        system_settings: SystemSettings,
+        route: str,
+        func: Callable,
+        kwargs: Dict[str, Any],
+        exec_info: Union[
+            Tuple[Type[BaseException], BaseException, TracebackType],
+            Tuple[None, None, None],
+        ],
+        custom_headers: Optional[Dict[str, Any]] = None,
+    ):
+        """Log command output and relevant information.
+
+        Parameters
+        ----------
+        user_settings : UserSettings
+            User Settings object.
+        system_settings : SystemSettings
+            System Settings object.
+        route : str
+            Route for the command.
+        func : Callable
+            Callable representing the executed function.
+        kwargs : Dict[str, Any]
+            Keyword arguments passed to the function.
+        exec_info : Union[
+            Tuple[Type[BaseException], BaseException, TracebackType],
+            Tuple[None, None, None],
+        ]
+            Exception information, by default None
+        """
+        self._user_settings = user_settings
+        self._system_settings = system_settings
+        self._logging_settings = LoggingSettings(
+            user_settings=self._user_settings,
+            system_settings=self._system_settings,
+        )
+        self._handlers_manager.update_handlers(self._logging_settings)
+
+        if not self._logging_settings.logging_suppress:
+
+            if "login" in route:
+                self._log_startup(route, custom_headers)
+            else:
+
+                # Remove CommandContext if any
+                kwargs.pop("cc", None)
+
+                # Get provider for posthog logs
+                passed_model = kwargs.get("provider_choices", DummyProvider())
+                provider = (
+                    passed_model.provider
+                    if hasattr(passed_model, "provider")
+                    else "not_passed_to_kwargs"
+                )
+
+                # Truncate kwargs if too long
+                kwargs = {k: str(v)[:300] for k, v in kwargs.items()}
+                # Get execution info
+                error = None if all(i is None for i in exec_info) else str(exec_info[1])
+
+                # Construct message
+                message_label = "ERROR" if error else "CMD"
+                log_message = json.dumps(
+                    {
+                        "route": route,
+                        "input": kwargs,
+                        "error": error,
+                        "provider": provider,
+                        "custom_headers": custom_headers,
+                    },
+                    default=to_jsonable_python,
+                )
+                log_message = f"{message_label}: {log_message}"
+                log_level = self._logger.error if error else self._logger.info
+                log_level(
+                    log_message,
+                    extra={"func_name_override": func.__name__},
+                    exc_info=exec_info,
+                )

openbb_platform/core/openbb_core/app/logs/models/logging_settings.py

@@ -0,0 +1,58 @@
+"""Logging settings."""
+
+from pathlib import Path
+from typing import List, Optional
+
+from openbb_core.app.logs.utils.utils import get_app_id, get_log_dir, get_session_id
+from openbb_core.app.model.system_settings import SystemSettings
+from openbb_core.app.model.user_settings import UserSettings
+
+
+# pylint: disable=too-many-instance-attributes
+class LoggingSettings:
+    """Logging settings."""
+
+    def __init__(
+        self,
+        user_settings: Optional[UserSettings] = None,
+        system_settings: Optional[SystemSettings] = None,
+    ):
+        """Initialize the logging settings."""
+        user_settings = user_settings if user_settings is not None else UserSettings()
+        system_settings = (
+            system_settings if system_settings is not None else SystemSettings()
+        )
+        user_data_directory = (
+            str(Path.home() / "OpenBBUserData")
+            if not user_settings.preferences
+            else user_settings.preferences.data_directory
+        )
+        hub_session = (
+            user_settings.profile.hub_session if user_settings.profile else None
+        )
+        if hub_session:
+            user_id = hub_session.user_uuid
+            user_email = hub_session.email
+            user_primary_usage = hub_session.primary_usage
+        else:
+            user_id, user_email, user_primary_usage = None, None, None
+
+        # System
+        self.app_name: str = system_settings.logging_app_name
+        self.sub_app_name: str = system_settings.logging_sub_app
+        self.app_id: str = get_app_id(user_data_directory)
+        self.session_id: str = get_session_id()
+        self.frequency: str = system_settings.logging_frequency
+        self.handler_list: List[str] = system_settings.logging_handlers
+        self.rolling_clock: bool = system_settings.logging_rolling_clock
+        self.verbosity: int = system_settings.logging_verbosity
+        self.platform: str = system_settings.platform
+        self.python_version: str = system_settings.python_version
+        self.platform_version: str = system_settings.version
+        self.logging_suppress: bool = system_settings.logging_suppress
+        self.log_collect: bool = system_settings.log_collect
+        # User
+        self.user_id: Optional[str] = user_id
+        self.user_logs_directory: Path = get_log_dir(user_data_directory)
+        self.user_email: Optional[str] = user_email
+        self.user_primary_usage: Optional[str] = user_primary_usage

openbb_platform/core/openbb_core/app/logs/utils/expired_files.py

@@ -0,0 +1,30 @@
+"""Expired files management utilities."""
+
+import contextlib
+from datetime import datetime
+from pathlib import Path
+from typing import List
+
+
+def get_timestamp_from_x_days(x: int) -> float:
+    """Get the timestamp from x days ago."""
+    timestamp_from_x_days = datetime.now().timestamp() - x * 86400
+    return timestamp_from_x_days
+
+
+def get_expired_file_list(directory: Path, before_timestamp: float) -> List[Path]:
+    """Get the list of expired files from a directory."""
+    expired_files = []
+    if directory.is_dir():  # Check if the directory exists and is a directory
+        for file in directory.iterdir():
+            if file.is_file() and file.lstat().st_mtime < before_timestamp:
+                expired_files.append(file)
+
+    return expired_files
+
+
+def remove_file_list(file_list: List[Path]):
+    """Remove a list of files."""
+    for file in file_list:
+        with contextlib.suppress(PermissionError):
+            file.unlink(missing_ok=True)

openbb_platform/core/openbb_core/app/logs/utils/utils.py

@@ -0,0 +1,74 @@
+"""Utility functions for logging."""
+
+import time
+import uuid
+import warnings
+from pathlib import Path, PosixPath
+
+
+def get_session_id() -> str:
+    """UUID of the current session."""
+    session_id = str(uuid.uuid4()) + "-" + str(int(time.time()))
+    return session_id
+
+
+def get_app_id(contextual_user_data_directory: str) -> str:
+    """Get UUID of the current installation."""
+    try:
+        app_id = get_log_dir(contextual_user_data_directory).stem
+    except OSError as e:
+        if e.errno == 30:
+            warnings.warn("Please move the application into a writable location.")
+            warnings.warn(
+                "Note for macOS users: copy `OpenBB Terminal` folder outside the DMG."
+            )
+        raise e
+    except Exception as e:
+        raise e
+
+    return app_id
+
+
+def get_log_dir(contextual_user_data_directory: str) -> PosixPath:
+    """Retrieve application's log directory."""
+    log_dir = create_log_dir_if_not_exists(contextual_user_data_directory)
+    logging_uuid = create_log_uuid_if_not_exists(log_dir)
+    uuid_log_dir = create_uuid_dir_if_not_exists(log_dir, logging_uuid)
+
+    return uuid_log_dir
+
+
+def create_log_dir_if_not_exists(contextual_user_data_directory: str) -> Path:
+    """Create a log directory for the current installation."""
+    log_dir = Path(contextual_user_data_directory).joinpath("logs").absolute()
+    if not log_dir.is_dir():
+        log_dir.mkdir(parents=True, exist_ok=True)
+
+    return log_dir
+
+
+def create_log_uuid_if_not_exists(log_dir: Path) -> str:
+    """Create a log id file for the current logging session."""
+    log_id = get_log_id(log_dir)
+    if not log_id.is_file():
+        logging_id = f"{uuid.uuid4()}"
+        log_id.write_text(logging_id, encoding="utf-8")
+    else:
+        logging_id = log_id.read_text(encoding="utf-8").rstrip()
+
+    return logging_id
+
+
+def get_log_id(log_dir):
+    """Get the log id file."""
+    return (log_dir / ".logid").absolute()
+
+
+def create_uuid_dir_if_not_exists(log_dir, logging_id) -> PosixPath:
+    """Create a directory for the current logging session."""
+    uuid_log_dir = (log_dir / logging_id).absolute()
+
+    if not uuid_log_dir.is_dir():
+        uuid_log_dir.mkdir(parents=True, exist_ok=True)
+
+    return uuid_log_dir

openbb_platform/core/openbb_core/app/model/__init__.py

@@ -0,0 +1 @@
+"""OpenBB Core App Model."""

openbb_platform/core/openbb_core/app/model/abstract/__init__.py

@@ -0,0 +1 @@
+"""OpenBB Core App Abstract Model."""

openbb_platform/core/openbb_core/app/model/abstract/error.py

@@ -0,0 +1,12 @@
+"""OpenBB Error."""
+
+from typing import Optional, Union
+
+
+class OpenBBError(Exception):
+    """OpenBB Error."""
+
+    def __init__(self, original: Optional[Union[str, Exception]] = None):
+        """Initialize the OpenBBError."""
+        self.original = original
+        super().__init__(str(original))

openbb_platform/core/openbb_core/app/model/abstract/results.py

@@ -0,0 +1,5 @@
+"""OpenBB Core App Model Abstract Results."""
+
+from pydantic import BaseModel
+
+Results = BaseModel

openbb_platform/core/openbb_core/app/model/abstract/singleton.py

@@ -0,0 +1,20 @@
+"""Singleton metaclass implementation."""
+
+from typing import Dict, Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class SingletonMeta(type, Generic[T]):
+    """Singleton metaclass."""
+
+    # TODO : check if we want to update this to be thread safe
+    _instances: Dict[T, T] = {}
+
+    def __call__(cls: "SingletonMeta", *args, **kwargs):
+        """Singleton pattern implementation."""
+        if cls not in cls._instances:
+            instance = super().__call__(*args, **kwargs)
+            cls._instances[cls] = instance
+
+        return cls._instances[cls]

openbb_platform/core/openbb_core/app/model/abstract/tagged.py

@@ -0,0 +1,10 @@
+"""OpenBB Core App Abstract Model Tagged."""
+
+from pydantic import BaseModel, Field
+from uuid_extensions import uuid7str  # type: ignore
+
+
+class Tagged(BaseModel):
+    """Model for Tagged."""
+
+    id: str = Field(default_factory=uuid7str, alias="_id")

openbb_platform/core/openbb_core/app/model/abstract/warning.py

@@ -0,0 +1,24 @@
+"""Module for warnings."""
+
+from warnings import WarningMessage
+
+from pydantic import BaseModel
+
+
+class Warning_(BaseModel):
+    """Model for Warning."""
+
+    category: str
+    message: str
+
+
+def cast_warning(w: WarningMessage) -> Warning_:
+    """Cast a warning to a pydantic model."""
+    return Warning_(
+        category=w.category.__name__,
+        message=str(w.message),
+    )
+
+
+class OpenBBWarning(Warning):
+    """Base class for OpenBB warnings."""

openbb_platform/core/openbb_core/app/model/api_settings.py

@@ -0,0 +1,57 @@
+"""FastAPI configuration settings model."""
+
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, computed_field
+
+
+class Cors(BaseModel):
+    """Cors model for FastAPI configuration."""
+
+    model_config = ConfigDict(frozen=True)
+
+    allow_origins: List[str] = Field(default_factory=lambda: ["*"])
+    allow_methods: List[str] = Field(default_factory=lambda: ["*"])
+    allow_headers: List[str] = Field(default_factory=lambda: ["*"])
+
+
+class Servers(BaseModel):
+    """Servers model for FastAPI configuration."""
+
+    model_config = ConfigDict(frozen=True)
+
+    url: str = ""
+    description: str = "Local OpenBB development server"
+
+
+class APISettings(BaseModel):
+    """Settings model for FastAPI configuration."""
+
+    model_config = ConfigDict(frozen=True)
+
+    version: str = "1"
+    title: str = "OpenBB Platform API"
+    description: str = "Investment research for everyone, anywhere."
+    terms_of_service: str = "http://example.com/terms/"
+    contact_name: str = "OpenBB Team"
+    contact_url: str = "https://openbb.co"
+    contact_email: str = "hello@openbb.co"
+    license_name: str = "AGPLv3"
+    license_url: str = "https://github.com/OpenBB-finance/OpenBB/blob/develop/LICENSE"
+    servers: List[Servers] = Field(default_factory=lambda: [Servers()])
+    cors: Cors = Field(default_factory=Cors)
+    custom_headers: Optional[Dict[str, str]] = Field(
+        default=None, description="Custom headers and respective default value."
+    )
+
+    @computed_field  # type: ignore[misc]
+    @property
+    def prefix(self) -> str:
+        """Return the API prefix."""
+        return f"/api/v{self.version}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of the model."""
+        return f"{self.__class__.__name__}\n\n" + "\n".join(
+            f"{k}: {v}" for k, v in self.model_dump().items()
+        )

openbb_platform/core/openbb_core/app/model/charts/chart.py

@@ -0,0 +1,30 @@
+"""OpenBB Core Chart model."""
+
+from typing import Any, Dict, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class Chart(BaseModel):
+    """Model for Chart."""
+
+    content: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="Raw textual representation of the chart.",
+    )
+    format: Optional[str] = Field(
+        default=None,
+        description="Complementary attribute to the `content` attribute. It specifies the format of the chart.",
+    )
+    fig: Optional[Any] = Field(
+        default=None,
+        description="The figure object.",
+        json_schema_extra={"exclude_from_api": True},
+    )
+    model_config = ConfigDict(validate_assignment=True)
+
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return f"{self.__class__.__name__}\n\n" + "\n".join(
+            f"{k}: {v}" for k, v in self.model_dump().items()
+        )

openbb_platform/core/openbb_core/app/model/charts/charting_settings.py

@@ -0,0 +1,62 @@
+"""Charting settings."""
+
+import importlib
+from pathlib import Path
+from typing import TYPE_CHECKING, Optional
+
+from openbb_core.app.logs.utils.utils import get_app_id
+from openbb_core.env import Env
+
+if TYPE_CHECKING:
+    from openbb_core.app.model.system_settings import SystemSettings
+    from openbb_core.app.model.user_settings import UserSettings
+
+
+# pylint: disable=too-many-instance-attributes
+class ChartingSettings:
+    """Charting settings."""
+
+    def __init__(
+        self,
+        user_settings: Optional["UserSettings"] = None,
+        system_settings: Optional["SystemSettings"] = None,
+    ):
+        """Initialize charting settings."""
+        user_settings_module = importlib.import_module(
+            "openbb_core.app.model.user_settings", "UserSettings"
+        )
+        system_settings_module = importlib.import_module(
+            "openbb_core.app.model.system_settings", "SystemSettings"
+        )
+
+        UserSettings = user_settings_module.UserSettings
+        SystemSettings = system_settings_module.SystemSettings
+        user_settings = user_settings or UserSettings()
+        system_settings = system_settings or SystemSettings()
+
+        user_data_directory = (
+            str(Path.home() / "OpenBBUserData")
+            if not user_settings.preferences
+            else user_settings.preferences.data_directory
+        )
+
+        # System
+        self.log_collect: bool = system_settings.log_collect
+        self.version: str = system_settings.version
+        self.python_version: str = system_settings.python_version
+        self.test_mode = system_settings.test_mode
+        self.app_id: str = get_app_id(user_data_directory)
+        self.debug_mode: bool = system_settings.debug_mode or Env().DEBUG_MODE
+        self.headless: bool = system_settings.headless
+        # User
+        self.user_email: Optional[str] = getattr(
+            user_settings.profile.hub_session, "email", None
+        )
+        self.user_uuid: Optional[str] = getattr(
+            user_settings.profile.hub_session, "user_uuid", None
+        )
+        self.user_exports_directory = user_settings.preferences.export_directory
+        self.user_styles_directory = user_settings.preferences.user_styles_directory
+        # Theme
+        self.chart_style: str = user_settings.preferences.chart_style
+        self.table_style = user_settings.preferences.table_style

openbb_platform/core/openbb_core/app/model/command_context.py

@@ -0,0 +1,12 @@
+"""Command Context."""
+
+from openbb_core.app.model.system_settings import SystemSettings
+from openbb_core.app.model.user_settings import UserSettings
+from pydantic import BaseModel, Field
+
+
+class CommandContext(BaseModel):
+    """Command Context."""
+
+    user_settings: UserSettings = Field(default_factory=UserSettings)
+    system_settings: SystemSettings = Field(default_factory=SystemSettings)

openbb_platform/core/openbb_core/app/model/credentials.py

@@ -0,0 +1,146 @@
+"""Credentials model and its utilities."""
+
+import json
+import traceback
+import warnings
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+from openbb_core.app.constants import USER_SETTINGS_PATH
+from openbb_core.app.extension_loader import ExtensionLoader
+from openbb_core.app.model.abstract.warning import OpenBBWarning
+from openbb_core.app.provider_interface import ProviderInterface
+from openbb_core.env import Env
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    SecretStr,
+    create_model,
+)
+from pydantic.functional_serializers import PlainSerializer
+from typing_extensions import Annotated
+
+
+class LoadingError(Exception):
+    """Error loading extension."""
+
+
+# @model_serializer blocks model_dump with pydantic parameters (include, exclude)
+OBBSecretStr = Annotated[
+    SecretStr,
+    PlainSerializer(
+        lambda x: x.get_secret_value(), return_type=str, when_used="json-unless-none"
+    ),
+]
+
+
+class CredentialsLoader:
+    """Here we create the Credentials model."""
+
+    credentials: Dict[str, List[str]] = {}
+
+    def format_credentials(self, additional: dict) -> Dict[str, Tuple[object, None]]:
+        """Prepare credentials map to be used in the Credentials model."""
+        formatted: Dict[str, Tuple[object, None]] = {}
+        for c_origin, c_list in self.credentials.items():
+            for c_name in c_list:
+                if c_name in formatted:
+                    warnings.warn(
+                        message=f"Skipping '{c_name}', credential already in use.",
+                        category=OpenBBWarning,
+                    )
+                    continue
+                formatted[c_name] = (
+                    Optional[OBBSecretStr],
+                    Field(default=None, description=c_origin, alias=c_name.upper()),
+                )
+
+        if additional:
+            for key, value in additional.items():
+                if key in formatted:
+                    continue
+                formatted[key] = (
+                    Optional[OBBSecretStr],
+                    Field(default=value, description=key, alias=key.upper()),
+                )
+
+        return dict(sorted(formatted.items()))
+
+    def from_obbject(self) -> None:
+        """Load credentials from OBBject extensions."""
+        for ext_name, ext in ExtensionLoader().obbject_objects.items():  # type: ignore[attr-defined]
+            try:
+                if ext_name in self.credentials:
+                    warnings.warn(
+                        message=f"Skipping '{ext_name}', name already in user.",
+                        category=OpenBBWarning,
+                    )
+                    continue
+                self.credentials[ext_name] = ext.credentials
+            except Exception as e:
+                msg = f"Error loading extension: {ext_name}\n"
+                if Env().DEBUG_MODE:
+                    traceback.print_exception(type(e), e, e.__traceback__)
+                    raise LoadingError(msg + f"\033[91m{e}\033[0m") from e
+                warnings.warn(
+                    message=msg,
+                    category=OpenBBWarning,
+                )
+
+    def from_providers(self) -> None:
+        """Load credentials from providers."""
+        self.credentials = ProviderInterface().credentials
+
+    def load(self) -> BaseModel:
+        """Load credentials from providers."""
+        # We load providers first to give them priority choosing credential names
+        self.from_providers()
+        self.from_obbject()
+        path = Path(USER_SETTINGS_PATH)
+        additional: dict = {}
+
+        if path.exists():
+            with open(USER_SETTINGS_PATH, encoding="utf-8") as f:
+                data = json.load(f)
+                if "credentials" in data:
+                    additional = data["credentials"]
+
+        model = create_model(
+            "Credentials",
+            __config__=ConfigDict(validate_assignment=True, populate_by_name=True),
+            **self.format_credentials(additional),  # type: ignore
+        )
+        model.origins = self.credentials
+        return model
+
+
+_Credentials = CredentialsLoader().load()
+
+
+class Credentials(_Credentials):  # type: ignore
+    """Credentials model used to store provider credentials."""
+
+    model_config = ConfigDict(extra="allow")
+
+    def __repr__(self) -> str:
+        """Define the string representation of the credentials."""
+        return (
+            self.__class__.__name__
+            + "\n\n"
+            + "\n".join([f"{k}: {v}" for k, v in sorted(self.__dict__.items())])
+        )
+
+    def show(self):
+        """Unmask credentials and print them."""
+        print(  # noqa: T201
+            self.__class__.__name__
+            + "\n\n"
+            + "\n".join(
+                [f"{k}: {v}" for k, v in sorted(self.model_dump(mode="json").items())]
+            )
+        )
+
+    def update(self, incoming: "Credentials"):
+        """Update current credentials."""
+        self.__dict__.update(incoming.model_dump(exclude_none=True))

openbb_platform/core/openbb_core/app/model/defaults.py

@@ -0,0 +1,56 @@
+"""Defaults model."""
+
+from typing import Any
+from warnings import warn
+
+from openbb_core.app.model.abstract.warning import OpenBBWarning
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+
+class Defaults(BaseModel):
+    """Defaults."""
+
+    model_config = ConfigDict(validate_assignment=True, populate_by_name=True)
+
+    commands: dict[str, dict[str, Any]] = Field(
+        default_factory=dict,
+        alias="routes",
+    )
+
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return f"{self.__class__.__name__}\n\n" + "\n".join(
+            f"{k}: {v}" for k, v in self.model_dump().items()
+        )
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_before(cls, values: dict) -> dict:
+        """Validate model (before)."""
+        key = "commands"
+        if "routes" in values:
+            if not values.get("routes"):
+                del values["routes"]
+            show_warnings = values.get("preferences", {}).get("show_warnings")
+            if show_warnings is False or show_warnings in ["False", "false"]:
+                warn(
+                    message="The 'routes' key is deprecated within 'defaults' of 'user_settings.json'."
+                    + " Suppress this warning by updating the key to 'commands'.",
+                    category=OpenBBWarning,
+                )
+                key = "routes"
+
+        new_values: dict = {"commands": {}}
+        for k, v in values.get(key, {}).items():
+            clean_k = k.strip("/").replace("/", ".")
+            provider = v.get("provider") if v else None
+            if isinstance(provider, str):
+                v["provider"] = [provider]
+            new_values["commands"][clean_k] = v
+
+        return new_values
+
+    def update(self, incoming: "Defaults"):
+        """Update current defaults."""
+        incoming_commands = incoming.model_dump(exclude_none=True).get("commands", {})
+        self.__dict__["commands"].update(incoming_commands)

openbb_platform/core/openbb_core/app/model/example.py

@@ -0,0 +1,229 @@
+"""Example class to represent endpoint examples."""
+
+from abc import abstractmethod
+from datetime import date, datetime, timedelta
+from typing import Any, Dict, List, Literal, Optional, Union, _GenericAlias  # type: ignore
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    computed_field,
+    model_validator,
+)
+
+QUOTE_TYPES = {str, date}
+
+
+class Example(BaseModel):
+    """Example model."""
+
+    scope: str
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    @abstractmethod
+    def to_python(self, **kwargs) -> str:
+        """Return a Python code representation of the example."""
+
+
+class APIEx(Example):
+    """API Example model."""
+
+    scope: Literal["api"] = "api"
+    description: Optional[str] = Field(
+        default=None, description="Optional description unless more than 3 parameters"
+    )
+    parameters: Dict[str, Union[str, int, float, bool, List[str], List[Dict[str, Any]]]]
+
+    @computed_field  # type: ignore[misc]
+    @property
+    def provider(self) -> Optional[str]:
+        """Return the provider from the parameters."""
+        return self.parameters.get("provider")  # type: ignore
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_model(cls, values: dict) -> dict:
+        """Validate model."""
+        parameters = values.get("parameters", {})
+        if "provider" not in parameters and "data" not in parameters:
+            raise ValueError("API example must specify a provider.")
+
+        provider = parameters.get("provider")
+        if provider and not isinstance(provider, str):
+            raise ValueError("Provider must be a string.")
+
+        return values
+
+    @staticmethod
+    def _unpack_type(type_: type) -> set:
+        """Unpack types from types, example Union[List[str], int] -> {typing._GenericAlias, int}."""
+        if (
+            hasattr(type_, "__args__")
+            and type(type_)  # pylint: disable=unidiomatic-typecheck
+            is not _GenericAlias
+        ):
+            return set().union(*map(APIEx._unpack_type, type_.__args__))  # type: ignore
+        return {type_} if isinstance(type_, type) else {type(type_)}
+
+    @staticmethod
+    def _shift(i: int) -> float:
+        """Return a transformation of the integer."""
+        return 2 * (i + 1) / (2 * i) % 1 + 1
+
+    @staticmethod
+    def mock_data(
+        dataset: Literal["timeseries", "panel"],
+        size: int = 5,
+        sample: Optional[Dict[str, Any]] = None,
+        multiindex: Optional[Dict[str, Any]] = None,
+    ) -> List[Dict]:
+        """Generate mock data from a sample.
+
+        Parameters
+        ----------
+        dataset : str
+            The type of data to return:
+            - 'timeseries': Time series data
+            - 'panel': Panel data (multiindex)
+
+        size : int
+            The size of the data to return, default is 5.
+        sample : Optional[Dict[str, Any]], optional
+            A sample of the data to return, by default None.
+        multiindex_names : Optional[List[str]], optional
+            The names of the multiindex, by default None.
+
+        Timeseries default sample:
+        {
+            "date": "2023-01-01",
+            "open": 110.0,
+            "high": 120.0,
+            "low": 100.0,
+            "close": 115.0,
+            "volume": 10000,
+        }
+
+        Panel default sample:
+        {
+            "portfolio_value": 100000,
+            "risk_free_rate": 0.02,
+        }
+        multiindex: {"asset_manager": "AM", "time": 0}
+
+        Returns
+        -------
+        List[Dict]
+            A list of dictionaries with the mock data.
+        """
+        if dataset == "timeseries":
+            sample = sample or {
+                "date": "2023-01-01",
+                "open": 110.0,
+                "high": 120.0,
+                "low": 100.0,
+                "close": 115.0,
+                "volume": 10000,
+            }
+            result = []
+            for i in range(1, size + 1):
+                s = APIEx._shift(i)
+                obs = {}
+                for k, v in sample.items():
+                    if k == "date":
+                        obs[k] = (
+                            datetime.strptime(v, "%Y-%m-%d") + timedelta(days=i)
+                        ).strftime("%Y-%m-%d")
+                    else:
+                        obs[k] = round(v * s, 2)
+                result.append(obs)
+            return result
+        if dataset == "panel":
+            sample = sample or {
+                "portfolio_value": 100000.0,
+                "risk_free_rate": 0.02,
+            }
+            multiindex = multiindex or {"asset_manager": "AM", "time": 0}
+            multiindex_names = list(multiindex.keys())
+            idx_1 = multiindex_names[0]
+            idx_2 = multiindex_names[1]
+            items_per_idx = 2
+            item: Dict[str, Any] = {
+                "is_multiindex": True,
+                "multiindex_names": str(multiindex_names),
+            }
+            # Iterate over the number of items to create and add them to the result
+            result = []
+            for i in range(1, size + 1):
+                item[idx_1] = f"{idx_1}_{i}"
+                for j in range(items_per_idx):
+                    item[idx_2] = j
+                    for k, v in sample.items():
+                        if isinstance(v, str):
+                            item[k] = f"{v}_{j}"
+                        else:
+                            item[k] = round(v * APIEx._shift(i + j), 2)
+                    result.append(item.copy())
+            return result
+        raise ValueError(f"Dataset '{dataset}' not found.")
+
+    def to_python(self, **kwargs) -> str:
+        """Return a Python code representation of the example."""
+        indentation = kwargs.get("indentation", "")
+        func_path = kwargs.get("func_path", ".func_router.func_name")
+        param_types: Dict[str, type] = kwargs.get("param_types", {})
+        prompt = kwargs.get("prompt", "")
+
+        eg = ""
+        if self.description:
+            eg += f"{indentation}{prompt}# {self.description}\n"
+
+        eg += f"{indentation}{prompt}obb{func_path}("
+        for k, v in self.parameters.items():
+            if k in param_types and (type_ := param_types.get(k)):
+                if QUOTE_TYPES.intersection(self._unpack_type(type_)):
+                    eg += f"{k}='{v}', "
+                else:
+                    eg += f"{k}={v}, "
+            else:
+                eg += f"{k}={v}, "
+
+        eg = indentation + eg.strip(", ") + ")\n"
+
+        return eg
+
+
+class PythonEx(Example):
+    """Python Example model."""
+
+    scope: Literal["python"] = "python"
+    description: str
+    code: List[str]
+
+    def to_python(self, **kwargs) -> str:
+        """Return a Python code representation of the example."""
+        indentation = kwargs.get("indentation", "")
+        prompt = kwargs.get("prompt", "")
+
+        eg = ""
+        if self.description:
+            eg += f"{indentation}{prompt}# {self.description}\n"
+
+        for line in self.code:
+            eg += f"{indentation}{prompt}{line}\n"
+
+        return eg
+
+
+def filter_list(
+    examples: List[Example],
+    providers: List[str],
+) -> List[Example]:
+    """Filter list of examples."""
+    return [
+        e
+        for e in examples
+        if (isinstance(e, APIEx) and (not e.provider or e.provider in providers))
+        or e.scope != "api"
+    ]

openbb_platform/core/openbb_core/app/model/extension.py

@@ -0,0 +1,79 @@
+"""Extension class for OBBject extensions."""
+
+import warnings
+from typing import Callable, List, Optional
+
+
+class Extension:
+    """
+    Serves as OBBject extension entry point and must be created by each extension package.
+
+    See https://docs.openbb.co/platform/development/developer-guidelines/obbject_extensions.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        credentials: Optional[List[str]] = None,
+        description: Optional[str] = None,
+    ) -> None:
+        """Initialize the extension.
+
+        Parameters
+        ----------
+        name : str
+            Name of the extension.
+        credentials : Optional[List[str]], optional
+            List of required credentials, by default None
+        description: Optional[str]
+            Extension description.
+        """
+        self.name = name
+        self.credentials = credentials or []
+        self.description = description
+
+    @property
+    def obbject_accessor(self) -> Callable:
+        """Extend an OBBject, inspired by pandas."""
+        # pylint: disable=import-outside-toplevel
+        # Avoid circular imports
+
+        from openbb_core.app.model.obbject import OBBject
+
+        return self.register_accessor(self.name, OBBject)
+
+    @staticmethod
+    def register_accessor(name, cls) -> Callable:
+        """Register a custom accessor."""
+
+        def decorator(accessor):
+            if hasattr(cls, name):
+                warnings.warn(
+                    f"registration of accessor '{repr(accessor)}' under name "
+                    f"'{repr(name)}' for type '{repr(cls)}' is overriding a preexisting "
+                    f"attribute with the same name.",
+                    UserWarning,
+                )
+            setattr(cls, name, CachedAccessor(name, accessor))
+            # pylint: disable=protected-access
+            cls.accessors.add(name)
+            return accessor
+
+        return decorator
+
+
+class CachedAccessor:
+    """CachedAccessor."""
+
+    def __init__(self, name: str, accessor) -> None:
+        """Initialize the cached accessor."""
+        self._name = name
+        self._accessor = accessor
+
+    def __get__(self, obj, cls):
+        """Get the cached accessor."""
+        if obj is None:
+            return self._accessor
+        accessor_obj = self._accessor(obj)
+        object.__setattr__(obj, self._name, accessor_obj)
+        return accessor_obj

openbb_platform/core/openbb_core/app/model/field.py

@@ -0,0 +1,28 @@
+"""Custom field for OpenBB."""
+
+from typing import Any, List, Optional
+
+from pydantic.fields import FieldInfo
+
+
+class OpenBBField(FieldInfo):
+    """Custom field for OpenBB."""
+
+    def __repr__(self):
+        """Override FieldInfo __repr__."""
+        # We use repr() to avoid decoding special characters like \n
+        if self.choices:
+            return f"OpenBBField(description={repr(self.description)}, choices={repr(self.choices)})"
+        return f"OpenBBField(description={repr(self.description)})"
+
+    def __init__(self, description: str, choices: Optional[List[Any]] = None):
+        """Initialize OpenBBField."""
+        json_schema_extra = {"choices": choices} if choices else None
+        super().__init__(description=description, json_schema_extra=json_schema_extra)  # type: ignore[arg-type]
+
+    @property
+    def choices(self) -> Optional[List[Any]]:
+        """Custom choices."""
+        if self.json_schema_extra:
+            return self.json_schema_extra.get("choices")  # type: ignore[union-attr,return-value]
+        return None

openbb_platform/core/openbb_core/app/model/hub/hub_session.py

@@ -0,0 +1,27 @@
+"""Model for HubSession."""
+
+from typing import Optional
+
+from pydantic import BaseModel, SecretStr, field_serializer
+
+
+class HubSession(BaseModel):
+    """Model for HubSession."""
+
+    username: Optional[str] = None
+    email: str
+    primary_usage: str
+    user_uuid: str
+    token_type: str
+    access_token: SecretStr
+
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return f"{self.__class__.__name__}\n\n" + "\n".join(
+            f"{k}: {v}" for k, v in self.model_dump().items()
+        )
+
+    @field_serializer("access_token", when_used="json-unless-none")
+    def _dump_secret(self, v):
+        """Dump secret."""
+        return v.get_secret_value()

openbb_platform/core/openbb_core/app/model/hub/hub_user_settings.py

@@ -0,0 +1,22 @@
+"""Hub user settings model."""
+
+from typing import Any, Dict, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+class HubUserSettings(BaseModel):
+    """Hub user settings model."""
+
+    features_settings: Dict[str, Any] = Field(default_factory=dict)
+    features_keys: Dict[str, Optional[str]] = Field(default_factory=dict)
+    # features_sources: Dict[str, Any]
+    # features_terminal_style: Dict[str, Union[str, Dict[str, str]]]
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    @field_validator("features_keys", mode="before", check_fields=False)
+    @classmethod
+    def to_lower(cls, d: dict) -> dict:
+        """Convert dict keys to lowercase."""
+        return {k.lower(): v for k, v in d.items()}

openbb_platform/core/openbb_core/app/model/metadata.py

@@ -0,0 +1,142 @@
+"""Metadata model."""
+
+from datetime import datetime
+from typing import Any, Dict, Optional, Sequence, Union
+
+from openbb_core.provider.abstract.data import Data
+from pydantic import BaseModel, Field, field_validator
+
+
+class Metadata(BaseModel):
+    """Metadata of a command execution."""
+
+    arguments: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Arguments of the command.",
+    )
+    duration: int = Field(
+        description="Execution duration in nano second of the command."
+    )
+    route: str = Field(description="Route of the command.")
+    timestamp: datetime = Field(description="Execution starting timestamp.")
+
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return f"{self.__class__.__name__}\n\n" + "\n".join(
+            f"{k}: {v}" for k, v in self.model_dump().items()
+        )
+
+    @field_validator("arguments")
+    @classmethod
+    def scale_arguments(cls, v):
+        """Scale arguments.
+
+        This function is meant to limit the size of the input arguments of a command.
+        If the type is one of the following: `Data`, `List[Data]`, `DataFrame`, `List[DataFrame]`,
+        `Series`, `List[Series]` or `ndarray`, the value of the argument is swapped by a dictionary
+        containing the type and the columns. If the type is not one of the previous, the
+        value is kept or trimmed to 80 characters.
+        """
+        # pylint: disable=import-outside-toplevel
+        from inspect import isclass  # noqa
+        from numpy import ndarray  # noqa
+        from pandas import DataFrame, Series  # noqa
+
+        arguments: Dict[str, Any] = {}
+        for item in ["provider_choices", "standard_params", "extra_params"]:
+            arguments[item] = {}
+            # The item could be class or it could a dictionary.
+            v_item = (
+                v.__dict__.get(item, {}) if not isinstance(v, dict) else v.get(item, {})
+            )
+            # The item might not be a dictionary yet.
+            v_item = v_item if isinstance(v_item, dict) else v_item.__dict__
+            for arg, arg_val in v_item.items():
+                new_arg_val: Optional[Union[str, dict[str, Sequence[Any]]]] = None
+
+                # Data
+                if isclass(type(arg_val)) and issubclass(type(arg_val), Data):
+                    new_arg_val = {
+                        "type": f"{type(arg_val).__name__}",
+                        "columns": list(arg_val.model_dump().keys()),
+                    }
+
+                # List[Data]
+                if isinstance(arg_val, list) and issubclass(type(arg_val[0]), Data):
+                    _columns = [list(d.model_dump().keys()) for d in arg_val]
+                    ld_columns = (
+                        item for sublist in _columns for item in sublist
+                    )  # flatten
+                    new_arg_val = {
+                        "type": f"List[{type(arg_val[0]).__name__}]",
+                        "columns": list(set(ld_columns)),
+                    }
+
+                # DataFrame
+                elif isinstance(arg_val, DataFrame):
+                    df_columns = (
+                        list(arg_val.index.names) + arg_val.columns.tolist()
+                        if any(index is not None for index in list(arg_val.index.names))
+                        else arg_val.columns.tolist()
+                    )
+                    new_arg_val = {
+                        "type": f"{type(arg_val).__name__}",
+                        "columns": df_columns,
+                    }
+
+                # List[DataFrame]
+                elif isinstance(arg_val, list) and issubclass(
+                    type(arg_val[0]), DataFrame
+                ):
+                    ldf_columns = [
+                        (
+                            list(df.index.names) + df.columns.tolist()
+                            if any(index is not None for index in list(df.index.names))
+                            else df.columns.tolist()
+                        )
+                        for df in arg_val
+                    ]
+                    new_arg_val = {
+                        "type": f"List[{type(arg_val[0]).__name__}]",
+                        "columns": ldf_columns,
+                    }
+
+                # Series
+                elif isinstance(arg_val, Series):
+                    new_arg_val = {
+                        "type": f"{type(arg_val).__name__}",
+                        "columns": list(arg_val.index.names) + [arg_val.name],
+                    }
+
+                # List[Series]
+                elif isinstance(arg_val, list) and isinstance(arg_val[0], Series):
+                    ls_columns = [
+                        (
+                            list(series.index.names) + [series.name]
+                            if any(
+                                index is not None for index in list(series.index.names)
+                            )
+                            else series.name
+                        )
+                        for series in arg_val
+                    ]
+                    new_arg_val = {
+                        "type": f"List[{type(arg_val[0]).__name__}]",
+                        "columns": ls_columns,
+                    }
+
+                # ndarray
+                elif isinstance(arg_val, ndarray):
+                    new_arg_val = {
+                        "type": f"{type(arg_val).__name__}",
+                        "columns": list(arg_val.dtype.names or []),
+                    }
+
+                else:
+                    str_repr_arg_val = str(arg_val)
+                    if len(str_repr_arg_val) > 80:
+                        new_arg_val = str_repr_arg_val[:80]
+
+                arguments[item][arg] = new_arg_val or arg_val
+
+        return arguments