Move model from sec to finra (#5692)

openbb_platform/providers/finra/openbb_finra/__init__.py

@@ -1,4 +1,5 @@
 """FINRA provider module."""
+from openbb_finra.models.equity_short_interest import FinraShortInterestFetcher
 from openbb_finra.models.otc_aggregate import FinraOTCAggregateFetcher
 from openbb_provider.abstract.provider import Provider
 
@@ -7,5 +8,8 @@
  website="https://finra.org",
  description="Financial Industry Regulatory Authority.",
  required_credentials=None,
- fetcher_dict={"OTCAggregate": FinraOTCAggregateFetcher},
+ fetcher_dict={
+ "OTCAggregate": FinraOTCAggregateFetcher,
+ "EquityShortInterest": FinraShortInterestFetcher,
+ },
 )

openbb_platform/providers/sec/openbb_sec/models/equity_short_interest.py → openbb_platform/providers/finra/openbb_finra/models/equity_short_interest.py

@@ -1,22 +1,22 @@
-"""SEC Company Filings fetcher."""
+"""FINRA Company Filings fetcher."""
 
 import sqlite3
 from typing import Any, Dict, List, Optional
 
+from openbb_finra.utils.data_storage import DB_PATH, prepare_data
 from openbb_provider.abstract.fetcher import Fetcher
 from openbb_provider.standard_models.equity_short_interest import (
  ShortInterestData,
  ShortInterestQueryParams,
 )
-from openbb_sec.utils.data_storage import DB_PATH, prepare_data
 
 
-class SecShortInterestQueryParams(ShortInterestQueryParams):
- """SEC Company Filings Query Params."""
+class FinraShortInterestQueryParams(ShortInterestQueryParams):
+ """Finra Company Filings Query Params."""
 
 
-class SecShortInterestData(ShortInterestData):
- """SEC Short Interest Data."""
+class FinraShortInterestData(ShortInterestData):
+ """Finra Short Interest Data."""
 
  __alias_dict__ = {
  "symbol": "symbolCode",
@@ -32,23 +32,23 @@ class SecShortInterestData(ShortInterestData):
  }
 
 
-class SecShortInterestFetcher(
- Fetcher[SecShortInterestQueryParams, List[SecShortInterestData]]
+class FinraShortInterestFetcher(
+ Fetcher[FinraShortInterestQueryParams, List[FinraShortInterestData]]
 ):
- """SEC Short Interest Fetcher."""
+ """Finra Short Interest Fetcher."""
 
  @staticmethod
- def transform_query(params: Dict[str, Any]) -> SecShortInterestQueryParams:
+ def transform_query(params: Dict[str, Any]) -> FinraShortInterestQueryParams:
  """Transform query params."""
- return SecShortInterestQueryParams(**params)
+ return FinraShortInterestQueryParams(**params)
 
  @staticmethod
  def extract_data(
- query: SecShortInterestQueryParams,
+ query: FinraShortInterestQueryParams,
  credentials: Optional[Dict[str, str]],
  **kwargs: Any,
  ) -> List[Dict]:
- """Extracts the data from the SEC endpoint."""
+ """Extracts the data from the Finra endpoint."""
 
  # Put the data in the cache
  prepare_data()
@@ -81,6 +81,6 @@ def extract_data(
  ]
 
  @staticmethod
- def transform_data(data: List[Dict], **kwargs: Any) -> List[SecShortInterestData]:
+ def transform_data(data: List[Dict], **kwargs: Any) -> List[FinraShortInterestData]:
  """Transforms the data."""
- return [SecShortInterestData.model_validate(d) for d in data]
+ return [FinraShortInterestData.model_validate(d) for d in data]

openbb_platform/providers/sec/openbb_sec/utils/data_storage.py → openbb_platform/providers/finra/openbb_finra/utils/data_storage.py

@@ -7,10 +7,11 @@
 from pathlib import Path
 
 import requests
-from openbb_sec.utils.helpers import get_short_interest_dates
+from openbb_core.app.utils import get_user_cache_directory
+from openbb_finra.utils.helpers import get_short_interest_dates
 from pandas import read_csv
 
-DB_PATH = Path.home() / ".openbb_platform/caches/sec_short_volume.db"
+DB_PATH = Path(get_user_cache_directory()) / "caches/sec_short_volume.db"
 DB_PATH.parent.mkdir(parents=True, exist_ok=True)
 
 

openbb_platform/providers/finra/openbb_finra/utils/helpers.py

@@ -1,4 +1,6 @@
 # Helper functions for FINRA API
+import datetime
+from typing import List
 
 import requests
 
@@ -101,3 +103,71 @@ def get_full_data(symbol, tier: str = "T1", is_ats: bool = True):
  data.append(response.json()[0])
 
  return data
+
+
+def get_adjusted_date(year, month, day):
+ # Get the date
+ date = datetime.date(year, month, day)
+
+ # If the date is a Saturday, subtract one day
+ if date.weekday() == 5:
+ date -= datetime.timedelta(days=1)
+ # If the date is a Sunday, subtract two days
+ elif date.weekday() == 6:
+ date -= datetime.timedelta(days=2)
+
+ return date
+
+
+def get_short_interest_dates() -> List[str]:
+ """
+ Get a list of dates for which the short interest data is available. It is reported on the 15th and the
+ last day of each month,but if the date falls on a weekend, the date is adjusted to the closest friday.
+ """
+
+ def get_adjusted_date(year, month, day):
+ """If the date falls on a weekend, find the closest date"""
+ # Get the date
+ date = datetime.date(year, month, day)
+
+ # If the date is a Saturday, subtract one day
+ if date.weekday() == 5:
+ date -= datetime.timedelta(days=1)
+ # If the date is a Sunday, subtract two days
+ elif date.weekday() == 6:
+ date -= datetime.timedelta(days=2)
+
+ return date
+
+ start_year = 2021
+ today = datetime.date.today() # Get today's date
+ end_year = today.year
+ dates_list = []
+
+ for yr in range(start_year, end_year + 1):
+ start_month = 7 if yr == start_year else 1
+ end_month = 12 if yr < today.year else today.month - 1
+ for month in range(start_month, end_month + 1): # Start from July for 2021
+ # Date for the 15th of the month
+ date_15 = get_adjusted_date(yr, month, 15)
+ dates_list.append(date_15.strftime("%Y%m%d"))
+
+ # Date for the last day of the month
+ if month == 2: # February
+ last_day = (
+ 29 if (yr % 4 == 0 and yr % 100 != 0) or (yr % 400 == 0) else 28
+ )
+ elif month in [4, 6, 9, 11]: # Months with 30 days
+ last_day = 30
+ else: # Months with 31 days
+ last_day = 31
+
+ last_date = get_adjusted_date(yr, month, last_day)
+ dates_list.append(last_date.strftime("%Y%m%d"))
+
+ # Manually replace '20220415' with '20220414' due to holiday
+ if "20220415" in dates_list:
+ index = dates_list.index("20220415")
+ dates_list[index] = "20220414"
+
+ return dates_list