stock_system/src/screener/t_atr_ema.py

from screener.user_input import get_interval_choice, get_date_range
import os
from datetime import datetime, timedelta
import pandas as pd
from db.db_connection import create_client
from trading.position_calculator import PositionCalculator
from utils.data_utils import get_stock_data, validate_signal_date, print_signal, save_signals_to_csv
from indicators.three_atr_ema import ThreeATREMAIndicator

def check_atr_ema_bullish_signal(df: pd.DataFrame) -> bool:
    """Check for bullish signal based on ATR EMA indicator"""
    # Get latest values from DataFrame
    last_price = df.iloc[-1]
    previous_price = df.iloc[-2]  # Get the previous row for comparison
    indicator = ThreeATREMAIndicator()
    results = indicator.calculate(df)
    indicator = ThreeATREMAIndicator()
    results = indicator.calculate(df)
    last_bands = results.iloc[-1]
    
    print(f"\nSunnyBands Indicators:")
    print(f"DMA: ${last_bands['dma']:.2f}")
    print(f"Upper Band: ${last_bands['upper_band']:.2f}")
    print(f"Lower Band: ${last_bands['lower_band']:.2f}")
    print(f"Bullish Signal: {'Yes' if last_bands['signal'] else 'No'}")

def check_atr_ema_buy_condition(df: pd.DataFrame) -> tuple:
    """Check if price is below EMA and moving up through lower ATR band"""
    # Get latest values from DataFrame
    last_price = df.iloc[-1]
    previous_price = df.iloc[-2]  # Get the previous row for comparison
    results = ThreeATREMAIndicator().calculate(df)  # Ensure results are calculated here

    # Check if price is below EMA and has started moving up
    ema = results['ema'].iloc[-1]
    lower_band = results['lower_band'].iloc[-1]
    
    signal = (
        last_price['close'] < ema and
        previous_price['close'] <= lower_band and
        last_price['close'] > previous_price['close']
    )
    
    return signal, last_price['date'] if signal else None, results.iloc[-1]

def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None:
    print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}")
    
    # Get time interval
    interval = get_interval_choice()
    
    start_date, end_date = get_date_range()
    start_ts = int(start_date.timestamp() * 1000000000)
    end_ts = int(end_date.timestamp() * 1000000000)
    
    client = create_client()
    
    try:
        query = f"""
            WITH latest_data AS (
                SELECT 
                    ticker,
                    argMax(close, window_start) as last_close,
                    sum(volume) as total_volume,
                    max(window_start) as last_update
                FROM stock_db.stock_prices
                WHERE window_start BETWEEN {start_ts} AND {end_ts}
                AND toDateTime(window_start/1000000000) <= now()
                GROUP BY ticker
                HAVING last_close BETWEEN {min_price} AND {max_price}
                AND total_volume >= {min_volume}
            )
            SELECT 
                ticker,
                last_close,
                total_volume,
                last_update
            FROM latest_data
            ORDER BY ticker
        """
        
        result = client.query(query)
        qualified_stocks = [(row[0], row[1], row[2], row[3]) for row in result.result_rows]
        
        if not qualified_stocks:
            print("No stocks found matching criteria.")
            return

        print(f"\nFound {len(qualified_stocks)} stocks matching criteria")

        # Initialize indicators
        indicator = ThreeATREMAIndicator()
        calculator = None
        if portfolio_size and portfolio_size > 0:
            calculator = PositionCalculator(
                account_size=portfolio_size,
                risk_percentage=1.0,
                stop_loss_percentage=7.0  # Explicitly set 7% stop
            )
            
        bullish_signals = []
        
        for ticker, current_price, current_volume, last_update in qualified_stocks:
            try:
                # Get historical data based on interval
                df = get_stock_data(ticker, start_date, end_date, interval)
                
                if df.empty or len(df) < 50:  # Need at least 50 bars for the indicator
                    continue
                
                results = indicator.calculate(df)
                
                # Check for signals
                signal, signal_date, indicator_values = check_atr_ema_buy_condition(df)
                if signal:
                    target_price = indicator_values['upper_band']
                    
                    if calculator:
                        position = calculator.calculate_position_size(current_price, target_price)
                        if position['shares'] > 0:
                            signal_data = {
                                'ticker': ticker,
                                'entry_price': current_price,
                                'target_price': target_price,
                                'signal_date': signal_date,
                                'volume': current_volume,
                                'last_update': datetime.fromtimestamp(last_update/1000000000),
                                'shares': position['shares'],
                                'position_size': position['position_value'],
                                'stop_loss': position['stop_loss'],
                                'risk_amount': position['potential_loss'],
                                'profit_amount': position['potential_profit'],
                                'risk_reward_ratio': position['risk_reward_ratio']
                            }
                            bullish_signals.append(signal_data)
                            dollar_risk = signal_data['risk'] * -1
                            print(f"\n🟢 {ticker} @ ${current_price:.2f} on {signal_date.strftime('%Y-%m-%d %H:%M')}")
                            print(f"   Size: {signal_data['shares']} shares (${signal_data['position_size']:.2f})")
                            print(f"   Stop: ${signal_data['stop_loss']:.2f} (7%) | Target: ${target_price:.2f}")
                            print(f"   Risk/Reward: 1:{signal_data['r_r']:.1f} | Risk: ${dollar_risk:.2f}")
                            print(f"   Potential Profit: ${signal_data['reward']:.2f}")
                
            except Exception as e:
                print(f"Error processing {ticker}: {str(e)}")
                continue
        
        save_signals_to_csv(bullish_signals, 'atr_ema')
            
    except Exception as e:
        print(f"Error during scan: {str(e)}")