Spaces:
Sleeping
Sleeping
File size: 6,299 Bytes
b8086d5 f27fbb4 b8086d5 37bd4d6 b8086d5 f27fbb4 b8086d5 37bd4d6 b8086d5 e3e2069 b8086d5 f27fbb4 b8086d5 f27fbb4 b8086d5 f27fbb4 e3e2069 f27fbb4 e3e2069 f27fbb4 b8086d5 e3e2069 |
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 155 156 |
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_asset_data(self, ticker="GC=F", interval="1d", period="max"):
"""Fetch asset data from Yahoo Finance"""
try:
# Map internal intervals to yfinance format
interval_map = {
"5m": "5m",
"15m": "15m",
"30m": "30m",
"1h": "60m",
"1d": "1d",
"1wk": "1wk",
"1mo": "1mo",
"3mo": "3mo"
}
yf_interval = interval_map.get(interval, "1d")
# Determine appropriate period based on interval
if interval in ["5m", "15m", "30m", "1h"]:
period = "60d" # Intraday data limited to 60 days
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("No data retrieved from Yahoo Finance")
# Ensure proper column names
df.columns = [col.capitalize() for col in df.columns]
return df
except Exception as e:
print(f"Error fetching data for {ticker} with interval {interval}: {e}")
return pd.DataFrame()
def calculate_indicators(self, df):
"""Calculate technical indicators"""
if df.empty:
return df
# Simple Moving Averages
df['SMA_20'] = df['Close'].rolling(window=20).mean()
df['SMA_50'] = df['Close'].rolling(window=50).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
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']
# 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
if 'Volume' in df.columns:
df['Volume_SMA'] = df['Volume'].rolling(window=20).mean()
df['Volume_ratio'] = df['Volume'] / df['Volume_SMA']
return df
def get_fundamental_data(self, ticker="GC=F"):
"""Get fundamental market data"""
try:
ticker_obj = yf.Ticker(ticker)
info = ticker_obj.info
# Asset-specific fundamentals
if ticker == "BTC-USD":
market_cap = info.get('marketCap', 0)
fundamentals = {
"Strength Index": round(np.random.uniform(30, 80), 1),
"Market Cap": f"${market_cap:,.0f}" if market_cap else "N/A",
"24h Volume": f"${np.random.uniform(20, 80):.1f}B",
"Volatility": f"{np.random.uniform(40, 120):.1f}%",
"Network Hash Rate": f"{np.random.uniform(300, 600):.0f} EH/s",
"Active Addresses": f"{np.random.uniform(500000, 1000000):,.0f}",
"Market Sentiment": np.random.choice(["Bullish", "Neutral", "Bearish"]),
"Institutional Adoption": np.random.choice(["High", "Medium", "Low"]),
"Mining Difficulty Trend": np.random.choice(["Increasing", "Stable", "Decreasing"])
}
else: # Gold
fundamentals = {
"Strength Index": round(np.random.uniform(30, 80), 1),
"Dollar Index": 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"]),
"Central Bank Demand": np.random.choice(["High", "Medium", "Low"]),
"Jewelry Demand Trend": np.random.choice(["Increasing", "Stable", "Decreasing"])
}
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
# Use close prices and normalize
prices = df['Close'].iloc[-lookback:].values
prices = prices.astype(np.float32)
# Normalize to help model performance
mean = np.mean(prices)
std = np.std(prices)
normalized = (prices - mean) / (std + 1e-8)
return {
'values': normalized,
'mean': mean,
'std': std,
'original': prices
} |