diff --git a/src/pages/screener/technical_scanner_page.py b/src/pages/screener/technical_scanner_page.py index 0108dd0..5e81920 100644 --- a/src/pages/screener/technical_scanner_page.py +++ b/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" ) diff --git a/src/screener/t_candlestick.py b/src/screener/t_candlestick.py new file mode 100644 index 0000000..2abecf8 --- /dev/null +++ b/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 []