timeframe.py

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