113 lines
4.8 KiB
Python
113 lines
4.8 KiB
Python
import requests
|
|
import re
|
|
from typing import Dict, Optional
|
|
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 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36'
|
|
})
|
|
|
|
def fetch_investing_com(self, asset_code: str) -> Optional[float]:
|
|
"""인베스팅닷컴 (윈도우 앱 방식 정규식 적용)"""
|
|
try:
|
|
url = f"https://www.investing.com/currencies/{asset_code.lower().replace('/', '-')}"
|
|
if asset_code == "USD/DXY":
|
|
url = "https://www.investing.com/indices/usdollar"
|
|
|
|
# allow_redirects를 True로 하여 주소 변경에 대응
|
|
response = self.session.get(url, timeout=10, allow_redirects=True)
|
|
html = response.text
|
|
|
|
# 윈도우에서 가장 잘 되던 패턴 순서대로 시도
|
|
patterns = [
|
|
r'data-test="instrument-price-last">([\d,.]+)<',
|
|
r'last_last">([\d,.]+)<',
|
|
r'instrument-price-last">([\d,.]+)<'
|
|
]
|
|
for pattern in patterns:
|
|
p = re.search(pattern, html)
|
|
if p:
|
|
return float(p.group(1).replace(',', ''))
|
|
except Exception as e:
|
|
print(f"⚠️ Investing 수집 실패 ({asset_code}): {e}")
|
|
return None
|
|
|
|
def fetch_binance(self) -> Optional[float]:
|
|
"""바이낸스 BTC/USDT (보내주신 윈도우 코드 로직)"""
|
|
url = "https://api.binance.com/api/v3/ticker/price"
|
|
try:
|
|
response = requests.get(url, params={"symbol": "BTCUSDT"}, timeout=5)
|
|
response.raise_for_status()
|
|
return float(response.json()["price"])
|
|
except Exception as e:
|
|
print(f"❌ Binance API 실패: {e}")
|
|
return None
|
|
|
|
def fetch_upbit(self) -> Optional[float]:
|
|
"""업비트 BTC/KRW (보내주신 윈도우 코드 로직)"""
|
|
url = "https://api.upbit.com/v1/ticker"
|
|
try:
|
|
response = requests.get(url, params={"markets": "KRW-BTC"}, timeout=5)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return float(data[0]["trade_price"]) if data else None
|
|
except Exception as e:
|
|
print(f"❌ Upbit API 실패: {e}")
|
|
return None
|
|
|
|
def fetch_usd_krw(self) -> Optional[float]:
|
|
"""USD/KRW 환율 (DNS 에러 방지 이중화)"""
|
|
# 방법 1: 두나무 CDN (원래 주소)
|
|
try:
|
|
url = "https://quotation-api-cdn.dunamu.com/v1/forex/recent?codes=FRX.KRWUSD"
|
|
res = requests.get(url, timeout=3)
|
|
if res.status_code == 200:
|
|
return float(res.json()[0]["basePrice"])
|
|
except:
|
|
pass # 실패하면 바로 인베스팅닷컴으로 전환
|
|
|
|
# 방법 2: 인베스팅닷컴에서 환율 가져오기 (가장 확실한 백업)
|
|
return self.fetch_investing_com("USD/KRW")
|
|
|
|
def fetch_krx_gold(self) -> Optional[float]:
|
|
"""금 시세 (네이버 금융 모바일)"""
|
|
try:
|
|
url = "https://m.stock.naver.com/marketindex/metals/M04020000"
|
|
res = requests.get(url, timeout=5)
|
|
m = re.search(r'\"closePrice\":\"([\d,]+)\"', res.text)
|
|
return float(m.group(1).replace(",", "")) if m else None
|
|
except:
|
|
return None
|
|
|
|
def fetch_all(self) -> Dict[str, Dict]:
|
|
print(f"📊 [{time.strftime('%H:%M:%S')}] 수집 시작...")
|
|
|
|
# 1. 환율 먼저 수집 (계산의 핵심)
|
|
usd_krw = self.fetch_usd_krw()
|
|
|
|
# 2. 나머지 자산 수집
|
|
results = {
|
|
"XAU/USD": {"가격": self.fetch_investing_com("XAU/USD"), "단위": "USD/oz"},
|
|
"XAU/CNY": {"가격": self.fetch_investing_com("XAU/CNY"), "단위": "CNY/oz"},
|
|
"XAU/GBP": {"가격": self.fetch_investing_com("XAU/GBP"), "단위": "GBP/oz"},
|
|
"USD/DXY": {"가격": self.fetch_investing_com("USD/DXY"), "단위": "Index"},
|
|
"USD/KRW": {"가격": usd_krw, "단위": "KRW"},
|
|
"BTC/USD": {"가격": self.fetch_binance(), "단위": "USDT"},
|
|
"BTC/KRW": {"가격": self.fetch_upbit(), "단위": "KRW"},
|
|
"KRX/GLD": {"가격": self.fetch_krx_gold(), "단위": "KRW/g"},
|
|
}
|
|
|
|
# 3. XAU/KRW 계산
|
|
xau_krw = None
|
|
if results["XAU/USD"]["가격"] and usd_krw:
|
|
xau_krw = round((results["XAU/USD"]["가격"] / 31.1034768) * usd_krw, 0)
|
|
results["XAU/KRW"] = {"가격": xau_krw, "단위": "KRW/g"}
|
|
|
|
success_count = sum(1 for v in results.values() if v['가격'] is not None)
|
|
print(f"✅ 수집 완료 (성공: {success_count}/9)")
|
|
return results
|
|
|
|
fetcher = DataFetcher() |