feat: Add SunnyBand scanner functionality to screener module

src/screener/t_sunnyband.py

@@ -0,0 +1,139 @@
+from datetime import datetime, timedelta
+import pandas as pd
+from db.db_connection import create_client
+from indicators.sunny_bands import SunnyBands
+
+def get_stock_data(ticker: str, start_date: datetime, end_date: datetime) -> pd.DataFrame:
+    """Fetch stock data from the database"""
+    client = create_client()
+    
+    query = f"""
+        SELECT 
+            date,
+            open,
+            high,
+            low,
+            close,
+            volume
+        FROM stock_db.stock_prices_daily
+        WHERE ticker = '{ticker}'
+        AND date BETWEEN '{start_date.date()}' AND '{end_date.date()}'
+        ORDER BY date ASC
+    """
+    
+    result = client.query(query)
+    
+    return pd.DataFrame(
+        result.result_rows,
+        columns=['date', 'open', 'high', 'low', 'close', 'volume']
+    )
+
+def get_valid_tickers(min_price: float, max_price: float, min_volume: int) -> list:
+    """Get tickers that meet the price and volume criteria"""
+    client = create_client()
+    yesterday = (datetime.now() - timedelta(days=1)).date()
+    
+    query = f"""
+        SELECT DISTINCT ticker
+        FROM stock_db.stock_prices_daily
+        WHERE date = '{yesterday}'
+        AND close BETWEEN {min_price} AND {max_price}
+        AND volume >= {min_volume}
+    """
+    
+    result = client.query(query)
+    return [row[0] for row in result.result_rows]
+
+def run_sunny_scanner(min_price: float, max_price: float, min_volume: int) -> None:
+    """Run the SunnyBand scanner and save results"""
+    # Get date range (60 days of data for calculations)
+    end_date = datetime.now()
+    start_date = end_date - timedelta(days=60)
+    
+    # Get valid tickers
+    tickers = get_valid_tickers(min_price, max_price, min_volume)
+    
+    if not tickers:
+        print("No stocks found matching your criteria.")
+        return
+    
+    print(f"\nScanning {len(tickers)} stocks...")
+    
+    # Initialize results lists
+    bullish_signals = []
+    bearish_signals = []
+    
+    # Initialize SunnyBands indicator
+    sunny = SunnyBands()
+    
+    # Scan each ticker
+    for ticker in tickers:
+        try:
+            # Get price data
+            df = get_stock_data(ticker, start_date, end_date)
+            
+            if len(df) < 50:  # Need enough data for the indicator
+                continue
+            
+            # Calculate SunnyBands
+            results = sunny.calculate(df)
+            
+            # Check last day's signals
+            last_day = df.iloc[-1]
+            
+            if results['bullish_signal'].iloc[-1]:
+                bullish_signals.append({
+                    'ticker': ticker,
+                    'price': last_day['close'],
+                    'volume': last_day['volume'],
+                    'date': last_day['date'],
+                    'dma': results['dma'].iloc[-1],
+                    'lower_band': results['lower_band'].iloc[-1]
+                })
+            
+            elif results['bearish_signal'].iloc[-1]:
+                bearish_signals.append({
+                    'ticker': ticker,
+                    'price': last_day['close'],
+                    'volume': last_day['volume'],
+                    'date': last_day['date'],
+                    'dma': results['dma'].iloc[-1],
+                    'upper_band': results['upper_band'].iloc[-1]
+                })
+                
+        except Exception as e:
+            print(f"Error processing {ticker}: {str(e)}")
+    
+    # Save and display results
+    output_date = datetime.now().strftime("%Y%m%d")
+    
+    if bullish_signals:
+        print("\n🟢 Bullish Signals:")
+        df_bullish = pd.DataFrame(bullish_signals)
+        bullish_file = f'src/reports/sunny_bullish_{output_date}.csv'
+        df_bullish.to_csv(bullish_file, index=False)
+        print(f"\nSaved {len(bullish_signals)} bullish signals to {bullish_file}")
+        
+        for signal in bullish_signals:
+            print(f"\n{signal['ticker']}:")
+            print(f"Price: ${signal['price']:.2f}")
+            print(f"Volume: {signal['volume']:,}")
+            print(f"DMA: ${signal['dma']:.2f}")
+            print(f"Lower Band: ${signal['lower_band']:.2f}")
+    
+    if bearish_signals:
+        print("\n🔴 Bearish Signals:")
+        df_bearish = pd.DataFrame(bearish_signals)
+        bearish_file = f'src/reports/sunny_bearish_{output_date}.csv'
+        df_bearish.to_csv(bearish_file, index=False)
+        print(f"\nSaved {len(bearish_signals)} bearish signals to {bearish_file}")
+        
+        for signal in bearish_signals:
+            print(f"\n{signal['ticker']}:")
+            print(f"Price: ${signal['price']:.2f}")
+            print(f"Volume: {signal['volume']:,}")
+            print(f"DMA: ${signal['dma']:.2f}")
+            print(f"Upper Band: ${signal['upper_band']:.2f}")
+    
+    if not bullish_signals and not bearish_signals:
+        print("\nNo signals found for today.")