fundamental data

public_fundamental.py

import asyncio
from typing import List, Generator, Callable, Any, Dict, Union
from flask import jsonify
from flask_apispec import doc, marshal_with
from google.cloud import ndb
from src.cache.cache_manager import app_cache
from src.exceptions import status_codes
from src.models.fundamental import (FundamentalGeneral, GeneralAddress, GeneralContact, GeneralListings,
                                    GeneralOfficers,
                                    FundamentalHighlights, FundamentalValuation, FundamentalSharesStats,
                                    FundamentalTechnicals,
                                    FundamentalSplitsDividends, FundamentalAnalystRanks, InstitutionHolders,
                                    FundsHolders,
                                    FundamentalInsiderTransactions, FundamentalOutStandingShares,
                                    AnnualOutStandingShares, QuarterlyOutStandingShares,
                                    HistoryEarnings, TrendEarnings, AnnualEarnings, FundamentalFinancials,
                                    QuarterlyBalanceSheets, AnnualBalanceSheets, FundamentalDividendsByYear)

from src.schemas.fundamentals.generaldetails import GeneralResponseSchema
from src.schemas.fundamentals.public_schema import PublicFundamentalsResponseSchema
from src.views import ViewModel, CACHE_TIMEOUT_DAY, MResponse
from collections import ChainMap

class GetAllFundamentalGeneralAPI(ViewModel):
    """
    **GetAllFundamentalGeneralAPI**
        will return only general fundamental data for all companies listed in stock exchanges
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    @doc(description="return general fundamental data for all companies")
    @marshal_with(GeneralResponseSchema)
    @app_cache.cache.memoize(timeout=CACHE_TIMEOUT_DAY)
    def get(self) -> MResponse:
        """
            return general fundamental data for all companies
        :return: Tuple
        """

        fundamental_list = await ndb.transaction_async(FundamentalGeneral.get_all())
        _payload = [fundamental.to_dict() for fundamental in fundamental_list]
        _message: str = "fundamental data found"
        return jsonify(dict(status=True, message=_message, payload=_payload)), status_codes.status_ok_code


# noinspection DuplicatedCode
class GetCompanyFundamentalDataAPI(ViewModel):
    """
    **GetCompanyFundamentalDataAPI**
        Get complete company fundamental data
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    @doc(description="for each stock code return the fundamental data for the company")
    @marshal_with(PublicFundamentalsResponseSchema)
    @app_cache.cache.memoize(timeout=CACHE_TIMEOUT_DAY)
    async def get(self, stock_code: str) -> MResponse:

        payload = await ndb.transaction_async(callback=self.get_fundamental_data(stock_code=stock_code), retries=3)

        return jsonify(status=True, payload=payload), status_codes.status_ok_code

    # noinspection PyUnresolvedReferences
    # noinspection PyTypeChecker
    @ndb.toplevel
    async def get_fundamental_data(self, stock_code : str) -> dict[str, str]:
        """
            **get_fundamental_data**
                obtain the entire fundamental record for a specific company
        :param stock_code:
        :return:
        """

        _tasks, fundamental_inst = await self.create_tasks(stock_code=stock_code)
        # please ensure to use gather here as we need the results in order
        payload = fundamental_inst.to_dict()

        results = await asyncio.gather(*_tasks)
        _gen = self.instance_gen(results=results)

        payload = await self.create_payload(_gen=_gen, payload=payload)

        _message: str = f"fundamental data for {stock_code} found"
        return jsonify(doc(status=True, payload=payload, message=_message)), status_codes.status_ok_code

    @staticmethod
    async def create_payload(_gen, payload):
        """
            create fundamental data payload
        :param _gen:
        :param payload:
        :return: payload: Dict
        """
        general_address_inst = next(_gen)
        contacts_inst = next(_gen)
        listing_inst_list: list = next(_gen)
        officers_inst_list: list = next(_gen)
        highlights_inst = next(_gen)
        valuation_inst = next(_gen)
        share_stats = next(_gen)
        technicals_inst = next(_gen)
        splits_inst = next(_gen)
        fundamentals_analyst_ranks: list = next(_gen)
        institution_holders_list: list = next(_gen)
        funds_holders_list: list = next(_gen)
        insiders_transaction_list: list = next(_gen)
        outstanding_shares_list: list = next(_gen)
        annual_outstanding_shares_list: list = next(_gen)
        quarterly_outstanding_shares_list: list = next(_gen)
        historical_earnings_list: list = next(_gen)
        trend_earnings_list: list = next(_gen)
        annual_earnings_list: list = next(_gen)
        financials_list: list = next(_gen)
        quarterly_balance_sheets_list: list = next(_gen)
        annual_balance_sheets_list: list = next(_gen)
        updates = {key: value.to_dict() for key, value in [
            ('address', general_address_inst),
            ('contacts', contacts_inst),
            ('listing', [listing.to_dict() for listing in listing_inst_list]),
            ('officers', [officer.to_dict() for officer in officers_inst_list]),
            ('highlights', highlights_inst),
            ('valuation', valuation_inst),
            ('share_stats', share_stats),
            ('technicals', technicals_inst),
            ('splits', splits_inst),
            ('analyst_ranks', fundamentals_analyst_ranks),
            ('institution_holders', [holder.to_dict() for holder in institution_holders_list]),
            ('funds_holders', [holder.to_dict() for holder in funds_holders_list]),
            ('insider_transactions', [transaction.to_dict() for transaction in insiders_transaction_list]),
            ('outstanding_shares', [share.to_dict() for share in outstanding_shares_list]),
            ('annual_outstanding_shares', [share.to_dict() for share in annual_outstanding_shares_list]),
            ('quarterly_outstanding_shares', [share.to_dict() for share in quarterly_outstanding_shares_list]),
            ('historical_earnings', [earning.to_dict() for earning in historical_earnings_list]),
            ('trend_earnings', [earning.to_dict() for earning in trend_earnings_list]),
            ('annual_earnings', [earning.to_dict() for earning in annual_earnings_list]),
            ('financials', [financial.to_dict() for financial in financials_list]),
            ('annual_balance_sheets', [balance_sheet.to_dict() for balance_sheet in annual_balance_sheets_list]),
            ('quarterly_balance_sheets', [balance_sheet.to_dict() for balance_sheet in quarterly_balance_sheets_list]),
        ]}
        payload = dict(ChainMap(payload, updates))
        return payload

    async def create_tasks(self, stock_code) -> tuple[map, FundamentalGeneral]:
        """
            create tasks which will be used to generate the payload
        :param stock_code:
        :return:
        """
        tasks = [self.get_general_address, self.get_contacts, self.get_general_listing, self.get_general_officers,
                 self.get_fundamental_highlights, self.get_valuations, self.get_share_stats, self.get_technicals,
                 self.get_splits, self.get_analyst_ranks, self.get_institution_holders, self.get_funds_holders,
                 self.get_insiders_transactions, self.get_outstanding_shares, self.get_annual_outstanding_shares,
                 self.get_quarterly_outstanding_shares, self.get_historical_earnings, self.get_trend_earnings,
                 self.get_annual_earnings, self.get_fundamental_financials, self.get_quarterly_balance_sheets,
                 self.get_annual_balance_sheets]
        fundamental_inst = FundamentalGeneral.query(FundamentalGeneral.stock_symbol == stock_code).get_async()
        _id: str = fundamental_inst.fundamental_id
        _tasks = map(lambda _task: _task(_id), tasks)
        return _tasks, fundamental_inst

    @staticmethod
    def instance_gen(results) -> Generator[Union[Dict[str, Any], List[Dict[str, Any]]]]:
        """
            this method will generator the relevant dictionary or list of dictionaries
            depending on the results of the query
        :param results:
        :return:
        """
        for result in results:
            yield result

    @staticmethod
    @ndb.tasklet
    async def get_annual_balance_sheets(_id: str) -> List[AnnualBalanceSheets]:
        annual_sheets = yield AnnualBalanceSheets.query(AnnualBalanceSheets.fundamental_id == _id).fetch_async()
        yield [sheet for sheet in annual_sheets]
        raise ndb.Return(annual_sheets)

    @staticmethod
    @ndb.tasklet
    async def get_quarterly_balance_sheets(_id: str) -> List[QuarterlyBalanceSheets]:
        quarterly_sheets = yield QuarterlyBalanceSheets.query(
            QuarterlyBalanceSheets.fundamental_id == _id).fetch_async()
        yield [quarter for quarter in quarterly_sheets]
        raise ndb.Return(quarterly_sheets)

    @staticmethod
    @ndb.tasklet
    async def get_fundamental_financials(_id: str) -> List[FundamentalFinancials]:
        financials_list = yield FundamentalFinancials.query(FundamentalFinancials.fundamental_id == _id).fetch_async()
        yield [finance for finance in financials_list]
        raise ndb.Return(financials_list)

    @staticmethod
    @ndb.tasklet
    async def get_annual_earnings(_id: str) -> List[AnnualEarnings]:
        annual_earnings = yield AnnualEarnings.query(AnnualEarnings.fundamental_id == _id).fetch_async()
        yield [earning for earning in annual_earnings]
        raise ndb.Return(annual_earnings)

    @staticmethod
    @ndb.tasklet
    async def get_trend_earnings(_id: str) -> List[TrendEarnings]:
        trend_earnings = yield TrendEarnings.query(TrendEarnings.fundamental_id == _id).fetch_async()
        yield [trend for trend in trend_earnings]
        raise ndb.Return(trend_earnings)

    @staticmethod
    @ndb.tasklet
    async def get_historical_earnings(_id: str) -> List[HistoryEarnings]:
        historical_earnings = yield HistoryEarnings.query(HistoryEarnings.fundamental_id == _id).fetch_async()
        yield [historical for historical in historical_earnings]
        raise ndb.Return(historical_earnings)

    @staticmethod
    @ndb.tasklet
    async def get_quarterly_outstanding_shares(_id: str) -> List[QuarterlyOutStandingShares]:
        quarterly_outstanding_shares = yield QuarterlyOutStandingShares.query(
            QuarterlyOutStandingShares.fundamental_id == _id).fetch_async()
        yield [share for share in quarterly_outstanding_shares]
        raise ndb.Return(quarterly_outstanding_shares)

    @staticmethod
    @ndb.tasklet
    async def get_annual_outstanding_shares(_id: str) -> List[AnnualOutStandingShares]:
        annual_outstanding = yield AnnualOutStandingShares.query(
            AnnualOutStandingShares.fundamental_id == _id).fetch_async()
        yield [share for share in annual_outstanding]
        raise ndb.Return(annual_outstanding)

    @staticmethod
    @ndb.tasklet
    async def get_outstanding_shares(_id: str) -> List[FundamentalOutStandingShares]:
        out_standing_shares = yield FundamentalOutStandingShares.query(
            FundamentalOutStandingShares.fundamental_id == _id).fetch_async()
        yield [share for share in out_standing_shares]
        raise ndb.Return(out_standing_shares)

    @staticmethod
    @ndb.tasklet
    async def get_insiders_transactions(_id: str) -> List[FundamentalInsiderTransactions]:
        insiders_transactions = yield FundamentalInsiderTransactions.query(
            FundamentalInsiderTransactions.fundamental_id == _id).fetch_async()
        yield [insider for insider in insiders_transactions]
        raise ndb.Return(insiders_transactions)


    @staticmethod
    @ndb.tasklet
    async def get_funds_holders(_id: str) -> List[FundsHolders]:
        funds_holders = yield FundsHolders.query(FundsHolders.fundamental_id == _id).fetch_async()
        yield [funds for funds in funds_holders]
        raise ndb.Return(funds_holders)

    @staticmethod
    @ndb.tasklet
    async def get_institution_holders(_id: str) -> List[InstitutionHolders]:
        institution_holders_list = yield InstitutionHolders.query(
            InstitutionHolders.fundamental_id == _id).fetch_async()
        yield [institution_ for institution_ in institution_holders_list]
        raise ndb.Return(institution_holders_list)

    @staticmethod
    @ndb.tasklet
    async def get_analyst_ranks(_id: str) -> FundamentalAnalystRanks:
        analyst_rank = yield FundamentalAnalystRanks.query(
            FundamentalAnalystRanks.fundamental_id == _id).get_async()
        raise ndb.Return(analyst_rank)


    @staticmethod
    @ndb.tasklet
    async def get_splits(_id: str) -> FundamentalSplitsDividends:
        splits_dividends = yield FundamentalSplitsDividends.query(
            FundamentalSplitsDividends.fundamental_id == _id).get_async()

        _by_year_list = yield GetCompanyFundamentalDataAPI.splits_by_year(splits_dividends)
        splits_dividends.update(number_of_dividends_by_year=[_by_year.to_dict() for _by_year in _by_year_list])

        raise ndb.Return(splits_dividends)

    @staticmethod
    @ndb.tasklet
    async def splits_by_year(splits_dividends):
        _by_year_list = yield FundamentalDividendsByYear.query(
        FundamentalDividendsByYear.split_dividend_id == splits_dividends.split_dividend_id).fetch_async()
        yield [_year for _year in _by_year_list]
        raise ndb.Return(_by_year_list)


    @staticmethod
    @ndb.tasklet
    async def get_technicals(_id: str) -> FundamentalTechnicals:
        fun_technicals = yield FundamentalTechnicals.query(FundamentalTechnicals.fundamental_id == _id).get_async()
        raise ndb.Return(fun_technicals)

    @staticmethod
    @ndb.tasklet
    async def get_share_stats(_id: str) -> FundamentalSharesStats:
        share_stats = yield FundamentalSharesStats.query(FundamentalSharesStats.fundamental_id == _id).get_async()
        raise ndb.Return(share_stats)

    @staticmethod
    @ndb.tasklet
    async def get_valuations(_id: str) -> FundamentalValuation:
        valuation = yield FundamentalValuation.query(FundamentalValuation.fundamental_id == _id).get_async()
        raise ndb.Return(valuation)

    @staticmethod
    @ndb.tasklet
    async def get_fundamental_highlights(_id: str) -> FundamentalHighlights:
        highlights_inst = yield FundamentalHighlights.query(FundamentalHighlights.fundamental_id == _id).fetch_async()
        raise ndb.Return(highlights_inst)

    @staticmethod
    @ndb.tasklet
    async def get_general_officers(_id: str) -> List[GeneralOfficers]:
        general_officers = yield GeneralOfficers.query(GeneralOfficers.fundamental_id == _id).fetch_async()
        yield [officer for officer in general_officers]
        raise ndb.Return(general_officers)

    @staticmethod
    @ndb.tasklet
    async def get_general_listing(_id: str) -> GeneralListings:
        general_listings = yield GeneralListings.query(GeneralListings.fundamental_id == _id).get_async()
        raise ndb.Return(general_listings)

    @staticmethod
    @ndb.tasklet
    async def get_contacts(_id: str) -> GeneralContact:
        contacts_inst = yield GeneralContact.query(GeneralContact.fundamental_id == _id).get_async()
        raise ndb.Return(contacts_inst)

    @staticmethod
    @ndb.tasklet
    async def get_general_address(_id: str) -> GeneralAddress:
        general_address_inst = yield GeneralAddress.query(GeneralAddress.fundamental_id == _id).get_async()
        raise ndb.Return(general_address_inst)