jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-27 12:40:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Feature] Add SEC As Provider To equity.ownership.insider_trading (#6947)

Browse Source 
* add sec to equity.ownership.insider_trading

* integration test params

* lint

* forgot that file in the commit

* grammar police

* alias map

* more lint

* more alias mapping

* pylint

---------

Co-authored-by: Igor Radovanovic <74266147+IgorWounds@users.noreply.github.com>
This commit is contained in:
Danglewood
2024-11-25 13:17:04 -08:00
committed by GitHub
parent 7d8202fc8a
commit bde8aba61e
11 changed files with 12345 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
openbb_platform/extensions/equity/integration/test_equity_api.py
View File

@@ -579,6 +579,16 @@ def test_equity_fundamental_income_growth(params, headers):
"sort_by": "updated_on",
}
),
(
{
"provider": "sec",
"symbol": "AAPL",
"limit": 10,
"start_date": "2024-06-30",
"end_date": "2024-09-30",
"use_cache": True,
}
),
],
)
@pytest.mark.integration

10
openbb_platform/extensions/equity/integration/test_equity_python.py
View File

@@ -468,6 +468,16 @@ def test_equity_fundamental_income_growth(params, obb):
"sort_by": "updated_on",
}
),
(
{
"provider": "sec",
"symbol": "AAPL",
"limit": 10,
"start_date": "2024-06-30",
"end_date": "2024-09-30",
"use_cache": True,
}
),
],
)
@pytest.mark.integration

190
openbb_platform/openbb/assets/reference.json vendored
View File

@@ -27405,6 +27405,32 @@
"optional": true,
"choices": null
}
],
"sec": [
{
"name": "start_date",
"type": "Union[date, str]",
"description": "Start date of the data, in YYYY-MM-DD format. Wide date ranges can result in long download times. Recommended to use a smaller date range, default is 120 days ago.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "end_date",
"type": "Union[date, str]",
"description": "End date of the data, in YYYY-MM-DD format. Default is today.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "use_cache",
"type": "bool",
"description": "Persist the data locally for future use. Default is True. Each form submission is an individual download and the SEC limits the number of concurrent downloads. This prevents the same file from being downloaded multiple times.",
"default": true,
"optional": true,
"choices": null
}
]
},
"returns": {
@@ -27416,7 +27442,7 @@
},
{
"name": "provider",
"type": "Optional[Literal['fmp', 'intrinio']]",
"type": "Optional[Literal['fmp', 'intrinio', 'sec']]",
"description": "Provider name."
},
{
@@ -27682,6 +27708,168 @@
"optional": true,
"choices": null
}
],
"sec": [
{
"name": "company_name",
"type": "str",
"description": "Name of the company.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "form",
"type": "Union[int, str]",
"description": "Form type.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "director",
"type": "bool",
"description": "Whether the owner is a director.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "officer",
"type": "bool",
"description": "Whether the owner is an officer.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "ten_percent_owner",
"type": "bool",
"description": "Whether the owner is a 10% owner.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "other",
"type": "bool",
"description": "Whether the owner is classified as other.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "other_text",
"type": "str",
"description": "Text for other classification.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "transaction_timeliness",
"type": "str",
"description": "Timeliness of the transaction.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "ownership_type",
"type": "str",
"description": "Type of ownership, direct or indirect.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "nature_of_ownership",
"type": "str",
"description": "Nature of the ownership.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "exercise_date",
"type": "date",
"description": "Date of exercise.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "expiration_date",
"type": "date",
"description": "Date of expiration for the derivative.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "deemed_execution_date",
"type": "date",
"description": "Deemed execution date.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "underlying_security_title",
"type": "str",
"description": "Title of the underlying security.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "underlying_security_shares",
"type": "float",
"description": "Number of underlying shares associated with the derivative.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "underlying_security_value",
"type": "float",
"description": "Value of the underlying security.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "conversion_exercise_price",
"type": "float",
"description": "Price of conversion or exercise of the securities.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "transaction_value",
"type": "float",
"description": "Total value of the transaction.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "value_owned",
"type": "float",
"description": "Value of the securities owned after the transaction.",
"default": null,
"optional": true,
"choices": null
},
{
"name": "footnote",
"type": "str",
"description": "Footnote for the transaction.",
"default": null,
"optional": true,
"choices": null
}
]
},
"model": "InsiderTrading"

65
openbb_platform/openbb/package/equity_ownership.py
View File

@@ -155,9 +155,9 @@ class ROUTER_equity_ownership(Container):
int, OpenBBField(description="The number of data entries to return.")
] = 500,
provider: Annotated[
Optional[Literal["fmp", "intrinio"]],
Optional[Literal["fmp", "intrinio", "sec"]],
OpenBBField(
description="The provider to use, by default None. If None, the priority list configured in the settings is used. Default priority: fmp, intrinio."
description="The provider to use, by default None. If None, the priority list configured in the settings is used. Default priority: fmp, intrinio, sec."
),
] = None,
**kwargs
@@ -170,25 +170,28 @@ class ROUTER_equity_ownership(Container):
Symbol to get data for.
limit : int
The number of data entries to return.
provider : Optional[Literal['fmp', 'intrinio']]
The provider to use, by default None. If None, the priority list configured in the settings is used. Default priority: fmp, intrinio.
provider : Optional[Literal['fmp', 'intrinio', 'sec']]
The provider to use, by default None. If None, the priority list configured in the settings is used. Default priority: fmp, intrinio, sec.
transaction_type : Optional[Literal['award', 'conversion', 'return', 'expire_short', 'in_kind', 'gift', 'expire_long', 'discretionary', 'other', 'small', 'exempt', 'otm', 'purchase', 'sale', 'tender', 'will', 'itm', 'trust']]
Type of the transaction. (provider: fmp)
start_date : Optional[datetime.date]
Start date of the data, in YYYY-MM-DD format. (provider: intrinio)
Start date of the data, in YYYY-MM-DD format. (provider: intrinio);
Start date of the data, in YYYY-MM-DD format. Wide date ranges can result in long download times. Recommended to use a smaller date range, default is 120 days ago. (provider: sec)
end_date : Optional[datetime.date]
End date of the data, in YYYY-MM-DD format. (provider: intrinio)
End date of the data, in YYYY-MM-DD format. (provider: intrinio, sec)
ownership_type : Optional[Literal['D', 'I']]
Type of ownership. (provider: intrinio)
sort_by : Optional[Literal['filing_date', 'updated_on']]
Field to sort by. (provider: intrinio)
use_cache : bool
Persist the data locally for future use. Default is True. Each form submission is an individual download and the SEC limits the number of concurrent downloads. This prevents the same file from being downloaded multiple times. (provider: sec)
Returns
-------
OBBject
results : List[InsiderTrading]
Serializable results.
provider : Optional[Literal['fmp', 'intrinio']]
provider : Optional[Literal['fmp', 'intrinio', 'sec']]
Provider name.
warnings : Optional[List[Warning_]]
List of warnings.
@@ -230,33 +233,59 @@ class ROUTER_equity_ownership(Container):
form_type : Optional[str]
Form type of the insider trading. (provider: fmp)
company_name : Optional[str]
Name of the company. (provider: intrinio)
Name of the company. (provider: intrinio, sec)
conversion_exercise_price : Optional[float]
Conversion/Exercise price of the shares. (provider: intrinio)
Conversion/Exercise price of the shares. (provider: intrinio);
Price of conversion or exercise of the securities. (provider: sec)
deemed_execution_date : Optional[date]
Deemed execution date of the trade. (provider: intrinio)
Deemed execution date of the trade. (provider: intrinio);
Deemed execution date. (provider: sec)
exercise_date : Optional[date]
Exercise date of the trade. (provider: intrinio)
Exercise date of the trade. (provider: intrinio);
Date of exercise. (provider: sec)
expiration_date : Optional[date]
Expiration date of the derivative. (provider: intrinio)
Expiration date of the derivative. (provider: intrinio);
Date of expiration for the derivative. (provider: sec)
underlying_security_title : Optional[str]
Name of the underlying non-derivative security related to this derivative transaction. (provider: intrinio)
Name of the underlying non-derivative security related to this derivative transaction. (provider: intrinio);
Title of the underlying security. (provider: sec)
underlying_shares : Optional[Union[int, float]]
Number of underlying shares related to this derivative transaction. (provider: intrinio)
nature_of_ownership : Optional[str]
Nature of ownership of the insider trading. (provider: intrinio)
Nature of ownership of the insider trading. (provider: intrinio);
Nature of the ownership. (provider: sec)
director : Optional[bool]
Whether the owner is a director. (provider: intrinio)
Whether the owner is a director. (provider: intrinio, sec)
officer : Optional[bool]
Whether the owner is an officer. (provider: intrinio)
Whether the owner is an officer. (provider: intrinio, sec)
ten_percent_owner : Optional[bool]
Whether the owner is a 10% owner. (provider: intrinio)
Whether the owner is a 10% owner. (provider: intrinio, sec)
other_relation : Optional[bool]
Whether the owner is having another relation. (provider: intrinio)
derivative_transaction : Optional[bool]
Whether the owner is having a derivative transaction. (provider: intrinio)
report_line_number : Optional[int]
Report line number of the insider trading. (provider: intrinio)
form : Optional[Union[int, str]]
Form type. (provider: sec)
other : Optional[bool]
Whether the owner is classified as other. (provider: sec)
other_text : Optional[str]
Text for other classification. (provider: sec)
transaction_timeliness : Optional[str]
Timeliness of the transaction. (provider: sec)
ownership_type : Optional[str]
Type of ownership, direct or indirect. (provider: sec)
underlying_security_shares : Optional[float]
Number of underlying shares associated with the derivative. (provider: sec)
underlying_security_value : Optional[float]
Value of the underlying security. (provider: sec)
transaction_value : Optional[float]
Total value of the transaction. (provider: sec)
value_owned : Optional[float]
Value of the securities owned after the transaction. (provider: sec)
footnote : Optional[str]
Footnote for the transaction. (provider: sec)
Examples
--------
@@ -272,7 +301,7 @@ class ROUTER_equity_ownership(Container):
"provider": self._get_provider(
provider,
"equity.ownership.insider_trading",
("fmp", "intrinio"),
("fmp", "intrinio", "sec"),
)
},
standard_params={

2
openbb_platform/providers/sec/openbb_sec/__init__.py
View File

@@ -8,6 +8,7 @@ from openbb_sec.models.equity_ftd import SecEquityFtdFetcher
from openbb_sec.models.equity_search import SecEquitySearchFetcher
from openbb_sec.models.etf_holdings import SecEtfHoldingsFetcher
from openbb_sec.models.form_13FHR import SecForm13FHRFetcher
from openbb_sec.models.insider_trading import SecInsiderTradingFetcher
from openbb_sec.models.institutions_search import SecInstitutionsSearchFetcher
from openbb_sec.models.latest_financial_reports import SecLatestFinancialReportsFetcher
from openbb_sec.models.rss_litigation import SecRssLitigationFetcher
@@ -29,6 +30,7 @@ sec_provider = Provider(
"EtfHoldings": SecEtfHoldingsFetcher,
"Filings": SecCompanyFilingsFetcher,
"Form13FHR": SecForm13FHRFetcher,
"InsiderTrading": SecInsiderTradingFetcher,
"InstitutionsSearch": SecInstitutionsSearchFetcher,
"LatestFinancialReports": SecLatestFinancialReportsFetcher,
"RssLitigation": SecRssLitigationFetcher,

2
openbb_platform/providers/sec/openbb_sec/models/company_filings.py
View File

@@ -314,7 +314,7 @@ class SecCompanyFilingsFetcher(
filings = filings[filings["reportDate"] >= query.start_date]
if query.end_date:
filings = filings[filings["reportDate"] <= query.end_date]
base_url = f"https://www.sec.gov/Archives/edgar/data/{query.cik}/"
base_url = f"https://www.sec.gov/Archives/edgar/data/{str(int(query.cik))}/" # type: ignore
filings["primaryDocumentUrl"] = (
base_url
+ filings["accessionNumber"].str.replace("-", "")

221
openbb_platform/providers/sec/openbb_sec/models/insider_trading.py Normal file
View File

@@ -0,0 +1,221 @@
"""SEC Insider Trading Model."""
# pylint: disable =unused-argument
from datetime import date as dateType
from typing import Any, Optional, Union
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.insider_trading import (
InsiderTradingData,
InsiderTradingQueryParams,
)
from openbb_core.provider.utils.descriptions import QUERY_DESCRIPTIONS
from pydantic import Field, field_validator
TRANSACTION_CODE_MAP = {
"A": "Grant, award or other acquisition pursuant to Rule 16b-3(d)",
"C": "Conversion of derivative security",
"D": "Disposition to the issuer of issuer equity securities pursuant to Rule 16b-3(e)",
"E": "Expiration of short derivative position",
"F": (
"Payment of exercise price or tax liability by delivering or withholding securities incident to the receipt, "
"exercise or vesting of a security issued in accordance with Rule 16b-3"
),
"G": "Bona fide gift",
"H": "Expiration (or cancellation) of long derivative position with value received",
"I": (
"Discretionary transaction in accordance with Rule 16b-3(f) "
"resulting in acquisition or disposition of issuer securities"
),
"J": "Other acquisition or disposition (describe transaction)",
"L": "Small acquisition under Rule 16a-6",
"M": "Exercise or conversion of derivative security exempted pursuant to Rule 16b-3",
"O": "Exercise of out-of-the-money derivative security",
"P": "Open market or private purchase of non-derivative or derivative security",
"S": "Open market or private sale of non-derivative or derivative security",
"U": "Disposition pursuant to a tender of shares in a change of control transaction",
"W": "Acquisition or disposition by will or the laws of descent and distribution",
"X": "Exercise of in-the-money or at-the-money derivative security",
"Z": "Deposit into or withdrawal from voting trust",
}
TIMELINESS_MAP = {
"E": "Early",
"L": "Late",
"Empty": "On-time",
}
class SecInsiderTradingQueryParams(InsiderTradingQueryParams):
"""SEC Insider Trading Query Params.
Source: https://www.sec.gov/Archives/edgar/data/
"""
start_date: Optional[dateType] = Field(
default=None,
description=QUERY_DESCRIPTIONS.get("start_date", "")
+ " Wide date ranges can result in long download times."
+ " Recommended to use a smaller date range, default is 120 days ago.",
)
end_date: Optional[dateType] = Field(
default=None,
description=QUERY_DESCRIPTIONS.get("end_date", "") + " Default is today.",
)
use_cache: bool = Field(
default=True,
description="Persist the data locally for future use. Default is True."
+ " Each form submission is an individual download and the SEC limits the number of concurrent downloads."
+ " This prevents the same file from being downloaded multiple times.",
)
class SecInsiderTradingData(InsiderTradingData):
"""SEC Insider Trading Data."""
company_name: Optional[str] = Field(
default=None, description="Name of the company."
)
form: Optional[Union[str, int]] = Field(default=None, description="Form type.")
director: Optional[bool] = Field(
default=None, description="Whether the owner is a director."
)
officer: Optional[bool] = Field(
default=None, description="Whether the owner is an officer."
)
ten_percent_owner: Optional[bool] = Field(
default=None, description="Whether the owner is a 10% owner."
)
other: Optional[bool] = Field(
default=None, description="Whether the owner is classified as other."
)
other_text: Optional[str] = Field(
default=None, description="Text for other classification."
)
transaction_timeliness: Optional[str] = Field(
default=None, description="Timeliness of the transaction."
)
ownership_type: Optional[str] = Field(
default=None, description="Type of ownership, direct or indirect."
)
nature_of_ownership: Optional[str] = Field(
default=None, description="Nature of the ownership."
)
exercise_date: Optional[dateType] = Field(
default=None, description="Date of exercise."
)
expiration_date: Optional[dateType] = Field(
default=None, description="Date of expiration for the derivative."
)
deemed_execution_date: Optional[dateType] = Field(
default=None, description="Deemed execution date."
)
underlying_security_title: Optional[str] = Field(
default=None, description="Title of the underlying security."
)
underlying_security_shares: Optional[float] = Field(
default=None,
description="Number of underlying shares associated with the derivative.",
)
underlying_security_value: Optional[float] = Field(
default=None, description="Value of the underlying security."
)
conversion_exercise_price: Optional[float] = Field(
default=None, description="Price of conversion or exercise of the securities."
)
transaction_value: Optional[float] = Field(
default=None, description="Total value of the transaction."
)
value_owned: Optional[float] = Field(
default=None, description="Value of the securities owned after the transaction."
)
footnote: Optional[str] = Field(
default=None, description="Footnote for the transaction."
)
@field_validator("symbol", mode="before", check_fields=False)
@classmethod
def _to_upper(cls, v):
"""Convert symbol to uppercase."""
return v.upper() if v else None
@field_validator("ownership_type", mode="before", check_fields=False)
@classmethod
def _map_ownership_type(cls, v):
"""Map ownership type to description."""
if not v:
return None
return "Direct" if v.strip() == "D" else "Indirect" if v.strip() == "I" else v
@field_validator("acquisition_or_disposition", mode="before", check_fields=False)
@classmethod
def _map_acquisition_disposition(cls, v):
"""Map acquisition or disposition to description."""
if not v:
return None
return (
"Acquisition"
if v.strip() == "A"
else "Disposition" if v.strip() == "D" else v
)
@field_validator("transaction_type", mode="before", check_fields=False)
@classmethod
def _map_transaction_code(cls, v):
"""Map transaction code to description."""
return TRANSACTION_CODE_MAP.get(v, v) if v else None
@field_validator("transaction_timeliness", mode="before", check_fields=False)
@classmethod
def _map_timeliness(cls, v):
"""Map timeliness code to description."""
return TIMELINESS_MAP.get(v, v) if v else None
class SecInsiderTradingFetcher(
Fetcher[SecInsiderTradingQueryParams, list[SecInsiderTradingData]]
):
"""SEC Insider Trading Fetcher."""
@staticmethod
def transform_query(params: dict[str, Any]) -> SecInsiderTradingQueryParams:
"""Transform query parameters."""
# pylint: disable=import-outside-toplevel
from datetime import datetime, timedelta
start_date = params.get("start_date")
end_date = params.get("end_date")
if not start_date and not end_date:
params["start_date"] = (datetime.now() - timedelta(days=120)).date()
params["end_date"] = datetime.now().date()
return SecInsiderTradingQueryParams(**params)
@staticmethod
async def aextract_data(
query: SecInsiderTradingQueryParams,
credentials: Optional[dict[str, str]],
**kwargs: Any,
) -> list[dict]:
"""Extract the data from the SEC archives."""
# pylint: disable=import-outside-toplevel
from openbb_sec.utils.form4 import get_form_4
return await get_form_4(
query.symbol,
query.start_date,
query.end_date,
query.limit,
query.use_cache,
)
@staticmethod
def transform_data(
query: SecInsiderTradingQueryParams,
data: list[dict],
**kwargs: Any,
) -> list[SecInsiderTradingData]:
"""Transform the data."""
return [SecInsiderTradingData.model_validate(d) for d in data]

649
openbb_platform/providers/sec/openbb_sec/utils/form4.py Normal file
View File

@@ -0,0 +1,649 @@
"""Module for handling Form 4 data, by company, from the SEC."""
import logging
from datetime import date as dateType
from typing import Optional
from openbb_core.app.model.abstract.error import OpenBBError
SEC_HEADERS: dict[str, str] = {
"User-Agent": "Jesus Window Washing jesus@stainedglass.com",
"Accept-Encoding": "gzip, deflate",
"Host": "www.sec.gov",
}
field_map = {
"filing_date": "filing_date",
"symbol": "symbol",
"form": "form",
"owner": "owner_name",
"owner_cik": "owner_cik",
"issuer": "company_name",
"issuer_cik": "company_cik",
"isDirector": "director",
"isOfficer": "officer",
"isTenPercentOwner": "ten_percent_owner",
"isOther": "other",
"otherText": "other_text",
"officerTitle": "owner_title",
"securityTitle": "security_type",
"transactionDate": "transaction_date",
"footnote": "footnote",
"transactionShares": "securities_transacted",
"transactionPricePerShare": "transaction_price",
"transactionTotalValue": "transaction_value",
"transactionCode": "transaction_type",
"transactionAcquiredDisposedCode": "acquisition_or_disposition",
"sharesOwnedFollowingTransaction": "securities_owned",
"valueOwnedFollowingTransaction": "value_owned",
"transactionTimeliness": "transaction_timeliness",
"directOrIndirectOwnership": "ownership_type",
"natureOfOwnership": "nature_of_ownership",
"conversionOrExercisePrice": "conversion_exercise_price",
"exerciseDate": "exercise_date",
"expirationDate": "expiration_date",
"deemedExecutionDate": "deemed_execution_date",
"underlyingSecurityTitle": "underlying_security_title",
"underlyingSecurityShares": "underlying_security_shares",
"underlyingSecurityValue": "underlying_security_value",
}
timeliness_map = {
"E": "Early",
"L": "Late",
"Empty": "On-time",
}
transaction_code_map = {
"A": "Grant, award or other acquisition pursuant to Rule 16b-3(d)",
"C": "Conversion of derivative security",
"D": "Disposition to the issuer of issuer equity securities pursuant to Rule 16b-3(e)",
"E": "Expiration of short derivative position",
"F": (
"Payment of exercise price or tax liability by delivering or withholding securities incident to the receipt, "
"exercise or vesting of a security issued in accordance with Rule 16b-3"
),
"G": "Bona fide gift",
"H": "Expiration (or cancellation) of long derivative position with value received",
"I": (
"Discretionary transaction in accordance with Rule 16b-3(f) "
"resulting in acquisition or disposition of issuer securities"
),
"J": "Other acquisition or disposition (describe transaction)",
"L": "Small acquisition under Rule 16a-6",
"M": "Exercise or conversion of derivative security exempted pursuant to Rule 16b-3",
"O": "Exercise of out-of-the-money derivative security",
"P": "Open market or private purchase of non-derivative or derivative security",
"S": "Open market or private sale of non-derivative or derivative security",
"U": "Disposition pursuant to a tender of shares in a change of control transaction",
"W": "Acquisition or disposition by will or the laws of descent and distribution",
"X": "Exercise of in-the-money or at-the-money derivative security",
"Z": "Deposit into or withdrawal from voting trust",
}
def get_logger():
"""Get the logger."""
logger_instance = logging.getLogger("openbb.sec")
handler = logging.StreamHandler()
handler.setLevel(logging.INFO)
formatter = logging.Formatter("\n%(message)s\n")
handler.setFormatter(formatter)
logger_instance.addHandler(handler)
logger_instance.setLevel(logging.INFO)
return logger_instance
logger = get_logger()
def setup_database(conn):
"""Create a caching database for Form 4 data."""
create_table_query = """
CREATE TABLE IF NOT EXISTS form4_data (
filing_date DATE,
symbol TEXT,
form TEXT,
owner_name TEXT,
owner_cik TEXT,
company_name TEXT,
company_cik TEXT,
director BOOLEAN,
officer BOOLEAN,
ten_percent_owner BOOLEAN,
other BOOLEAN,
other_text TEXT,
owner_title TEXT,
security_type TEXT,
transaction_date DATE,
transaction_type TEXT,
acquisition_or_disposition TEXT,
footnote TEXT,
securities_transacted REAL,
transaction_price MONEY,
transaction_value MONEY,
securities_owned REAL,
value_owned MONEY,
transaction_timeliness TEXT,
ownership_type TEXT,
nature_of_ownership TEXT,
conversion_exercise_price MONEY,
exercise_date DATE,
expiration_date DATE,
deemed_execution_date DATE,
underlying_security_title TEXT,
underlying_security_shares REAL,
underlying_security_value MONEY,
filing_url TEXT NOT NULL
);
"""
conn.execute(create_table_query)
conn.commit()
def add_missing_column(conn, column_name):
"""Add a missing column to the form4_data table."""
missing_type = (
"MONEY"
if "price" in column_name or "value" in column_name
else (
"REAL"
if "shares" in column_name
else (
"BOOLEAN"
if "is_" in column_name
else "DATE" if "date" in column_name else "TEXT"
)
)
)
cursor = conn.cursor()
cursor.execute(f"ALTER TABLE form4_data ADD COLUMN {column_name} {missing_type}")
conn.commit()
def compress_db(db_path):
"""Compress the database file."""
# pylint: disable=import-outside-toplevel
import gzip
import shutil
with open(db_path, "rb") as f_in, gzip.open(f"{db_path}.gz", "wb") as f_out:
shutil.copyfileobj(f_in, f_out)
def decompress_db(db_path):
"""Decompress the database file."""
# pylint: disable=import-outside-toplevel
import gzip
import shutil
with gzip.open(f"{db_path}.gz", "rb") as f_in, open(db_path, "wb") as f_out:
shutil.copyfileobj(f_in, f_out)
def close_db(conn, db_path):
"""Sort the table by "date" before closing the connection and compressing the database."""
# pylint: disable=import-outside-toplevel
import os
conn.execute(
"CREATE TABLE IF NOT EXISTS form4_data_sorted AS SELECT * FROM form4_data ORDER BY filing_date"
)
conn.execute("DROP TABLE form4_data")
conn.execute("ALTER TABLE form4_data_sorted RENAME TO form4_data")
conn.commit()
conn.close()
compress_db(db_path)
os.remove(db_path)
async def get_form_4_urls(
symbol,
start_date: Optional[dateType] = None,
end_date: Optional[dateType] = None,
use_cache: bool = True,
):
"""Get the form 4 URLs for a symbol."""
# pylint: disable=import-outside-toplevel
from openbb_sec.models.company_filings import SecCompanyFilingsFetcher
fetcher = SecCompanyFilingsFetcher()
form_4 = await fetcher.fetch_data(
dict(
symbol=symbol,
form_type="4",
provider="sec",
limit=0,
use_cache=use_cache,
),
{},
)
start_date = (
start_date
if isinstance(start_date, dateType)
else (
dateType.fromisoformat(start_date) # type: ignore
if start_date and isinstance(start_date, str)
else None
)
)
end_date = (
end_date
if isinstance(end_date, dateType)
else (
dateType.fromisoformat(end_date) # type: ignore
if end_date and isinstance(end_date, str)
else None
)
)
urls: list = []
for item in form_4:
if start_date and item.filing_date < start_date:
continue
if end_date and item.report_date > end_date:
continue
to_replace = f"{item.primary_doc.split('/')[0]}/"
form_url = item.report_url.replace(to_replace, "")
if form_url.endswith(".xml"):
urls.append(form_url)
return urls
def clean_xml(xml_content):
"""Clean the XML content."""
# pylint: disable=import-outside-toplevel
import re
xml_content = re.sub(r"\\", "", xml_content)
xml_content = xml_content.replace("/s/ ", "")
xml_content = re.sub(r"&(?!amp;|lt;|gt;|quot;|apos;)", "&amp;", xml_content)
return xml_content
async def get_form_4_data(url) -> dict:
"""Get the form 4 data."""
# pylint: disable=import-outside-toplevel
from warnings import warn # noqa
from xmltodict import parse
from openbb_core.provider.utils.helpers import amake_request
async def response_callback(response, _):
"""Response callback function."""
return await response.read()
response = await amake_request(
url,
headers=SEC_HEADERS,
response_callback=response_callback,
timeout=30,
) # type: ignore
response_text = response.decode("utf-8")
if "Traffic Limit" in response_text:
raise OpenBBError(
"You've exceeded the SEC's traffic limit. Access will be limited for 10 minutes."
" Reduce the number of requests by using a more specific date range."
)
cleaned_response = clean_xml(response_text)
try:
xml_data = parse(cleaned_response)
except Exception as e:
warn(f"Error parsing XML from {url}: {e}")
return {}
return (
xml_data.get("ownershipDocument") if xml_data.get("ownershipDocument") else {}
)
async def parse_form_4_data( # noqa: PLR0915, PLR0912 # pylint: disable=too-many-branches
data,
):
"""Parse the Form 4 data."""
owner = data.get("reportingOwner", {})
owners = ""
ciks = ""
if isinstance(owner, list):
owners = ";".join(
[d.get("reportingOwnerId", {}).get("rptOwnerName") for d in owner]
)
ciks = ";".join(
[d.get("reportingOwnerId", {}).get("rptOwnerCik") for d in owner]
)
issuer = data.get("issuer", {})
owner_relationship = (
owner.get("reportingOwnerRelationship", {})
if isinstance(owner, dict)
else (
owner[0].get("reportingOwnerRelationship", {})
if isinstance(owner, list)
else {}
)
)
signature_data = data.get("ownerSignature")
if signature_data and isinstance(signature_data, dict):
signature_date = signature_data.get("signatureDate")
elif signature_data and isinstance(signature_data, list):
signature_date = signature_data[0].get("signatureDate")
else:
signature_date = None
footnotes = data.get("footnotes", {})
if footnotes:
footnote_items = footnotes.get("footnote")
if isinstance(footnote_items, dict):
footnote_items = [footnote_items]
footnotes = {item["@id"]: item["#text"] for item in footnote_items}
metadata = {
"filing_date": signature_date or data.get("periodOfReport"),
"symbol": issuer.get("issuerTradingSymbol", "").upper(),
"form": data.get("documentType"),
"owner": (
owners if owners else owner.get("reportingOwnerId", {}).get("rptOwnerName")
),
"owner_cik": (
ciks if ciks else owner.get("reportingOwnerId", {}).get("rptOwnerCik")
),
"issuer": issuer.get("issuerName"),
"issuer_cik": issuer.get("issuerCik"),
**owner_relationship,
}
results: list = []
if data.get("nonDerivativeTable") and (
data["nonDerivativeTable"].get("nonDerivativeTransaction")
or data["nonDerivativeTable"].get("nonDerivativeHolding")
):
temp_table = data["nonDerivativeTable"]
tables = (
temp_table["nonDerivativeTransaction"]
if temp_table.get("nonDerivativeTransaction")
else temp_table["nonDerivativeHolding"]
)
parsed_table1: list = []
if isinstance(tables, dict):
tables = [tables]
for table in tables:
if isinstance(table, str):
continue
new_row = {**metadata}
for key, value in table.items():
if key == "transactionCoding":
new_row["transaction_type"] = value.get("transactionCode")
new_row["form"] = (
value.get("transactionFormType") or metadata["form"]
)
elif isinstance(value, dict):
if "footnoteId" in value:
if isinstance(value["footnoteId"], list):
ids = [item["@id"] for item in value["footnoteId"]]
footnotes = (
"; ".join(
[
footnotes.get(footnote_id, "")
for footnote_id in ids
]
)
if isinstance(footnotes, dict)
else footnotes
)
new_row["footnote"] = footnotes
else:
footnote_id = value["footnoteId"]["@id"]
new_row["footnote"] = (
(
footnotes
if isinstance(footnotes, str)
else footnotes.get(footnote_id)
)
if footnotes
else None
)
for k, v in value.items():
if k == "value":
new_row[key] = v
if isinstance(v, dict):
if "footnoteId" in v:
if isinstance(v["footnoteId"], list):
ids = [item["@id"] for item in v["footnoteId"]]
footnotes = (
footnotes
if isinstance(footnotes, str)
else (
"; ".join(
[
footnotes.get(footnote_id)
for footnote_id in ids
]
)
if footnotes
else None
)
)
new_row["footnote"] = footnotes
else:
footnote_id = v["footnoteId"]["@id"]
new_row["footnote"] = (
(
footnotes
if isinstance(footnotes, str)
else footnotes.get(footnote_id)
)
if footnotes
else None
)
for k1, v1 in v.items():
if k1 == "value":
new_row[k] = v1
if new_row:
parsed_table1.append(new_row)
results.extend(parsed_table1)
if (
data.get("derivativeTable")
and data["derivativeTable"].get("derivativeTransaction")
) or data.get("derivativeSecurity"):
parsed_table2: list = []
tables = (
data["derivativeSecurity"]
if data.get("derivativeSecurity")
else data["derivativeTable"]["derivativeTransaction"]
)
if isinstance(tables, dict):
tables = [tables]
for table in tables:
if isinstance(table, str):
continue
new_row = {**metadata}
for key, value in table.items():
if key == "transactionCoding":
new_row["transaction_type"] = value.get("transactionCode")
new_row["form"] = (
value.get("transactionFormType") or metadata["form"]
)
elif isinstance(value, dict):
for k, v in value.items():
if k == "value":
new_row[key] = v
if isinstance(v, dict):
for k1, v1 in v.items():
if k1 == "value":
new_row[k] = v1
t_value = new_row.pop("transactionValue", None)
if t_value:
new_row["transactionTotalValue"] = t_value
parsed_table2.append(new_row)
results.extend(parsed_table2)
return results
async def download_data(urls, use_cache: bool = True): # noqa: PLR0915
"""Get the Form 4 data from a list of URLs."""
# pylint: disable=import-outside-toplevel
import asyncio # noqa
import os
import sqlite3
from numpy import nan
from openbb_core.app.utils import get_user_cache_directory
from pandas import DataFrame
results: list = []
non_cached_urls: list = []
try:
if use_cache is True:
db_dir = f"{get_user_cache_directory()}/sql"
db_path = f"{db_dir}/sec_form4.db"
# Decompress the database file
if os.path.exists(f"{db_path}.gz"):
decompress_db(db_path)
os.makedirs(db_dir, exist_ok=True)
try:
conn = sqlite3.connect(db_path)
setup_database(conn)
cached_data = get_cached_data(urls, conn)
cached_urls = {entry["filing_url"] for entry in cached_data}
for url in urls:
if url not in cached_urls:
non_cached_urls.append(url)
except sqlite3.DatabaseError as e:
logger.info("Error connecting to the database.")
retry_input = input(
"Would you like to retry with a new database? (y/n): "
)
if retry_input.lower() == "y":
faulty_db_path = f"{db_path}.faulty"
os.rename(db_path, faulty_db_path)
logger.info("Renamed faulty database to %s", faulty_db_path)
db_path = f"{db_dir}/sec_form4.db"
conn = sqlite3.connect(db_path)
setup_database(conn)
cached_data = get_cached_data(urls, conn)
cached_urls = {entry["filing_url"] for entry in cached_data}
for url in urls:
if url not in cached_urls:
non_cached_urls.append(url)
else:
raise OpenBBError(e) from e
results.extend(cached_data)
elif use_cache is False:
non_cached_urls = urls
async def get_one(url):
"""Get the data for one URL."""
data = await get_form_4_data(url)
result = await parse_form_4_data(data)
if not result and use_cache is True:
df = DataFrame([{"filing_url": url}])
df.to_sql("form4_data", conn, if_exists="append", index=False)
if result:
df = DataFrame(result)
df.loc[:, "filing_url"] = url
df = df.replace({nan: None}).rename(columns=field_map)
try:
if use_cache is True:
df.to_sql("form4_data", conn, if_exists="append", index=False)
except sqlite3.DatabaseError as e:
if "no column named" in str(e):
missing_column = (
str(e).split("no column named ")[1].split(" ")[0]
)
missing_column = field_map.get(missing_column, missing_column)
add_missing_column(conn, missing_column)
df.to_sql("form4_data", conn, if_exists="append", index=False)
else:
raise OpenBBError(e) from e
results.extend(df.replace({nan: None}).to_dict(orient="records"))
time_estimate = (len(non_cached_urls) / 7) * 1.8
logger.info(
"Found %d total filings and %d"
" uncached entries to download, estimated download time: %d seconds.",
len(urls),
len(non_cached_urls),
round(time_estimate),
)
min_warn_time = 10
if time_estimate > min_warn_time:
logger.info(
"Warning: This function is not intended for mass data collection."
" Long download times are due to limitations with concurrent downloads from the SEC."
"\n\nReduce the number of requests by using a more specific date range."
)
if len(non_cached_urls) > 0:
async with asyncio.Semaphore(8):
for url_chunk in [
non_cached_urls[i : i + 8]
for i in range(0, len(non_cached_urls), 8)
]:
await asyncio.gather(*[get_one(url) for url in url_chunk])
await asyncio.sleep(1.125)
if use_cache is True:
close_db(conn, db_path)
results = [entry for entry in results if entry.get("filing_date")]
return sorted(results, key=lambda x: x["filing_date"], reverse=True)
except Exception as e: # pylint: disable=broad-except
if use_cache is True:
close_db(conn, db_path)
raise OpenBBError(
f"Unexpected error while downloading and processing data -> {e.__class__.__name__}: {e}"
) from e
def get_cached_data(urls, conn):
"""Retrieve cached data for a list of URLs."""
# pylint: disable=import-outside-toplevel
from numpy import nan
from pandas import read_sql
placeholders = ", ".join("?" for _ in urls)
query = f"SELECT * FROM form4_data WHERE filing_url IN ({placeholders})" # noqa
df = read_sql(query, conn, params=urls)
return df.replace({nan: None}).to_dict(orient="records") if not df.empty else []
async def get_form_4(
symbol,
start_date: Optional[dateType] = None,
end_date: Optional[dateType] = None,
limit: Optional[int] = None,
use_cache: bool = True,
) -> list[dict]:
"""Get the Form 4 data by ticker symbol or CIK number."""
# pylint: disable=import-outside-toplevel
import asyncio
try:
urls = await get_form_4_urls(symbol, start_date, end_date, use_cache)
if limit is not None:
urls = urls[:limit]
data = await download_data(urls, use_cache)
except asyncio.TimeoutError as e:
raise OpenBBError(
"A timeout error occurred while downloading the data. Please try again."
) from e
if not data:
raise OpenBBError(f"No Form 4 data was returned for {symbol}.")
return data

5600
openbb_platform/providers/sec/tests/record/http/test_sec_fetchers/test_sec_insider_trading_fetcher_urllib3_v1.yaml vendored Normal file
View File

File diff suppressed because it is too large Load Diff

5600
openbb_platform/providers/sec/tests/record/http/test_sec_fetchers/test_sec_insider_trading_fetcher_urllib3_v2.yaml vendored Normal file
View File

File diff suppressed because it is too large Load Diff

16
openbb_platform/providers/sec/tests/test_sec_fetchers.py
View File

@@ -11,6 +11,7 @@ from openbb_sec.models.equity_ftd import SecEquityFtdFetcher
from openbb_sec.models.equity_search import SecEquitySearchFetcher
from openbb_sec.models.etf_holdings import SecEtfHoldingsFetcher
from openbb_sec.models.form_13FHR import SecForm13FHRFetcher
from openbb_sec.models.insider_trading import SecInsiderTradingFetcher
from openbb_sec.models.institutions_search import SecInstitutionsSearchFetcher
from openbb_sec.models.latest_financial_reports import SecLatestFinancialReportsFetcher
from openbb_sec.models.rss_litigation import SecRssLitigationFetcher
@@ -170,3 +171,18 @@ def test_sec_latest_financial_reports_fetcher(credentials=test_credentials):
fetcher = SecLatestFinancialReportsFetcher()
result = fetcher.test(params, credentials)
assert result is None
@pytest.mark.record_http
def test_sec_insider_trading_fetcher(credentials=test_credentials):
"""Test the SEC Insider Trading fetcher."""
params = {
"symbol": "AAPL",
"start_date": date(2024, 6, 30),
"end_date": date(2024, 9, 30),
"use_cache": False,
}
fetcher = SecInsiderTradingFetcher()
result = fetcher.test(params, credentials)
assert result is None