[Feature]try to add fmp senate trading data api

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

@@ -0,0 +1,45 @@
+"""Government Trades Standard Model."""
+
+from datetime import date as dateType
+from typing import Literal, Optional
+
+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,
+    QUERY_DESCRIPTIONS,
+)
+from pydantic import Field, NonNegativeInt, field_validator
+
+
+class GovernmentTradesQueryParams(QueryParams):
+    """Government Trades Query."""
+
+    symbol: Optional[str] = Field(
+        default=None, description=QUERY_DESCRIPTIONS.get("symbol", "")
+    )
+    chamber: Literal["house", "senate", "all"] = Field(
+        default="all", description="Government Chamber."
+    )
+    limit: Optional[NonNegativeInt] = Field(
+        default=100, description=QUERY_DESCRIPTIONS.get("limit", "")
+    )
+
+    @field_validator("symbol", mode="before", check_fields=False)
+    @classmethod
+    def to_upper(cls, v: str) -> str:
+        """Convert field to uppercase."""
+        return v.upper() if v else None
+
+
+class GovernmentTradesData(Data):
+    """Government Trades data."""
+
+    symbol: Optional[str] = Field(description=DATA_DESCRIPTIONS.get("symbol", ""))
+    date: dateType = Field(description=DATA_DESCRIPTIONS.get("date", ""))
+    transaction_date: Optional[dateType] = Field(
+        default=None, description="Date of Transaction."
+    )
+    representative: Optional[str] = Field(
+        default=None, description="Name of Representative."
+    )

openbb_platform/extensions/equity/integration/test_equity_api.py

@@ -2156,3 +2156,35 @@ def test_equity_discovery_latest_financial_reports(params, headers):
     result = requests.get(url, headers=headers, timeout=10)
     assert isinstance(result, requests.Response)
     assert result.status_code == 200
+
+
+@parametrize(
+    "params",
+    [
+        (
+            {
+                "chamber": "all",
+                "symbol": "AAPL",
+                "provider": "fmp",
+            }
+        ),
+        (
+            {
+                "chamber": "all",
+                "limit": 300,
+                "provider": "fmp",
+            }
+        ),
+    ],
+
+)
+@pytest.mark.integration
+def test_equity_ownership_government_trades(params, headers):
+    """Test the equity ownership government trades endpoint."""
+    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/equity/ownership/government_trades?{query_str}"
+    result = requests.get(url, headers=headers, timeout=10)
+    assert isinstance(result, requests.Response)
+    assert result.status_code == 200

openbb_platform/extensions/equity/integration/test_equity_python.py

@@ -2017,3 +2017,33 @@ def test_equity_discovery_latest_financial_reports(params, obb):
     assert result
     assert isinstance(result, OBBject)
     assert len(result.results) > 0
+
+
+@parametrize(
+    "params",
+    [
+        (
+            {
+                "chamber": "all",
+                "symbol": "AAPL",
+                "provider": "fmp",
+            }
+        ),
+        (
+            {
+                "chamber": "all",
+                "limit": 300,
+                "provider": "fmp",
+            }
+        ),
+    ],
+)
+@pytest.mark.integration
+def test_equity_ownership_government_trades(params, obb):
+    """Test the equity ownership government trades endpoint."""
+    params = {p: v for p, v in params.items() if v}
+
+    result = obb.equity.ownership.government_trades(**params)
+    assert result
+    assert isinstance(result, OBBject)
+    assert len(result.results) > 0

openbb_platform/extensions/equity/openbb_equity/ownership/ownership_router.py

@@ -112,3 +112,22 @@ async def form_13f(
     their investment strategy from competitors and the public.
     """
     return await OBBject.from_query(Query(**locals()))
+
+
+@router.command(
+    model="GovernmentTrades",
+    examples=[
+        APIEx(parameters={"symbol": "AAPL", "chamber": "all", "provider": "fmp"}),
+        APIEx(parameters={"limit": 500, "chamber": "all", "provider": "fmp"}),
+    ],
+)
+async def government_trades(
+    cc: CommandContext,
+    provider_choices: ProviderChoices,
+    standard_params: StandardParams,
+    extra_params: ExtraParams,
+) -> OBBject:
+    """Obtain government transaction data, including data from the Senate
+    and the House of Representatives.
+    """
+    return await OBBject.from_query(Query(**locals()))

openbb_platform/providers/fmp/openbb_fmp/__init__.py

@@ -41,6 +41,7 @@
 from openbb_fmp.models.financial_ratios import FMPFinancialRatiosFetcher
 from openbb_fmp.models.forward_ebitda_estimates import FMPForwardEbitdaEstimatesFetcher
 from openbb_fmp.models.forward_eps_estimates import FMPForwardEpsEstimatesFetcher
+from openbb_fmp.models.government_trades import FMPGovernmentTradesFetcher
 from openbb_fmp.models.historical_dividends import FMPHistoricalDividendsFetcher
 from openbb_fmp.models.historical_employees import FMPHistoricalEmployeesFetcher
 from openbb_fmp.models.historical_eps import FMPHistoricalEpsFetcher
@@ -137,6 +138,7 @@
         "WorldNews": FMPWorldNewsFetcher,
         "EtfHistorical": FMPEquityHistoricalFetcher,
         "YieldCurve": FMPYieldCurveFetcher,
+        "GovernmentTrades": FMPGovernmentTradesFetcher,
     },
     repr_name="Financial Modeling Prep (FMP)",
     deprecated_credentials={"API_KEY_FINANCIALMODELINGPREP": "fmp_api_key"},

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

@@ -0,0 +1,168 @@
+"""Government Trades Model."""
+
+import asyncio
+import math
+from typing import Any, Dict, List, Optional
+from warnings import warn
+
+from openbb_core.provider.abstract.fetcher import Fetcher
+from openbb_core.provider.standard_models.government_trades import (
+    GovernmentTradesData,
+    GovernmentTradesQueryParams,
+)
+from openbb_core.provider.utils.errors import EmptyDataError
+from openbb_core.provider.utils.helpers import amake_request
+from openbb_fmp.utils.helpers import create_url, response_callback
+from pydantic import Field
+
+
+class FMPGovernmentTradesQueryParams(GovernmentTradesQueryParams):
+    """Government Trades Query Parameters.
+
+    Source: https://financialmodelingprep.com/api/v4/senate-trading?symbol=AAPL
+    Source: https://financialmodelingprep.com/api/v4/senate-trading-rss-feed?page=0
+    Source: https://financialmodelingprep.com/api/v4/senate-disclosure?symbol=AAPL
+    Source: https://financialmodelingprep.com/api/v4/senate-disclosure-rss-feed?page=0
+    """
+
+
+class FMPGovernmentTradesData(GovernmentTradesData):
+    """Government Trades Data Model."""
+
+    __alias_dict__ = {
+        "symbol": "ticker",
+        "transaction_date": "transactionDate",
+        "representative": "office",
+    }
+    link: Optional[str] = Field(
+        default=None, description="Link to the transaction document."
+    )
+    owner: Optional[str] = Field(
+        default=None, description="Ownership status (e.g., Spouse, Joint)."
+    )
+    asset_type: Optional[str] = Field(
+        default=None, description="Type of asset involved in the transaction."
+    )
+    asset_description: Optional[str] = Field(
+        default=None, description="Description of the asset."
+    )
+    type: Optional[str] = Field(
+        default=None, description="Type of transaction (e.g., Sale, Purchase)."
+    )
+    amount: Optional[str] = Field(default=None, description="Transaction amount range.")
+    comment: Optional[str] = Field(
+        default=None, description="Additional comments on the transaction."
+    )
+
+
+class FMPGovernmentTradesFetcher(
+    Fetcher[
+        FMPGovernmentTradesQueryParams,
+        List[FMPGovernmentTradesData],
+    ]
+):
+    """Fetches and transforms data from the Government Trades endpoints."""
+
+    @staticmethod
+    def transform_query(params: Dict[str, Any]) -> FMPGovernmentTradesQueryParams:
+        """Transform the query params."""
+        return FMPGovernmentTradesQueryParams(**params)
+
+    @staticmethod
+    async def aextract_data(
+        query: FMPGovernmentTradesQueryParams,
+        credentials: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> List[Dict]:
+        """Return the raw data from the Government Trades endpoint."""
+        symbols = []
+        if query.symbol:
+            symbols = query.symbol.split(",")
+        results: List[Dict] = []
+        chamber_url_dict = {
+            "house": ["senate-disclosure"],
+            "senate": ["senate-trading"],
+            "all": ["senate-disclosure", "senate-trading"],
+        }
+        api_key = credentials.get("fmp_api_key") if credentials else ""
+        keys_to_remove = {
+            "comment",
+            "district",
+            "capitalGainsOver200USD",
+            "disclosureYear",
+            "firstName",
+            "lastName",
+        }
+        keys_to_rename = {"dateRecieved": "date", "disclosureDate": "date"}
+
+        async def get_one(url):
+            """Get data for one URL."""
+            result = await amake_request(url, response_callback=response_callback, **kwargs)
+            processed_list = []
+
+            for entry in result:
+                new_entry = {
+                    keys_to_rename.get(k, k): v
+                    for k, v in entry.items()
+                    if k not in keys_to_remove
+                }
+                processed_list.append(new_entry)
+            if not processed_list or len(processed_list) == 0:
+                warn(f"Symbol Error: No data found for {url}")
+            if processed_list:
+                results.extend(processed_list)
+
+        if symbols:
+            urls_list = []
+            for symbol in symbols:
+                query.symbol = symbol
+                url = [
+                    create_url(
+                        4,
+                        f"{i}",
+                        api_key=api_key,
+                        query=query,
+                        exclude=["chamber", "limit"],
+                    )
+                    for i in chamber_url_dict[query.chamber]
+                ]
+                urls_list.extend(url)
+            await asyncio.gather(*[get_one(url) for url in urls_list])
+        else:
+            urls_list = []
+            pages = math.ceil(query.limit / 100)
+            for page in range(pages):
+                query.page = page
+                url = [
+                    create_url(
+                        4,
+                        f"{i}-rss-feed",
+                        api_key=api_key,
+                        query=query,
+                        exclude=["chamber", "limit"],
+                    )
+                    for i in chamber_url_dict[query.chamber]
+                ]
+                urls_list.extend(url)
+            await asyncio.gather(*[get_one(url) for url in urls_list])
+        if not results:
+            raise EmptyDataError("No data returned for the given symbol.")
+
+        return results
+
+    @staticmethod
+    def transform_data(
+        query: FMPGovernmentTradesQueryParams, data: List[Dict], **kwargs: Any
+    ) -> List[FMPGovernmentTradesData]:
+        """Return the transformed data."""
+
+        return sorted(
+            [
+                FMPGovernmentTradesData(
+                    **{k: v for k, v in d.items() if v and v != "--"}
+                )
+                for d in data
+            ],
+            key=lambda x: x.date,
+            reverse=True,
+        )

openbb_platform/providers/fmp/tests/test_fmp_fetchers.py

@@ -71,6 +71,7 @@
 from openbb_fmp.models.treasury_rates import FMPTreasuryRatesFetcher
 from openbb_fmp.models.world_news import FMPWorldNewsFetcher
 from openbb_fmp.models.yield_curve import FMPYieldCurveFetcher
+from openbb_fmp.models.government_trades import FMPGovernmentTradesFetcher
 
 test_credentials = UserService().default_user_settings.credentials.model_dump(
     mode="json"
@@ -762,3 +763,17 @@ def test_fmp_historical_market_cap_fetcher(credentials=test_credentials):
     fetcher = FmpHistoricalMarketCapFetcher()
     result = fetcher.test(params, credentials)
     assert result is None
+
+
+@pytest.mark.record_http
+def test_fmp_government_trades_fetcher(credentials=test_credentials):
+    """Test FMP government trades fetcher.
+    params limit only functions when there is no parameter symbol.
+    """
+    params = {
+        "chamber": "senate",
+        "limit": 1,
+    }
+    fetcher = FMPGovernmentTradesFetcher()
+    result = fetcher.test(params, credentials)
+    assert result is None