Add ticker scanner module with Telegram integration

- Create modular ticker scanner system for monitoring growing stocks
- Implement TickerAnalyzer orchestrator with parallel data downloading
- Add comprehensive growth metrics analysis (CAGR, Sharpe, acceleration)
- Fix scheduler bugs and enable hourly monitoring
- Integrate TradingView link generation for each ticker
- Add /scan command to Telegram bot for on-demand scanning
- Add logging throughout ticker scanner modules
- Refactor legacy code into maintainable components

The scanner analyzes top 20 fast and stable growing tickers from Yahoo Finance,
ranks them by velocity score, and sends formatted reports to Telegram.

src/core/ticker_scanner/__init__.py

@@ -0,0 +1,19 @@
+"""
+Ticker Scanner Module
+Monitors and analyzes stock tickers for growth potential
+"""
+
+from src.core.ticker_scanner.ticker_analyzer import TickerAnalyzer
+from src.core.ticker_scanner.scheduler import Scheduler
+from src.core.ticker_scanner.growth_speed_analyzer import GrowthSpeedAnalyzer
+from src.core.ticker_scanner.core_enums import StockExchange, GrowthCategory
+from src.core.ticker_scanner.growth_metrics import GrowthSpeedMetrics
+
+__all__ = [
+    'TickerAnalyzer',
+    'Scheduler',
+    'GrowthSpeedAnalyzer',
+    'StockExchange',
+    'GrowthCategory',
+    'GrowthSpeedMetrics',
+]

src/core/ticker_scanner/growth_metrics.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+
+from src.core.ticker_scanner.core_enums import GrowthCategory
+
+
+@dataclass
+class GrowthSpeedMetrics:
+    """Detailed growth velocity analysis"""
+    # Linear metrics
+    linear_slope: float  # Daily price change slope
+    linear_r_squared: float  # Goodness of fit
+    # Compound metrics
+    cagr: float  # Compound Annual Growth Rate
+    total_return: float  # Total percentage return
+    # Acceleration metrics
+    acceleration: float  # Second derivative of growth
+    recent_momentum: float  # Last 6 months vs overall trend
+    # Volatility-adjusted
+    sharpe_ratio: float  # Risk-adjusted return
+    sortino_ratio: float  # Downside risk-adjusted return
+    # Speed classification
+    growth_velocity_score: float  # 0-100 composite score
+    growth_category: GrowthCategory

src/core/ticker_scanner/growth_speed_analyzer.py

@@ -0,0 +1,116 @@
+import numpy as np
+import pandas as pd
+from sklearn.linear_model import LinearRegression
+
+from src.core.ticker_scanner.core_enums import GrowthCategory
+from src.core.ticker_scanner.growth_metrics import GrowthSpeedMetrics
+from src.telegram_bot.logger import main_logger as logger
+
+
+class GrowthSpeedAnalyzer:
+    """Advanced growth velocity and acceleration analysis"""
+
+    @staticmethod
+    def analyze(prices: np.ndarray, dates: pd.DatetimeIndex) -> GrowthSpeedMetrics:
+        """Comprehensive growth speed analysis"""
+
+        # Time series setup
+        n = len(prices)
+        x = np.arange(n).reshape(-1, 1)
+        y = prices.reshape(-1, 1)
+
+        # 1. Linear regression metrics
+        model = LinearRegression().fit(x, y)
+        linear_slope = model.coef_[0][0]
+        predictions = model.predict(x)
+
+        ss_res = np.sum((y - predictions) ** 2)
+        ss_tot = np.sum((y - np.mean(y)) ** 2)
+        r_squared = 1 - (ss_res / ss_tot) if ss_tot > 0 else 0
+
+        # 2. Compound growth metrics
+        years = (dates[-1] - dates[0]).days / 365.25
+        total_return = ((prices[-1] / prices[0]) - 1) * 100
+        cagr = (pow(prices[-1] / prices[0], 1 / years) - 1) * 100 if years > 0 else 0
+
+        # 3. Acceleration (second derivative)
+        returns = np.diff(np.log(prices))
+        if len(returns) > 1:
+            acceleration = np.polyfit(range(len(returns)), returns, 1)[0] * 252  # Annualized
+        else:
+            acceleration = 0
+
+        # 4. Recent momentum (last 6 months vs overall)
+        six_months_ago = max(0, n - 126)  # ~6 months of trading days
+        recent_return = (prices[-1] / prices[six_months_ago] - 1) * 100 if six_months_ago < n else 0
+        recent_momentum = recent_return - (total_return / years * 0.5) if years > 0 else 0
+
+        # 5. Risk-adjusted metrics
+        daily_returns = np.diff(prices) / prices[:-1]
+
+        if len(daily_returns) > 0:
+            mean_return = np.mean(daily_returns) * 252  # Annualized
+            std_return = np.std(daily_returns) * np.sqrt(252)  # Annualized
+
+            sharpe_ratio = mean_return / std_return if std_return > 0 else 0
+
+            # Sortino ratio (downside deviation)
+            downside_returns = daily_returns[daily_returns < 0]
+            downside_std = np.std(downside_returns) * np.sqrt(252) if len(downside_returns) > 0 else std_return
+            sortino_ratio = mean_return / downside_std if downside_std > 0 else 0
+        else:
+            sharpe_ratio = sortino_ratio = 0
+
+        # 6. Composite growth velocity score (0-100)
+        velocity_score = GrowthSpeedAnalyzer._calculate_velocity_score(
+            cagr, r_squared, acceleration, sharpe_ratio, recent_momentum
+        )
+
+        # 7. Categorize growth speed
+        growth_category = GrowthSpeedAnalyzer._categorize_growth(velocity_score)
+
+        return GrowthSpeedMetrics(
+            linear_slope=linear_slope,
+            linear_r_squared=r_squared,
+            cagr=cagr,
+            total_return=total_return,
+            acceleration=acceleration,
+            recent_momentum=recent_momentum,
+            sharpe_ratio=sharpe_ratio,
+            sortino_ratio=sortino_ratio,
+            growth_velocity_score=velocity_score,
+            growth_category=growth_category
+        )
+
+    @staticmethod
+    def _calculate_velocity_score(cagr: float, r_squared: float,
+                                  acceleration: float, sharpe: float,
+                                  momentum: float) -> float:
+        """Calculate composite 0-100 growth velocity score"""
+        # Normalize components to 0-1 scale
+        cagr_norm = min(max(cagr / 50, 0), 1)  # Cap at 50% CAGR
+        r_squared_norm = max(r_squared, 0)
+        accel_norm = min(max((acceleration + 1) / 2, 0), 1)  # Center at 0
+        sharpe_norm = min(max(sharpe / 3, 0), 1)  # Cap at 3.0
+        momentum_norm = min(max((momentum + 50) / 100, 0), 1)  # -50 to +50 range
+        # Weighted combination
+        score = (
+            0.35 * cagr_norm +
+            0.20 * r_squared_norm +
+            0.20 * accel_norm +
+            0.15 * sharpe_norm +
+            0.10 * momentum_norm
+        ) * 100
+        return round(score, 2)
+
+    @staticmethod
+    def _categorize_growth(velocity_score: float) -> GrowthCategory:
+        """Classify growth speed"""
+        if velocity_score >= 75:
+            return GrowthCategory.EXPLOSIVE
+        elif velocity_score >= 60:
+            return GrowthCategory.STRONG
+        elif velocity_score >= 40:
+            return GrowthCategory.MODERATE
+        else:
+            return GrowthCategory.SLOW

src/core/ticker_scanner/parallel_data_downloader.py

@@ -14,6 +14,7 @@ import yfinance as yf
 
 from src.core.ticker_scanner.core_enums import StockExchange
 from src.core.ticker_scanner.tickers_provider import TickersProvider
+from src.telegram_bot.logger import main_logger as logger
 
 
 MAX_WORKERS = 8                 # Number of parallel processes
@@ -26,20 +27,25 @@ MIN_DATA_POINTS = 50            # Minimum number of price points required
 def fetch_prices(ticker: str, max_retries: int = MAX_RETRIES) -> dict[str, Any]:
     """
     Download all-time closing prices for a single ticker safely.
-    Returns dict {'ticker': ticker, 'prices': ndarray} or None if failed.
+    Returns dict {'ticker': ticker, 'prices': ndarray, 'dates': DatetimeIndex} or None if failed.
     """
     for attempt in range(max_retries):
         try:
             df = yf.download(ticker, period="max", progress=False, auto_adjust=True)
-            closes = df["Close"].dropna().values
+            closes = df["Close"].dropna()
             if len(closes) < MIN_DATA_POINTS:
                 return None
-            return {"ticker": ticker, "prices": closes}
+            return {
+                "ticker": ticker,
+                "prices": closes.values,
+                "dates": closes.index
+            }
         except yf.shared.YFRateLimitError:
             wait = SLEEP_BETWEEN_RETRIES + random.random()
-            print(f"⚠️ Rate limited for {ticker}. Waiting {wait:.1f}s and retrying...")
+            logger.warning(f"Rate limited for {ticker}. Waiting {wait:.1f}s and retrying...")
             time.sleep(wait)
-        except Exception:
+        except Exception as e:
+            logger.debug(f"Failed to fetch {ticker}: {e}")
             return None
     return None
 
@@ -57,23 +63,22 @@ def batch(iterable: list[str], n: int = BATCH_SIZE):
 def download_tickers_parallel(tickers: list[str], max_workers: int = MAX_WORKERS) -> list[dict[str, Any]]:
     """
     Download a large list of tickers in parallel batches.
-    Returns a list of {'ticker': ..., 'prices': ...} dicts.
+    Returns a list of {'ticker': ..., 'prices': ..., 'dates': ...} dicts.
     """
     all_results = []
     all_failed = []
 
     for batch_num, ticker_batch in enumerate(batch(tickers, BATCH_SIZE), start=1):
-        print(f"🔹 Processing batch {batch_num}: {len(ticker_batch)} tickers")
+        logger.info(f"Processing batch {batch_num}: {len(ticker_batch)} tickers")
         results, failed = process_batch(ticker_batch, max_workers)
         all_results.extend(results)
         all_failed.extend(failed)
         # small sleep between batches to reduce rate-limit chance
         time.sleep(1 + random.random())
 
-    print(f"\n✅ Total downloaded: {len(all_results)}")
+    logger.info(f"Total downloaded: {len(all_results)}")
     if all_failed:
-        print(f"❌ Total failed: {len(all_failed)}")
-        print("Failed tickers:", all_failed)
+        logger.warning(f"Total failed: {len(all_failed)} - {all_failed[:10]}")  # Show first 10
     return all_results
 
 def process_batch(ticker_batch: list[str], max_workers: int) -> tuple[list[dict[str, Any]], list[Any]]:
@@ -97,13 +102,24 @@ def process_batch(ticker_batch: list[str], max_workers: int) -> tuple[list[dict[
                 failed.append(ticker)
     return results, failed
 
-def run_parallel_data_downloader():
-    all_tickers = TickersProvider().get_tickers(StockExchange.NASDAQ)
-    tickers = all_tickers[:200]  # Limit to first 200 for testing
-    print("🚀 Starting parallel download...")
+def run_parallel_data_downloader(exchange: StockExchange = StockExchange.NASDAQ,
+                                 limit: int = 200) -> list[dict[str, Any]]:
+    """
+    Main function to download ticker data in parallel.
+
+    Args:
+        exchange: Stock exchange to download from
+        limit: Maximum number of tickers to download
+
+    Returns:
+        List of dicts with ticker, prices, and dates
+    """
+    all_tickers = TickersProvider().get_tickers(exchange)
+    tickers = all_tickers[:limit]
+    logger.info(f"Starting parallel download for {len(tickers)} tickers from {exchange.value}...")
     data = download_tickers_parallel(tickers)
-    for d in data:
-        print(f"{d['ticker']}: {len(d['prices'])} price points")
+    logger.info(f"Downloaded {len(data)} tickers successfully")
+    return data
 
 
 if __name__ == "__main__":

src/core/ticker_scanner/scheduler.py

@@ -1,17 +1,21 @@
 import time
 import signal
+import asyncio
 
 import schedule
 
 from src.telegram_bot.logger import main_logger as logger
+from src.core.ticker_scanner.growth_speed_analyzer import GrowthSpeedAnalyzer
+from src.core.ticker_scanner.parallel_data_downloader import run_parallel_data_downloader
+from src.core.ticker_scanner.ticker_analyzer import TickerAnalyzer
 
 
 class Scheduler:
     """Schedule and manage periodic analysis"""
-    def __init__(self, exchange: Exchange = Exchange.SP500,
-                 schedule_time: str = "18:00"):
+    def __init__(self, exchange: str = "NASDAQ", interval_hours: int = 1, telegram_bot_service=None):
         self.exchange = exchange
-        self.schedule_time = schedule_time
+        self.interval_hours = interval_hours
+        self.telegram_bot_service = telegram_bot_service
         self.running = True
         # Setup signal handlers for graceful shutdown
         signal.signal(signal.SIGINT, self._signal_handler)
@@ -22,31 +26,33 @@ class Scheduler:
         logger.info("Received shutdown signal. Cleaning up...")
         self.running = False
 
-    def run_scheduled_job(self):
+    async def run_scheduled_job(self):
         """Execute the analysis job"""
         try:
-            analyzer = AsyncTrendAnalyzer(self.exchange)
-            asyncio.run(analyzer.run_analysis())
-            analyzer.cleanup()
+            logger.info(f"Starting scheduled analysis for {self.exchange}")
+            analyzer = TickerAnalyzer(
+                exchange=self.exchange,
+                telegram_bot_service=self.telegram_bot_service
+            )
+            await analyzer.run_analysis()
+            logger.info(f"Completed scheduled analysis for {self.exchange}")
         except Exception as e:
             logger.error(f"Scheduled job failed: {e}", exc_info=True)
 
-    def start(self):
+    async def start(self):
         """Start the scheduler"""
-        logger.info(f"Starting scheduled analyzer for {self.exchange.value}")
-        logger.info(f"Schedule: Daily at {self.schedule_time}")
-
+        logger.info(f"Starting scheduled analyzer for {self.exchange}")
+        logger.info(f"Schedule: Every {self.interval_hours} hour(s)")
         # Schedule the job
-        schedule.every().day.at(self.schedule_time).do(self.run_scheduled_job)
-
+        schedule.every(self.interval_hours).hours.do(
+            lambda: asyncio.create_task(self.run_scheduled_job())
+        )
         # Run immediately on startup
         logger.info("Running initial analysis...")
-        self.run_scheduled_job()
-
+        await self.run_scheduled_job()
         # Main scheduler loop
         logger.info("Scheduler active. Press Ctrl+C to stop.")
         while self.running:
             schedule.run_pending()
-            time.sleep(60)  # Check every minute
-
+            await asyncio.sleep(60)  # Check every minute
         logger.info("Scheduler stopped gracefully")

src/core/ticker_scanner/tickers_provider.py

@@ -1,6 +1,7 @@
 import pandas as pd
 
 from src.core.ticker_scanner.core_enums import StockExchange
+from src.telegram_bot.logger import main_logger as logger
 
 
 class TickersProvider:
@@ -21,8 +22,12 @@ class TickersProvider:
         return tickers
 
     def get_tickers(self, exchange: StockExchange) -> list[str]:
+        logger.info(f"Fetching tickers for {exchange.value}")
         if exchange == exchange.NASDAQ:
-            return self.load_active_nasdaq_tickers()
+            tickers = self.load_active_nasdaq_tickers()
         elif exchange == exchange.NYSE:
-            return self.load_active_nyse_tickers()
-        return []
+            tickers = self.load_active_nyse_tickers()
+        else:
+            tickers = []
+        logger.info(f"Found {len(tickers)} tickers for {exchange.value}")
+        return tickers

src/core/ticker_scanner/trend_analyzer.py

@@ -0,0 +1,48 @@
+"""
+DEPRECATED: This file has been refactored into modular components.
+
+Please use the new modular system instead:
+- TickerAnalyzer: Main orchestrator (ticker_analyzer.py)
+- GrowthSpeedAnalyzer: Growth metrics analysis (growth_speed_analyzer.py)
+- Scheduler: Scheduling system (scheduler.py)
+- parallel_data_downloader: Data downloading (parallel_data_downloader.py)
+
+Example usage:
+    from src.core.ticker_scanner import TickerAnalyzer, Scheduler
+
+    # One-time analysis
+    analyzer = TickerAnalyzer(exchange="NASDAQ", limit=200)
+    await analyzer.run_analysis()
+
+    # Scheduled analysis
+    scheduler = Scheduler(exchange="NASDAQ", interval_hours=1)
+    await scheduler.start()
+"""
+
+import logging
+
+logger = logging.getLogger(__name__)
+logger.warning(
+    "trend_analyzer.py is deprecated. "
+    "Use TickerAnalyzer from ticker_analyzer.py instead."
+)
+
+# Legacy code archived below for reference
+"""
+Original code has been refactored into:
+
+1. ticker_analyzer.py - Main orchestrator (TickerAnalyzer class)
+2. growth_speed_analyzer.py - Growth analysis logic
+3. scheduler.py - Scheduling system
+4. parallel_data_downloader.py - Data downloading
+5. core_enums.py - Enums and constants
+6. growth_metrics.py - Data classes for metrics
+
+The new system provides:
+- Better separation of concerns
+- Modular and testable components
+- Proper logging integration
+- Type hints and documentation
+- Telegram integration support
+- TradingView link generation
+"""

src/core/ticker_scanner/trend_metrics.py

@@ -0,0 +1,29 @@
+from dataclasses import dataclass
+from datetime import datetime
+
+from src.core.ticker_scanner.growth_metrics import GrowthSpeedMetrics
+
+
+@dataclass
+class TrendMetrics:
+    """Comprehensive trend analysis"""
+    ticker: str
+    # Growth speed
+    growth_speed: GrowthSpeedMetrics
+    # Trend characteristics
+    positive_years_pct: float
+    consecutive_positive_years: int
+    max_drawdown: float
+    recovery_speed: float  # Days to recover from max drawdown
+    # Consistency metrics
+    volatility: float
+    beta: float | None  # vs market if available
+    trend_strength: float  # Custom composite 0-1
+    # Data quality
+    data_points: int
+    years_of_data: float
+    first_date: datetime
+    last_date: datetime
+    # Rankings
+    rank_by_velocity: int | None = None
+    rank_by_consistency: int | None = None

src/telegram_bot/telegram_bot_service.py

@@ -24,6 +24,7 @@ from src.services.async_trading_grid_calculator import generate_grid_message
 from src.core.fundamental_analysis.async_fundamental_analyzer import AsyncFundamentalAnalyzer
 from src.api.insiders.insider_trading_aggregator import InsiderTradingAggregator
 from src.telegram_bot.logger import main_logger as logger
+from src.core.ticker_scanner import TickerAnalyzer
 
 
 class TelegramBotService:
@@ -186,6 +187,7 @@ class TelegramBotService:
             response += "/fa TICKER - Fundamental analysis report (e.g., /fa AAPL)\n"
             response += "/insiders - Provides key insider's trades\n"
             response += "/insiders NVDA 30 - Insider's trades for the last 30 days\n"
+            response += "/scan EXCHANGE - Scan for top 20 growing tickers (e.g., /scan NASDAQ)\n"
 
         elif base_command == "/status":
             response = "✅ <b>Bot Status: Online</b>\n\n"
@@ -234,6 +236,11 @@ class TelegramBotService:
                                                text=None, user_name=user_name)
             return
 
+        elif base_command == "/scan":
+            await self.handle_scan_command(chat_id=chat_id, command_parts=command_parts,
+                                           text=None, user_name=user_name)
+            return
+
         else:
             response = f"❓ Unknown command: {command}\n\n"
             response += "Type /help to see available commands."
@@ -652,3 +659,47 @@ class TelegramBotService:
         else:
             logger.info(f"No recent trades found for {ticker}")
             await self.send_message_via_proxy(chat_id,f"No recent trades found for {ticker}")
+
+    async def handle_scan_command(
+            self, chat_id: int, command_parts: list[str], text: str | None, user_name: str
+    ) -> None:
+        """Ticker scanner command handler"""
+        # Default to NASDAQ
+        exchange = "NASDAQ"
+        if len(command_parts) >= 2:
+            exchange = command_parts[1].upper()
+        # Validate exchange
+        valid_exchanges = ["NASDAQ", "NYSE"]
+        if exchange not in valid_exchanges:
+            await self.send_message_via_proxy(
+                chat_id,
+                f"❌ Invalid exchange: {exchange}\n\n"
+                f"Supported exchanges: {', '.join(valid_exchanges)}\n\n"
+                f"Examples:\n• /scan NASDAQ\n• /scan NYSE"
+            )
+            return
+        # Send loading message
+        loading_msg = f"🔍 <b>Scanning {exchange} for top growing tickers...</b>\n\n"
+        loading_msg += "⏳ This may take a few minutes:\n"
+        loading_msg += "📥 Downloading historical data...\n"
+        loading_msg += "📊 Analyzing growth metrics...\n"
+        loading_msg += "🏆 Ranking tickers..."
+        await self.send_message_via_proxy(chat_id, loading_msg)
+        try:
+            # Create analyzer and run analysis
+            analyzer = TickerAnalyzer(
+                exchange=exchange,
+                telegram_bot_service=self,
+                limit=200  # Limit to 200 tickers for reasonable execution time
+            )
+            logger.info(f"Starting ticker scan for {exchange}")
+            top_tickers = await analyzer.run_analysis()
+            # Format and send results
+            message = analyzer._format_telegram_message(top_tickers)
+            await self.send_message_via_proxy(chat_id, message)
+            logger.info(f"Ticker scan completed for {exchange}")
+        except Exception as e:
+            logger.error(f"Error in ticker scanner: {e}", exc_info=True)
+            error_msg = f"❌ An error occurred during ticker scanning:\n\n{str(e)}\n\n"
+            error_msg += "Please try again later."
+            await self.send_message_via_proxy(chat_id, error_msg)