# stocks.py from datetime import datetime, timedelta import numpy as np import json from functools import lru_cache # Importações dos novos módulos from data.database import DatabaseManager from data.api_client import NewsAPIClient from analysis.technical import TechnicalAnalyzer from analysis.fundamental import FundamentalAnalyzer from analysis.sentiment import SentimentAnalyzer from strategies.backtrader import BacktraderIntegration # Database configuration db_manager = DatabaseManager() # Analyzers configuration technical_analyzer = TechnicalAnalyzer() fundamental_analyzer = FundamentalAnalyzer() sentiment_analyzer = SentimentAnalyzer() # Confidence calculator class ConfidenceCalculator: def __init__(self): self.weights = { 'sentiment': 0.4, 'technical': 0.3, 'fundamental': 0.3 } def calculate(self, sentiment, technical, fundamental): sentiment_score = self._normalie_sentiment(sentiment) technical_score = self._normalize_technical(technical) fundamental_score = self._normalize_fundamental(fundamental) weighted_score = ( sentiment_score * self.weights['sentiment'] + technical_score * self.weights['technical'] + fundamental_score * self.weights['fundamental'] ) return { 'total_confidence': weighted_score, 'components': { 'sentiment': sentiment_score, 'technical': technical_score, 'fundamental': fundamental_score } } def _normalie_sentiment(self, sentiment): """ Normaliza o sentimento em um único score de confiança. Parâmetros: sentiment (dict): {'negative': float, 'neutral': float, 'positive': float} Retorno: float: score normalizado entre 0 e 1. """ # Definição dos pesos para cada categoria weight_positive = 1.0 # Sentimento positivo contribui mais weight_neutral = 0.5 # Neutro tem impacto médio weight_negative = 0.0 # Negativo reduz a confiança # Cálculo do score ponderado sentiment_score = ( sentiment['positive'] * weight_positive + sentiment['neutral'] * weight_neutral + sentiment['negative'] * weight_negative ) # Garantindo que o score fique entre 0 e 1 return max(0.0, min(1.0, sentiment_score)) def _normalize_technical(self, tech): if tech is None: return 0.5 rsi_score = 1 - abs(tech['rsi'] - 50) / 50 price_score = np.tanh(tech['price_vs_sma'] * 100) return 0.6 * rsi_score + 0.4 * price_score def _normalize_fundamental(self, fund): if not fund: return 0.5 pe_ratio = fund.get('pe_ratio', 0) sector_pe = fund.get('sector_pe') revenue_growth = fund.get('revenue_growth', 0) if sector_pe is None: pe_score = 0.5 else: pe_score = 1 if pe_ratio < sector_pe else 0.5 growth_score = min(revenue_growth / 20, 1) return 0.5 * pe_score + 0.5 * growth_score # Analysis pipeline class AnalysisPipeline: def __init__(self, sentiment_threshold=0.6, confidence_threshold=0.7): self.confidence_calc = ConfidenceCalculator() self.sentiment_threshold = sentiment_threshold self.confidence_threshold = confidence_threshold def set_sentiment_threshold(self, sentiment_threshold): self.sentiment_threshold = sentiment_threshold def set_confidence_threshold(self, confidence_threshold): self.confidence_threshold = confidence_threshold def analyze_company(self, ticker, news_api_key=None, fetch_new=True): try: # fundamental analysis fundamental = fundamental_analyzer.analyze(ticker) shortName = fundamental['shortName'].split()[0] if fetch_new: db_manager.save_data(ticker, 'financials', fundamental) # collecting last news news = self._get_news(shortName, ticker, news_api_key, fetch_new) # technical analysis technical = technical_analyzer.analyze(ticker) if not fundamental or not news or technical is None: print(f"Insufficient data for: {ticker}") return None # Sentiment analysis sentiment = sentiment_analyzer.analyze(news) # Confidence calculation confidence = self.confidence_calc.calculate( sentiment, technical, self._prepare_fundamental(fundamental) ) # Recommendation generation recommendation = self.generate_recommendation( sentiment, technical, fundamental, confidence ) return { 'recommendation': recommendation, 'confidence': confidence, 'technical': technical, 'fundamental': fundamental, 'sentiment': sentiment } except Exception as e: print(f"Error in analysis: {e}") return None def _get_news(self, shortName, ticker, api_key, fetch_new): if fetch_new and api_key: try: news_client = NewsAPIClient(api_key) articles = news_client.get_news(shortName) # Flatten the list of articles db_manager.save_data(ticker, 'news', articles) return articles except Exception as e: print(f"Error fetching news: {e}") return db_manager.get_historical_data(ticker)['news'] else: return db_manager.get_historical_data(ticker)['news'] def _prepare_fundamental(self, fundamental): return { 'pe_ratio': fundamental.get('trailingPE', 0), 'sector_pe': fundamental.get('sectorPE'), 'revenue_growth': fundamental.get('revenueGrowth', 0) } def generate_recommendation(self, sentiment, technical, fundamental, confidence): pe_ratio = fundamental.get('trailingPE', 0) sector_pe = fundamental.get('sectorPE') if confidence['total_confidence'] < 0.4: return 'NEUTRAL' if sector_pe is not None and sector_pe > 0: if pe_ratio < sector_pe * 0.7: return 'BUY' elif pe_ratio > sector_pe * 1.3: return 'SELL' if confidence['total_confidence'] > self.confidence_threshold and sentiment['positive'] > self.sentiment_threshold: return 'BUY' if technical and 'trend' in technical: return 'HOLD' if technical['trend'] == 'bullish' else 'SELL' return 'NEUTRAL' # Main function if __name__ == "__main__": pipeline = AnalysisPipeline() print("\n=== Stock Market Analysis ===") ticker = input("Enter the company ticker (e.g., AAPL): ").strip().upper() fetch_new = input("Fetch new data from the internet? (y/n): ").lower() fetch_new_bool = fetch_new in ['y', 'yes'] initial_investment = float(input("Initial Investment (USD): ")) years_back = int(input("Historical Period (years): ")) commission = float(input("Commission per trade: ")) news_api_key = 'YOUR_NEWSAPI_KEY_HERE' print(f"\nRunning analysis for {ticker}...") result = pipeline.analyze_company( ticker=ticker, news_api_key=news_api_key if news_api_key else None, fetch_new=fetch_new_bool ) if result: end_date = datetime.now() start_date = end_date - timedelta(days=years_back * 365) bt_integration = BacktraderIntegration(result) bt_integration.add_data_feed(ticker, start_date, end_date) final_value = bt_integration.run_simulation(initial_investment, commission) print("\n=== Analysis Result ===") print(f"Recommendation: {result['recommendation']}") print(f"Confidence: {result['confidence']['total_confidence']:.2%}") print(f"Simulation Return: {(final_value / initial_investment - 1) * 100:.2f}%") print("\nDetails:") print(f"1. Sentiment: {json.dumps(result['sentiment'], indent=2)}") print(f"2. Technical Analysis: RSI {result['technical']['rsi']:.1f}, Price vs SMA50: {result['technical']['price_vs_sma']:.2%}") print(f"3. Fundamental: P/E {result['fundamental'].get('trailingPE', 'N/A')} vs Sector {result['fundamental'].get('sectorPE', 'N/A')}") print(f"4. Confidence Components: {json.dumps(result['confidence']['components'], indent=2)}") else: print("\nUnable to complete analysis. Please check:") print("- Internet connection") print("- Ticker symbol") print("- Availability of historical data")