diff --git a/src/screener/t_atr_ema_v2.py b/src/screener/t_atr_ema_v2.py index f333773..8e0082a 100644 --- a/src/screener/t_atr_ema_v2.py +++ b/src/screener/t_atr_ema_v2.py @@ -70,32 +70,15 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int, min_volume (int): Minimum trading volume portfolio_size (float, optional): Portfolio size for position sizing """ - print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}") - - interval = get_interval_choice() - - start_date, end_date = get_date_range() - start_ts = int(start_date.timestamp() * 1000000000) - end_ts = int(end_date.timestamp() * 1000000000) - try: - qualified_stocks = get_qualified_stocks(start_date, end_date, min_price, max_price, min_volume) + # Initialize scanner components + interval, start_date, end_date, qualified_stocks, calculator = initialize_scanner( + min_price, max_price, min_volume, portfolio_size + ) if not qualified_stocks: - print("No stocks found matching criteria.") return - - print(f"\nFound {len(qualified_stocks)} stocks matching criteria") - - # Initialize position calculator if portfolio size provided - calculator = None - if portfolio_size and portfolio_size > 0: - calculator = PositionCalculator( - account_size=portfolio_size, - risk_percentage=1.0, - stop_loss_percentage=7.0 - ) - + entry_signals = [] for ticker, current_price, current_volume, last_update, stock_type in qualified_stocks: @@ -107,28 +90,11 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int, signals = check_entry_signal(df) for signal, signal_date, signal_data in signals: - entry_data = { - 'ticker': ticker, - 'entry_price': signal_data['price'], - 'target_price': signal_data['ema'], - 'volume': current_volume, - 'signal_date': signal_date, - 'stock_type': stock_type, - 'last_update': datetime.fromtimestamp(last_update/1000000000) - } - - if calculator: - position = calculator.calculate_position_size(entry_data['entry_price']) - potential_profit = (entry_data['target_price'] - entry_data['entry_price']) * position['shares'] - entry_data.update({ - 'shares': position['shares'], - 'position_size': position['position_value'], - 'stop_loss': position['stop_loss'], - 'risk_amount': position['potential_loss'], - 'profit_amount': potential_profit, - 'risk_reward_ratio': abs(potential_profit / position['potential_loss']) if position['potential_loss'] != 0 else 0 - }) - + signal_data['date'] = signal_date + entry_data = process_signal_data( + ticker, signal_data, current_volume, + last_update, stock_type, calculator + ) entry_signals.append(entry_data) print_signal(entry_data) diff --git a/src/utils/data_utils.py b/src/utils/data_utils.py index 148db30..3815edb 100644 --- a/src/utils/data_utils.py +++ b/src/utils/data_utils.py @@ -134,6 +134,83 @@ def save_signals_to_csv(signals: list, scanner_name: str) -> None: df_signals.to_csv(output_file, index=False) print(f"\nSaved {len(signals)} signals to {output_file}") +def initialize_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> tuple: + """ + Initialize common scanner components + + Args: + min_price (float): Minimum stock price + max_price (float): Maximum stock price + min_volume (int): Minimum trading volume + portfolio_size (float, optional): Portfolio size for position sizing + + Returns: + tuple: (interval, start_date, end_date, qualified_stocks, calculator) + """ + print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}") + + interval = get_interval_choice() + start_date, end_date = get_date_range() + + qualified_stocks = get_qualified_stocks(start_date, end_date, min_price, max_price, min_volume) + + if not qualified_stocks: + print("No stocks found matching criteria.") + return None, None, None, None, None + + print(f"\nFound {len(qualified_stocks)} stocks matching criteria") + + # Initialize position calculator if portfolio size provided + calculator = None + if portfolio_size and portfolio_size > 0: + calculator = PositionCalculator( + account_size=portfolio_size, + risk_percentage=1.0, + stop_loss_percentage=7.0 + ) + + return interval, start_date, end_date, qualified_stocks, calculator + +def process_signal_data(ticker: str, signal_data: dict, current_volume: int, + last_update: int, stock_type: str, calculator: PositionCalculator = None) -> dict: + """ + Process and format signal data consistently + + Args: + ticker (str): Stock ticker + signal_data (dict): Raw signal data + current_volume (int): Current trading volume + last_update (int): Last update timestamp + stock_type (str): Stock type/label + calculator (PositionCalculator, optional): Position calculator instance + + Returns: + dict: Processed signal data + """ + entry_data = { + 'ticker': ticker, + 'entry_price': signal_data['price'], + 'target_price': signal_data.get('ema', signal_data.get('upper_band')), # Handle both ATR and Sunny + 'volume': current_volume, + 'signal_date': signal_data.get('date', datetime.now()), + 'stock_type': stock_type, + 'last_update': datetime.fromtimestamp(last_update/1000000000) + } + + if calculator: + position = calculator.calculate_position_size(entry_data['entry_price']) + potential_profit = (entry_data['target_price'] - entry_data['entry_price']) * position['shares'] + entry_data.update({ + 'shares': position['shares'], + 'position_size': position['position_value'], + 'stop_loss': position['stop_loss'], + 'risk_amount': position['potential_loss'], + 'profit_amount': potential_profit, + 'risk_reward_ratio': abs(potential_profit / position['potential_loss']) if position['potential_loss'] != 0 else 0 + }) + + return entry_data + def get_stock_data(ticker: str, start_date: datetime, end_date: datetime, interval: str) -> pd.DataFrame: """ Fetch and resample stock data based on the chosen interval