Spaces:
Sleeping
Sleeping
File size: 6,045 Bytes
b8086d5 ac24887 b8086d5 cd38f99 b8086d5 cd38f99 b8086d5 ac24887 b8086d5 f27fbb4 cd38f99 f27fbb4 b8086d5 cd38f99 b8086d5 ac24887 b8086d5 ac24887 b8086d5 ac24887 b8086d5 ac24887 cd38f99 b8086d5 f27fbb4 b8086d5 f27fbb4 ac24887 f27fbb4 ac24887 f27fbb4 ac24887 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 |
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:
# FIX: Gunakan string interval yang didukung langsung oleh Yahoo/yfinance
interval_map = {
"5m": "5m",
"15m": "15m",
"30m": "30m",
"1h": "1h",
"4h": "4h",
"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)
# Mengatasi masalah data kripto yang mungkin tidak tersedia dalam periode lama
if ticker == "BTC-USD" and period == "max":
period = "5y"
df = ticker_obj.history(interval=yf_interval, period=period)
if df.empty:
# Perbarui pesan error untuk memperjelas log
raise ValueError(f"No data retrieved from Yahoo Finance for {ticker} at interval {interval}. Check if ticker/interval is supported.")
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):
if df.empty:
return df
df['SMA_5'] = df['Close'].rolling(window=5).mean()
df['SMA_20'] = df['Close'].rolling(window=20).mean()
df['EMA_12'] = df['Close'].ewm(span=12, adjust=False).mean()
df['EMA_26'] = df['Close'].ewm(span=26, adjust=False).mean()
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']
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)
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))
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)
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()
df['Volume_SMA'] = df['Volume'].rolling(window=20).mean()
df['Volume_ratio'] = df['Volume'] / df['Volume_SMA']
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
df['DL'] = 30
return df
def get_fundamental_data(self, ticker="GC=F"):
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):
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
} |