Spaces:
Sleeping
Sleeping
File size: 2,550 Bytes
bfa4e15 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 |
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
import pytz
def is_market_open():
"""Check if US market is currently open"""
now = datetime.now(pytz.timezone('America/New_York'))
# Market hours are 9:30 AM - 4:00 PM Eastern Time, Monday to Friday
is_weekday = now.weekday() < 5
is_market_hours = 9.5 <= now.hour + (now.minute / 60) <= 16
# Pre-market (4:00 AM - 9:30 AM) and After-hours (4:00 PM - 8:00 PM)
is_extended_hours = (4 <= now.hour + (now.minute / 60) <= 20)
return is_weekday and (is_market_hours or is_extended_hours)
def fetch_market_data(symbol, period='1d', interval='15m'):
"""Fetch market data from Yahoo Finance"""
try:
# Check if we have a valid symbol
if not symbol:
raise Exception("Please enter a valid trading symbol")
# Always fetch 1 day of data to ensure we have enough history
ticker = yf.Ticker(symbol)
df = ticker.history(period='1d', interval='15m', prepost=True)
if df.empty:
raise Exception(f"No data available for {symbol}. Please verify the symbol is correct.")
# Get the market status
market_status = "Market Open" if is_market_open() else "Market Closed"
# Trim data based on selected timeframe
now = datetime.now(pytz.timezone('America/New_York'))
if period.endswith('m'):
minutes = int(period[:-1])
cutoff_time = now - timedelta(minutes=minutes)
else:
hours = int(period[:-1])
cutoff_time = now - timedelta(hours=hours)
df = df[df.index >= cutoff_time]
if df.empty:
raise Exception(f"No recent data available. {market_status}.")
return df
except Exception as e:
if "symbol may be delisted" in str(e).lower():
raise Exception(f"Symbol {symbol} not found. Please verify the symbol is correct.")
raise Exception(f"Data fetch error: {str(e)}")
def calculate_performance_metrics(predictions, actual):
"""Calculate prediction performance metrics"""
if len(predictions) == 0 or len(actual) == 0:
return {
'accuracy': 0,
'success_rate': 0,
'total_predictions': 0
}
correct_predictions = sum(p == a for p, a in zip(predictions, actual))
return {
'accuracy': correct_predictions / len(predictions),
'success_rate': correct_predictions / len(predictions) * 100,
'total_predictions': len(predictions)
} |