Bobby/stock_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Optimize sunny scanner with direct database query and simplified processing

Browse Source
This commit is contained in:
Bobby Abellana (aider) 2025-02-07 00:24:08 -08:00 committed by Bobby Abellana
parent e7a32dd9ab
commit 7faf56a712
No known key found for this signature in database
GPG Key ID: 647714CC45F3647B
1 changed files with 81 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

152
src/screener/t_sunnyband.py
View File

@ -277,81 +277,91 @@ def run_sunny_scanner(min_price: float, max_price: float, min_volume: int, portf
interval = get_interval_choice() interval = get_interval_choice()
end_date = datetime.now() end_date = datetime.now()
start_date = end_date - timedelta(days=1) start_date = end_date - timedelta(days=1)
lookback_start = end_date - timedelta(days=60)
tickers = get_valid_tickers(min_price, max_price, min_volume, interval) # First get the data from database
if not tickers: client = create_client()
print("No stocks found matching criteria.")
return
print(f"\nScanning {len(tickers)} qualified stocks...") # Query to get stocks meeting criteria
query = f"""
SELECT DISTINCT ticker, close
FROM stock_db.stock_prices_daily
WHERE date = (SELECT max(date) FROM stock_db.stock_prices_daily)
AND close BETWEEN {min_price} AND {max_price}
AND volume >= {min_volume}
ORDER BY ticker
"""
sunny = SunnyBands() try:
calculator = None result = client.query(query)
if portfolio_size and portfolio_size > 0: qualified_stocks = [(row[0], row[1]) for row in result.result_rows]
calculator = PositionCalculator(account_size=portfolio_size)
if not qualified_stocks:
bullish_signals = [] print("No stocks found matching criteria.")
bearish_signals = [] return
for ticker in tickers:
try:
df = get_stock_data(ticker, start_date, end_date, interval)
if df.empty or len(df) < 50:
continue
results = sunny.calculate(df)
last_day = df.iloc[-1]
if results['bullish_signal'].iloc[-1]: print(f"\nFound {len(qualified_stocks)} stocks matching criteria")
entry_price = last_day['close']
dma = results['dma'].iloc[-1] # Initialize indicators
upper_band = results['upper_band'].iloc[-1] sunny = SunnyBands()
band_range = upper_band - dma calculator = None
target_price = upper_band + band_range if portfolio_size and portfolio_size > 0:
calculator = PositionCalculator(account_size=portfolio_size)
bullish_signals = []
bearish_signals = []
# Process each qualified stock
for ticker, current_price in qualified_stocks:
try:
df = get_stock_data(ticker, start_date, end_date, interval)
if df.empty or len(df) < 50:
continue
results = sunny.calculate(df)
signal_data = { if results['bullish_signal'].iloc[-1]:
'ticker': ticker, target_price = results['upper_band'].iloc[-1]
'entry': entry_price,
'target': target_price 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,
'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" Shares: {signal_data['shares']} | Risk: ${abs(signal_data['risk']):.2f} | "
f"Reward: ${signal_data['reward']:.2f} | R/R: {signal_data['r_r']:.2f}")
if calculator: elif results['bearish_signal'].iloc[-1]:
position = calculator.calculate_position_size(entry_price, target_price) bearish_signals.append({
signal_data.update({ 'ticker': ticker,
'shares': position['shares'], 'price': current_price
'position_size': position['position_value'],
'stop_loss': position['stop_loss'],
'risk': position['potential_loss'],
'reward': position['potential_profit'],
'r_r': position['risk_reward_ratio']
}) })
print(f"\n🔴 {ticker} at ${current_price:.2f}")
bullish_signals.append(signal_data)
print(f"\n🟢 {ticker} Entry: ${entry_price:.2f} Target: ${target_price:.2f}") except Exception as e:
if calculator: continue
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}") # Save results
output_date = datetime.now().strftime("%Y%m%d")
elif results['bearish_signal'].iloc[-1]: if bullish_signals:
bearish_signals.append({ df_bullish = pd.DataFrame(bullish_signals)
'ticker': ticker, df_bullish.to_csv(f'reports/sunny_bullish_{output_date}.csv', index=False)
'price': last_day['close']
}) if bearish_signals:
print(f"\n🔴 {ticker} at ${last_day['close']:.2f}") df_bearish = pd.DataFrame(bearish_signals)
df_bearish.to_csv(f'reports/sunny_bearish_{output_date}.csv', index=False)
except Exception as e:
continue print(f"\nFound {len(bullish_signals)} bullish and {len(bearish_signals)} bearish signals")
# Save results more concisely except Exception as e:
output_date = datetime.now().strftime("%Y%m%d") print(f"Error during scan: {str(e)}")
if bullish_signals:
df_bullish = pd.DataFrame(bullish_signals)
df_bullish.to_csv(f'reports/sunny_bullish_{output_date}.csv', index=False)
if bearish_signals:
df_bearish = pd.DataFrame(bearish_signals)
df_bearish.to_csv(f'reports/sunny_bearish_{output_date}.csv', index=False)
print(f"\nFound {len(bullish_signals)} bullish and {len(bearish_signals)} bearish signals")
print("Results saved to reports directory")