application / timeframe.py
Dooratre's picture
Create timeframe.py
d05d6ed verified
raw
history blame
10.4 kB
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