diff --git a/src/screener/t_atr_ema.py b/src/screener/t_atr_ema.py index 7acbcf8..1fab2aa 100644 --- a/src/screener/t_atr_ema.py +++ b/src/screener/t_atr_ema.py @@ -4,7 +4,7 @@ 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 +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: @@ -144,18 +144,7 @@ def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, por 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") + save_signals_to_csv(bullish_signals, 'atr_ema') except Exception as e: print(f"Error during scan: {str(e)}") diff --git a/src/screener/t_atr_ema_v2.py b/src/screener/t_atr_ema_v2.py index f4489e1..a13e150 100644 --- a/src/screener/t_atr_ema_v2.py +++ b/src/screener/t_atr_ema_v2.py @@ -3,7 +3,7 @@ import pandas as pd import os from db.db_connection import create_client from trading.position_calculator import PositionCalculator -from utils.data_utils import get_stock_data, validate_signal_date +from utils.data_utils import get_stock_data, validate_signal_date, print_signal, save_signals_to_csv from screener.user_input import get_interval_choice, get_date_range from indicators.three_atr_ema import ThreeATREMAIndicator @@ -148,32 +148,13 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int, entry_signals.append(entry_data) # Print signal information with date - print(f"\nšŸ” {ticker} @ ${entry_data['price']:.2f} on {signal_date.strftime('%Y-%m-%d %H:%M')}") - if calculator: - print(f" Size: {entry_data['shares']} shares (${entry_data['position_size']:.2f})") - print(f" Stop: ${entry_data['stop_loss']:.2f} (7%)") - print(f" Risk: ${abs(entry_data['risk']):.2f}") - # Add potential profit calculation and display - target_price = entry_data['ema'] # Use stored EMA as target - potential_profit = (target_price - entry_data['price']) * entry_data['shares'] - print(f" Target: ${target_price:.2f} | Profit: ${potential_profit:.2f}") + print_signal(entry_data) except Exception as e: print(f"Error processing {ticker}: {str(e)}") continue - # Save results - if entry_signals: - output_dir = 'reports' - os.makedirs(output_dir, exist_ok=True) - output_date = datetime.now().strftime("%Y%m%d_%H%M") - output_file = f'{output_dir}/atr_ema_v2_{output_date}.csv' - - df_signals = pd.DataFrame(entry_signals) - df_signals.to_csv(output_file, index=False) - print(f"\nSaved {len(entry_signals)} signals to {output_file}") - else: - print("\nNo entry signals found") + save_signals_to_csv(entry_signals, 'atr_ema_v2') except Exception as e: print(f"Error during scan: {str(e)}") diff --git a/src/screener/t_sunnyband.py b/src/screener/t_sunnyband.py index 8b54ed3..db1cd84 100644 --- a/src/screener/t_sunnyband.py +++ b/src/screener/t_sunnyband.py @@ -6,7 +6,7 @@ from db.db_connection import create_client from indicators.sunny_bands import SunnyBands from trading.position_calculator import PositionCalculator from screener.user_input import get_interval_choice, get_date_range -from utils.data_utils import get_stock_data, validate_signal_date +from utils.data_utils import get_stock_data, validate_signal_date, print_signal, save_signals_to_csv @@ -271,11 +271,7 @@ def run_sunny_scanner(min_price: float, max_price: float, min_volume: int, portf dollar_risk = position['potential_loss'] * -1 signal_date = validate_signal_date(df.iloc[-1]['date']) # Get and validate the date signal_data['signal_date'] = signal_date # Add to signal data - print(f"\n🟢 {ticker} @ ${current_price:.2f} on {signal_date.strftime('%Y-%m-%d %H:%M')}") - print(f" Size: {position['shares']} shares (${position['position_value']:.2f})") - print(f" Stop: ${signal_data['stop_loss']:.2f} (-7%) | Target: ${target_price:.2f}") - print(f" Risk/Reward: 1:{position['risk_reward_ratio']:.1f} | Risk: ${dollar_risk:.2f}") - print(f" Potential Profit: ${position['potential_profit']:.2f}") + print_signal(signal_data, "🟢") except ValueError as e: print(f"Skipping {ticker} position: {str(e)}") continue @@ -295,18 +291,7 @@ def run_sunny_scanner(min_price: float, max_price: float, min_volume: int, portf 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}/sunny_signals_{output_date}.csv' - df_bullish.to_csv(output_file, index=False) - print(f"\nSaved {len(bullish_signals)} signals to {output_file}") - else: - print("\nNo signals found") + save_signals_to_csv(bullish_signals, 'sunny') except Exception as e: print(f"Error during scan: {str(e)}") diff --git a/src/utils/data_utils.py b/src/utils/data_utils.py index a36662e..99bad09 100644 --- a/src/utils/data_utils.py +++ b/src/utils/data_utils.py @@ -17,6 +17,44 @@ def validate_signal_date(signal_date: datetime) -> datetime: return current_date return signal_date +def print_signal(signal_data: dict, signal_type: str = "šŸ”") -> None: + """ + Print standardized signal output + + Args: + signal_data (dict): Dictionary containing signal information + signal_type (str): Emoji indicator for signal type (default: šŸ”) + """ + try: + print(f"\n{signal_type} {signal_data['ticker']} @ ${signal_data['entry_price']:.2f} on {signal_data['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: ${signal_data['target_price']:.2f}") + print(f" Risk/Reward: 1:{signal_data['risk_reward_ratio']:.1f} | Risk: ${abs(signal_data['risk_amount']):.2f}") + print(f" Potential Profit: ${signal_data['profit_amount']:.2f}") + except KeyError as e: + print(f"Error printing signal: Missing key {e}") + +def save_signals_to_csv(signals: list, scanner_name: str) -> None: + """ + Save signals to CSV file with standardized format and naming + + Args: + signals (list): List of signal dictionaries + scanner_name (str): Name of the scanner for file naming + """ + if not signals: + print("\nNo signals found") + return + + output_dir = 'reports' + os.makedirs(output_dir, exist_ok=True) + output_date = datetime.now().strftime("%Y%m%d_%H%M") + output_file = f'{output_dir}/{scanner_name}_{output_date}.csv' + + df_signals = pd.DataFrame(signals) + df_signals.to_csv(output_file, index=False) + print(f"\nSaved {len(signals)} signals to {output_file}") + def get_stock_data(ticker: str, start_date: datetime, end_date: datetime, interval: str) -> pd.DataFrame: """ Fetch stock data from the database with enhanced fallback logic