From 8d5a194ec62296712a393462d1deb6349d9bec80 Mon Sep 17 00:00:00 2001 From: "Bobby Abellana (aider)" Date: Thu, 6 Feb 2025 23:23:45 -0800 Subject: [PATCH] refactor: Simplify run_sunny_scanner with focused output and concise processing --- src/screener/t_sunnyband.py | 224 ++++++------------------------------ 1 file changed, 38 insertions(+), 186 deletions(-) diff --git a/src/screener/t_sunnyband.py b/src/screener/t_sunnyband.py index ffb7055..df41769 100644 --- a/src/screener/t_sunnyband.py +++ b/src/screener/t_sunnyband.py @@ -272,234 +272,86 @@ def view_stock_details(ticker: str, interval: str, start_date: datetime, end_dat print(f"Error analyzing {ticker}: {str(e)}") def run_sunny_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None: - """Run the SunnyBand scanner and save results""" - print(f"\nInitializing scan for stocks between ${min_price:.2f} and ${max_price:.2f}") - print(f"Minimum volume: {min_volume:,}") + print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}") - # Get user's preferred interval interval = get_interval_choice() - - # Set date range to look back from current time end_date = datetime.now() - start_date = end_date - timedelta(days=1) # Look at last trading day for signals - lookback_start = end_date - timedelta(days=60) # For DMA calculation + start_date = end_date - timedelta(days=1) + lookback_start = end_date - timedelta(days=60) - print(f"\nAnalyzing data from {lookback_start.date()} to {end_date.date()}") - print(f"Looking for signals in the last trading day") - - # Get valid tickers - print("\nFetching qualified stocks...") tickers = get_valid_tickers(min_price, max_price, min_volume, interval) - if not tickers: - print("No stocks found matching your criteria.") + print("No stocks found matching criteria.") return - print(f"\nFound {len(tickers)} stocks to scan") - print("Looking for SunnyBand crossovers...") - print("This may take a few minutes...") + print(f"\nScanning {len(tickers)} qualified stocks...") - # Initialize results lists - bullish_signals = [] - bearish_signals = [] - errors = [] - - # Initialize SunnyBands indicator and position calculator sunny = SunnyBands() calculator = None if portfolio_size and portfolio_size > 0: calculator = PositionCalculator(account_size=portfolio_size) - print(f"\nInitialized position calculator with portfolio size: ${portfolio_size:,.2f}") - # Track progress - total = len(tickers) - processed = 0 + bullish_signals = [] + bearish_signals = [] - # Scan each ticker for ticker in tickers: - processed += 1 - if processed % 10 == 0: # Show progress every 10 stocks - print(f"Progress: {processed}/{total} stocks processed ({(processed/total)*100:.1f}%)") - try: - # Get price data df = get_stock_data(ticker, start_date, end_date, interval) - - if df.empty: + if df.empty or len(df) < 50: continue - if len(df) < 50: # Need enough data for the indicator - continue - - # Calculate SunnyBands results = sunny.calculate(df) - - # Debug data alignment - print(f"\nDebug: Data for {ticker}") - print("Last 3 candles:") - for i in [-3, -2, -1]: - candle = df.iloc[i] - bands = results.iloc[i] - print(f"\nDate/Time: {candle['date']}") - print(f"OHLC: ${candle['open']:.2f}, ${candle['high']:.2f}, ${candle['low']:.2f}, ${candle['close']:.2f}") - print(f"DMA: ${bands['dma']:.2f}") - print(f"Bands: Lower=${bands['lower_band']:.2f}, Upper=${bands['upper_band']:.2f}") - print(f"Signals: Bullish={bands['bullish_signal']}, Bearish={bands['bearish_signal']}") - - # Check last day's signals last_day = df.iloc[-1] if results['bullish_signal'].iloc[-1]: - print("\nDebug: Processing bullish signal") # Debug line + entry_price = last_day['close'] + dma = results['dma'].iloc[-1] + upper_band = results['upper_band'].iloc[-1] + band_range = upper_band - dma + target_price = upper_band + band_range + signal_data = { 'ticker': ticker, - 'price': last_day['close'], - 'volume': last_day['volume'], - 'date': last_day['date'], - 'dma': results['dma'].iloc[-1], - 'lower_band': results['lower_band'].iloc[-1], - 'upper_band': results['upper_band'].iloc[-1] + 'entry': entry_price, + 'target': target_price } - # Add position sizing if calculator exists if calculator: - print(f"Debug: Calculator exists, calculating position for price: ${last_day['close']:.2f}") # Debug line - try: - entry_price = last_day['close'] - upper_band = results['upper_band'].iloc[-1] - dma = results['dma'].iloc[-1] - - # Calculate band distances - band_range = upper_band - dma # Distance from middle to upper band - - # Set target as one full band range above the upper band - target_price = upper_band + band_range - - print(f"Debug: Entry: ${entry_price:.2f}, DMA: ${dma:.2f}, Upper Band: ${upper_band:.2f}, Target: ${target_price:.2f}") - position = calculator.calculate_position_size( - entry_price=entry_price, - target_price=target_price - ) - # Format debug position output with rounded values - debug_position = {k: round(float(v), 2) if isinstance(v, (float, np.float64)) else v - for k, v in position.items()} - print(f"Debug: Position calculation result: {debug_position}") # Debug line - signal_data.update({ - 'shares': position['shares'], - 'position_value': position['position_value'], - 'stop_loss': position['stop_loss'], - 'potential_profit': position['potential_profit'], - 'potential_loss': position['potential_loss'], - 'risk_reward_ratio': position['risk_reward_ratio'] - }) - except ValueError as e: - print(f"Position sizing error for {ticker}: {str(e)}") + position = calculator.calculate_position_size(entry_price, target_price) + signal_data.update({ + '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"šŸŸ¢ Bullish Signal: {ticker} at ${last_day['close']:.2f}") + print(f"\nšŸŸ¢ {ticker} Entry: ${entry_price:.2f} Target: ${target_price:.2f}") + if calculator: + print(f" Shares: {signal_data['shares']} | Risk: ${abs(signal_data['risk']):.2f} | " + f"Reward: ${signal_data['reward']:.2f} | R/R: {signal_data['r_r']:.2f}") elif results['bearish_signal'].iloc[-1]: - signal_data = { + bearish_signals.append({ 'ticker': ticker, - 'price': last_day['close'], - 'volume': last_day['volume'], - 'date': last_day['date'], - 'dma': results['dma'].iloc[-1], - 'upper_band': results['upper_band'].iloc[-1] - } - bearish_signals.append(signal_data) - print(f"šŸ”“ Bearish Signal: {ticker} at ${last_day['close']:.2f}") + 'price': last_day['close'] + }) + print(f"\nšŸ”“ {ticker} at ${last_day['close']:.2f}") except Exception as e: - errors.append(f"{ticker}: {str(e)}") continue - # Save and display results + # Save results more concisely output_date = datetime.now().strftime("%Y%m%d") - - print(f"\nScan Complete! Processed {total} stocks.") - - if errors: - print(f"\nEncountered {len(errors)} errors during scan:") - for error in errors[:5]: # Show first 5 errors - print(error) - if len(errors) > 5: - print(f"...and {len(errors) - 5} more errors") - if bullish_signals: - print(f"\nšŸŸ¢ Found {len(bullish_signals)} Bullish Signals:") df_bullish = pd.DataFrame(bullish_signals) - # Create reports directory if it doesn't exist - os.makedirs('reports', exist_ok=True) - - bullish_file = f'reports/sunny_bullish_{output_date}.csv' - df_bullish.to_csv(bullish_file, index=False) - print(f"Saved to {bullish_file}") - - for signal in bullish_signals: - print(f"\n{signal['ticker']}:") - print(f"Entry Price: ${signal['price']:.2f}") - print(f"Volume: {signal['volume']:,}") - print(f"Target (Upper Band): ${signal['upper_band']:.2f}") - - if 'shares' in signal: - # Convert numpy float64 to regular float and round to 2 decimal places - position_value = round(float(signal['position_value']), 2) - stop_loss = round(float(signal['stop_loss']), 2) - potential_loss = round(float(signal['potential_loss']), 2) - potential_profit = round(float(signal['potential_profit']), 2) - risk_reward = round(float(signal['risk_reward_ratio']), 2) - target_price = round(float(signal['upper_band']), 2) - - # Calculate percentage gains/losses and round to 1 decimal place - profit_percentage = round((potential_profit / position_value) * 100, 1) - loss_percentage = round((abs(potential_loss) / position_value) * 100, 1) - - print("\nPosition Details:") - print(f"Shares: {signal['shares']:,}") - print(f"Position Size: ${position_value:,.2f}") - print(f"Entry Price: ${signal['price']:.2f}") - print(f"Stop Loss: ${stop_loss:.2f} (-6%)") - print(f"Target Price: ${target_price:.2f}") - print(f"Risk Amount: ${abs(potential_loss):,.2f} ({loss_percentage:.1f}%)") - print(f"Potential Profit: ${potential_profit:,.2f} ({profit_percentage:.1f}%)") - print(f"Risk/Reward Ratio: {risk_reward:.2f}") + df_bullish.to_csv(f'reports/sunny_bullish_{output_date}.csv', index=False) if bearish_signals: - print(f"\nšŸ”“ Found {len(bearish_signals)} Bearish Signals:") df_bearish = pd.DataFrame(bearish_signals) - # Create reports directory if it doesn't exist - os.makedirs('reports', exist_ok=True) - - bearish_file = f'reports/sunny_bearish_{output_date}.csv' - df_bearish.to_csv(bearish_file, index=False) - print(f"Saved to {bearish_file}") - - for signal in bearish_signals: - print(f"\n{signal['ticker']}:") - print(f"Price: ${signal['price']:.2f}") - print(f"Volume: {signal['volume']:,}") - print(f"DMA: ${signal['dma']:.2f}") - print(f"Upper Band: ${signal['upper_band']:.2f}") + df_bearish.to_csv(f'reports/sunny_bearish_{output_date}.csv', index=False) - if not bullish_signals and not bearish_signals: - print("\nNo signals found for today.") - else: - while True: - view_choice = input("\nWould you like to view detailed data for a stock? (Enter ticker or 'n' to exit): ").upper() - if view_choice == 'N': - break - - # Check if ticker is in our signals - found = False - for signals in [bullish_signals, bearish_signals]: - for signal in signals: - if signal['ticker'] == view_choice: - found = True - view_stock_details(view_choice, interval, start_date, end_date) - break - if found: - break - - if not found: - print(f"Ticker {view_choice} not found in signals list.") + print(f"\nFound {len(bullish_signals)} bullish and {len(bearish_signals)} bearish signals") + print("Results saved to reports directory")