67 lines
2.7 KiB
Python
67 lines
2.7 KiB
Python
import requests
|
|
import logging
|
|
from time import sleep
|
|
from typing import List, Dict, Optional
|
|
from datetime import datetime, date
|
|
import json
|
|
|
|
class FinancialsCollector:
|
|
def __init__(self, api_key: str):
|
|
self.api_key = api_key
|
|
self.base_url = "https://financialmodelingprep.com/api/v3"
|
|
self.base_url_v4 = "https://financialmodelingprep.com/api/v4"
|
|
self.session = self._create_session()
|
|
self.rate_limit_pause = 0.2
|
|
|
|
def _create_session(self) -> requests.Session:
|
|
session = requests.Session()
|
|
return session
|
|
|
|
def _make_request(self, url: str, params: Dict) -> List[Dict]:
|
|
try:
|
|
sleep(self.rate_limit_pause)
|
|
response = self.session.get(url, params=params, timeout=10)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
logging.error(f"API request error for {url}: {str(e)}")
|
|
return []
|
|
|
|
def _parse_date(self, date_str: str) -> Optional[date]:
|
|
try:
|
|
formats = ['%Y-%m-%d', '%Y-%m', '%Y']
|
|
for fmt in formats:
|
|
try:
|
|
return datetime.strptime(date_str, fmt).date()
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
except Exception as e:
|
|
logging.error(f"Error parsing date {date_str}: {e}")
|
|
return None
|
|
|
|
def get_latest_financial_statement_date(self, ticker: str) -> Optional[date]:
|
|
try:
|
|
params = {'apikey': self.api_key, 'symbol': ticker}
|
|
response = self._make_request(f"{self.base_url_v4}/financial-reports-dates", params)
|
|
if response and isinstance(response, list) and len(response) > 0:
|
|
date_str = response[0].get('date')
|
|
if date_str:
|
|
return self._parse_date(date_str)
|
|
return None
|
|
except Exception as e:
|
|
logging.error(f"Error getting latest financial date for {ticker}: {e}")
|
|
return None
|
|
|
|
def get_income_statement(self, ticker: str, period: str = 'quarter') -> List[Dict]:
|
|
params = {'apikey': self.api_key, 'period': period}
|
|
return self._make_request(f"{self.base_url}/income-statement/{ticker}", params)
|
|
|
|
def get_balance_sheet(self, ticker: str, period: str = 'quarter') -> List[Dict]:
|
|
params = {'apikey': self.api_key, 'period': period}
|
|
return self._make_request(f"{self.base_url}/balance-sheet-statement/{ticker}", params)
|
|
|
|
def get_cash_flow(self, ticker: str, period: str = 'quarter') -> List[Dict]:
|
|
params = {'apikey': self.api_key, 'period': period}
|
|
return self._make_request(f"{self.base_url}/cash-flow-statement/{ticker}", params)
|