add core moduler of the ticker scanner

pyproject.toml

@@ -69,6 +69,7 @@ dependencies = [
     "hmmlearn>=0.3.3",
     "xgboost>=3.1.1",
     "optuna>=4.5.0",
+    "schedule>=1.2.2",
 ]
 
 [build-system]

src/core/ticker_scanner/core_enums.py

@@ -0,0 +1,24 @@
+from enum import Enum
+
+
+class StockExchange(Enum):
+    NYSE = "NYSE"  # New York Stock Exchange
+    NASDAQ = "NASDAQ"  # NASDAQ
+    LSE = "LSE"  # London Stock Exchange
+    TSE = "TSE"  # Tokyo Stock Exchange
+    SSE = "SSE"  # Shanghai Stock Exchange
+    HKEX = "HKEX"  # Hong Kong Stock Exchange
+    BSE = "BSE"  # Bombay Stock Exchange
+    NSE = "NSE"  # National Stock Exchange of India
+    ASX = "ASX"  # Australian Securities Exchange
+    TSX = "TSX"  # Toronto Stock Exchange
+    SIX = "SIX"  # Swiss Exchange
+    FWB = "FWB"  # Frankfurt Stock Exchange
+
+
+class GrowthCategory(Enum):
+    """Growth speed classification"""
+    EXPLOSIVE = "explosive"  # Top 10%
+    STRONG = "strong"        # Top 25%
+    MODERATE = "moderate"    # Top 50%
+    SLOW = "slow"            # Bottom 50%

src/core/ticker_scanner/parallel_data_downloader.py

@@ -0,0 +1,110 @@
+"""
+parallel_yf_downloader.py
+Parallel downloading of ticker historical prices using multiprocessing,
+with retry and rate-limit handling and batching.
+"""
+
+import time
+import random
+from itertools import islice
+from typing import Any
+from concurrent.futures import ProcessPoolExecutor, as_completed
+
+import yfinance as yf
+
+from src.core.ticker_scanner.core_enums import StockExchange
+from src.core.ticker_scanner.tickers_provider import TickersProvider
+
+
+MAX_WORKERS = 8                 # Number of parallel processes
+MAX_RETRIES = 3                 # Retry count on failure
+SLEEP_BETWEEN_RETRIES = 1.0     # Seconds between retries
+BATCH_SIZE = 50                 # Number of tickers per batch
+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.
+    """
+    for attempt in range(max_retries):
+        try:
+            df = yf.download(ticker, period="max", progress=False, auto_adjust=True)
+            closes = df["Close"].dropna().values
+            if len(closes) < MIN_DATA_POINTS:
+                return None
+            return {"ticker": ticker, "prices": closes}
+        except yf.shared.YFRateLimitError:
+            wait = SLEEP_BETWEEN_RETRIES + random.random()
+            print(f"⚠️ Rate limited for {ticker}. Waiting {wait:.1f}s and retrying...")
+            time.sleep(wait)
+        except Exception:
+            return None
+    return None
+
+def batch(iterable: list[str], n: int = BATCH_SIZE):
+    """
+    Yield successive n-sized batches from iterable.
+    """
+    it = iter(iterable)
+    while True:
+        chunk = list(islice(it, n))
+        if not chunk:
+            break
+        yield chunk
+
+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.
+    """
+    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")
+        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)}")
+    if all_failed:
+        print(f"❌ Total failed: {len(all_failed)}")
+        print("Failed tickers:", all_failed)
+    return all_results
+
+def process_batch(ticker_batch: list[str], max_workers: int) -> tuple[list[dict[str, Any]], list[Any]]:
+    """
+    Process a batch of tickers in parallel using multiprocessing.
+    Returns tuple (successful_results, failed_tickers)
+    """
+    results = []
+    failed = []
+    with ProcessPoolExecutor(max_workers=max_workers) as executor:
+        futures = {executor.submit(fetch_prices, t): t for t in ticker_batch}
+        for future in as_completed(futures):
+            ticker = futures[future]
+            try:
+                res = future.result()
+                if res:
+                    results.append(res)
+                else:
+                    failed.append(ticker)
+            except Exception:
+                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...")
+    data = download_tickers_parallel(tickers)
+    for d in data:
+        print(f"{d['ticker']}: {len(d['prices'])} price points")
+
+
+if __name__ == "__main__":
+    run_parallel_data_downloader()

src/core/ticker_scanner/scheduler.py

@@ -0,0 +1,52 @@
+import time
+import signal
+
+import schedule
+
+from src.telegram_bot.logger import main_logger as logger
+
+
+class Scheduler:
+    """Schedule and manage periodic analysis"""
+    def __init__(self, exchange: Exchange = Exchange.SP500,
+                 schedule_time: str = "18:00"):
+        self.exchange = exchange
+        self.schedule_time = schedule_time
+        self.running = True
+        # Setup signal handlers for graceful shutdown
+        signal.signal(signal.SIGINT, self._signal_handler)
+        signal.signal(signal.SIGTERM, self._signal_handler)
+
+    def _signal_handler(self, signum, frame):
+        """Handle shutdown signals"""
+        logger.info("Received shutdown signal. Cleaning up...")
+        self.running = False
+
+    def run_scheduled_job(self):
+        """Execute the analysis job"""
+        try:
+            analyzer = AsyncTrendAnalyzer(self.exchange)
+            asyncio.run(analyzer.run_analysis())
+            analyzer.cleanup()
+        except Exception as e:
+            logger.error(f"Scheduled job failed: {e}", exc_info=True)
+
+    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}")
+
+        # Schedule the job
+        schedule.every().day.at(self.schedule_time).do(self.run_scheduled_job)
+
+        # Run immediately on startup
+        logger.info("Running initial analysis...")
+        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
+
+        logger.info("Scheduler stopped gracefully")

src/core/ticker_scanner/tickers_provider.py

@@ -0,0 +1,28 @@
+import pandas as pd
+
+from src.core.ticker_scanner.core_enums import StockExchange
+
+
+class TickersProvider:
+    def load_active_nasdaq_tickers(self) -> list[str]:
+        url = "ftp://ftp.nasdaqtrader.com/SymbolDirectory/nasdaqtraded.txt"
+        df = pd.read_csv(url, sep="|")
+        # Keep only active tickers (Test Issue == 'N')
+        df_active = df[df["Test Issue"] == "N"]
+        tickers = df_active["NASDAQ Symbol"].tolist()
+        return tickers
+
+    def load_active_nyse_tickers(self) -> list[str]:
+        url = "https://eodhistoricaldata.com/api/exchange-symbol-list/NYSE.csv"
+        df = pd.read_csv(url)
+        # Keep only active, common stocks
+        df_active = df[(df['Type'] == 'Common Stock') & (df['Delisted'] != 1)]
+        tickers = df_active['Code'].tolist()
+        return tickers
+
+    def get_tickers(self, exchange: StockExchange) -> list[str]:
+        if exchange == exchange.NASDAQ:
+            return self.load_active_nasdaq_tickers()
+        elif exchange == exchange.NYSE:
+            return self.load_active_nyse_tickers()
+        return []