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