Spaces:
Sleeping
Sleeping
| # Use a pipeline as a high-level helper | |
| from transformers import pipeline | |
| import json | |
| import requests | |
| import datetime | |
| import sqlite3 | |
| import pandas as pd | |
| import streamlit as st | |
| import os | |
| import streamlit as st | |
| from requests.exceptions import HTTPError, RequestException | |
| # API DOC: https://finnhub.io/docs/api/introduction | |
| def get_finnhub_data(example: str) -> json: | |
| """ | |
| Pass in the "example" string from the API documentation. It changes for every endpoint. | |
| :param1 example: '/company-news?symbol=AAPL&from=2023-08-15&to=2023-08-20' | |
| """ | |
| base_url = 'https://finnhub.io/api/v1//' | |
| token = f"&token={os.environ['finnhub_token']}" | |
| try: | |
| request = requests.get(f"{base_url}{example}{token}") | |
| request.raise_for_status() # This will raise an HTTPError if the response was an error | |
| return request.json() | |
| except HTTPError as http_err: | |
| st.write(f"HTTP error occurred: {http_err}") # Python 3.6+ | |
| # Consider logging the error or handling it further based on your needs | |
| except RequestException as err: | |
| st.write(f"Other error occurred: {err}") # Python 3.6+ | |
| # Handle other types of exceptions (e.g., network issues) | |
| except Exception as e: | |
| st.write(f"Unexpected error: {e}") | |
| # Catch-all for any other exceptions, which is useful for debugging | |
| return {} # Return | |
| nlp = pipeline("sentiment-analysis", model="ProsusAI/finbert") | |
| def sentiment_analysis(headline:str) -> str: | |
| """ | |
| Pass in a headline and get back a sentiment analysis of the text. | |
| This only works for one headline at a time; Should be expanded to work for a list of headlines. | |
| :param1 headline: Text string: 'Apple is the best company in the world' | |
| """ | |
| return nlp(headline) | |
| # --------------------------------- get news articles for a company --------------------------------- # | |
| conn = sqlite3.connect('fin_data.db') | |
| c = conn.cursor() | |
| #c.execute("DROP TABLE IF EXISTS company_news") | |
| c.execute("""CREATE TABLE IF NOT EXISTS company_news ( | |
| id integer primary key autoincrement, | |
| ticker text, | |
| category text, | |
| headline text, | |
| date_stamp text, | |
| sentiment_label text, | |
| sentiment_score real | |
| )""") | |
| ticker = st.text_input(label='Entering ticker will add last 5 days of news sentiment data to database.') | |
| start_date_str = (datetime.datetime.today() - datetime.timedelta(days=5)).strftime('%Y-%m-%d') | |
| end_date_str = datetime.datetime.today().strftime('%Y-%m-%d') | |
| dates = set(pd.date_range(start=start_date_str, end=end_date_str).strftime('%Y-%m-%d')) | |
| if st.button('Load') and ticker: | |
| c.execute('SELECT DISTINCT(ticker), date_stamp FROM company_news where ticker = ?', (ticker,)) | |
| distinct_tickers = c.fetchall() | |
| distinct_ticker_symbols = [ticker[0] for ticker in distinct_tickers] | |
| existing_dates = set([ticker[1] for ticker in distinct_tickers]) | |
| unique_dates = dates - existing_dates | |
| if unique_dates: | |
| for date in unique_dates: | |
| try: | |
| res_news = get_finnhub_data(f"/company-news?symbol={ticker}&from={date}&to={date}") | |
| except: | |
| st.error('Invalid Ticker.') | |
| st.write(f"Proessing {len(res_news)} headlines for ", ticker,' on ', date) | |
| for item in res_news: | |
| dt_object = datetime.datetime.fromtimestamp(item['datetime']).strftime("%Y-%m-%d") | |
| sentiment = sentiment_analysis(item['headline']) | |
| sentiment_label = sentiment[0]['label'] | |
| sentiment_score = sentiment[0]['score'] | |
| #st.write(sentiment_label, dt_object) | |
| query = """ | |
| INSERT INTO company_news (ticker, category, headline, date_stamp, sentiment_label, sentiment_score) | |
| VALUES (?, ?, ?, ?, ?, ?) | |
| """ | |
| data = (ticker, item['category'], item['headline'], dt_object, sentiment_label, sentiment_score) | |
| #Execute the query with the data | |
| c.execute(query, data) | |
| c.execute(""" | |
| SELECT * FROM company_news WHERE ticker = ? | |
| """, (ticker,)) | |
| # Fetch all results | |
| rows = c.fetchall() | |
| # Create a DataFrame | |
| df = pd.DataFrame(rows, columns=[description[0] for description in c.description]) | |
| st.write(df) | |
| # # --------------------------------- get insider sentiment --------------------------------- # | |
| # res_sentiment = get_finnhub_data(f'/stock/insider-sentiment?symbol={ticker}') | |
| # print(res_sentiment['data'][0].keys()) | |
| conn.commit() | |
| conn.close() |