import requests import json import time from datetime import datetime, timedelta import pandas as pd from tabulate import tabulate def fetch_financial_data(symbol="USDCAD=X", start_date=None, end_date=None, interval="1m", days_to_fetch=1): """ Fetch and process financial data from Yahoo Finance. Parameters: symbol (str): The ticker symbol (e.g., "USDCAD=X" for USD/CAD exchange rate) start_date (datetime): Start date for the data (optional, defaults to 1 day ago if not provided) end_date (datetime): End date for the data (optional, defaults to current time if not provided) interval (str): Data interval - Options include "1m", "5m", "15m", "30m", "60m", "1d", "1wk", "1mo" days_to_fetch (int): Number of days to fetch data for if end_date is not specified Returns: dict: A dictionary containing: - 'data': pandas DataFrame with the processed financial data - 'meta': dictionary with meta information - 'table': formatted table string - 'meta_info': formatted meta information string - 'stats': formatted summary statistics string - 'success': boolean indicating if the operation was successful - 'message': status message """ # Set default dates if not provided if start_date is None: start_date = datetime.now() - timedelta(days=days_to_fetch) if end_date is None: end_date = datetime.now() result = { 'data': None, 'meta': None, 'table': None, 'meta_info': None, 'stats': None, 'success': False, 'message': "" } try: # Convert datetime to Unix timestamp period1 = int(time.mktime(start_date.timetuple())) period2 = int(time.mktime(end_date.timetuple())) # Construct the URL url = f"https://query2.finance.yahoo.com/v8/finance/chart/{symbol}" # Parameters for the request params = { "period1": period1, "period2": period2, "interval": interval, "includePrePost": "true", "events": "div|split|earn", "lang": "en-US", "region": "US", "source": "cosaic" } # Headers to mimic a browser request headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36" } # Make the request response = requests.get(url, params=params, headers=headers) if response.status_code == 200: data = response.json() # Process the data if not data or 'chart' not in data or 'result' not in data['chart'] or not data['chart']['result']: result['message'] = "No valid data to process" return result # Extract the relevant data api_result = data['chart']['result'][0] meta = api_result['meta'] timestamps = api_result['timestamp'] quotes = api_result['indicators']['quote'][0] # Create a DataFrame df = pd.DataFrame({ 'timestamp': timestamps, 'open': quotes['open'], 'high': quotes['high'], 'low': quotes['low'], 'close': quotes['close'], 'volume': quotes['volume'] }) # Convert timestamps to datetime df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s') # Calculate additional metrics df['range'] = df['high'] - df['low'] df['change'] = df['close'] - df['open'] df['change_pct'] = (df['change'] / df['open'] * 100).round(3) # Mark if the candle is bullish (close > open) or bearish (close < open) df['trend'] = ['Bullish' if c > o else 'Bearish' if c < o else 'Neutral' for c, o in zip(df['close'], df['open'])] # Format the DataFrame for display display_df = df.copy() # Format the timestamp display_df['Date'] = display_df['timestamp'].dt.strftime('%Y-%m-%d') display_df['Time'] = display_df['timestamp'].dt.strftime('%H:%M:%S') # Round numerical values to 4 decimal places for col in ['open', 'high', 'low', 'close', 'range']: if col in display_df.columns: display_df[col] = display_df[col].round(4) # Add color indicators for trend (will be visible in HTML output) display_df['Change'] = display_df['change'].round(4) display_df['% Change'] = display_df['change_pct'].round(2).astype(str) + '%' # Select columns for display table_df = display_df[['Date', 'Time', 'open', 'high', 'low', 'close', 'range', 'Change', '% Change', 'trend']] # Rename columns for better readability table_df.columns = ['Date', 'Time', 'Open', 'High', 'Low', 'Close', 'Range', 'Change', '% Change', 'Trend'] # Create the table using tabulate table = tabulate(table_df, headers='keys', tablefmt='fancy_grid', showindex=False) # Add a title title = f"{symbol} Price Data ({interval} interval)" title_line = "=" * len(title) formatted_table = f"\n{title_line}\n{title}\n{title_line}\n\n{table}" # Format meta information meta_str = "META INFORMATION\n===============\n" # Extract key information key_info = { 'Currency': meta.get('currency'), 'Symbol': meta.get('symbol'), 'Exchange Name': meta.get('exchangeName'), 'Full Exchange Name': meta.get('fullExchangeName'), 'Instrument Type': meta.get('instrumentType'), 'First Trade Date': datetime.fromtimestamp(meta.get('firstTradeDate', 0)).strftime('%Y-%m-%d') if 'firstTradeDate' in meta else 'N/A', 'Regular Market Time': datetime.fromtimestamp(meta.get('regularMarketTime', 0)).strftime('%Y-%m-%d %H:%M:%S') if 'regularMarketTime' in meta else 'N/A', 'Timezone': meta.get('timezone'), 'Exchange Timezone': meta.get('exchangeTimezoneName'), 'Regular Market Price': meta.get('regularMarketPrice'), '52 Week High': meta.get('fiftyTwoWeekHigh'), '52 Week Low': meta.get('fiftyTwoWeekLow'), 'Day High': meta.get('regularMarketDayHigh'), 'Day Low': meta.get('regularMarketDayLow'), 'Volume': meta.get('regularMarketVolume'), 'Long Name': meta.get('longName'), 'Short Name': meta.get('shortName'), 'Previous Close': meta.get('previousClose'), 'Scale': meta.get('scale'), 'Price Hint': meta.get('priceHint') } # Format the key_info as a string for key, value in key_info.items(): if isinstance(value, float): meta_str += f"{key}: {value:.4f}\n" else: meta_str += f"{key}: {value}\n" # Add trading period information if 'currentTradingPeriod' in meta: meta_str += "\nCURRENT TRADING PERIOD\n=====================\n" for period_type, period_info in meta['currentTradingPeriod'].items(): start_time = datetime.fromtimestamp(period_info.get('start', 0)).strftime('%Y-%m-%d %H:%M:%S') end_time = datetime.fromtimestamp(period_info.get('end', 0)).strftime('%Y-%m-%d %H:%M:%S') meta_str += f"{period_type.capitalize()} Period: {start_time} to {end_time} ({period_info.get('timezone', 'Unknown')})\n" # Generate summary statistics stats = { 'Symbol': symbol, 'Long Name': meta.get('longName', symbol), 'Period Start': df['timestamp'].min().strftime("%Y-%m-%d %H:%M:%S"), 'Period End': df['timestamp'].max().strftime("%Y-%m-%d %H:%M:%S"), 'Data Points': len(df), 'Opening Price': df['open'].iloc[0], 'Closing Price': df['close'].iloc[-1], 'Current Price': meta.get('regularMarketPrice', df['close'].iloc[-1]), 'Overall Change': df['close'].iloc[-1] - df['open'].iloc[0], 'Overall % Change': ((df['close'].iloc[-1] / df['open'].iloc[0]) - 1) * 100, 'Highest Price': df['high'].max(), 'Lowest Price': df['low'].min(), 'Average Price': df[['open', 'high', 'low', 'close']].mean().mean(), 'Price Range': df['high'].max() - df['low'].min(), 'Day High (Meta)': meta.get('regularMarketDayHigh'), 'Day Low (Meta)': meta.get('regularMarketDayLow'), '52 Week High': meta.get('fiftyTwoWeekHigh'), '52 Week Low': meta.get('fiftyTwoWeekLow'), 'Bullish Candles': (df['trend'] == 'Bullish').sum(), 'Bearish Candles': (df['trend'] == 'Bearish').sum(), 'Neutral Candles': (df['trend'] == 'Neutral').sum() } # Format the stats as a string stats_str = "SUMMARY STATISTICS\n==================\n" for key, value in stats.items(): if isinstance(value, float): stats_str += f"{key}: {value:.4f}\n" else: stats_str += f"{key}: {value}\n" # Populate result dictionary result['data'] = df result['meta'] = meta result['table'] = formatted_table result['meta_info'] = meta_str result['stats'] = stats_str result['success'] = True result['message'] = f"Successfully processed {len(df)} data points." return result else: result['message'] = f"Failed to fetch data: Status code {response.status_code}" return result except Exception as e: result['message'] = f"Error: {str(e)}" return result