Create timeframe.py

timeframe.py

@@ -0,0 +1,232 @@
+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