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'
    },
    'MORNING_DOJI_STAR': {
        'function': talib.CDLMORNINGDOJISTAR,
        'description': 'Morning Doji Star Pattern'
    },
    'DRAGONFLY_DOJI': {
        'function': talib.CDLDRAGONFLYDOJI,
        'description': 'Dragonfly Doji Pattern'
    },
    'HARAMI': {
        'function': talib.CDLHARAMI,
        'description': 'Bullish Harami Pattern'
    },
    'INVERTED_HAMMER': {
        'function': talib.CDLINVERTEDHAMMER,
        'description': 'Inverted Hammer Pattern'
    },
    'THREE_INSIDE_UP': {
        'function': talib.CDL3INSIDE,
        'description': 'Three Inside Up Pattern'
    },
    'THREE_OUTSIDE_UP': {
        'function': talib.CDL3OUTSIDE,
        'description': 'Three Outside Up Pattern'
    },
    'BELT_HOLD': {
        'function': talib.CDLBELTHOLD,
        'description': 'Bullish Belt Hold Pattern'
    },
    'LADDER_BOTTOM': {
        'function': talib.CDLLADDERBOTTOM,
        'description': 'Ladder Bottom Pattern'
    },
    'MATCHING_LOW': {
        'function': talib.CDLMATCHINGLOW,
        'description': 'Matching Low Pattern'
    },
    'UNIQUE_THREE_RIVER': {
        'function': talib.CDLUNIQUE3RIVER,
        'description': 'Unique Three River Bottom 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 minimum bars for pattern recognition
        return []
    
    signals = []
    
    # 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 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:
            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,
                           selected_patterns: list = 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) < 14:  # Need minimum bars
                    continue
                    
                signals = check_entry_signal(df, selected_patterns)
                for signal, signal_date, signal_data in signals:
                    entry_data = {
                        'ticker': ticker,
                        'signal_date': signal_date,
                        'entry_price': signal_data['price'],
                        'patterns': signal_data['patterns'],
                        'pattern_count': signal_data['pattern_count'],
                        'volume': current_volume,
                        'last_update': last_update,
                        'stock_type': stock_type
                    }
                    bullish_signals.append(entry_data)
                    
                    # Custom print for candlestick signals
                    print(f"🕯️ {ticker}: {entry_data['patterns']} on {signal_date.strftime('%Y-%m-%d')} "
                          f"at ${signal_data['price']:.2f}")
                    
            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 []