import os import pandas as pd from datetime import datetime, timedelta from trading.position_calculator import PositionCalculator from utils.common_utils import get_user_input, get_stock_data, get_qualified_stocks from typing import Optional import requests def get_float_input(prompt: str) -> Optional[float]: return get_user_input(prompt, float) def get_current_prices(tickers: list) -> dict: """Get current prices for multiple tickers using yfinance""" if not tickers: return {} polygon_api_key = os.environ.get("POLYGON_API_KEY") if not polygon_api_key: print("Error: POLYGON_API_KEY environment variable not set.") return {} prices = {} try: for ticker in tickers: url = f"https://api.polygon.io/v2/last/trade/{ticker}?apiKey={polygon_api_key}" response = requests.get(url) if response.status_code == 200: data = response.json() try: prices[ticker] = data['results']['price'] except KeyError: prices[ticker] = 0.0 else: print(f"Error fetching price for {ticker}: {response.status_code} - {response.text}") prices[ticker] = 0.0 # Verify that we have prices for all tickers for ticker in tickers: if ticker not in prices or prices[ticker] == 0: print(f"Could not fetch price for {ticker}") prices[ticker] = 0.0 except Exception as e: print(f"Error fetching prices: {e}") # If there's an error, fall back to setting price to 0.0 for ticker in tickers: if ticker not in prices: prices[ticker] = 0.0 try: stock = yf.Ticker(ticker) price = stock.info.get('regularMarketPrice', 0.0) prices[ticker] = float(price) except: prices[ticker] = 0.0 return prices def validate_signal_date(signal_date: datetime) -> datetime: """ Validate and adjust signal date if needed Args: signal_date (datetime): Signal date to validate Returns: datetime: Valid signal date (not in future) """ current_date = datetime.now() if signal_date > current_date: return current_date return signal_date def print_signal(signal_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['stock_type']}) @ ${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 for {signal_data.get('ticker', 'Unknown')}: Missing key {e}") # Print available keys for debugging print(f"Available keys: {list(signal_data.keys())}") 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}")