156 lines
6.3 KiB
Plaintext
156 lines
6.3 KiB
Plaintext
import requests
|
|
import re
|
|
from typing import Dict, Optional
|
|
import time
|
|
from typing import Dict, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import update
|
|
from datetime import datetime
|
|
|
|
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'
|
|
})
|
|
# 이전 가격 저장용 (색상 변경 판단에 사용 예정)
|
|
self.last_prices = {}
|
|
self.daily_closing_prices = {} # 일일 종가 저장용 (목표 수익률 감지에 사용 예정)
|
|
|
|
def update_closing_prices(self):
|
|
"""매일 07:10에 호출되어 기준가를 스냅샷 찍음"""
|
|
results = self.fetch_all()
|
|
for key, data in results.items():
|
|
if data['가격'] is not None:
|
|
self.daily_closing_prices[key] = data['가격']
|
|
print(f"📌 [기준가 업데이트] 07:10 기준가 저장 완료: {self.daily_closing_prices}")
|
|
|
|
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()
|
|
|
|
# 임시 결과 저장
|
|
raw_results = {
|
|
"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": usd_krw,
|
|
"BTC/USD": self.fetch_binance(),
|
|
"BTC/KRW": self.fetch_upbit(),
|
|
"KRX/GLD": self.fetch_krx_gold(),
|
|
}
|
|
|
|
# XAU/KRW 계산
|
|
if raw_results["XAU/USD"] and usd_krw:
|
|
raw_results["XAU/KRW"] = round((raw_results["XAU/USD"] / 31.1034768) * usd_krw, 0)
|
|
else:
|
|
raw_results["XAU/KRW"] = None
|
|
|
|
final_results = {}
|
|
units = {
|
|
"XAU/USD": "USD/oz", "XAU/CNY": "CNY/oz", "XAU/GBP": "GBP/oz",
|
|
"USD/DXY": "Index", "USD/KRW": "KRW", "BTC/USD": "USDT",
|
|
"BTC/KRW": "KRW", "KRX/GLD": "KRW/g", "XAU/KRW": "KRW/g"
|
|
}
|
|
|
|
# 상태(색상) 결정 로직
|
|
for key, price in raw_results.items():
|
|
state = "stable"
|
|
|
|
if key in self.daily_closing_prices and price is not None:
|
|
closing = self.daily_closing_prices[key]
|
|
if price > closing:
|
|
state = "up"
|
|
elif price < closing:
|
|
state = "down"
|
|
|
|
final_results[key] = {
|
|
"가격": price,
|
|
"단위": units.get(key, ""),
|
|
"상태": state # up, down, stable 중 하나 전달
|
|
}
|
|
|
|
# 다음 비교를 위해 현재 가격 저장
|
|
if price is not None:
|
|
self.last_prices[key] = price
|
|
|
|
success_count = sum(1 for v in final_results.values() if v['가격'] is not None)
|
|
print(f"✅ 수집 완료 (성공: {success_count}/9)")
|
|
return final_results
|
|
|
|
fetcher = DataFetcher() |