Krikorios
48e60d015f
- Add advanced metrics dashboard with trade analytics - Add new trading components (EntryTypeAnalysis, MultiDayPositionTracker, NewsEventTracker, etc.) - Add strategy mode selector and trend confirmation - Add risk automation panel and slippage correlation analysis - Add daily trading plan enhancements with modal components - Add custom hooks (useApi, useLocalStorage, useAdvancedTradeMetrics) - Add broker service integration and trading API - Add test setup and vitest configuration - Include parquet data files for live market data - Add comprehensive documentation in docs/ folder
239 lines
8.5 KiB
Python
239 lines
8.5 KiB
Python
"""
|
|
Robust Gold Price Fetcher with Multiple Data Sources and Fallback
|
|
Ensures accurate real-time gold pricing with redundancy
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import httpx
|
|
from typing import Optional, Dict, Any
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GoldPriceFetcher:
|
|
"""
|
|
Multi-source gold price fetcher with automatic fallback
|
|
|
|
Data Sources (in priority order):
|
|
1. Alpha Vantage - GLD ETF (reliable, free tier)
|
|
2. Twelve Data API (if available)
|
|
3. Yahoo Finance (backup)
|
|
4. Static fallback to reasonable estimate
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.client = httpx.AsyncClient(timeout=10.0)
|
|
# GLD ETF tracks ~1/10th of gold spot price
|
|
self.gld_multiplier = 10.0
|
|
# Gold futures (GC) are 100oz contracts, but quote is per oz
|
|
self.gc_multiplier = 1.0
|
|
|
|
async def get_current_gold_price(self) -> Dict[str, Any]:
|
|
"""
|
|
Get current gold price with automatic fallback through multiple sources
|
|
|
|
Returns:
|
|
Dict with: price, source, timestamp, high_24h, low_24h, change_percent
|
|
"""
|
|
# Try Alpha Vantage GLD first (most reliable)
|
|
try:
|
|
result = await self._fetch_from_alpha_vantage_gld()
|
|
if result:
|
|
logger.info(f"✅ Gold price from Alpha Vantage GLD: ${result['price']:.2f}")
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"Alpha Vantage GLD failed: {e}")
|
|
|
|
# Try Twelve Data if available
|
|
try:
|
|
result = await self._fetch_from_twelve_data()
|
|
if result:
|
|
logger.info(f"✅ Gold price from Twelve Data: ${result['price']:.2f}")
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"Twelve Data failed: {e}")
|
|
|
|
# Try alternative free sources
|
|
try:
|
|
result = await self._fetch_from_metals_api()
|
|
if result:
|
|
logger.info(f"✅ Gold price from Metals-API: ${result['price']:.2f}")
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"Metals-API failed: {e}")
|
|
|
|
# Last resort: return estimated price with warning
|
|
logger.error("⚠️ All gold price sources failed, using estimated price")
|
|
return self._get_fallback_price()
|
|
|
|
async def _fetch_from_alpha_vantage_gld(self) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Fetch from Alpha Vantage using GLD ETF as proxy
|
|
GLD tracks gold at ~1/10th spot price
|
|
"""
|
|
api_key = settings.ALPHA_VANTAGE_API_KEY or "M1S58UEM42CQD31T"
|
|
url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol=GLD&apikey={api_key}"
|
|
|
|
response = await self.client.get(url)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if "Global Quote" not in data or not data["Global Quote"]:
|
|
return None
|
|
|
|
quote = data["Global Quote"]
|
|
gld_price = float(quote.get("05. price", 0))
|
|
|
|
if gld_price == 0:
|
|
return None
|
|
|
|
# Convert GLD price to gold spot price (multiply by 10)
|
|
gold_price = gld_price * self.gld_multiplier
|
|
|
|
return {
|
|
"price": gold_price,
|
|
"open": float(quote.get("02. open", 0)) * self.gld_multiplier,
|
|
"high": float(quote.get("03. high", 0)) * self.gld_multiplier,
|
|
"low": float(quote.get("04. low", 0)) * self.gld_multiplier,
|
|
"volume": int(quote.get("06. volume", 0)),
|
|
"previous_close": float(quote.get("08. previous close", 0)) * self.gld_multiplier,
|
|
"change": float(quote.get("09. change", 0)) * self.gld_multiplier,
|
|
"change_percent": quote.get("10. change percent", "0%"),
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"source": "Alpha Vantage (GLD ETF)",
|
|
"trading_day": quote.get("07. latest trading day", ""),
|
|
}
|
|
|
|
async def _fetch_from_twelve_data(self) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Fetch from Twelve Data API (if API key available)
|
|
They have direct XAU/USD forex pair
|
|
"""
|
|
# Twelve Data would require API key setup
|
|
# Placeholder for now
|
|
return None
|
|
|
|
async def _fetch_from_metals_api(self) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Fetch from Metals-API.com free tier
|
|
Provides direct gold spot prices
|
|
"""
|
|
try:
|
|
# Free tier endpoint (limited requests)
|
|
url = "https://metals-api.com/api/latest"
|
|
params = {
|
|
"access_key": "your_key_here", # Would need API key
|
|
"base": "USD",
|
|
"symbols": "XAU"
|
|
}
|
|
|
|
# Skip if no key configured
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
def _get_fallback_price(self) -> Dict[str, Any]:
|
|
"""
|
|
Return reasonable estimated gold price when all sources fail
|
|
Based on typical 2025 gold trading range
|
|
"""
|
|
# Conservative estimate for late 2025 gold prices
|
|
estimated_price = 3800.0 # Mid-range estimate
|
|
|
|
return {
|
|
"price": estimated_price,
|
|
"open": estimated_price,
|
|
"high": estimated_price * 1.01,
|
|
"low": estimated_price * 0.99,
|
|
"volume": 0,
|
|
"previous_close": estimated_price,
|
|
"change": 0.0,
|
|
"change_percent": "0%",
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"source": "FALLBACK_ESTIMATE",
|
|
"trading_day": datetime.utcnow().strftime("%Y-%m-%d"),
|
|
"warning": "⚠️ Using estimated price - all data sources unavailable"
|
|
}
|
|
|
|
async def get_intraday_data(self, interval: str = "5min", limit: int = 100) -> list[Dict[str, Any]]:
|
|
"""
|
|
Get intraday gold price data
|
|
|
|
Args:
|
|
interval: Time interval (1min, 5min, 15min, 30min, 60min)
|
|
limit: Number of data points to return
|
|
|
|
Returns:
|
|
List of OHLCV data points
|
|
"""
|
|
try:
|
|
return await self._fetch_intraday_alpha_vantage(interval, limit)
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch intraday data: {e}")
|
|
return []
|
|
|
|
async def _fetch_intraday_alpha_vantage(self, interval: str, limit: int) -> list[Dict[str, Any]]:
|
|
"""
|
|
Fetch intraday data from Alpha Vantage
|
|
Using GLD as proxy since XAU/USD intraday is premium
|
|
"""
|
|
api_key = settings.ALPHA_VANTAGE_API_KEY or "M1S58UEM42CQD31T"
|
|
url = f"https://www.alphavantage.co/query"
|
|
params = {
|
|
"function": "TIME_SERIES_INTRADAY",
|
|
"symbol": "GLD",
|
|
"interval": interval,
|
|
"apikey": api_key,
|
|
"outputsize": "compact" # Last 100 data points
|
|
}
|
|
|
|
response = await self.client.get(url, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
time_series_key = f"Time Series ({interval})"
|
|
if time_series_key not in data:
|
|
return []
|
|
|
|
time_series = data[time_series_key]
|
|
|
|
# Convert to OHLCV format and apply gold multiplier
|
|
result = []
|
|
for timestamp, values in list(time_series.items())[:limit]:
|
|
result.append({
|
|
"timestamp": timestamp,
|
|
"time": int(datetime.fromisoformat(timestamp.replace("Z", "+00:00")).timestamp()),
|
|
"open": float(values["1. open"]) * self.gld_multiplier,
|
|
"high": float(values["2. high"]) * self.gld_multiplier,
|
|
"low": float(values["3. low"]) * self.gld_multiplier,
|
|
"close": float(values["4. close"]) * self.gld_multiplier,
|
|
"volume": int(values["5. volume"]),
|
|
})
|
|
|
|
return sorted(result, key=lambda x: x["time"])
|
|
|
|
async def close(self):
|
|
"""Close HTTP client"""
|
|
await self.client.aclose()
|
|
|
|
|
|
# Global instance
|
|
gold_price_fetcher = GoldPriceFetcher()
|
|
|
|
|
|
# Convenience functions for backward compatibility
|
|
async def get_current_gold_price() -> Dict[str, Any]:
|
|
"""Get current gold spot price"""
|
|
return await gold_price_fetcher.get_current_gold_price()
|
|
|
|
|
|
async def get_gold_intraday(interval: str = "5min", limit: int = 100) -> list[Dict[str, Any]]:
|
|
"""Get intraday gold price data"""
|
|
return await gold_price_fetcher.get_intraday_data(interval, limit)
|