[Feature] Add Currency Snapshots With Data From FMP (#6162)

* add obb.currency.snapshots() endpoint and create new standard model

* add standard model

* static assets

* add couple of comments

* codespell

---------

Co-authored-by: Igor Radovanovic <74266147+IgorWounds@users.noreply.github.com>

openbb_platform/core/openbb_core/provider/standard_models/currency_snapshots.py

@@ -0,0 +1,82 @@
+"""Currency Snapshots Standard Model."""
+
+from typing import List, Literal, Optional, Union
+
+from pydantic import Field, field_validator
+
+from openbb_core.provider.abstract.data import Data
+from openbb_core.provider.abstract.query_params import QueryParams
+from openbb_core.provider.utils.descriptions import DATA_DESCRIPTIONS
+
+
+class CurrencySnapshotsQueryParams(QueryParams):
+    """Currency Snapshots Query Params."""
+
+    base: str = Field(description="The base currency symbol.", default="usd")
+    quote_type: Literal["direct", "indirect"] = Field(
+        description="Whether the quote is direct or indirect."
+        + " Selecting 'direct' will return the exchange rate"
+        + " as the amount of domestic currency required to buy one unit"
+        + " of the foreign currency."
+        + " Selecting 'indirect' (default) will return the exchange rate"
+        + " as the amount of foreign currency required to buy one unit"
+        + " of the domestic currency.",
+        default="indirect",
+    )
+    counter_currencies: Optional[Union[str, List[str]]] = Field(
+        description="An optional list of counter currency symbols to filter for."
+        + " None returns all.",
+        default=None,
+    )
+
+    @field_validator("base", mode="before", check_fields=False)
+    @classmethod
+    def to_upper(cls, v):
+        """Convert the base currency to uppercase."""
+        return v.upper()
+
+    @field_validator("counter_currencies", mode="before", check_fields=False)
+    @classmethod
+    def convert_string(cls, v):
+        """Convert the counter currencies to an upper case string list."""
+        if v is not None:
+            return ",".join(v).upper() if isinstance(v, list) else v.upper()
+        return None
+
+
+class CurrencySnapshotsData(Data):
+    """Currency Snapshots Data."""
+
+    base_currency: str = Field(description="The base, or domestic, currency.")
+    counter_currency: str = Field(description="The counter, or foreign, currency.")
+    last_rate: float = Field(
+        description="The exchange rate, relative to the base currency."
+        + " Rates are expressed as the amount of foreign currency"
+        + " received from selling one unit of the base currency,"
+        + " or the quantity of foreign currency required to purchase"
+        + " one unit of the domestic currency."
+        + " To inverse the perspective, set the 'quote_type' parameter as 'direct'.",
+    )
+    open: Optional[float] = Field(
+        description=DATA_DESCRIPTIONS.get("open", ""),
+        default=None,
+    )
+    high: Optional[float] = Field(
+        description=DATA_DESCRIPTIONS.get("high", ""),
+        default=None,
+    )
+    low: Optional[float] = Field(
+        description=DATA_DESCRIPTIONS.get("low", ""),
+        default=None,
+    )
+    close: Optional[float] = Field(
+        description=DATA_DESCRIPTIONS.get("close", ""),
+        default=None,
+    )
+    volume: Optional[int] = Field(
+        description=DATA_DESCRIPTIONS.get("volume", ""), default=None
+    )
+    prev_close: Optional[float] = Field(
+        description=DATA_DESCRIPTIONS.get("prev_close", ""),
+        default=None,
+    )

openbb_platform/extensions/currency/integration/test_currency_api.py

@@ -165,3 +165,27 @@ def test_currency_reference_rates(params, headers):
     result = requests.get(url, headers=headers, timeout=10)
     assert isinstance(result, requests.Response)
     assert result.status_code == 200
+
+
+@parametrize(
+    "params",
+    [
+        (
+            {
+                "provider": "fmp",
+                "base": "USD,XAU",
+                "counter_currencies": "EUR,JPY,GBP",
+                "quote_type": "indirect",
+            }
+        ),
+    ],
+)
+@pytest.mark.integration
+def test_currency_snapshots(params, headers):
+    params = {p: v for p, v in params.items() if v}
+
+    query_str = get_querystring(params, [])
+    url = f"http://0.0.0.0:8000/api/v1/currency/snapshots?{query_str}"
+    result = requests.get(url, headers=headers, timeout=10)
+    assert isinstance(result, requests.Response)
+    assert result.status_code == 200

openbb_platform/extensions/currency/integration/test_currency_python.py

@@ -154,3 +154,24 @@ def test_currency_reference_rates(params, obb):
     assert result
     assert isinstance(result, OBBject)
     assert len(result.model_dump()["results"].items()) > 0
+
+
+@parametrize(
+    "params",
+    [
+        (
+            {
+                "provider": "fmp",
+                "base": "USD,XAU",
+                "counter_currencies": "EUR,JPY,GBP",
+                "quote_type": "indirect",
+            }
+        ),
+    ],
+)
+@pytest.mark.integration
+def test_currency_snapshots(params, obb):
+    result = obb.currency.snapshots(**params)
+    assert result
+    assert isinstance(result, OBBject)
+    assert len(result.results) > 0

openbb_platform/extensions/currency/openbb_currency/currency_router.py

@@ -66,3 +66,22 @@ async def reference_rates(
     impacting global trade, loans, and investments.
     """
     return await OBBject.from_query(Query(**locals()))
+
+
+@router.command(
+    model="CurrencySnapshots",
+    exclude_auto_examples=True,
+    examples=[
+        "obb.currency.snapshots(",
+        'provider="fmp", base="USD,XAU", counter_currencies="EUR,JPY,GBP", quote_type="indirect"',
+        ")",
+    ],
+)
+async def snapshots(
+    cc: CommandContext,
+    provider_choices: ProviderChoices,
+    standard_params: StandardParams,
+    extra_params: ExtraParams,
+) -> OBBject:
+    """Snapshots of currency exchange rates from an indirect or direct perspective of a base currency."""
+    return await OBBject.from_query(Query(**locals()))

openbb_platform/openbb/assets/module_map.json

@@ -8,6 +8,7 @@
     "currency_price": "/currency/price",
     "currency_price_historical": "/currency/price/historical",
     "currency_search": "/currency/search",
+    "currency_snapshots": "/currency/snapshots",
     "derivatives": "/derivatives",
     "derivatives_futures": "/derivatives/futures",
     "derivatives_futures_curve": "/derivatives/futures/curve",

openbb_platform/openbb/package/currency.py

@@ -1,17 +1,20 @@
 ### THIS FILE IS AUTO-GENERATED. DO NOT EDIT. ###
 
-from typing import Literal, Optional
+from typing import List, Literal, Optional, Union
 
+from openbb_core.app.model.custom_parameter import OpenBBCustomParameter
 from openbb_core.app.model.obbject import OBBject
 from openbb_core.app.static.container import Container
 from openbb_core.app.static.utils.decorators import exception_handler, validate
 from openbb_core.app.static.utils.filters import filter_inputs
+from typing_extensions import Annotated
 
 
 class ROUTER_currency(Container):
     """/currency
     /price
     search
+    snapshots
     """
 
     def __repr__(self) -> str:
@@ -136,3 +139,120 @@ class ROUTER_currency(Container):
                 extra_params=kwargs,
             )
         )
+
+    @exception_handler
+    @validate
+    def snapshots(
+        self,
+        base: Annotated[
+            Union[str, List[str]],
+            OpenBBCustomParameter(
+                description="The base currency symbol. Multiple items allowed for provider(s): fmp."
+            ),
+        ] = "usd",
+        quote_type: Annotated[
+            Literal["direct", "indirect"],
+            OpenBBCustomParameter(
+                description="Whether the quote is direct or indirect. Selecting 'direct' will return the exchange rate as the amount of domestic currency required to buy one unit of the foreign currency. Selecting 'indirect' (default) will return the exchange rate as the amount of foreign currency required to buy one unit of the domestic currency."
+            ),
+        ] = "indirect",
+        counter_currencies: Annotated[
+            Union[str, List[str], None],
+            OpenBBCustomParameter(
+                description="An optional list of counter currency symbols to filter for. None returns all."
+            ),
+        ] = None,
+        provider: Optional[Literal["fmp"]] = None,
+        **kwargs
+    ) -> OBBject:
+        """Snapshots of currency exchange rates from an indirect or direct perspective of a base currency.
+
+        Parameters
+        ----------
+        base : Union[str, List[str]]
+            The base currency symbol. Multiple items allowed for provider(s): fmp.
+        quote_type : Literal['direct', 'indirect']
+            Whether the quote is direct or indirect. Selecting 'direct' will return the exchange rate as the amount of domestic currency required to buy one unit of the foreign currency. Selecting 'indirect' (default) will return the exchange rate as the amount of foreign currency required to buy one unit of the domestic currency.
+        counter_currencies : Union[str, List[str], None]
+            An optional list of counter currency symbols to filter for. None returns all.
+        provider : Optional[Literal['fmp']]
+            The provider to use for the query, by default None.
+            If None, the provider specified in defaults is selected or 'fmp' if there is
+            no default.
+
+        Returns
+        -------
+        OBBject
+            results : List[CurrencySnapshots]
+                Serializable results.
+            provider : Optional[Literal['fmp']]
+                Provider name.
+            warnings : Optional[List[Warning_]]
+                List of warnings.
+            chart : Optional[Chart]
+                Chart object.
+            extra : Dict[str, Any]
+                Extra info.
+
+        CurrencySnapshots
+        -----------------
+        base_currency : str
+            The base, or domestic, currency.
+        counter_currency : str
+            The counter, or foreign, currency.
+        last_rate : float
+            The exchange rate, relative to the base currency. Rates are expressed as the amount of foreign currency received from selling one unit of the base currency, or the quantity of foreign currency required to purchase one unit of the domestic currency. To inverse the perspective, set the 'quote_type' parameter as 'direct'.
+        open : Optional[float]
+            The open price.
+        high : Optional[float]
+            The high price.
+        low : Optional[float]
+            The low price.
+        close : Optional[float]
+            The close price.
+        volume : Optional[int]
+            The trading volume.
+        prev_close : Optional[float]
+            The previous close price.
+        change : Optional[float]
+            The change in the price from the previous close. (provider: fmp)
+        change_percent : Optional[float]
+            The change in the price from the previous close, as a normalized percent. (provider: fmp)
+        ma50 : Optional[float]
+            The 50-day moving average. (provider: fmp)
+        ma200 : Optional[float]
+            The 200-day moving average. (provider: fmp)
+        year_high : Optional[float]
+            The 52-week high. (provider: fmp)
+        year_low : Optional[float]
+            The 52-week low. (provider: fmp)
+        last_rate_timestamp : Optional[datetime]
+            The timestamp of the last rate. (provider: fmp)
+
+        Example
+        -------
+        >>> from openbb import obb
+        >>> obb.currency.snapshots(
+        >>> provider="fmp", base="USD,XAU", counter_currencies="EUR,JPY,GBP", quote_type="indirect"
+        >>> )
+        """  # noqa: E501
+
+        return self._run(
+            "/currency/snapshots",
+            **filter_inputs(
+                provider_choices={
+                    "provider": self._get_provider(
+                        provider,
+                        "/currency/snapshots",
+                        ("fmp",),
+                    )
+                },
+                standard_params={
+                    "base": base,
+                    "quote_type": quote_type,
+                    "counter_currencies": counter_currencies,
+                },
+                extra_params=kwargs,
+                extra_info={"base": {"multiple_items_allowed": ["fmp"]}},
+            )
+        )

openbb_platform/providers/fmp/openbb_fmp/__init__.py

@@ -17,6 +17,7 @@ from openbb_fmp.models.crypto_historical import FMPCryptoHistoricalFetcher
 from openbb_fmp.models.crypto_search import FMPCryptoSearchFetcher
 from openbb_fmp.models.currency_historical import FMPCurrencyHistoricalFetcher
 from openbb_fmp.models.currency_pairs import FMPCurrencyPairsFetcher
+from openbb_fmp.models.currency_snapshots import FMPCurrencySnapshotsFetcher
 from openbb_fmp.models.discovery_filings import FMPDiscoveryFilingsFetcher
 from openbb_fmp.models.earnings_call_transcript import FMPEarningsCallTranscriptFetcher
 from openbb_fmp.models.economic_calendar import FMPEconomicCalendarFetcher
@@ -86,6 +87,7 @@ fmp_provider = Provider(
         "CryptoSearch": FMPCryptoSearchFetcher,
         "CurrencyHistorical": FMPCurrencyHistoricalFetcher,
         "CurrencyPairs": FMPCurrencyPairsFetcher,
+        "CurrencySnapshots": FMPCurrencySnapshotsFetcher,
         "DiscoveryFilings": FMPDiscoveryFilingsFetcher,
         "EarningsCallTranscript": FMPEarningsCallTranscriptFetcher,
         "EconomicCalendar": FMPEconomicCalendarFetcher,

openbb_platform/providers/fmp/openbb_fmp/models/currency_snapshots.py

@@ -0,0 +1,158 @@
+"""FMP Currency Snapshots Model."""
+
+# pylint: disable=unused-argument
+
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+from openbb_core.provider.abstract.fetcher import Fetcher
+from openbb_core.provider.standard_models.currency_snapshots import (
+    CurrencySnapshotsData,
+    CurrencySnapshotsQueryParams,
+)
+from openbb_core.provider.utils.errors import EmptyDataError
+from openbb_core.provider.utils.helpers import amake_request
+from pandas import DataFrame, concat
+from pydantic import Field, field_validator
+
+
+class FMPCurrencySnapshotsQueryParams(CurrencySnapshotsQueryParams):
+    """FMP Currency Snapshots Query.
+
+    Source: https://site.financialmodelingprep.com/developer/docs#exchange-prices-quote
+    """
+
+    __json_schema_extra__ = {"base": ["multiple_items_allowed"]}
+
+
+class FMPCurrencySnapshotsData(CurrencySnapshotsData):
+    """FMP Currency Snapshots Data."""
+
+    __alias_dict__ = {
+        "last_rate": "price",
+        "high": "dayHigh",
+        "low": "dayLow",
+        "ma50": "priceAvg50",
+        "ma200": "priceAvg200",
+        "year_high": "yearHigh",
+        "year_low": "yearLow",
+        "prev_close": "previousClose",
+        "change_percent": "changesPercentage",
+        "last_rate_timestamp": "timestamp",
+    }
+
+    change: Optional[float] = Field(
+        description="The change in the price from the previous close.", default=None
+    )
+    change_percent: Optional[float] = Field(
+        description="The change in the price from the previous close, as a normalized percent.",
+        default=None,
+        json_schema_extra={"x-unit_measurement": "percent", "x-frontend_multiply": 100},
+    )
+    ma50: Optional[float] = Field(
+        description="The 50-day moving average.", default=None
+    )
+    ma200: Optional[float] = Field(
+        description="The 200-day moving average.", default=None
+    )
+    year_high: Optional[float] = Field(description="The 52-week high.", default=None)
+    year_low: Optional[float] = Field(description="The 52-week low.", default=None)
+    last_rate_timestamp: Optional[datetime] = Field(
+        description="The timestamp of the last rate.", default=None
+    )
+
+    @field_validator("change_percent", mode="before", check_fields=False)
+    @classmethod
+    def normalize_percent(cls, v):
+        """Normalize the percent."""
+        return v / 100 if v is not None else None
+
+
+class FMPCurrencySnapshotsFetcher(
+    Fetcher[FMPCurrencySnapshotsQueryParams, List[FMPCurrencySnapshotsData]]
+):
+    """FMP Currency Snapshots Fetcher."""
+
+    @staticmethod
+    def transform_query(params: Dict[str, Any]) -> FMPCurrencySnapshotsQueryParams:
+        """Transform the query parameters."""
+        return FMPCurrencySnapshotsQueryParams(**params)
+
+    @staticmethod
+    async def aextract_data(
+        query: FMPCurrencySnapshotsQueryParams,
+        credentials: Optional[Dict[str, str]],
+        **kwargs: Any,
+    ) -> List[Dict]:
+        """Extract the data from the FMP endpoint."""
+
+        api_key = credentials.get("fmp_api_key") if credentials else ""
+
+        url = f"https://financialmodelingprep.com/api/v3/quotes/forex?apikey={api_key}"
+
+        return await amake_request(url, **kwargs)  # type: ignore
+
+    @staticmethod
+    def transform_data(
+        query: FMPCurrencySnapshotsQueryParams,
+        data: List[Dict],
+        **kwargs: Any,
+    ) -> List[FMPCurrencySnapshotsData]:
+        """Filter by the query parameters and validate the model."""
+
+        if not data:
+            raise EmptyDataError("No data was returned from the FMP endpoint.")
+
+        # Drop all the zombie columns FMP returns.
+        df = (
+            DataFrame(data)
+            .dropna(how="all", axis=1)
+            .drop(columns=["exchange", "avgVolume"])
+        )
+
+        new_df = DataFrame()
+
+        # Filter for the base currencies requested and the quote_type.
+        for symbol in query.base.split(","):
+            temp = (
+                df.query("`symbol`.str.startswith(@symbol)")
+                if query.quote_type == "indirect"
+                else df.query("`symbol`.str.endswith(@symbol)")
+            ).rename(columns={"symbol": "base_currency", "name": "counter_currency"})
+            temp["base_currency"] = symbol
+            temp["counter_currency"] = (
+                [d.split("/")[1] for d in temp["counter_currency"]]
+                if query.quote_type == "indirect"
+                else [d.split("/")[0] for d in temp["counter_currency"]]
+            )
+            # Filter for the counter currencies, if requested.
+            if query.counter_currencies is not None:
+                counter_currencies = (  # noqa: F841  # pylint: disable=unused-variable
+                    query.counter_currencies
+                    if isinstance(query.counter_currencies, list)
+                    else query.counter_currencies.split(",")
+                )
+                temp = (
+                    temp.query("`counter_currency`.isin(@counter_currencies)")
+                    .set_index("counter_currency")
+                    # Sets the counter currencies in the order they were requested.
+                    .filter(items=counter_currencies, axis=0)
+                    .reset_index()
+                )
+            # If there are no records, don't concatenate.
+            if len(temp) > 0:
+                # Convert the Unix timestamp to a datetime.
+                temp.timestamp = temp.timestamp.apply(
+                    lambda x: datetime.fromtimestamp(x)
+                )
+                new_df = concat([new_df, temp])
+            if len(new_df) == 0:
+                raise EmptyDataError(
+                    "No data was found using the applied filters. Check the parameters."
+                )
+            # Fill and replace any NaN values with NoneType.
+            new_df = new_df.fillna("N/A").replace("N/A", None)
+        return [
+            FMPCurrencySnapshotsData.model_validate(d)
+            for d in new_df.reset_index(drop=True).to_dict(orient="records")
+        ]

openbb_platform/providers/fmp/tests/test_fmp_fetchers.py

@@ -21,6 +21,7 @@ from openbb_fmp.models.crypto_historical import FMPCryptoHistoricalFetcher
 from openbb_fmp.models.crypto_search import FMPCryptoSearchFetcher
 from openbb_fmp.models.currency_historical import FMPCurrencyHistoricalFetcher
 from openbb_fmp.models.currency_pairs import FMPCurrencyPairsFetcher
+from openbb_fmp.models.currency_snapshots import FMPCurrencySnapshotsFetcher
 from openbb_fmp.models.discovery_filings import FMPDiscoveryFilingsFetcher
 from openbb_fmp.models.earnings_call_transcript import FMPEarningsCallTranscriptFetcher
 from openbb_fmp.models.economic_calendar import FMPEconomicCalendarFetcher
@@ -654,3 +655,16 @@ def test_fmp_etf_equity_exposure_fetcher(credentials=test_credentials):
     fetcher = FMPEtfEquityExposureFetcher()
     result = fetcher.test(params, credentials)
     assert result is None
+
+
+@pytest.mark.record_http
+def test_fmp_currency_snapshots_fetcher(credentials=test_credentials):
+    params = {
+        "base": "XAU",
+        "quote_type": "indirect",
+        "counter_currencies": "USD,EUR,GBP,JPY,HKD,AUD,CAD,CHF,SEK,NZD,SGD",
+    }
+
+    fetcher = FMPCurrencySnapshotsFetcher()
+    result = fetcher.test(params, credentials)
+    assert result is None