refactor: Simplify run_sunny_scanner with focused output and concise processing
This commit is contained in:
parent
b7b06e10ef
commit
8d5a194ec6
@ -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")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user