windpacer/AssetPilot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Very Good Web Server Work

Browse Source
This commit is contained in:
Wind
2026-02-13 18:40:48 +09:00
parent 18fa480c84
commit d4f1ca87ab
12 changed files with 1369 additions and 1352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

255
asset_pilot_docker/app/fetcher.py
View File

@@ -1,113 +1,200 @@
import requests
import httpx
import asyncio
import re
from typing import Dict, Optional
import time
from datetime import datetime
from typing import Dict, Optional
from sqlalchemy.orm import Session
from sqlalchemy import update
# 프로젝트 구조에 따라 .models 또는 models에서 Asset을 가져옵니다.
try:
from .models import Asset
except ImportError:
from models import Asset
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'
})
# 인베스팅닷컴은 헤더가 없으면 403 에러를 뱉습니다. 브라우저와 동일하게 설정합니다.
self.headers = {
'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',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'Accept-Language': 'ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7',
'Cache-Control': 'no-cache',
'Pragma': 'no-cache'
}
self.daily_closing_prices = {}
def fetch_investing_com(self, asset_code: str) -> Optional[float]:
"""인베스팅닷컴 (윈도우 앱 방식 정규식 적용)"""
async def fetch_google_finance(self, client: httpx.AsyncClient, asset_code: str) -> Optional[float]:
"""인베스팅보다 빠르고 야후보다 안정적인 구글 파이낸스 우회"""
try:
url = f"https://www.investing.com/currencies/{asset_code.lower().replace('/', '-')}"
# 구글 파이낸스 환율/지수 URL
symbol = "USD-KRW" if asset_code == "USD/KRW" else "INDEXDXY:CURRENCY" if asset_code == "USD/DXY" else None
if not symbol: return None
url = f"https://www.google.com/finance/quote/{symbol}"
# 구글은 헤더만 있으면 응답 속도가 정말 빠릅니다.
res = await client.get(url, timeout=5)
if res.status_code != 200: return None
# 구글 특유의 가격 클래스 추출 (정규식)
m = re.search(r'data-last-price="([\d,.]+)"', res.text)
if m:
return float(m.group(1).replace(",", ""))
except Exception as e:
print(f"⚠️ Google Finance 에러 ({asset_code}): {e}")
return None
async def fetch_investing_com(self, client: httpx.AsyncClient, asset_code: str) -> Optional[float]:
"""인베스팅닷컴 비동기 수집 (USD/KRW 포함 전용)"""
try:
# 자산별 URL 매핑
if asset_code == "USD/DXY":
url = "https://www.investing.com/indices/usdollar"
# elif asset_code == "USD/KRW":
# url = "https://kr.investing.com/currencies/usd-krw"
else:
url = f"https://www.investing.com/currencies/{asset_code.lower().replace('/', '-')}"
# allow_redirects를 True로 하여 주소 변경에 대응
response = self.session.get(url, timeout=10, allow_redirects=True)
# 인베스팅은 쿠키와 리다이렉트가 중요하므로 follow_redirects 사용
response = await client.get(url, timeout=15, follow_redirects=True)
if response.status_code != 200:
print(f"⚠️ Investing 응답 에러 ({asset_code}): {response.status_code}")
return None
html = response.text
# 윈도우에서 가장 잘 되던 패턴 순서대로 시도
# 인베스팅닷컴의 다양한 HTML 구조에 대응하는 정규식 (우선순위 순)
patterns = [
r'data-test="instrument-price-last">([\d,.]+)<',
r'last_last">([\d,.]+)<',
r'instrument-price-last">([\d,.]+)<'
r'data-test="instrument-price-last">([\d,.]+)<', # 최신 메인 패턴
r'last_last">([\d,.]+)<', # 구형/세부 페이지 패턴
r'instrument-price-last">([\d,.]+)<', # 클래식 패턴
r'class="[^"]*text-2xl[^"]*">([\d,.]+)<' # 비상용 패턴
]
for pattern in patterns:
p = re.search(pattern, html)
if p:
return float(p.group(1).replace(',', ''))
val_str = p.group(1).replace(',', '')
return float(val_str)
print(f"⚠️ Investing 패턴 매칭 실패 ({asset_code})")
except Exception as e:
print(f"⚠️ Investing 수집 실패 ({asset_code}): {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"
async def fetch_binance(self, client: httpx.AsyncClient) -> Optional[float]:
"""바이낸스 BTC/USDT"""
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
url = "https://api.binance.com/api/v3/ticker/price?symbol=BTCUSD"
res = await client.get(url, timeout=5)
return float(res.json()["price"])
except: return None
def fetch_upbit(self) -> Optional[float]:
"""업비트 BTC/KRW (보내주신 윈도우 코드 로직)"""
url = "https://api.upbit.com/v1/ticker"
async def fetch_upbit(self, client: httpx.AsyncClient) -> Optional[float]:
"""업비트 BTC/KRW"""
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
url = "https://api.upbit.com/v1/ticker?markets=KRW-BTC"
res = await client.get(url, timeout=5)
return float(res.json()[0]["trade_price"])
except: 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]:
"""금 시세 (네이버 금융 모바일)"""
async def fetch_krx_gold(self, client: httpx.AsyncClient) -> 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
res = await client.get(url, timeout=5)
m = re.search(r'\"closePrice\":\"([\d,.]+)\"', res.text)
if m:
return float(m.group(1).replace(",", ""))
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
async def update_closing_prices(self, db: Session):
"""매일 아침 기준가를 스냅샷 찍어 메모리에 저장"""
results = await self.update_realtime_prices(db)
for symbol, price in results.items():
if price:
self.daily_closing_prices[symbol] = price
print(f"📌 [기준가 업데이트] 완료: {self.daily_closing_prices}")
async def update_realtime_prices(self, db: Session) -> Dict:
"""[핵심] 비동기 수집 후 DB 즉시 업데이트"""
start_time = time.time()
async with httpx.AsyncClient(headers=self.headers, follow_redirects=True) as client:
# 1. 병렬 수집 태스크 정의 (USD/KRW도 인베스팅닷컴 함수 사용)
tasks = {
"XAU/USD": self.fetch_investing_com(client, "XAU/USD"),
"XAU/CNY": self.fetch_investing_com(client, "XAU/CNY"),
"XAU/GBP": self.fetch_investing_com(client, "XAU/GBP"),
"USD/DXY": self.fetch_investing_com(client, "USD/DXY"),
"USD/KRW": self.fetch_google_finance(client, "USD/KRW"),
"BTC/USD": self.fetch_binance(client),
"BTC/KRW": self.fetch_upbit(client),
"KRX/GLD": self.fetch_krx_gold(client),
}
keys = list(tasks.keys())
values = await asyncio.gather(*tasks.values(), return_exceptions=True)
# 결과 가공
raw_results = {}
for i, val in enumerate(values):
symbol = keys[i]
if isinstance(val, (int, float)):
raw_results[symbol] = val
else:
raw_results[symbol] = None
print(f"❌ {symbol} 수집 실패: {val}")
# [수정 구간] 2. XAU/KRW 계산 및 프리미엄용 전일종가 매핑
if raw_results.get("XAU/USD") and raw_results.get("USD/KRW"):
# (현재가) 국제 금 시세 실시간 계산
raw_results["XAU/KRW"] = round((raw_results["XAU/USD"] / 31.1034768) * raw_results["USD/KRW"], 0)
# [추가] XAU/KRW의 '전일종가' 필드에 'KRX/GLD 현재가'를 강제로 주입
# 이렇게 하면 화면에서 (국제계산가 - 국내현물가)가 실시간 변동액으로 표시됨
if raw_results.get("KRX/GLD"):
from app.models import UserAsset, Asset
asset_xau = db.query(Asset).filter(Asset.symbol == "XAU/KRW").first()
if asset_xau:
db.execute(
update(UserAsset)
.where(UserAsset.asset_id == asset_xau.id)
.values(previous_close=raw_results["KRX/GLD"])
)
# 2) [핵심 추가] 메모리에 저장된 기준가도 실시간 KRX 가격으로 갱신!
# 그래야 밑에 있는 '3. DB 업데이트 수행' 루프에서 state(up/down)가 실시간으로 계산됩니다.
self.daily_closing_prices["XAU/KRW"] = raw_results["KRX/GLD"]
else:
raw_results["XAU/KRW"] = None
# 3. DB 업데이트 수행
for symbol, price in raw_results.items():
if price is not None:
state = "stable"
if symbol in self.daily_closing_prices:
closing = self.daily_closing_prices[symbol]
if price > closing: state = "up"
elif price < closing: state = "down"
# SQL 실행
db.execute(
update(Asset)
.where(Asset.symbol == symbol)
.values(
current_price=price,
price_state=state,
last_updated=datetime.now()
)
)
db.commit()
print(f"✅ [{datetime.now().strftime('%H:%M:%S')}] 수집 및 DB 저장 완료 ({time.time()-start_time:.2f}s)")
return raw_results
# 싱글톤 인스턴스 생성
fetcher = DataFetcher()

142
asset_pilot_docker/app/fetcher.py.claude
View File

@@ -1,142 +0,0 @@
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()

6
asset_pilot_docker/app/models.py
View File

@@ -15,6 +15,12 @@ class Asset(Base):
category = Column(String(50)) # 귀금속, 암호화폐, 환율 등
created_at = Column(DateTime, default=datetime.utcnow)
# --- 새로 추가한 실시간 데이터 컬럼 ---
current_price = Column(Float) # 실시간 현재가
price_state = Column(String(20), default="stable") # up, down, stable
last_updated = Column(DateTime) # 마지막 수집 시각
# ------------------------------------
# 관계
user_assets = relationship("UserAsset", back_populates="asset")
price_history = relationship("PriceHistory", back_populates="asset")