Spaces:
Build error
Build error
File size: 6,163 Bytes
b8086d5 2c99719 b8086d5 a469221 b8086d5 2c99719 b8086d5 2c99719 b8086d5 2c99719 b8086d5 a469221 b8086d5 a469221 b8086d5 a469221 b8086d5 a469221 b8086d5 2c99719 b8086d5 2c99719 a469221 2c99719 a469221 2c99719 a469221 2c99719 b8086d5 a469221 b8086d5 |
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 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class DataProcessor:
def __init__(self):
self.fundamentals_cache = {}
def get_market_data(self, ticker="GC=F", interval="1d"):
"""Fetch market data from Yahoo Finance for a given ticker"""
try:
interval_map = {
"5m": "5m",
"15m": "15m",
"30m": "30m",
"1h": "60m",
"4h": "240m",
"1d": "1d",
"1wk": "1wk",
"1mo": "1mo",
"3mo": "3mo"
}
yf_interval = interval_map.get(interval, "1d")
if interval in ["5m", "15m", "30m", "1h", "4h"]:
period = "60d"
elif interval in ["1d"]:
period = "1y"
elif interval in ["1wk"]:
period = "2y"
else:
period = "max"
ticker_obj = yf.Ticker(ticker)
df = ticker_obj.history(interval=yf_interval, period=period)
if df.empty:
raise ValueError(f"No data retrieved from Yahoo Finance for {ticker}")
df.columns = [col.capitalize() for col in df.columns]
return df
except Exception as e:
print(f"Error fetching data for {ticker}: {e}")
return pd.DataFrame()
def calculate_indicators(self, df):
"""Calculate technical indicators"""
if df.empty:
return df
# Simple Moving Averages (5, 20 as requested)
df['SMA_5'] = df['Close'].rolling(window=5).mean()
df['SMA_20'] = df['Close'].rolling(window=20).mean()
# Exponential Moving Averages
df['EMA_12'] = df['Close'].ewm(span=12, adjust=False).mean()
df['EMA_26'] = df['Close'].ewm(span=26, adjust=False).mean()
# MACD (12, 26, 9)
df['MACD'] = df['EMA_12'] - df['EMA_26']
df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
df['MACD_histogram'] = df['MACD'] - df['MACD_signal']
# Split histogram into positive and negative for plotting
df['MACD_bar_positive'] = df['MACD_histogram'].where(df['MACD_histogram'] > 0, 0)
df['MACD_bar_negative'] = df['MACD_histogram'].where(df['MACD_histogram'] < 0, 0)
# RSI
delta = df['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# Bollinger Bands
df['BB_middle'] = df['Close'].rolling(window=20).mean()
bb_std = df['Close'].rolling(window=20).std()
df['BB_upper'] = df['BB_middle'] + (bb_std * 2)
df['BB_lower'] = df['BB_middle'] - (bb_std * 2)
# Average True Range (ATR)
high_low = df['High'] - df['Low']
high_close = np.abs(df['High'] - df['Close'].shift())
low_close = np.abs(df['Low'] - df['Close'].shift())
ranges = pd.concat([high_low, high_close, low_close], axis=1)
true_range = ranges.max(axis=1)
df['ATR'] = true_range.rolling(window=14).mean()
# Volume indicators
df['Volume_SMA'] = df['Volume'].rolling(window=20).mean()
df['Volume_ratio'] = df['Volume'] / df['Volume_SMA']
# Stochastic Oscillator (14, 3)
low_14 = df['Low'].rolling(window=14).min()
high_14 = df['High'].rolling(window=14).max()
df['%K'] = 100 * (df['Close'] - low_14) / (high_14 - low_14)
df['%D'] = df['%K'].rolling(window=3).mean()
df['%SD'] = df['%D'].rolling(window=3).mean()
df['UL'] = 70 # Upper limit
df['DL'] = 30 # Lower limit
return df
def get_fundamental_data(self, ticker="GC=F"):
"""Get fundamental gold market data (now generalized/mocked)"""
try:
if ticker == "BTC-USD":
fundamentals = {
"Crypto Volatility Index": round(np.random.uniform(50, 150), 1),
"Dominance Index": f"{np.random.uniform(40, 60):.2f}%",
"Fear & Greed Index": np.random.choice(["Extreme Fear", "Fear", "Neutral", "Greed", "Extreme Greed"]),
"Hash Rate Trend": np.random.choice(["Increasing", "Stable", "Decreasing"]),
"Institutional Flow (Net)": f"{np.random.uniform(-100, 100):,.0f}M USD",
"Market Sentiment": np.random.choice(["Bullish", "Neutral", "Bearish"]),
}
else:
fundamentals = {
"Gold Strength Index": round(np.random.uniform(30, 80), 1),
"Dollar Index (DXY)": round(np.random.uniform(90, 110), 1),
"Real Interest Rate": f"{np.random.uniform(-2, 5):.2f}%",
"Gold Volatility": f"{np.random.uniform(10, 40):.1f}%",
"Commercial Hedgers (Net)": f"{np.random.uniform(-50000, 50000):,.0f}",
"Managed Money (Net)": f"{np.random.uniform(-100000, 100000):,.0f}",
"Market Sentiment": np.random.choice(["Bullish", "Neutral", "Bearish"]),
}
return fundamentals
except Exception as e:
print(f"Error fetching fundamentals: {e}")
return {"Error": str(e)}
def prepare_for_chronos(self, df, lookback=100):
"""Prepare data for Chronos model"""
if df.empty or len(df) < lookback:
return None
prices = df['Close'].iloc[-lookback:].values
prices = prices.astype(np.float32)
mean = np.mean(prices)
std = np.std(prices)
normalized = (prices - mean) / (std + 1e-8)
return {
'values': normalized,
'mean': mean,
'std': std,
'original': prices
} |