You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
531 lines
21 KiB
Python
531 lines
21 KiB
Python
"""
|
|
Stock Oracle - Investment Data Analyzer
|
|
Python client integration with Stock Oracle API for comprehensive financial analysis.
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Union
|
|
import logging
|
|
import warnings
|
|
from stock_oracle_client import StockOracleClient, StockOracleAPIError
|
|
|
|
warnings.filterwarnings('ignore')
|
|
|
|
class StockOracleAnalyzer:
|
|
"""
|
|
Stock Oracle - Comprehensive financial data analyzer using Stock Oracle API
|
|
"""
|
|
|
|
def __init__(self, api_url: str = "http://localhost:18001"):
|
|
"""
|
|
Initialize the analyzer with Stock Oracle API
|
|
|
|
Args:
|
|
api_url: Stock Oracle API URL
|
|
"""
|
|
self.client = StockOracleClient(api_url)
|
|
self.logger = self._setup_logging()
|
|
|
|
# Test connection
|
|
try:
|
|
health = self.client.get_health()
|
|
self.logger.info(f"Connected to Stock Oracle API: {health.get('status')}")
|
|
except Exception as e:
|
|
self.logger.error(f"Failed to connect to Stock Oracle API: {e}")
|
|
|
|
def _setup_logging(self) -> logging.Logger:
|
|
"""Set up logging configuration"""
|
|
logging.basicConfig(level=logging.INFO)
|
|
return logging.getLogger(__name__)
|
|
|
|
def get_company_data(self, ticker: str, period: str = "5y") -> Dict:
|
|
"""
|
|
Retrieve comprehensive company data from Stock Oracle API
|
|
|
|
Args:
|
|
ticker: Stock ticker symbol
|
|
period: Time period for data (e.g., "1y", "3y", "5y")
|
|
|
|
Returns:
|
|
Dict containing financial data, price data, and analysis
|
|
"""
|
|
try:
|
|
self.logger.info(f"Retrieving data for {ticker}")
|
|
|
|
# Get financial data
|
|
financial_data = self.client.get_financial_data(
|
|
ticker=ticker,
|
|
period=period,
|
|
include_metrics=True
|
|
)
|
|
|
|
# Get price data
|
|
price_data = self.client.get_price_data(
|
|
ticker=ticker,
|
|
period=period
|
|
)
|
|
|
|
# Get news and social media data
|
|
try:
|
|
news_social_data = self.client.get_news_social_data(
|
|
ticker=ticker,
|
|
days_back=30,
|
|
max_articles=50
|
|
)
|
|
except Exception as e:
|
|
self.logger.warning(f"Could not retrieve news/social data: {e}")
|
|
news_social_data = None
|
|
|
|
# Combine all data
|
|
data = {
|
|
'ticker': ticker,
|
|
'company_name': financial_data.get('company', {}).get('name'),
|
|
'financial_data': financial_data,
|
|
'price_data': price_data,
|
|
'news_social_data': news_social_data,
|
|
'analysis_summary': self._generate_analysis_summary(financial_data, price_data, news_social_data)
|
|
}
|
|
|
|
return data
|
|
|
|
except StockOracleAPIError as e:
|
|
self.logger.error(f"API Error retrieving data for {ticker}: {str(e)}")
|
|
return None
|
|
except Exception as e:
|
|
self.logger.error(f"Error retrieving data for {ticker}: {str(e)}")
|
|
return None
|
|
|
|
def _generate_analysis_summary(self, financial_data: Dict, price_data: Dict, news_social_data: Optional[Dict]) -> Dict:
|
|
"""Generate analysis summary from all available data"""
|
|
|
|
summary = {
|
|
'financial_health': self._assess_financial_health(financial_data),
|
|
'price_trends': self._analyze_price_trends(price_data),
|
|
'sentiment_analysis': self._analyze_sentiment(news_social_data) if news_social_data else None,
|
|
'key_metrics': self._extract_key_metrics(financial_data),
|
|
'investment_score': None # Will be calculated based on all factors
|
|
}
|
|
|
|
# Calculate overall investment score (0-100)
|
|
summary['investment_score'] = self._calculate_investment_score(summary)
|
|
|
|
return summary
|
|
|
|
def _assess_financial_health(self, financial_data: Dict) -> Dict:
|
|
"""Assess financial health based on financial metrics"""
|
|
|
|
health_score = 0
|
|
max_score = 100
|
|
assessment = {}
|
|
|
|
try:
|
|
latest_data = financial_data.get('financial_data', [])
|
|
if not latest_data:
|
|
return {'score': 0, 'assessment': 'No financial data available'}
|
|
|
|
latest = latest_data[0] # Most recent data
|
|
|
|
# Revenue growth assessment
|
|
if len(latest_data) >= 2:
|
|
current_revenue = latest.get('revenue', 0)
|
|
previous_revenue = latest_data[1].get('revenue', 0)
|
|
if previous_revenue > 0:
|
|
revenue_growth = (current_revenue - previous_revenue) / previous_revenue
|
|
assessment['revenue_growth'] = revenue_growth
|
|
if revenue_growth > 0.15: # >15% growth
|
|
health_score += 25
|
|
elif revenue_growth > 0.05: # >5% growth
|
|
health_score += 15
|
|
elif revenue_growth > 0: # Positive growth
|
|
health_score += 10
|
|
|
|
# Profitability assessment
|
|
if 'pe_ratio' in latest and latest['pe_ratio'] and latest['pe_ratio'] > 0:
|
|
pe = latest['pe_ratio']
|
|
assessment['pe_ratio'] = pe
|
|
if 10 <= pe <= 25: # Reasonable P/E range
|
|
health_score += 25
|
|
elif 5 <= pe < 40: # Acceptable range
|
|
health_score += 15
|
|
|
|
# ROE assessment
|
|
if 'roe' in latest and latest['roe']:
|
|
roe = latest['roe']
|
|
assessment['roe'] = roe
|
|
if roe > 0.20: # >20% ROE
|
|
health_score += 25
|
|
elif roe > 0.15: # >15% ROE
|
|
health_score += 20
|
|
elif roe > 0.10: # >10% ROE
|
|
health_score += 15
|
|
|
|
# Debt assessment
|
|
if 'debt_to_equity' in latest and latest['debt_to_equity']:
|
|
debt_to_equity = latest['debt_to_equity']
|
|
assessment['debt_to_equity'] = debt_to_equity
|
|
if debt_to_equity < 0.3: # Low debt
|
|
health_score += 25
|
|
elif debt_to_equity < 0.6: # Moderate debt
|
|
health_score += 15
|
|
elif debt_to_equity < 1.0: # High but manageable debt
|
|
health_score += 5
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error in financial health assessment: {e}")
|
|
|
|
return {
|
|
'score': min(health_score, max_score),
|
|
'assessment': assessment,
|
|
'grade': self._score_to_grade(min(health_score, max_score))
|
|
}
|
|
|
|
def _analyze_price_trends(self, price_data: Dict) -> Dict:
|
|
"""Analyze price trends and momentum"""
|
|
|
|
try:
|
|
prices = price_data.get('price_data', [])
|
|
if len(prices) < 10:
|
|
return {'trend': 'insufficient_data'}
|
|
|
|
# Get closing prices
|
|
closes = [p['close'] for p in prices]
|
|
dates = [p['date'] for p in prices]
|
|
|
|
# Calculate basic metrics
|
|
current_price = closes[-1]
|
|
price_52w_high = max(closes)
|
|
price_52w_low = min(closes)
|
|
|
|
# Calculate moving averages
|
|
ma_50 = np.mean(closes[-50:]) if len(closes) >= 50 else np.mean(closes)
|
|
ma_20 = np.mean(closes[-20:]) if len(closes) >= 20 else np.mean(closes)
|
|
|
|
# Calculate price momentum (30-day return)
|
|
month_ago_price = closes[-30] if len(closes) >= 30 else closes[0]
|
|
momentum_30d = (current_price - month_ago_price) / month_ago_price
|
|
|
|
# Calculate volatility (standard deviation)
|
|
returns = [(closes[i] - closes[i-1]) / closes[i-1] for i in range(1, len(closes))]
|
|
volatility = np.std(returns) * np.sqrt(252) # Annualized volatility
|
|
|
|
# Determine trend
|
|
trend = 'neutral'
|
|
if current_price > ma_50 and ma_20 > ma_50:
|
|
trend = 'bullish'
|
|
elif current_price < ma_50 and ma_20 < ma_50:
|
|
trend = 'bearish'
|
|
|
|
return {
|
|
'trend': trend,
|
|
'current_price': current_price,
|
|
'price_52w_high': price_52w_high,
|
|
'price_52w_low': price_52w_low,
|
|
'from_52w_high': (current_price - price_52w_high) / price_52w_high,
|
|
'from_52w_low': (current_price - price_52w_low) / price_52w_low,
|
|
'ma_50': ma_50,
|
|
'ma_20': ma_20,
|
|
'momentum_30d': momentum_30d,
|
|
'volatility': volatility
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error in price trend analysis: {e}")
|
|
return {'trend': 'error', 'error': str(e)}
|
|
|
|
def _analyze_sentiment(self, news_social_data: Dict) -> Dict:
|
|
"""Analyze news and social media sentiment"""
|
|
|
|
try:
|
|
news_articles = news_social_data.get('news', {}).get('articles', [])
|
|
social_posts = news_social_data.get('social_media', {}).get('posts', [])
|
|
|
|
# Simple sentiment analysis based on keywords
|
|
positive_words = ['growth', 'profit', 'success', 'strong', 'beat', 'exceed', 'bullish', 'upgrade']
|
|
negative_words = ['loss', 'decline', 'fall', 'weak', 'miss', 'bearish', 'downgrade', 'concern']
|
|
|
|
total_sentiment = 0
|
|
total_items = 0
|
|
|
|
# Analyze news sentiment
|
|
for article in news_articles:
|
|
title = article.get('title', '').lower()
|
|
summary = article.get('summary', '').lower()
|
|
text = f"{title} {summary}"
|
|
|
|
sentiment = 0
|
|
for word in positive_words:
|
|
sentiment += text.count(word)
|
|
for word in negative_words:
|
|
sentiment -= text.count(word)
|
|
|
|
total_sentiment += sentiment
|
|
total_items += 1
|
|
|
|
# Analyze social media sentiment (including Reddit scores)
|
|
social_sentiment = 0
|
|
for post in social_posts:
|
|
title = post.get('title', '').lower()
|
|
content = post.get('content', '').lower()
|
|
text = f"{title} {content}"
|
|
|
|
sentiment = 0
|
|
for word in positive_words:
|
|
sentiment += text.count(word)
|
|
for word in negative_words:
|
|
sentiment -= text.count(word)
|
|
|
|
# Factor in Reddit score if available
|
|
score = post.get('score', 0)
|
|
if score > 100:
|
|
sentiment += 1 # High-scoring posts are generally positive
|
|
elif score < -10:
|
|
sentiment -= 1
|
|
|
|
social_sentiment += sentiment
|
|
total_items += 1
|
|
|
|
# Calculate overall sentiment
|
|
if total_items > 0:
|
|
overall_sentiment = (total_sentiment + social_sentiment) / total_items
|
|
else:
|
|
overall_sentiment = 0
|
|
|
|
# Classify sentiment
|
|
if overall_sentiment > 0.5:
|
|
sentiment_label = 'positive'
|
|
elif overall_sentiment < -0.5:
|
|
sentiment_label = 'negative'
|
|
else:
|
|
sentiment_label = 'neutral'
|
|
|
|
return {
|
|
'overall_sentiment': overall_sentiment,
|
|
'sentiment_label': sentiment_label,
|
|
'news_articles_count': len(news_articles),
|
|
'social_posts_count': len(social_posts),
|
|
'total_items_analyzed': total_items
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error in sentiment analysis: {e}")
|
|
return {'sentiment_label': 'error', 'error': str(e)}
|
|
|
|
def _extract_key_metrics(self, financial_data: Dict) -> Dict:
|
|
"""Extract key financial metrics"""
|
|
|
|
try:
|
|
latest_data = financial_data.get('financial_data', [])
|
|
if not latest_data:
|
|
return {}
|
|
|
|
latest = latest_data[0]
|
|
|
|
return {
|
|
'revenue': latest.get('revenue'),
|
|
'net_income': latest.get('net_income'),
|
|
'eps': latest.get('eps'),
|
|
'pe_ratio': latest.get('pe_ratio'),
|
|
'pb_ratio': latest.get('pb_ratio'),
|
|
'roe': latest.get('roe'),
|
|
'roa': latest.get('roa'),
|
|
'debt_to_equity': latest.get('debt_to_equity'),
|
|
'market_cap': latest.get('market_cap'),
|
|
'period_date': latest.get('period_date')
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error extracting key metrics: {e}")
|
|
return {}
|
|
|
|
def _calculate_investment_score(self, summary: Dict) -> int:
|
|
"""Calculate overall investment score (0-100)"""
|
|
|
|
try:
|
|
score = 0
|
|
|
|
# Financial health (40% weight)
|
|
financial_score = summary.get('financial_health', {}).get('score', 0)
|
|
score += financial_score * 0.4
|
|
|
|
# Price trends (30% weight)
|
|
price_trends = summary.get('price_trends', {})
|
|
if price_trends.get('trend') == 'bullish':
|
|
score += 30
|
|
elif price_trends.get('trend') == 'neutral':
|
|
score += 15
|
|
# Bearish trend adds 0
|
|
|
|
# Sentiment (20% weight)
|
|
sentiment = summary.get('sentiment_analysis')
|
|
if sentiment:
|
|
if sentiment.get('sentiment_label') == 'positive':
|
|
score += 20
|
|
elif sentiment.get('sentiment_label') == 'neutral':
|
|
score += 10
|
|
# Negative sentiment adds 0
|
|
|
|
# Momentum bonus (10% weight)
|
|
momentum = price_trends.get('momentum_30d', 0)
|
|
if momentum > 0.10: # >10% monthly return
|
|
score += 10
|
|
elif momentum > 0.05: # >5% monthly return
|
|
score += 7
|
|
elif momentum > 0: # Positive return
|
|
score += 3
|
|
|
|
return min(int(score), 100)
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error calculating investment score: {e}")
|
|
return 0
|
|
|
|
def _score_to_grade(self, score: int) -> str:
|
|
"""Convert numeric score to letter grade"""
|
|
if score >= 90:
|
|
return 'A+'
|
|
elif score >= 85:
|
|
return 'A'
|
|
elif score >= 80:
|
|
return 'A-'
|
|
elif score >= 75:
|
|
return 'B+'
|
|
elif score >= 70:
|
|
return 'B'
|
|
elif score >= 65:
|
|
return 'B-'
|
|
elif score >= 60:
|
|
return 'C+'
|
|
elif score >= 55:
|
|
return 'C'
|
|
elif score >= 50:
|
|
return 'C-'
|
|
elif score >= 40:
|
|
return 'D'
|
|
else:
|
|
return 'F'
|
|
|
|
def analyze_multiple_companies(self, tickers: List[str], period: str = "3y") -> Dict:
|
|
"""Analyze multiple companies and return comparative data"""
|
|
|
|
results = {}
|
|
|
|
for ticker in tickers:
|
|
self.logger.info(f"Analyzing {ticker}...")
|
|
company_data = self.get_company_data(ticker, period)
|
|
if company_data:
|
|
results[ticker] = company_data
|
|
|
|
return results
|
|
|
|
def create_summary_report(self, analysis_results: Dict) -> pd.DataFrame:
|
|
"""Create a summary report comparing multiple companies"""
|
|
|
|
summary_data = []
|
|
|
|
for ticker, data in analysis_results.items():
|
|
if not data:
|
|
continue
|
|
|
|
analysis = data.get('analysis_summary', {})
|
|
key_metrics = analysis.get('key_metrics', {})
|
|
financial_health = analysis.get('financial_health', {})
|
|
price_trends = analysis.get('price_trends', {})
|
|
sentiment = analysis.get('sentiment_analysis', {})
|
|
|
|
row = {
|
|
'Ticker': ticker,
|
|
'Company': data.get('company_name', ''),
|
|
'Investment Score': analysis.get('investment_score', 0),
|
|
'Financial Grade': financial_health.get('grade', 'N/A'),
|
|
'Price Trend': price_trends.get('trend', 'N/A'),
|
|
'Sentiment': sentiment.get('sentiment_label', 'N/A') if sentiment else 'N/A',
|
|
# Financial metrics
|
|
'Revenue (M)': key_metrics.get('revenue', 0) / 1_000_000 if key_metrics.get('revenue') else None,
|
|
'Net Income (M)': key_metrics.get('net_income', 0) / 1_000_000 if key_metrics.get('net_income') else None,
|
|
'EPS': key_metrics.get('eps'),
|
|
'P/E Ratio': key_metrics.get('pe_ratio'),
|
|
'ROE (%)': key_metrics.get('roe') * 100 if key_metrics.get('roe') else None,
|
|
'Debt/Equity': key_metrics.get('debt_to_equity'),
|
|
# Price metrics
|
|
'Current Price': price_trends.get('current_price'),
|
|
'30d Momentum (%)': price_trends.get('momentum_30d', 0) * 100 if price_trends.get('momentum_30d') else None,
|
|
'Volatility (%)': price_trends.get('volatility', 0) * 100 if price_trends.get('volatility') else None,
|
|
# News metrics
|
|
'News Articles': sentiment.get('news_articles_count', 0) if sentiment else 0,
|
|
'Social Posts': sentiment.get('social_posts_count', 0) if sentiment else 0,
|
|
}
|
|
|
|
summary_data.append(row)
|
|
|
|
df = pd.DataFrame(summary_data)
|
|
|
|
# Sort by investment score descending
|
|
if not df.empty and 'Investment Score' in df.columns:
|
|
df = df.sort_values('Investment Score', ascending=False)
|
|
|
|
return df
|
|
|
|
|
|
# Example usage
|
|
if __name__ == "__main__":
|
|
# Initialize analyzer
|
|
analyzer = StockOracleAnalyzer("http://localhost:18001")
|
|
|
|
# Analyze a single company
|
|
print("Analyzing Apple Inc. (AAPL)...")
|
|
aapl_data = analyzer.get_company_data("AAPL", period="2y")
|
|
|
|
if aapl_data:
|
|
print(f"Successfully retrieved data for {aapl_data['company_name']}")
|
|
analysis = aapl_data['analysis_summary']
|
|
|
|
print(f"\n📊 Investment Score: {analysis['investment_score']}/100")
|
|
print(f"📈 Financial Grade: {analysis['financial_health']['grade']}")
|
|
print(f"📉 Price Trend: {analysis['price_trends']['trend']}")
|
|
if analysis['sentiment_analysis']:
|
|
print(f"💭 Sentiment: {analysis['sentiment_analysis']['sentiment_label']}")
|
|
|
|
print(f"\n💰 Key Metrics:")
|
|
metrics = analysis['key_metrics']
|
|
if metrics.get('revenue'):
|
|
print(f" Revenue: ${metrics['revenue']/1e9:.2f}B")
|
|
if metrics.get('net_income'):
|
|
print(f" Net Income: ${metrics['net_income']/1e9:.2f}B")
|
|
if metrics.get('pe_ratio'):
|
|
print(f" P/E Ratio: {metrics['pe_ratio']:.2f}")
|
|
if metrics.get('roe'):
|
|
print(f" ROE: {metrics['roe']*100:.2f}%")
|
|
|
|
# Analyze multiple companies
|
|
print("\n" + "="*80)
|
|
print("Analyzing multiple companies...")
|
|
|
|
tickers = ["AAPL", "MSFT", "GOOGL", "TSLA", "NVDA"]
|
|
results = analyzer.analyze_multiple_companies(tickers, period="1y")
|
|
|
|
# Create summary report
|
|
summary_df = analyzer.create_summary_report(results)
|
|
|
|
if not summary_df.empty:
|
|
print("\n📋 Investment Summary Report:")
|
|
print("=" * 120)
|
|
|
|
# Display key columns
|
|
display_cols = [
|
|
'Ticker', 'Company', 'Investment Score', 'Financial Grade',
|
|
'Price Trend', 'Sentiment', 'P/E Ratio', 'ROE (%)', '30d Momentum (%)'
|
|
]
|
|
|
|
available_cols = [col for col in display_cols if col in summary_df.columns]
|
|
print(summary_df[available_cols].to_string(index=False))
|
|
|
|
# Top recommendation
|
|
if 'Investment Score' in summary_df.columns:
|
|
top_pick = summary_df.iloc[0]
|
|
print(f"\n🏆 Top Investment Recommendation: {top_pick['Ticker']} ({top_pick['Company']})")
|
|
print(f" Investment Score: {top_pick['Investment Score']}/100")
|
|
print(f" Financial Grade: {top_pick['Financial Grade']}")
|
|
else:
|
|
print("No data available for analysis.") |