66 add methods for all company info

alphavantage_api_client/__init__.py

@@ -1,4 +1,5 @@
 from alphavantage_api_client.client import AlphavantageClient
 from alphavantage_api_client.models import GlobalQuote, Quote, AccountingReport, CompanyOverview, RealGDP, \
-    CsvNotSupported, TickerSearch, MarketStatus, MarketMovers, NewsAndSentiment
+    CsvNotSupported, TickerSearch, MarketStatus, MarketMovers, NewsAndSentiment, EarningsCalendar\
+    , EarningsCalendarItem,IpoCalendarItem, IpoCalendar
 from alphavantage_api_client.ticker import Ticker

alphavantage_api_client/client.py

@@ -6,11 +6,13 @@
 from .response_validation_rules import ValidationRuleChecks
 import json
 from alphavantage_api_client.models import GlobalQuote, Quote, AccountingReport, CompanyOverview, RealGDP, \
-    CsvNotSupported, TickerSearch, MarketStatus, NewsAndSentiment, MarketMovers
+    CsvNotSupported, TickerSearch, MarketStatus, NewsAndSentiment, MarketMovers, EarningsCalendar\
+    , IpoCalendarItem, IpoCalendar
 import copy
 import logging
 import hashlib
 from typing import Optional, Union
+import csv
 
 
 class ApiKeyNotFound(Exception):
@@ -681,6 +683,50 @@ def get_top_gainers_and_losers(self) -> MarketMovers:
 
         return MarketMovers.parse_obj(json_response)
 
+    def get_earnings_calendar(self, event: dict) -> EarningsCalendar:
+        """
+        This API returns a list of company earnings expected in the next 3, 6, or 12 months.
+        Returns:
+
+        """
+        defaults = {
+            "function": "EARNINGS_CALENDAR",
+            "horizon": "3month",
+            "datatype": "csv"
+        }
+        json_request = self.__create_api_request_from__(defaults, event)
+        json_response = self.get_data_from_alpha_vantage(json_request, self.__retry__)
+        records = json_response['csv'].split("\n")
+        header = records.pop(0)
+        headers = header.split(",")
+        reader = csv.DictReader(records, delimiter=',')
+        reader.fieldnames = headers
+        items = list(reader)
+        json_response['data'] = items
+
+        return EarningsCalendar.parse_obj(json_response)
+
+    def get_ipo_calendar(self) -> IpoCalendar:
+        """
+        This API returns a list of company earnings expected in the next 3, 6, or 12 months.
+        Returns:
+
+        """
+        json_request = {
+            "function": "IPO_CALENDAR",
+            "datatype": "csv"
+        }
+
+        json_response = self.get_data_from_alpha_vantage(json_request, self.__retry__)
+        records = json_response['csv'].split("\n")
+        header = records.pop(0)
+        headers = header.split(",")
+        reader = csv.DictReader(records, delimiter=',')
+        reader.fieldnames = headers
+        items = list(reader)
+        json_response['data'] = items
+
+        return IpoCalendar.parse_obj(json_response)
 
     def get_news_and_sentiment(self, event: dict) -> NewsAndSentiment:
         """

alphavantage_api_client/models.py

@@ -43,6 +43,30 @@ class MarketMovers(BaseResponse):
     meta_data: Optional[str] = Field(str, alias='metadata')
     most_actively_traded: list[dict]
 
+class EarningsCalendarItem(BaseModel):
+    symbol: str
+    name: str
+    report_date: str = Field(str,alias='reportDate')
+    fiscal_date_ending: str = Field(str,alias='fiscalDateEnding')
+    estimate: Optional[float]
+    currency: Optional[str]
+
+class IpoCalendarItem(BaseModel):
+    symbol: str
+    name: str
+    ipo_date: str = Field(str,alias="ipoDate")
+    price_range_low: Optional[float] = Field(float, alias="priceRangeLow")
+    price_range_high: Optional[float] = Field(float, alias="priceRangeHigh")
+    currency: Optional[str]
+    exchange: Optional[str]
+
+class IpoCalendar(BaseResponse):
+    data: Optional[list[IpoCalendarItem]] = Field([])
+
+class EarningsCalendar(BaseResponse):
+    symbol: str
+    data: Optional[list[EarningsCalendarItem]] = Field([])
+
 class Quote(BaseQuote):
     """
     data is this clients abstraction of the response from alpha vantage. Time Series, Technical Indicator

tests/test_single_client_cache_integration.py

@@ -4,7 +4,7 @@
     , NewsAndSentiment
 import logging
 import json
-
+import csv
 client = AlphavantageClient().should_retry_once().use_simple_cache()
 
 
@@ -529,3 +529,52 @@ def test_get_market_movers():
         assert "change_amount" in item, "change_amount not found within result"
         assert "change_percentage" in item, "change_percentage not found within result"
         assert "volume" in item, "volume not found within result"
+
+@pytest.mark.integration
+def test_get_earnings_calendar():
+    symbols = ["IBM", "AAPL", "AMZN", "MSFT", "TSLA", "SYM"]
+
+    for symbol in symbols:
+        event = {
+            "symbol": symbol
+        }
+        earnings_calendar = client.get_earnings_calendar(event)
+        assert earnings_calendar.success, f"success was found to be True which is unexpected: {earnings_calendar.error_message}"
+        assert not earnings_calendar.limit_reached, f"limit_reached is true {earnings_calendar.error_message}"
+        assert len(earnings_calendar.csv), "csv is not defined within response"
+        assert len(earnings_calendar.data), "data is not defined within response"
+
+        for item in earnings_calendar.data:
+            print(item.json())
+
+@pytest.mark.integration
+def test_get_earnings_calendar():
+    symbols = ["IBM", "AAPL", "AMZN", "MSFT", "TSLA", "SYM"]
+
+    for symbol in symbols:
+        event = {
+            "symbol": symbol
+        }
+        earnings_calendar = client.get_earnings_calendar(event)
+        assert earnings_calendar.success, f"success was found to be True which is unexpected: {earnings_calendar.error_message}"
+        assert not earnings_calendar.limit_reached, f"limit_reached is true {earnings_calendar.error_message}"
+        assert len(earnings_calendar.csv), "csv is not defined within response"
+        assert len(earnings_calendar.data), "data is not defined within response"
+
+        for item in earnings_calendar.data:
+            print(item.json())
+
+def test_get_ipo_calendar():
+
+    ipo_calendar = client.get_ipo_calendar()
+    assert ipo_calendar.success, f"success was found to be True which is unexpected: {ipo_calendar.error_message}"
+    assert not ipo_calendar.limit_reached, f"limit_reached is true {ipo_calendar.error_message}"
+    assert len(ipo_calendar.csv), "csv is not defined within response"
+    assert len(ipo_calendar.data), "data is not defined within response"
+
+    for item in ipo_calendar.data:
+        assert len(item.symbol), "csv is not defined within response"
+        assert len(item.ipo_date), "ipo_date is not defined within response"
+        assert len(item.name), "name is not defined within response"
+
+