add stock predictor, fix news iterator, etc

.gitignore

@@ -1,6 +1,20 @@
-.env
+# Ignore Python cache files
+__pycache__/
+# Ignore virtual environment directories
 .venv/
+# Ignore environment variable files
+.env
+# Ignore lock files
 uv.lock
-src/__pycache__/
+# Ignore egg-info directories
 src/your_project_name.egg-info/
-src/news_sentoment_analyzer.egg-info/
+src/news_sentoment_analyzer.egg-info/
+# Ignore Python bytecode files
+*.pyc
+*.pyo
+# Ignore Jupyter Notebook checkpoints
+.ipynb_checkpoints/
+# Ignore IDE specific files
+.idea/
+# Ignore logs
+logs/

pyproject.toml

@@ -1,5 +1,6 @@
 [project]
-name = "news_sentoment_analyzer"
+requires-python = ">=3.11"
+name = "financial_news_bot"
 version = "0.1.0"
 description = "new sentiment analyzer using lexicon based and machine learning techniques"
 authors = [
@@ -63,6 +64,8 @@ dependencies = [
     "isort>=5.10.0",
     "python-telegram-bot>=20.0",
     "dotenv>=0.9.9",
+    "keras>=3.11.2",
+    "tensorflow>=2.20.0",
 ]
 
 [build-system]

src/core/risk_management/risk_analyzer.py

@@ -25,7 +25,7 @@ class RiskAnalyzer:
             http_options=types.HttpOptions(api_version='v1')
         )
 
-    def get_stock_data(self, ticker: str, period: str = "1y") -> pd.DataFrame:
+    def get_stock_data(self, ticker: str, period: str = "3mo") -> pd.DataFrame:
         try:
             stock = yf.Ticker(ticker)
             data = stock.history(period=period)

src/services/async_stock_price_predictor.py

@@ -4,6 +4,7 @@ import asyncio
 from datetime import datetime, timezone, timedelta
 from collections import defaultdict
 from typing import Any
+import warnings
 
 import numpy as np
 import pandas as pd
@@ -14,6 +15,7 @@ from sklearn.preprocessing import MinMaxScaler
 from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline
 from huggingface_hub import hf_hub_download
 from concurrent.futures import ProcessPoolExecutor
+import yfinance as yf
 
 from src.telegram_bot.logger import main_logger as logger
 
@@ -301,6 +303,46 @@ class AsyncStockPricePredictor:
             logger.error(f"Error fetching stock data for {ticker}: {e}")
             raise RuntimeError(f"Failed to fetch stock data: {e}")
 
+    @staticmethod
+    async def fetch_prices(ticker: str, period: str = "6mo", interval: str = "1d") -> pd.DataFrame | None:
+        """
+        Fetch historical stock price data from Yahoo Finance.
+
+        Args:
+            ticker: Stock ticker symbol (e.g., 'AAPL')
+            period: Time period for data (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)
+            interval: Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)
+
+        Returns:
+            DataFrame with OHLCV data or None if error occurs
+        """
+        try:
+            logger.info(f"Fetching data for {ticker}")
+
+            # Suppress yfinance warnings
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                df = yf.download(ticker, period=period, interval=interval, progress=False)
+
+            if df.empty:
+                logger.error(f"No data found for ticker {ticker}")
+                return None
+
+            # Select relevant columns
+            df = df[["Open", "High", "Low", "Close", "Volume"]].copy()
+            df.dropna(inplace=True)
+
+            if len(df) < 60:
+                logger.warning(f"Insufficient data for {ticker}. Got {len(df)} days, need at least 60")
+                return None
+
+            logger.info(f"Successfully fetched {len(df)} data points for {ticker}")
+            return df
+
+        except Exception as e:
+            logger.error(f"Error fetching data for {ticker}: {e}")
+            return None
+
     async def fetch_news(self, ticker: str) -> list[dict[str, Any]]:
         """Fetch recent news for a stock ticker."""
         url = f"https://query1.finance.yahoo.com/v6/finance/news?symbols={ticker.upper()}"

src/services/news_iterator.py

@@ -1,8 +1,8 @@
 import os
 import json
-import time
 import re
-from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+from datetime import datetime
 import asyncio
 from dataclasses import dataclass
 
@@ -121,12 +121,12 @@ class CompanyNewsPostsIterator:
     NEWS_PING_PERIOD = 120 # seconds
 
     def __init__(self, finhub_api_key: str | None, google_api_key: str | None, sentiments_only: list[str] | None = None,
-                 company_symbol: str = "NVDA", date: str | None = None):
+                 company_symbol: str = "NVDA", time_zone: str = "America/New_York"):
         self.finhub_api_key = finhub_api_key
         self.finnhub_client = self.build_finnhub_client()
         self.sentiment_analyzer = SentimentAnalyzer(google_api_key=google_api_key)
         self.company_symbol = company_symbol
-        self.date = date or self.default_date()
+        self.time_zone = time_zone
         self.sentiments_only = sentiments_only or ['positive', 'negative']
         self.news_posts = []
         self.latest_timestamp = None
@@ -161,9 +161,10 @@ class CompanyNewsPostsIterator:
 
     async def read_news_posts(self):
         loop = asyncio.get_event_loop()
+        date = self.news_date()
         news_posts = await loop.run_in_executor(
             None,
-            lambda: self.finnhub_client.company_news(self.company_symbol, _from=self.date, to=self.date)
+            lambda: self.finnhub_client.company_news(self.company_symbol, _from=date, to=date)
         )
         decorated_news_posts = [NewsPostDecorator.from_dict(news_post) for news_post in news_posts]
         self.news_posts = self.unread_posts(decorated_news_posts)
@@ -171,15 +172,15 @@ class CompanyNewsPostsIterator:
     def build_finnhub_client(self):
         return finnhub.Client(api_key=self.finhub_api_key)
 
-    def default_date(self):
-        return datetime.now().strftime('%Y-%m-%d')
-
     def unread_posts(self, posts):
         if not self.latest_timestamp:
             return posts
         idx = next((i for i, post in enumerate(posts) if post.datetime == self.latest_timestamp), 0)
         return posts[:idx]
 
+    def news_date(self):
+        return datetime.now(ZoneInfo(self.time_zone)).strftime('%Y-%m-%d')
+
 
 if __name__ == '__main__':
     iterator = CompanyNewsPostsIterator(

src/services/stock_predictor.py

@@ -0,0 +1,451 @@
+"""
+Async Stock Price Predictor using Amazon Chronos T5-Small Time Series Model
+
+Required installations:
+pip install chronos-forecasting yfinance torch numpy pandas aiohttp asyncio
+
+Usage:
+python stock_predictor.py
+"""
+
+import yfinance as yf
+import torch
+import numpy as np
+from chronos import ChronosPipeline
+import pandas as pd
+import logging
+import asyncio
+import aiohttp
+from concurrent.futures import ThreadPoolExecutor
+from typing import Optional, Tuple, List, Dict
+from datetime import datetime
+import warnings
+import time
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class AsyncStockPredictor:
+    """
+    An async stock price predictor using Amazon Chronos T5 time series model.
+
+    This class fetches historical stock data asynchronously and uses the Chronos model
+    to predict future stock prices and movement trends with concurrent processing.
+    """
+
+    def __init__(self, model_name: str = "amazon/chronos-t5-small", max_workers: int = 4):
+        """
+        Initialize the async stock predictor with Chronos model.
+
+        Args:
+            model_name: Name of the Chronos model to use
+            max_workers: Maximum number of worker threads for CPU-intensive tasks
+        """
+        self.model_name = model_name
+        self.max_workers = max_workers
+        self.executor = ThreadPoolExecutor(max_workers=max_workers)
+        self.pipeline = None
+
+    async def initialize(self):
+        """Initialize the model asynchronously."""
+        try:
+            logger.info(f"Loading Chronos model: {self.model_name}")
+            # Run model loading in thread pool to avoid blocking
+            self.pipeline = await asyncio.get_event_loop().run_in_executor(
+                self.executor, self._load_model
+            )
+            logger.info("Chronos model loaded successfully")
+        except Exception as e:
+            logger.error(f"Error loading model: {e}")
+            raise
+
+    def _load_model(self):
+        """Load the Chronos model (CPU intensive, runs in thread pool)."""
+        try:
+            return ChronosPipeline.from_pretrained(
+                self.model_name,
+                device_map="auto",
+                torch_dtype=torch.bfloat16,
+            )
+        except Exception as e:
+            logger.warning(f"Failed to load with optimized settings: {e}")
+            logger.info("Attempting to load with default settings...")
+            return ChronosPipeline.from_pretrained(self.model_name)
+
+    async def fetch_prices_async(self, ticker: str, period: str = "6mo", interval: str = "1d") -> Optional[
+        pd.DataFrame]:
+        """
+        Fetch historical stock price data asynchronously.
+
+        Args:
+            ticker: Stock ticker symbol (e.g., 'AAPL')
+            period: Time period for data (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)
+            interval: Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)
+
+        Returns:
+            DataFrame with OHLCV data or None if error occurs
+        """
+        try:
+            logger.info(f"Fetching data for {ticker}")
+
+            # Run yfinance download in thread pool to avoid blocking
+            df = await asyncio.get_event_loop().run_in_executor(
+                self.executor, self._fetch_data_sync, ticker, period, interval
+            )
+
+            if df is None or df.empty:
+                logger.error(f"No data found for ticker {ticker}")
+                return None
+
+            # Select relevant columns
+            df = df[["Open", "High", "Low", "Close", "Volume"]].copy()
+            df.dropna(inplace=True)
+
+            if len(df) < 30:
+                logger.warning(f"Insufficient data for {ticker}. Got {len(df)} days, need at least 30")
+                return None
+
+            logger.info(f"Successfully fetched {len(df)} data points for {ticker}")
+            return df
+
+        except Exception as e:
+            logger.error(f"Error fetching data for {ticker}: {e}")
+            return None
+
+    def _fetch_data_sync(self, ticker: str, period: str, interval: str) -> Optional[pd.DataFrame]:
+        """Synchronous data fetching (runs in thread pool)."""
+        try:
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                df = yf.download(ticker, period=period, interval=interval, progress=False)
+            return df
+        except Exception as e:
+            logger.error(f"Error in sync data fetch for {ticker}: {e}")
+            return None
+
+    async def predict_next_day_async(self, prices: pd.DataFrame, prediction_length: int = 1, num_samples: int = 20) -> \
+    Tuple[str, float, List[float]]:
+        """
+        Predict next day's price using Chronos time series model asynchronously.
+
+        Args:
+            prices: DataFrame with historical price data
+            prediction_length: Number of future periods to predict
+            num_samples: Number of sample predictions to generate
+
+        Returns:
+            Tuple of (trend_description, confidence_score, predicted_prices)
+        """
+        if self.pipeline is None:
+            return "❌ Model not initialized", 0.0, []
+
+        if prices is None or len(prices) < 30:
+            return "❓ Insufficient data", 0.0, []
+
+        try:
+            # Run prediction in thread pool as it's CPU intensive
+            result = await asyncio.get_event_loop().run_in_executor(
+                self.executor, self._predict_sync, prices, prediction_length, num_samples
+            )
+            return result
+
+        except Exception as e:
+            logger.error(f"Error during async prediction: {e}")
+            return "❌ Prediction error", 0.0, []
+
+    def _predict_sync(self, prices: pd.DataFrame, prediction_length: int, num_samples: int) -> Tuple[
+        str, float, List[float]]:
+        """Synchronous prediction (runs in thread pool)."""
+        try:
+            # Use closing prices as the time series
+            closes = prices["Close"].values
+            context_length = min(len(closes), 512)  # Chronos context limit
+            context = closes[-context_length:]
+
+            logger.info(f"Using {context_length} data points for prediction")
+
+            # Convert to tensor
+            context_tensor = torch.tensor(context, dtype=torch.float32).unsqueeze(0)
+
+            # Generate predictions
+            with torch.no_grad():
+                forecast = self.pipeline.predict(
+                    context=context_tensor,
+                    prediction_length=prediction_length,
+                    num_samples=num_samples
+                )
+
+            # Extract predictions
+            predictions = forecast[0, :, 0].numpy()  # First batch, all samples, first prediction step
+
+            # Calculate statistics
+            mean_prediction = np.mean(predictions)
+            std_prediction = np.std(predictions)
+
+            current_price = float(closes[-1])
+            price_change_pct = ((mean_prediction - current_price) / current_price) * 100
+
+            # Determine trend based on prediction
+            if price_change_pct > 2.0:
+                trend = "🚀 Strong Growth Expected"
+                confidence = min(0.9, abs(price_change_pct) / 10.0)
+            elif price_change_pct > 0.5:
+                trend = "📈 Moderate Growth Expected"
+                confidence = min(0.7, abs(price_change_pct) / 5.0)
+            elif price_change_pct < -2.0:
+                trend = "📉 Strong Decline Expected"
+                confidence = min(0.9, abs(price_change_pct) / 10.0)
+            elif price_change_pct < -0.5:
+                trend = "📉 Moderate Decline Expected"
+                confidence = min(0.7, abs(price_change_pct) / 5.0)
+            else:
+                trend = "➡️ Sideways Movement Expected"
+                confidence = 0.5
+
+            # Adjust confidence based on prediction variance
+            variance_factor = min(1.0, std_prediction / current_price)
+            confidence = max(0.1, confidence * (1 - variance_factor))
+
+            logger.info(f"Prediction: ${mean_prediction:.2f} ({price_change_pct:+.2f}%) - {trend}")
+
+            return trend, confidence, predictions.tolist()
+
+        except Exception as e:
+            logger.error(f"Error in sync prediction: {e}")
+            return "❌ Prediction error", 0.0, []
+
+    async def calculate_technical_indicators_async(self, prices: pd.DataFrame) -> dict:
+        """
+        Calculate basic technical indicators asynchronously.
+
+        Args:
+            prices: DataFrame with historical price data
+
+        Returns:
+            Dictionary with technical indicators
+        """
+        try:
+            # Run calculations in thread pool for consistency
+            indicators = await asyncio.get_event_loop().run_in_executor(
+                self.executor, self._calculate_indicators_sync, prices
+            )
+            return indicators
+        except Exception as e:
+            logger.error(f"Error calculating technical indicators: {e}")
+            return {}
+
+    def _calculate_indicators_sync(self, prices: pd.DataFrame) -> dict:
+        """Synchronous indicator calculation."""
+        try:
+            # Simple moving averages
+            sma_20 = prices['Close'].rolling(window=20).mean().iloc[-1]
+            sma_50 = prices['Close'].rolling(window=50).mean().iloc[-1]
+
+            # Price change
+            price_change = ((prices['Close'].iloc[-1] - prices['Close'].iloc[-2]) / prices['Close'].iloc[-2]) * 100
+
+            # Volume analysis
+            avg_volume = prices['Volume'].rolling(window=20).mean().iloc[-1]
+            current_volume = prices['Volume'].iloc[-1]
+            volume_ratio = current_volume / avg_volume if avg_volume > 0 else 1.0
+
+            return {
+                'sma_20': sma_20,
+                'sma_50': sma_50,
+                'price_change': price_change,
+                'volume_ratio': volume_ratio
+            }
+        except Exception as e:
+            logger.error(f"Error in sync indicator calculation: {e}")
+            return {}
+
+    async def analyze_stock_async(self, ticker: str) -> str:
+        """
+        Perform complete stock analysis asynchronously.
+
+        Args:
+            ticker: Stock ticker symbol
+
+        Returns:
+            Formatted analysis message
+        """
+        try:
+            # Fetch price data
+            prices = await self.fetch_prices_async(ticker)
+
+            if prices is None:
+                return f"❌ Could not fetch data for {ticker}"
+
+            # Run prediction and technical analysis concurrently
+            prediction_task = self.predict_next_day_async(prices)
+            indicators_task = self.calculate_technical_indicators_async(prices)
+
+            # Wait for both tasks to complete
+            (trend, confidence, predictions), indicators = await asyncio.gather(
+                prediction_task, indicators_task
+            )
+
+            # Create analysis message
+            message = await self.create_analysis_message_async(
+                ticker, prices, trend, confidence, predictions, indicators
+            )
+
+            return message
+
+        except Exception as e:
+            logger.error(f"Error analyzing {ticker}: {e}")
+            return f"❌ Error analyzing {ticker}: {e}"
+
+    async def create_analysis_message_async(self, ticker: str, prices: pd.DataFrame, trend: str,
+                                            confidence: float, predictions: List[float] = None,
+                                            indicators: dict = None) -> str:
+        """
+        Create a comprehensive analysis message asynchronously.
+
+        Args:
+            ticker: Stock ticker symbol
+            prices: DataFrame with price data
+            trend: Predicted trend
+            confidence: Prediction confidence score
+            predictions: List of predicted prices
+            indicators: Technical indicators dictionary
+
+        Returns:
+            Formatted analysis message
+        """
+        if prices is None or prices.empty:
+            return f"❌ Unable to analyze {ticker} - no data available"
+
+        try:
+            last_close = float(prices["Close"].iloc[-1])
+            last_date = prices.index[-1].strftime('%Y-%m-%d')
+
+            message_parts = [
+                f"📊 **Stock Analysis: {ticker}**",
+                f"📅 Date: {last_date}",
+                f"💰 Current Price: ${last_close:.2f}",
+                f"🔮 Prediction: {trend}",
+                f"🎯 Confidence: {confidence:.1%}",
+                ""
+            ]
+
+            # Add price prediction if available
+            if predictions and len(predictions) > 0:
+                mean_pred = np.mean(predictions)
+                min_pred = np.min(predictions)
+                max_pred = np.max(predictions)
+                price_change = ((mean_pred - last_close) / last_close) * 100
+
+                message_parts.extend([
+                    "🎲 **Price Predictions:**",
+                    f"• Expected Price: ${mean_pred:.2f} ({price_change:+.2f}%)",
+                    f"• Price Range: ${min_pred:.2f} - ${max_pred:.2f}",
+                    f"• Prediction Samples: {len(predictions)}",
+                    ""
+                ])
+
+            # Add technical indicators if available
+            if indicators:
+                message_parts.extend([
+                    "📈 **Technical Indicators:**",
+                    f"• 20-day SMA: ${indicators.get('sma_20', 0):.2f}",
+                    f"• 50-day SMA: ${indicators.get('sma_50', 0):.2f}",
+                    f"• Daily Change: {indicators.get('price_change', 0):.2f}%",
+                    f"• Volume Ratio: {indicators.get('volume_ratio', 0):.2f}x",
+                    ""
+                ])
+
+            message_parts.extend([
+                "⚠️ **Disclaimer:** This is AI-generated analysis, not financial advice.",
+                "Predictions are based on historical patterns and may not reflect future performance.",
+                "Always do your own research and consult financial advisors before investing."
+            ])
+
+            return "\n".join(message_parts)
+
+        except Exception as e:
+            logger.error(f"Error creating message: {e}")
+            return f"❌ Error creating analysis for {ticker}"
+
+    async def analyze_multiple_stocks(self, tickers: List[str]) -> Dict[str, str]:
+        """
+        Analyze multiple stocks concurrently.
+
+        Args:
+            tickers: List of stock ticker symbols
+
+        Returns:
+            Dictionary mapping tickers to analysis messages
+        """
+        tasks = [self.analyze_stock_async(ticker) for ticker in tickers]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        analysis_results = {}
+        for ticker, result in zip(tickers, results):
+            if isinstance(result, Exception):
+                analysis_results[ticker] = f"❌ Error analyzing {ticker}: {result}"
+            else:
+                analysis_results[ticker] = result
+
+        return analysis_results
+
+    async def close(self):
+        """Clean up resources."""
+        if hasattr(self, 'executor'):
+            self.executor.shutdown(wait=True)
+        logger.info("AsyncStockPredictor resources cleaned up")
+
+
+async def main():
+    """Main async function to demonstrate the stock predictor."""
+    predictor = AsyncStockPredictor()
+
+    try:
+        # Initialize the model
+        await predictor.initialize()
+
+        # List of tickers to analyze
+        tickers = ["AAPL", "GOOGL", "MSFT", "TSLA", "NVDA", "AMD"]
+
+        print(f"\n🚀 Starting concurrent analysis of {len(tickers)} stocks...")
+        start_time = time.time()
+
+        # Analyze all stocks concurrently
+        results = await predictor.analyze_multiple_stocks(tickers)
+
+        end_time = time.time()
+        total_time = end_time - start_time
+
+        # Display results
+        for ticker, analysis in results.items():
+            print(f"\n{'=' * 60}")
+            print(f"Analysis for {ticker}")
+            print('=' * 60)
+            print(analysis)
+
+        print(f"\n🏁 Analysis completed in {total_time:.2f} seconds")
+        print(f"⚡ Average time per stock: {total_time / len(tickers):.2f} seconds")
+
+    except Exception as e:
+        logger.error(f"Error in main execution: {e}")
+        print(f"❌ Application error: {e}")
+
+    finally:
+        # Clean up resources
+        await predictor.close()
+
+
+def run_async_analysis():
+    """Entry point for running the async analysis."""
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("\n🛑 Analysis interrupted by user")
+    except Exception as e:
+        print(f"❌ Fatal error: {e}")
+
+
+if __name__ == "__main__":
+    run_async_analysis()

src/telegram_bot/telegram_bot_service.py

@@ -19,6 +19,7 @@ from src.services.news_pooling_service import NewsPollingService
 from src.core.risk_management.risk_analyzer import RiskAnalyzer
 from src.services.news_iterator import CompanyNewsPostsIterator
 from src.services.async_stock_price_predictor import AsyncStockPricePredictor, handle_stock_prediction
+from src.services.stock_predictor import AsyncStockPredictor
 
 
 class TelegramBotService:
@@ -522,11 +523,16 @@ class TelegramBotService:
     ) -> None:
         ticker = 'NVDA'  # Default ticker if not specified
         # Show available backends
+        '''
         available_backends = AsyncStockPricePredictor.get_available_backends()
         main_logger.info(f"Available Keras backends: {available_backends}")
         main_logger.info(f"\nUsing Keras backend: {os.environ.get('KERAS_BACKEND')}")
+        '''
+        predictor = AsyncStockPredictor()
+        await predictor.initialize()
         ticker = command_parts[1].upper() if len(command_parts) == 2 else ticker
         await self.send_message_via_proxy(chat_id, f"Predicting the price for the ticker {ticker} ...")
+        '''
         # Initialize with Keras 3.0 and JAX backend
         predictor = AsyncStockPricePredictor(
             lstm_model_repo="jengyang/lstm-stock-prediction-model",
@@ -535,6 +541,23 @@ class TelegramBotService:
             max_workers=2,
             keras_backend="jax"  # Can also use "torch" or "tensorflow"
         )
-        result = await handle_stock_prediction(ticker, predictor)
-        main_logger.info(f'predicted price: {result}')
+        '''
+        # List of tickers to analyze
+        tickers = [ticker]
+        main_logger.info(f"\n🚀 Starting concurrent analysis of {len(tickers)} stocks...")
+        start_time = time.time()
+        # Analyze all stocks concurrently
+        results = await predictor.analyze_multiple_stocks(tickers)
+        end_time = time.time()
+        total_time = end_time - start_time
+        # Display results
+        for ticker, analysis in results.items():
+            main_logger.info(f"\n{'=' * 60}")
+            main_logger.info(f"Analysis for {ticker}")
+            main_logger.info('=' * 60)
+            main_logger.info(analysis)
+        main_logger.info(f"\n🏁 Analysis completed in {total_time:.2f} seconds")
+        main_logger.info(f"⚡ Average time per stock: {total_time / len(tickers):.2f} seconds")
+        #main_logger.info(f'predicted price: {result}')
+        result = results.get(ticker)
         await self.send_message_via_proxy(chat_id, result)