feat: Add support for ATR EMA scanner with enhanced position calculator integration

2025-02-07 21:19:57 -08:00 · 2025-02-07 21:19:57 -08:00 · d6f2e1b961
commit d6f2e1b961
parent 0a4d5c2a5d
1 changed files with 134 additions and 0 deletions
--- a/src/screener/t_atr_ema.py
+++ b/src/screener/t_atr_ema.py
@ -0,0 +1,134 @@
+import os
+from datetime import datetime, timedelta
+import pandas as pd
+from db.db_connection import create_client
+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]
+    last_bands = results.iloc[-1]  # You need to calculate results first
+    
+    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) -> bool:
+    """Check if price is below EMA and moving up through lower ATR band"""
+    # Get latest values from DataFrame
+    last_price = df.iloc[-1]
+    
+    # Check if price is below EMA and has started moving up
+    return (
+        last_price['close'] < ema &
+        last_price['close'].shift(1) <= lower_band &
+        last_price['close'] > last_price['close'].shift(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()
+    
+    end_date = datetime.now()
+    start_date = end_date - timedelta(days=1)  # Get last trading day
+    
+    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}
+                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)
+            
+        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
+                if results['bullish_signal'].iloc[-1]:
+                    target_price = results['upper_band'].iloc[-1]
+                    
+                    if calculator:
+                        position = calculator.calculate_position_size(current_price, target_price)
+                        if position['shares'] > 0:
+                            signal_data = {
+                                'ticker': ticker,
+                                'entry': current_price,
+                                'target': target_price,
+                                'volume': current_volume,
+                                'last_update': datetime.fromtimestamp(last_update/1000000000),
+                                'shares': position['shares'],
+                                'position_size': position['position_value'],
+                                'stop_loss': position['stop_loss'],
+                                'risk': position['potential_loss'],
+                                'reward': position['potential_profit'],
+                                'r_r': position['risk_reward_ratio']
+                            }
+                            bullish_signals.append(signal_data)
+                            print(f"\n🟢 {ticker} Entry: ${current_price:.2f} Target: ${target_price:.2f}")
+                            print(f"   Potential Profit: ${signal_data['reward']:.2f} | Potential Loss: ${abs(signal_data['risk']):.2f}")
+                
+            except Exception as e:
+                print(f"Error processing {ticker}: {str(e)}")
+                continue
+        
+        # Save results
+        output_dir = 'reports'
+        os.makedirs(output_dir, exist_ok=True)
+        output_date = datetime.now().strftime("%Y%m%d_%H%M")
+        
+        if bullish_signals:
+            df_bullish = pd.DataFrame(bullish_signals)
+            output_file = f'{output_dir}/atr_ema_bullish_{output_date}.csv'
+            df_bullish.to_csv(output_file, index=False)
+            print(f"\nSaved bullish signals to {output_file}")
+        else:
+            print("No bullish signals found")
+            
+    except Exception as e:
+        print(f"Error during scan: {str(e)}")