aggregator.py

from bezinga_caller import bezinga_get
from finub_caller import get_finhub
from marketaux_caller import get_marketaux
from newsapi_caller import get_newsapi
from newsdata_caller import get_newsdata
from vantage_caller import get_vantage
from transformers import pipeline


def get_articles_sentiment(ticker, model):
    pipe = pipeline("text-classification", model=model)

    # getting a list of article of given ticket from different sources
    try:
        bezinga_list = bezinga_get(ticker)
        bezinga_results = pipe(bezinga_list)
    except Exception as e:
        print(e)
        bezinga_results = False

    try:
        newsapi_list = get_newsapi(ticker)
        newsapi_results = pipe(newsapi_list)
    except Exception as e:
        print(e)
        newsapi_results = False
    try:
        newsdata_list = get_newsdata(ticker)
        newsdata_results = pipe(newsdata_list)
    except Exception as e:
        print(e)
        newsdata_results = False

    try:
        finhub_list = get_finhub(ticker)
        finhub_results = pipe(finhub_list)
    except Exception as e:
        print(e)
        finhub_results = False
        
    try:
        vantage_list = get_vantage(ticker)
        vantage_results = pipe(vantage_list)
    except Exception as e:
        print(e)
        vantage_results = False

    try:
        marketaux_list = get_marketaux(ticker)
        marketaux_results = pipe(marketaux_list)
    except Exception as e:
        print(e)
        marketaux_results = False
    
    

    # finhub_list = get_finhub(ticker)
    # marketaux_list = get_marketaux(ticker)
    # newsapi_list = get_newsapi(ticker)
    # newsdata_list = get_newsdata(ticker)
    # vantage_list = get_vantage(ticker)

    
    # calling ai model on each list
    # finhub_results = pipe(finhub_list)
    # marketaux_results = pipe(marketaux_list)
    # newsapi_results = pipe(newsapi_list)
    # newsdata_results = pipe(newsdata_list)
    # vantage_results = pipe(vantage_list)
    
    # replacing values for calculations and doing the sentiment for each source
    def replace_values(result):
    # Replace values in the label column
        for dict in result:
            if dict["label"] == "LABEL_1":
                dict["label"] = 2
            else:
                dict["label"] = 1
            
    try:
        replace_values(bezinga_results)

        bezinga_label_mean = float(sum(d['label'] for d in bezinga_results)) / len(bezinga_results)
        
        bezinga_positives = []
        bezinga_negatives = []

        for dict in bezinga_results:
            if dict["label"] == 2:
                bezinga_positives.append(dict)
            else:
                bezinga_negatives.append(dict)

        if len(bezinga_positives) > 0:
            bezinga_positive_score_mean = float(sum(d['score'] for d in bezinga_positives)) / len(bezinga_positives)
            
        if len(bezinga_negatives) > 0:
            bezinga_negative_score_mean = float(sum(d['score'] for d in bezinga_negatives)) / len(bezinga_negatives)
    except Exception as e:
        print(e)

        # finhub
    if finhub_results:
        replace_values(finhub_results)

        finhub_label_mean = float(sum(d['label'] for d in finhub_results)) / len(finhub_results)
        
        finhub_positives = []
        finhub_negatives = []

        for dict in finhub_results:
            if dict["label"] == 2:
                finhub_positives.append(dict)
            else:
                finhub_negatives.append(dict)

        if len(finhub_positives) > 0:
            finhub_positive_score_mean = float(sum(d['score'] for d in finhub_positives)) / len(finhub_positives)
            
        if len(finhub_negatives) > 0:
            finhub_negative_score_mean = float(sum(d['score'] for d in finhub_negatives)) / len(finhub_negatives)

    # marketaux
    if marketaux_results:
        replace_values(marketaux_results)

        marketaux_label_mean = float(sum(d['label'] for d in marketaux_results)) / len(marketaux_results)
        
        marketaux_positives = []
        marketaux_negatives = []

        for dict in marketaux_results:
            if dict["label"] == 2:
                marketaux_positives.append(dict)
            else:
                marketaux_negatives.append(dict)

        if len(marketaux_positives) > 0:
            marketaux_positive_score_mean = float(sum(d['score'] for d in marketaux_positives)) / len(marketaux_positives)
            
        if len(marketaux_negatives) > 0:
            marketaux_negative_score_mean = float(sum(d['score'] for d in marketaux_negatives)) / len(marketaux_negatives)
    
    # newsapi
    if newsapi_results:
        replace_values(newsapi_results)

        newsapi_label_mean = float(sum(d['label'] for d in newsapi_results) + 1) / (len(newsapi_results) + 2)
        
        newsapi_positives = []
        newsapi_negatives = []

        for dict in newsapi_results:
            if dict["label"] == 2:
                newsapi_positives.append(dict)
            else:
                newsapi_negatives.append(dict)

        if len(newsapi_positives) > 0:
            newsapi_positive_score_mean = float(sum(d['score'] for d in newsapi_positives)) / len(newsapi_positives)
            
        if len(newsapi_negatives) > 0:
            newsapi_negative_score_mean = float(sum(d['score'] for d in newsapi_negatives)) / len(newsapi_negatives)

    
    # newsdata
    if newsdata_results:
        replace_values(newsdata_results)

        newsdata_label_mean = float(sum(d['label'] for d in newsdata_results)) / len(newsdata_results)
        
        newsdata_positives = []
        newsdata_negatives = []

        for dict in newsdata_results:
            if dict["label"] == 2:
                newsdata_positives.append(dict)
            else:
                newsdata_negatives.append(dict)

        if len(newsdata_positives) > 0:
            newsdata_positive_score_mean = float(sum(d['score'] for d in newsdata_positives)) / len(newsdata_positives)
            
        if len(newsdata_negatives) > 0:
            newsdata_negative_score_mean = float(sum(d['score'] for d in newsdata_negatives)) / len(newsdata_negatives)

    # vantage
    if vantage_results:
        replace_values(vantage_results)

        vantage_label_mean = float(sum(d['label'] for d in vantage_results)) / len(vantage_results)
        
        vantage_positives = []
        vantage_negatives = []

        for dict in vantage_results:
            if dict["label"] == 2:
                vantage_positives.append(dict)
            else:
                vantage_negatives.append(dict)

        if len(vantage_positives) > 0:
            vantage_positive_score_mean = float(sum(d['score'] for d in vantage_positives)) / len(vantage_positives)
            
        if len(vantage_negatives) > 0:
            vantage_negative_score_mean = float(sum(d['score'] for d in vantage_negatives)) / len(vantage_negatives)

    results_dict = {
        "bezinga": {
            "bezinga_articles": len(bezinga_results) if bezinga_results else 0,
            "bezinga_positives": len(bezinga_positives) if bezinga_results else 0,
            "bezinga_negatives": len(bezinga_negatives) if bezinga_results else 0,
            "bezinga_sentiment_mean": bezinga_label_mean if bezinga_results else 0,
            "bezinga_positive_score_mean": bezinga_positive_score_mean if bezinga_results else 0,
            "bezinga_negative_score_mean": bezinga_negative_score_mean if bezinga_results else 0
        },
        "finhub": {
            "finhub_articles": len(finhub_results) if finhub_results else 0,
            "finhub_positives": len(finhub_positives) if finhub_results else 0,
            "finhub_negatives": len(finhub_negatives) if finhub_results else 0,
            "finhub_sentiment_mean": finhub_label_mean if finhub_results else 0,
            "finhub_positive_score_mean": finhub_positive_score_mean if finhub_results else 0,
            "finhub_negative_score_mean": finhub_negative_score_mean if finhub_results else 0
        },
        "marketaux": {
            "marketaux_articles": len(marketaux_results) if marketaux_results else 0,
            "marketaux_positives": len(marketaux_positives) if marketaux_results else 0,
            "marketaux_negatives": len(marketaux_negatives) if marketaux_results else 0,
            "marketaux_sentiment_mean": marketaux_label_mean if marketaux_results else 0,
            "marketaux_positive_score_mean": marketaux_positive_score_mean if marketaux_results else 0,
            "marketaux_negative_score_mean": marketaux_negative_score_mean if marketaux_results else 0
        },
        "newsapi": {
            "newsapi_articles": len(newsapi_results) if newsapi_results else 0,
            "newsapi_positives": len(newsapi_positives) if newsapi_results else 0,
            "newsapi_negatives": len(newsapi_negatives) if newsapi_results else 0,
            "newsapi_sentiment_mean": newsapi_label_mean if newsapi_results else 0,
            "newsapi_positive_score_mean": newsapi_positive_score_mean if newsapi_results else 0,
            "newsapi_negative_score_mean": newsapi_negative_score_mean if newsapi_results else 0
        },
        "newsdata": {
            "newsdata_articles": len(newsdata_results) if newsdata_results else 0,
            "newsdata_positives": len(newsdata_positives) if newsdata_results else 0,
            "newsdata_negatives": len(newsdata_negatives) if newsdata_results else 0,
            "newsdata_sentiment_mean": newsdata_label_mean if newsdata_results else 0,
            "newsdata_positive_score_mean": newsdata_positive_score_mean if newsdata_results else 0,
            "newsdata_negative_score_mean": newsdata_negative_score_mean if newsdata_results else 0
        },
        "vantage": {
            "vantage_articles": len(vantage_results) if vantage_results else 0,
            "vantage_positives": len(vantage_positives) if vantage_results else 0,
            "vantage_negatives": len(vantage_negatives) if vantage_results else 0,
            "vantage_sentiment_mean": vantage_label_mean if vantage_results else 0,
            "vantage_positive_score_mean": vantage_positive_score_mean if vantage_results else 0,
            "vantage_negative_score_mean": vantage_negative_score_mean if vantage_results else 0
        }
    }

    return results_dict