Spaces:
Running
Running
| import os | |
| from sec_api import ExtractorApi | |
| from functools import wraps | |
| from typing import Annotated | |
| from ..utils import SavePathType | |
| from ..data_source import FMPUtils | |
| CACHE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".cache") | |
| def init_sec_api(func): | |
| def wrapper(*args, **kwargs): | |
| global extractor | |
| if os.environ.get("SEC_API_KEY") is None: | |
| print("Please set the environment variable SEC_API_KEY to use sec_api.") | |
| return None | |
| else: | |
| extractor = ExtractorApi(os.environ["SEC_API_KEY"]) | |
| print("Sec Api initialized") | |
| return func(*args, **kwargs) | |
| return wrapper | |
| class SECUtils: | |
| def get_10k_section( | |
| ticker_symbol: Annotated[str, "ticker symbol"], | |
| fyear: Annotated[str, "fiscal year of the 10-K report"], | |
| section: Annotated[ | |
| str | int, | |
| "Section of the 10-K report to extract, should be in [1, 1A, 1B, 2, 3, 4, 5, 6, 7, 7A, 8, 9, 9A, 9B, 10, 11, 12, 13, 14, 15]", | |
| ], | |
| report_address: Annotated[ | |
| str, | |
| "URL of the 10-K report, if not specified, will get report url from fmp api", | |
| ] = None, | |
| save_path: SavePathType = None, | |
| ) -> str: | |
| """ | |
| Get a specific section of a 10-K report from the SEC API. | |
| """ | |
| if isinstance(section, int): | |
| section = str(section) | |
| if section not in [ | |
| "1A", | |
| "1B", | |
| "7A", | |
| "9A", | |
| "9B", | |
| ] + [str(i) for i in range(1, 16)]: | |
| raise ValueError( | |
| "Section must be in [1, 1A, 1B, 2, 3, 4, 5, 6, 7, 7A, 8, 9, 9A, 9B, 10, 11, 12, 13, 14, 15]" | |
| ) | |
| # os.makedirs(f"{self.project_dir}/10k", exist_ok=True) | |
| # report_name = f"{self.project_dir}/10k/section_{section}.txt" | |
| # if USE_CACHE and os.path.exists(report_name): | |
| # with open(report_name, "r") as f: | |
| # section_text = f.read() | |
| # else: | |
| if report_address is None: | |
| report_address = FMPUtils.get_sec_report(ticker_symbol, fyear) | |
| if report_address.startswith("Link: "): | |
| report_address = report_address.lstrip("Link: ").split()[0] | |
| else: | |
| return report_address # debug info | |
| cache_path = os.path.join(CACHE_PATH, f"sec_utils/{ticker_symbol}_{fyear}_{section}.txt") | |
| if os.path.exists(cache_path): | |
| with open(cache_path, "r") as f: | |
| section_text = f.read() | |
| else: | |
| section_text = extractor.get_section(report_address, section, "text") | |
| os.makedirs(os.path.dirname(cache_path), exist_ok=True) | |
| with open(cache_path, "w") as f: | |
| f.write(section_text) | |
| if save_path: | |
| os.makedirs(os.path.dirname(save_path), exist_ok=True) | |
| with open(save_path, "w") as f: | |
| f.write(section_text) | |
| return section_text | |