stock_system/src/screener/t_candlestick.py

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, selected_patterns: list = None) -> list:
    """
    Check for bullish candlestick patterns across the entire date range
    
    Args:
        df (pd.DataFrame): DataFrame with OHLCV data
        selected_patterns (list): List of patterns to scan for
        
    Returns:
        list: List of tuples (signal, date, signal_data) for each signal found
    """
    if len(df) < 14:  # Need at least 14 bars for ATR calculation
        return []
    
    signals = []
    
    # Calculate ATR first
    atr = talib.ATR(df['high'].values, df['low'].values, df['close'].values, timeperiod=14)
    
    # Use selected patterns or all patterns if none selected
    patterns_to_scan = {k: v for k, v in CANDLESTICK_PATTERNS.items() 
                       if selected_patterns is None or k in selected_patterns}
    
    # Calculate patterns
    pattern_signals = {}
    for pattern_name, pattern_info in patterns_to_scan.items():
        pattern_signals[pattern_name] = pattern_info['function'](
            df['open'].values, 
            df['high'].values,
            df['low'].values,
            df['close'].values
        )
    
    # Look for signals across all candles
    for i in range(14, len(df)):  # Start after ATR lookback period
        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:
            # Calculate basic target and stop levels using recent price action
            current_price = df.iloc[i]['close']
            recent_low = df.iloc[max(0, i-5):i+1]['low'].min()  # Look back 5 bars for stop
            
            # Default to percentage-based calculations
            atr_value = atr[i]
            if pd.isna(atr_value):
                target_price = current_price * 1.02  # 2% target
                stop_loss = current_price * 0.99    # 1% stop
            else:
                target_price = current_price + (2 * atr_value)
                stop_loss = max(recent_low, current_price - atr_value)
            
            signal_data = {
                'price': current_price,
                'patterns': ', '.join(found_patterns),
                'pattern_count': len(found_patterns),
                'target_price': target_price,
                'stop_loss': stop_loss
            }
            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,
                           selected_patterns: list = None) -> None:
    """
    Run candlestick pattern scanner to find bullish patterns
    
    Args:
        min_price (float): Minimum stock price
        max_price (float): Maximum stock price
        min_volume (int): Minimum volume
        portfolio_size (float, optional): Portfolio size for position sizing
        interval (str, optional): Time interval for data. Defaults to "1d"
        start_date (datetime, optional): Start date for scanning
        end_date (datetime, optional): End date for scanning
        selected_patterns (list, optional): List of patterns to scan for
    """
    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, selected_patterns)  # Pass selected_patterns here
                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 []