Create app.py

app.py

@@ -0,0 +1,199 @@
+import streamlit as st
+from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer
+import requests
+import os
+import tweepy  # For Twitter API
+from googleapiclient.discovery import build  # For YouTube API
+from datetime import datetime, timedelta
+import daal4py as d4p  # Intel DAAL
+import numpy as np
+
+# Alpha Vantage API Key (replace with your own key)
+ALPHA_VANTAGE_API_KEY = "your_alpha_vantage_api_key"
+TWITTER_BEARER_TOKEN = "your_twitter_bearer_token"
+YOUTUBE_API_KEY = "your_youtube_api_key"
+
+# Hugging Face sentiment analysis pipeline
+@st.cache_resource
+def load_sentiment_model():
+    return pipeline("sentiment-analysis")
+
+# Load LLaMA model for stock recommendations
+@st.cache_resource
+def load_llama_model():
+    model_name = "meta-llama/Llama-2-7b-chat-hf"  # Change to your desired model
+    tokenizer = AutoTokenizer.from_pretrained(model_name)
+    model = AutoModelForCausalLM.from_pretrained(model_name)
+    return tokenizer, model
+
+# Fetch stock data from Alpha Vantage API
+def fetch_stock_data(symbol):
+    url = f"https://www.alphavantage.co/query?function=TIME_SERIES_DAILY&symbol={symbol}&apikey={ALPHA_VANTAGE_API_KEY}"
+    response = requests.get(url)
+    return response.json().get("Time Series (Daily)", {})
+
+# Fetch company overview (for fundamental analysis) from Alpha Vantage API
+def fetch_fundamental_data(symbol):
+    url = f"https://www.alphavantage.co/query?function=OVERVIEW&symbol={symbol}&apikey={ALPHA_VANTAGE_API_KEY}"
+    response = requests.get(url)
+    return response.json()
+
+# Twitter API function to fetch recent tweets
+def fetch_tweets(stock_symbol, count=10):
+    headers = {"Authorization": f"Bearer {TWITTER_BEARER_TOKEN}"}
+    query = f"{stock_symbol} stock"
+    url = f"https://api.twitter.com/2/tweets/search/recent?query={query}&max_results={count}&tweet.fields=created_at,text"
+    response = requests.get(url, headers=headers)
+    tweets = response.json().get("data", [])
+    return [tweet["text"] for tweet in tweets]
+
+# YouTube API function to fetch recent comments
+def fetch_youtube_comments(stock_symbol, max_results=5):
+    youtube = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)
+    request = youtube.search().list(
+        q=f"{stock_symbol} stock",
+        part='snippet',
+        type='video',
+        maxResults=5
+    )
+    response = request.execute()
+    video_ids = [item['id']['videoId'] for item in response['items']]
+
+    comments = []
+    for video_id in video_ids:
+        comment_request = youtube.commentThreads().list(
+            part='snippet',
+            videoId=video_id,
+            maxResults=max_results
+        )
+        comment_response = comment_request.execute()
+        for comment in comment_response['items']:
+            comment_text = comment['snippet']['topLevelComment']['snippet']['textDisplay']
+            comments.append(comment_text)
+
+    return comments
+
+# Perform sentiment analysis on social media comments/tweets
+def perform_sentiment_analysis(comments):
+    sentiment_model = load_sentiment_model()
+    return sentiment_model(comments)
+
+# Compute Moving Average using Intel oneAPI AI Analytics Toolkit (daal4py)
+def compute_moving_average(prices, window=5):
+    price_array = np.array(prices, dtype=np.float64).reshape(-1, 1)
+
+    # Initialize Intel DAAL low-order moments algorithm (for moving average)
+    algorithm = d4p.low_order_moments()
+
+    moving_averages = []
+    for i in range(len(price_array) - window + 1):
+        window_data = price_array[i:i + window]
+        result = algorithm.compute(window_data)
+        moving_averages.append(result.mean[0])
+
+    return moving_averages
+
+# Technical analysis function using Intel oneAPI
+def technical_analysis(symbol):
+    data = fetch_stock_data(symbol)
+    
+    if data:
+        closing_prices = [float(v['4. close']) for v in data.values()]
+        dates = list(data.keys())
+
+        # Compute 5-day moving average
+        moving_averages = compute_moving_average(closing_prices)
+
+        latest_date = dates[0]
+        latest_price = closing_prices[0]
+        latest_moving_average = moving_averages[0] if moving_averages else "N/A"
+        
+        return {
+            "Date": latest_date,
+            "Closing Price": latest_price,
+            "5-Day Moving Average": latest_moving_average
+        }
+    return {}
+
+# Streamlit Web App
+def main():
+    st.title("Stock Analysis App with Social Media Sentiment Analysis")
+    st.write("""
+    This app provides stock analysis using:
+    - Sentiment Analysis (from Twitter and YouTube)
+    - Technical Analysis (with Intel oneAPI for performance)
+    - Fundamental Analysis
+    - Buy/Sell/Hold Recommendations
+    """)
+
+    company_symbol = st.text_input("Enter the stock symbol (e.g., AAPL, TSLA, GOOGL):")
+
+    if company_symbol:
+        try:
+            stock_data = fetch_stock_data(company_symbol)
+            fundamental_data = fetch_fundamental_data(company_symbol)
+            
+            if stock_data:
+                st.subheader("Asset Overview")
+                st.json(stock_data)
+
+                with st.expander("Technical Analysis (Intel oneAPI)"):
+                    st.subheader("Technical Analysis")
+                    tech_analysis = technical_analysis(company_symbol)
+                    st.write(tech_analysis)
+
+                with st.expander("Fundamental Analysis"):
+                    st.subheader("Fundamental Analysis")
+                    if fundamental_data:
+                        st.write(f"**Company Name**: {fundamental_data.get('Name', 'N/A')}")
+                        st.write(f"**Sector**: {fundamental_data.get('Sector', 'N/A')}")
+                        st.write(f"**Industry**: {fundamental_data.get('Industry', 'N/A')}")
+                        st.write(f"**Market Capitalization**: {fundamental_data.get('MarketCapitalization', 'N/A')}")
+                        st.write(f"**Earnings per Share (EPS)**: {fundamental_data.get('EPS', 'N/A')}")
+                        st.write(f"**Price to Earnings Ratio (P/E)**: {fundamental_data.get('PERatio', 'N/A')}")
+                        st.write(f"**Dividend Yield**: {fundamental_data.get('DividendYield', 'N/A')}")
+                        st.write(f"**Profit Margin**: {fundamental_data.get('ProfitMargin', 'N/A')}")
+                    else:
+                        st.write("No fundamental data available.")
+
+                with st.expander("Sentiment Analysis"):
+                    st.subheader("Sentiment Analysis")
+                    st.write("Fetching views from Twitter and YouTube...")
+
+                    # Fetch tweets and comments
+                    tweets = fetch_tweets(company_symbol)
+                    youtube_comments = fetch_youtube_comments(company_symbol)
+
+                    st.write("**Tweets**:")
+                    for tweet in tweets:
+                        st.write(f"- {tweet}")
+
+                    st.write("**YouTube Comments**:")
+                    for comment in youtube_comments:
+                        st.write(f"- {comment}")
+
+                    # Perform sentiment analysis on collected data
+                    all_comments = tweets + youtube_comments
+                    if all_comments:
+                        sentiment_results = perform_sentiment_analysis(all_comments)
+                        st.write("Sentiment Analysis Results:")
+                        for sentiment in sentiment_results:
+                            st.write(sentiment)
+
+                with st.expander("Recommendation"):
+                    st.subheader("Recommendation")
+                    tokenizer, llama_model = load_llama_model()
+                    stock_recommendation = llama_model.generate(
+                        tokenizer(company_symbol, return_tensors="pt").input_ids,
+                        max_new_tokens=50
+                    )
+                    recommendation_text = tokenizer.decode(stock_recommendation[0], skip_special_tokens=True)
+                    st.write(recommendation_text)
+            else:
+                st.error(f"No data available for the symbol entered.")
+
+        except Exception as e:
+            st.error(f"An error occurred: {e}")
+
+if __name__ == "__main__":
+    main()