Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| import pandas as pd | |
| import numpy as np | |
| from transformers import pipeline # For sentiment analysis | |
| from sklearn.ensemble import IsolationForest # For anomaly detection | |
| import yfinance as yf # For stock market data | |
| import requests # For API calls | |
| # Replace with your actual API keys | |
| NEWS_API_KEY = "0ac3b0aa2cec405b85ecdb0e8ca15823" # Get from https://newsapi.org/ | |
| TWITTER_BEARER_TOKEN = "YOUR_TWITTER_BEARER_TOKEN" # Get from https://developer.twitter.com/ | |
| # Initialize pre-trained sentiment analysis model from Hugging Face Transformers | |
| sentiment_analyzer = pipeline("sentiment-analysis") | |
| def fetch_news(keyword): | |
| """Fetches news articles using the NewsAPI.""" | |
| try: | |
| url = f"https://newsapi.org/v2/everything?q={keyword}&apiKey={NEWS_API_KEY}" | |
| response = requests.get(url) | |
| response.raise_for_status() # Raise an exception for HTTP errors | |
| articles = response.json().get("articles", []) | |
| return pd.DataFrame([{"title": article["title"], "description": article["description"]} for article in articles]) | |
| except requests.exceptions.RequestException as e: | |
| raise Exception(f"Failed to fetch news: {e}") | |
| except Exception as e: | |
| raise Exception(f"Error processing news data: {e}") | |
| def fetch_social_media_data(keyword): | |
| """Fetches social media data using Twitter API.""" | |
| try: | |
| url = f"https://api.twitter.com/2/tweets/search/recent?query={keyword}&tweet.fields=text&max_results=10" | |
| headers = {"Authorization": f"Bearer {TWITTER_BEARER_TOKEN}"} | |
| response = requests.get(url, headers=headers) | |
| response.raise_for_status() # Raise an exception for HTTP errors | |
| tweets = response.json().get("data", []) | |
| if tweets: # Handle case when no tweets are found | |
| return pd.DataFrame([{"text": tweet["text"]} for tweet in tweets]) | |
| else: | |
| return pd.DataFrame({"text": []}) # Return empty DataFrame if no tweets found | |
| except requests.exceptions.RequestException as e: | |
| raise Exception(f"Failed to fetch social media data: {e}") | |
| except Exception as e: | |
| raise Exception(f"Error processing social media data: {e}") | |
| def fetch_market_data(ticker, timeframe): | |
| """Fetches stock/crypto market data using yfinance.""" | |
| try: | |
| data = yf.download(ticker, period=timeframe, interval="1d") | |
| if data.empty: | |
| raise Exception(f"No market data found for ticker: {ticker}") | |
| return data.reset_index() | |
| except Exception as e: | |
| raise Exception(f"Failed to fetch market data for {ticker}: {str(e)}") | |
| def analyze_sentiment(text_list): | |
| """Performs sentiment analysis on a list of texts.""" | |
| if not text_list: # Handle empty text list | |
| return [] | |
| try: | |
| sentiments = sentiment_analyzer(text_list) | |
| scores = [item['score'] if item['label'] == 'POSITIVE' else -item['score'] for item in sentiments] | |
| return scores | |
| except Exception as e: | |
| raise Exception(f"Sentiment analysis error: {e}") | |
| def detect_anomalies(data): | |
| """Detects anomalies in time series data using Isolation Forest.""" | |
| if len(data) <= 1: # Need at least 2 data points for diff and anomaly detection | |
| return [] | |
| try: | |
| model = IsolationForest(contamination=0.1, random_state=42) | |
| anomalies = model.fit_predict(data.reshape(-1, 1)) | |
| return [i for i, val in enumerate(anomalies) if val == -1] | |
| except Exception as e: | |
| raise Exception(f"Anomaly detection error: {e}") | |
| def identify_opportunities(ticker, news_sentiment, social_sentiment, anomalies, market_data): | |
| """Identifies potential explosive growth opportunities.""" | |
| if np.mean(news_sentiment) > 0.3 and np.mean(social_sentiment) > 0.3 and len(anomalies) > 0: # Reduced sentiment threshold slightly | |
| return [ | |
| { | |
| "ticker": ticker, | |
| "potential_gain": np.random.randint(10, 50), # Simulated gain percentage | |
| "risk_level": "High", | |
| "disclaimer": "This is a speculative opportunity. Conduct thorough research." | |
| } | |
| ] | |
| return [] | |
| def analyze_market(ticker_or_keyword, timeframe="1d"): | |
| """ | |
| Analyzes news, social media, and market data for a given ticker or keyword. | |
| Args: | |
| ticker_or_keyword (str): The stock ticker symbol (e.g., "AAPL") or keyword (e.g., "AI"). | |
| timeframe (str): The time frame for analysis (e.g., "1d", "1w", "1m"). | |
| Returns: | |
| dict: A dictionary containing analysis results for Gradio display. | |
| """ | |
| try: | |
| # Data Collection | |
| news_df = fetch_news(ticker_or_keyword) | |
| social_media_df = fetch_social_media_data(ticker_or_keyword) | |
| market_df = fetch_market_data(ticker_or_keyword, timeframe) | |
| # Sentiment Analysis | |
| news_sentiment = analyze_sentiment(news_df["description"].fillna("").tolist()) if not news_df.empty else [] | |
| social_sentiment = analyze_sentiment(social_media_df["text"].tolist()) if not social_media_df.empty else [] | |
| # Anomaly Detection | |
| price_changes = market_df["Close"].pct_change().dropna().values if not market_df.empty else np.array([]) | |
| anomalies = detect_anomalies(price_changes) | |
| # Opportunity Identification | |
| opportunities = identify_opportunities( | |
| ticker_or_keyword, | |
| news_sentiment, | |
| social_sentiment, | |
| anomalies, | |
| market_df | |
| ) | |
| # Results Formatting for Gradio | |
| results_md = f"## Analysis Results for: {ticker_or_keyword}\n\n" | |
| results_md += f"**Average News Sentiment:** {np.mean(news_sentiment):.2f} \n" if news_sentiment else "**Average News Sentiment:** N/A (No news found) \n" | |
| results_md += f"**Average Social Sentiment:** {np.mean(social_sentiment):.2f} \n" if social_sentiment else "**Average Social Sentiment:** N/A (No social media data found) \n" | |
| results_md += f"**Anomalies Detected in Price Changes:** {len(anomalies)} \n\n" if price_changes.size > 0 else "**Anomalies Detected in Price Changes:** N/A (Insufficient market data) \n\n" | |
| if opportunities: | |
| results_md += "### Potential Explosive Growth Opportunities:\n" | |
| opportunities_df = pd.DataFrame(opportunities) | |
| results_md += opportunities_df.to_markdown(index=False) + "\n\n" | |
| else: | |
| results_md += "**No Explosive Growth Opportunities Identified based on current analysis.**\n\n" | |
| results_md += "---\n**Disclaimer:** This analysis is for informational purposes only and not financial advice. Investing in financial markets involves risk. Conduct thorough research and consult with a financial advisor before making investment decisions." | |
| return results_md | |
| except Exception as e: | |
| error_md = f"## Analysis Error for: {ticker_or_keyword}\n\n" | |
| error_md += f"**Error Details:** {str(e)}\n\n" | |
| error_md += "---\n**Disclaimer:** This analysis is for informational purposes only and not financial advice. Investing in financial markets involves risk. Conduct thorough research and consult with a financial advisor before making investment decisions." | |
| return error_md | |
| # Gradio Interface | |
| iface = gr.Interface( | |
| fn=analyze_market, | |
| inputs=[ | |
| gr.Textbox(label="Stock Ticker or Keyword (e.g., AAPL, BTC-USD, AI)"), | |
| gr.Dropdown(["1d", "1w", "1m"], label="Timeframe", value="1d"), | |
| ], | |
| outputs=gr.Markdown(label="Analysis Results"), | |
| title="Explosive Growth Opportunity Finder", | |
| description=( | |
| "This tool leverages AI to analyze news sentiment, social media trends, and market data to identify potential investment opportunities. " | |
| "Enter a stock ticker (e.g., AAPL), crypto symbol (e.g., BTC-USD), or a general keyword (e.g., AI) to analyze. " | |
| "**Disclaimer:** This is a highly speculative tool for educational purposes. " | |
| "It is not financial advice. Investing in financial markets involves significant risk. " | |
| "Always conduct your own thorough research and consult with a financial advisor before making any investment decisions." | |
| ), | |
| ) | |
| if __name__ == "__main__": | |
| iface.launch() |