Spaces:
Sleeping
Sleeping
| 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 | |