1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
"""FMP Insider Trading Model."""
# pylint: disable=unused-argument
import math
from typing import Any, Dict, List, Optional
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.insider_trading import (
InsiderTradingData,
InsiderTradingQueryParams,
)
from openbb_core.provider.utils.helpers import amake_requests, get_querystring
from openbb_fmp.utils.definitions import TRANSACTION_TYPES, TRANSACTION_TYPES_DICT
from pydantic import Field, model_validator
class FMPInsiderTradingQueryParams(InsiderTradingQueryParams):
"""FMP Insider Trading Query.
Source: https://site.financialmodelingprep.com/developer/docs/#Stock-Insider-Trading
"""
transaction_type: Optional[TRANSACTION_TYPES] = Field(
default=None,
description="Type of the transaction.",
alias="transactionType",
)
@model_validator(mode="after")
@classmethod
def validate_transaction_type(cls, values: "FMPInsiderTradingQueryParams"):
"""Validate the transaction type."""
if isinstance(values.transaction_type, list):
values.transaction_type = ",".join(values.transaction_type)
return values
class FMPInsiderTradingData(InsiderTradingData):
"""FMP Insider Trading Data."""
__alias_dict__ = {
"owner_cik": "reportingCik",
"owner_name": "reportingName",
"owner_title": "typeOfOwner",
"security_type": "securityName",
"transaction_price": "price",
"acquisition_or_disposition": "acquistionOrDisposition",
"filing_url": "link",
}
form_type: str = Field(description="Form type of the insider trading.")
class FMPInsiderTradingFetcher(
Fetcher[
FMPInsiderTradingQueryParams,
List[FMPInsiderTradingData],
]
):
"""Transform the query, extract and transform the data from the FMP endpoints."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> FMPInsiderTradingQueryParams:
"""Transform the query params."""
return FMPInsiderTradingQueryParams(**params)
@staticmethod
async def aextract_data(
query: FMPInsiderTradingQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the FMP endpoint."""
api_key = credentials.get("fmp_api_key") if credentials else ""
transaction_type = TRANSACTION_TYPES_DICT.get(query.transaction_type, None)
query = query.model_copy(update={"transaction_type": transaction_type})
base_url = "https://financialmodelingprep.com/api/v4/insider-trading"
query_str = get_querystring(query.model_dump(by_alias=True), ["page"])
pages = math.ceil(query.limit / 100)
urls = [
f"{base_url}?{query_str}&page={page}&apikey={api_key}"
for page in range(pages)
]
return await amake_requests(urls, raise_for_status=True, **kwargs)
@staticmethod
def transform_data(
query: FMPInsiderTradingQueryParams, data: List[Dict], **kwargs: Any
) -> List[FMPInsiderTradingData]:
"""Return the transformed data."""
data = sorted(data, key=lambda x: x["filingDate"], reverse=True)
return [FMPInsiderTradingData.model_validate(d) for d in data]
|
