feat: Implement candlestick pattern scanner with TA-Lib patterns

src/pages/screener/technical_scanner_page.py

@@ -12,7 +12,7 @@ def technical_scanner_page():
     with scanner_tab:
         scanner_type = st.selectbox(
             "Select Scanner",
-            ["SunnyBands", "ATR-EMA", "ATR-EMA v2", "Heikin-Ashi"],
+            ["SunnyBands", "ATR-EMA", "ATR-EMA v2", "Heikin-Ashi", "Candlestick"],
             key="tech_scanner_type"
         )
 

src/screener/t_candlestick.py

@@ -0,0 +1,129 @@
+import pandas as pd
+from datetime import datetime, timedelta
+import talib
+from db.db_connection import create_client
+from utils.data_utils import (
+    get_stock_data, validate_signal_date, print_signal, 
+    save_signals_to_csv, get_qualified_stocks
+)
+from utils.scanner_utils import initialize_scanner, process_signal_data
+from trading.position_calculator import PositionCalculator
+
+# Dictionary mapping pattern names to their functions and descriptions
+CANDLESTICK_PATTERNS = {
+    'BULLISH_ENGULFING': {
+        'function': talib.CDLENGULFING,
+        'description': 'Bullish Engulfing Pattern'
+    },
+    'HAMMER': {
+        'function': talib.CDLHAMMER,
+        'description': 'Hammer Pattern'
+    },
+    'MORNING_STAR': {
+        'function': talib.CDLMORNINGSTAR,
+        'description': 'Morning Star Pattern'
+    },
+    'PIERCING_LINE': {
+        'function': talib.CDLPIERCING,
+        'description': 'Piercing Line Pattern'
+    },
+    'THREE_WHITE_SOLDIERS': {
+        'function': talib.CDL3WHITESOLDIERS,
+        'description': 'Three White Soldiers Pattern'
+    }
+}
+
+def check_entry_signal(df: pd.DataFrame) -> list:
+    """
+    Check for bullish candlestick patterns
+    
+    Args:
+        df (pd.DataFrame): DataFrame with OHLCV data
+        
+    Returns:
+        list: List of tuples (signal, date, signal_data) for each signal found
+    """
+    if len(df) < 5:  # Need at least 5 bars for pattern recognition
+        return []
+    
+    signals = []
+    
+    # Calculate all patterns
+    pattern_signals = {}
+    for pattern_name, pattern_info in CANDLESTICK_PATTERNS.items():
+        pattern_signals[pattern_name] = pattern_info['function'](
+            df['open'].values, 
+            df['high'].values,
+            df['low'].values,
+            df['close'].values
+        )
+    
+    # Look for signals in the last candle
+    i = len(df) - 1
+    found_patterns = []
+    
+    for pattern_name, pattern_values in pattern_signals.items():
+        # Check if we have a bullish signal (value > 0)
+        if pattern_values[i] > 0:
+            found_patterns.append(CANDLESTICK_PATTERNS[pattern_name]['description'])
+    
+    if found_patterns:
+        signal_data = {
+            'price': df.iloc[i]['close'],
+            'patterns': ', '.join(found_patterns),
+            'pattern_count': len(found_patterns)
+        }
+        signals.append((True, df.iloc[i]['date'], signal_data))
+    
+    return signals
+
+def run_candlestick_scanner(min_price: float, max_price: float, min_volume: int, 
+                           portfolio_size: float = None, interval: str = "1d",
+                           start_date: datetime = None, end_date: datetime = None) -> None:
+    """
+    Run candlestick pattern scanner to find bullish patterns
+    """
+    try:
+        # Initialize scanner components
+        interval, start_date, end_date, qualified_stocks, calculator = initialize_scanner(
+            min_price=min_price,
+            max_price=max_price,
+            min_volume=min_volume,
+            portfolio_size=portfolio_size,
+            interval=interval,
+            start_date=start_date,
+            end_date=end_date
+        )
+        
+        if not qualified_stocks:
+            return
+            
+        bullish_signals = []
+        
+        for ticker, current_price, current_volume, last_update, stock_type in qualified_stocks:
+            try:
+                df = get_stock_data(ticker, start_date, end_date, interval)
+                
+                if df.empty or len(df) < 5:  # Need at least 5 bars
+                    continue
+                    
+                signals = check_entry_signal(df)
+                for signal, signal_date, signal_data in signals:
+                    signal_data['date'] = signal_date
+                    entry_data = process_signal_data(
+                        ticker, signal_data, current_volume, 
+                        last_update, stock_type, calculator
+                    )
+                    bullish_signals.append(entry_data)
+                    print_signal(entry_data, "🕯️")  # Candlestick emoji
+                    
+            except Exception as e:
+                print(f"Error processing {ticker}: {str(e)}")
+                continue
+        
+        save_signals_to_csv(bullish_signals, 'candlestick')
+        return bullish_signals
+            
+    except Exception as e:
+        print(f"Error during scan: {str(e)}")
+        return []