143 lines
5.4 KiB
Plaintext
143 lines
5.4 KiB
Plaintext
import requests
|
|
from typing import Dict, Optional
|
|
from bs4 import BeautifulSoup
|
|
import time
|
|
|
|
class DataFetcher:
|
|
"""모든 자산 가격 수집 클래스"""
|
|
|
|
def __init__(self):
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
|
|
})
|
|
self.investing_cache = {}
|
|
self.cache_time = 0
|
|
|
|
def fetch_investing_com(self, asset_code: str) -> Optional[float]:
|
|
"""Investing.com에서 가격 수집"""
|
|
# 간단한 캐싱 (5초)
|
|
if time.time() - self.cache_time < 5 and asset_code in self.investing_cache:
|
|
return self.investing_cache[asset_code]
|
|
|
|
asset_map = {
|
|
"XAU/USD": "8830",
|
|
"XAU/CNY": "2186",
|
|
"XAU/GBP": "8500",
|
|
"USD/DXY": "8827"
|
|
}
|
|
|
|
asset_id = asset_map.get(asset_code)
|
|
if not asset_id:
|
|
return None
|
|
|
|
try:
|
|
url = f"https://www.investing.com/currencies/{asset_code.lower().replace('/', '-')}"
|
|
response = self.session.get(url, timeout=5)
|
|
response.raise_for_status()
|
|
|
|
soup = BeautifulSoup(response.text, 'lxml')
|
|
price_elem = soup.select_one('[data-test="instrument-price-last"]')
|
|
|
|
if price_elem:
|
|
price_text = price_elem.text.strip().replace(',', '')
|
|
price = float(price_text)
|
|
self.investing_cache[asset_code] = price
|
|
return price
|
|
except Exception as e:
|
|
print(f"Investing.com 수집 실패 ({asset_code}): {e}")
|
|
|
|
return None
|
|
|
|
def fetch_binance(self) -> Optional[float]:
|
|
"""바이낸스 BTC/USDT 가격"""
|
|
try:
|
|
url = "https://api.binance.com/api/v3/ticker/price"
|
|
response = self.session.get(url, params={"symbol": "BTCUSDT"}, timeout=5)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return float(data["price"]) if "price" in data else None
|
|
except Exception as e:
|
|
print(f"Binance API 실패: {e}")
|
|
return None
|
|
|
|
def fetch_upbit(self) -> Optional[float]:
|
|
"""업비트 BTC/KRW 가격"""
|
|
try:
|
|
url = "https://api.upbit.com/v1/ticker"
|
|
response = self.session.get(url, params={"markets": "KRW-BTC"}, timeout=5)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data[0]["trade_price"] if data and "trade_price" in data[0] else None
|
|
except Exception as e:
|
|
print(f"Upbit API 실패: {e}")
|
|
return None
|
|
|
|
def fetch_usd_krw(self) -> Optional[float]:
|
|
"""USD/KRW 환율"""
|
|
try:
|
|
url = "https://quotation-api-cdn.dunamu.com/v1/forex/recent?codes=FRX.KRWUSD"
|
|
response = self.session.get(url, timeout=5)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data[0]["basePrice"] if data else None
|
|
except Exception as e:
|
|
print(f"USD/KRW 수집 실패: {e}")
|
|
return None
|
|
|
|
def fetch_krx_gold(self) -> Optional[float]:
|
|
"""한국거래소 금 현물 가격"""
|
|
try:
|
|
url = "http://www.goldpr.co.kr/gms/default.asp"
|
|
response = self.session.get(url, timeout=5)
|
|
response.encoding = 'euc-kr'
|
|
|
|
soup = BeautifulSoup(response.text, 'lxml')
|
|
|
|
# 금 현물 가격 파싱 (사이트 구조에 따라 조정 필요)
|
|
price_elem = soup.select_one('table tr:nth-of-type(2) td:nth-of-type(2)')
|
|
if price_elem:
|
|
price_text = price_elem.text.strip().replace(',', '').replace('원', '')
|
|
return float(price_text)
|
|
except Exception as e:
|
|
print(f"KRX 금 가격 수집 실패: {e}")
|
|
|
|
return None
|
|
|
|
def fetch_all(self) -> Dict[str, Dict]:
|
|
"""모든 자산 가격 수집"""
|
|
print("📊 데이터 수집 시작...")
|
|
|
|
# 개별 자산 수집
|
|
xau_usd = self.fetch_investing_com("XAU/USD")
|
|
xau_cny = self.fetch_investing_com("XAU/CNY")
|
|
xau_gbp = self.fetch_investing_com("XAU/GBP")
|
|
usd_dxy = self.fetch_investing_com("USD/DXY")
|
|
usd_krw = self.fetch_usd_krw()
|
|
btc_usd = self.fetch_binance()
|
|
btc_krw = self.fetch_upbit()
|
|
krx_gold = self.fetch_krx_gold()
|
|
|
|
# XAU/KRW 계산 (트로이온스 -> 그램당 원화)
|
|
xau_krw = None
|
|
if xau_usd and usd_krw:
|
|
xau_krw = round((xau_usd / 31.1034768) * usd_krw, 0)
|
|
|
|
results = {
|
|
"XAU/USD": {"가격": xau_usd, "단위": "USD/oz"},
|
|
"XAU/CNY": {"가격": xau_cny, "단위": "CNY/oz"},
|
|
"XAU/GBP": {"가격": xau_gbp, "단위": "GBP/oz"},
|
|
"USD/DXY": {"가격": usd_dxy, "단위": "Index"},
|
|
"USD/KRW": {"가격": usd_krw, "단위": "KRW"},
|
|
"BTC/USD": {"가격": btc_usd, "단위": "USDT"},
|
|
"BTC/KRW": {"가격": btc_krw, "단위": "KRW"},
|
|
"KRX/GLD": {"가격": krx_gold, "단위": "KRW/g"},
|
|
"XAU/KRW": {"가격": xau_krw, "단위": "KRW/g"},
|
|
}
|
|
|
|
print(f"✅ 데이터 수집 완료 (성공: {sum(1 for v in results.values() if v['가격'])}/9)")
|
|
return results
|
|
|
|
# 전역 인스턴스
|
|
fetcher = DataFetcher()
|