FastAPI app.on_event -> lifespan 변경
This commit is contained in:
Wind
@@ -7,7 +7,7 @@ from typing import Dict, Optional
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import update
|
||||
|
||||
# 프로젝트 구조에 따라 .models 또는 models에서 Asset을 가져옵니다.
|
||||
# [중요] 순환 참조 방지를 위해 상단에서는 Asset만 가져옵니다.
|
||||
try:
|
||||
from .models import Asset
|
||||
except ImportError:
|
||||
@@ -15,7 +15,7 @@ except ImportError:
|
||||
|
||||
class DataFetcher:
|
||||
def __init__(self):
|
||||
# 인베스팅닷컴은 헤더가 없으면 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',
|
||||
@@ -26,18 +26,15 @@ class DataFetcher:
|
||||
self.daily_closing_prices = {}
|
||||
|
||||
async def fetch_google_finance(self, client: httpx.AsyncClient, asset_code: str) -> Optional[float]:
|
||||
"""인베스팅보다 빠르고 야후보다 안정적인 구글 파이낸스 우회"""
|
||||
"""구글 파이낸스 환율/지수 수집"""
|
||||
try:
|
||||
# 구글 파이낸스 환율/지수 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(",", ""))
|
||||
@@ -46,38 +43,30 @@ class DataFetcher:
|
||||
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('/', '-')}"
|
||||
|
||||
# 인베스팅은 쿠키와 리다이렉트가 중요하므로 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'class="[^"]*text-2xl[^"]*">([\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:
|
||||
val_str = p.group(1).replace(',', '')
|
||||
return float(val_str)
|
||||
return float(p.group(1).replace(',', ''))
|
||||
|
||||
print(f"⚠️ Investing 패턴 매칭 실패 ({asset_code})")
|
||||
except Exception as e:
|
||||
@@ -85,7 +74,7 @@ class DataFetcher:
|
||||
return None
|
||||
|
||||
async def fetch_binance(self, client: httpx.AsyncClient) -> Optional[float]:
|
||||
"""바이낸스 BTC/USDT"""
|
||||
"""바이낸스 BTC/USD"""
|
||||
try:
|
||||
url = "https://api.binance.com/api/v3/ticker/price?symbol=BTCUSD"
|
||||
res = await client.get(url, timeout=5)
|
||||
@@ -111,7 +100,7 @@ class DataFetcher:
|
||||
except: return None
|
||||
|
||||
async def update_closing_prices(self, db: Session):
|
||||
"""매일 아침 기준가를 스냅샷 찍어 메모리에 저장"""
|
||||
"""매일 정해진 시간에 기준가 스냅샷 업데이트"""
|
||||
results = await self.update_realtime_prices(db)
|
||||
for symbol, price in results.items():
|
||||
if price:
|
||||
@@ -119,11 +108,11 @@ class DataFetcher:
|
||||
print(f"📌 [기준가 업데이트] 완료: {self.daily_closing_prices}")
|
||||
|
||||
async def update_realtime_prices(self, db: Session) -> Dict:
|
||||
"""[핵심] 비동기 수집 후 DB 즉시 업데이트"""
|
||||
"""[핵심] 비동기 수집 후 DB 즉시 업데이트 및 특수 로직 처리"""
|
||||
start_time = time.time()
|
||||
|
||||
async with httpx.AsyncClient(headers=self.headers, follow_redirects=True) as client:
|
||||
# 1. 병렬 수집 태스크 정의 (USD/KRW도 인베스팅닷컴 함수 사용)
|
||||
# 1. 병렬 수집
|
||||
tasks = {
|
||||
"XAU/USD": self.fetch_investing_com(client, "XAU/USD"),
|
||||
"XAU/CNY": self.fetch_investing_com(client, "XAU/CNY"),
|
||||
@@ -138,63 +127,55 @@ class DataFetcher:
|
||||
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}")
|
||||
raw_results = {keys[i]: (v if isinstance(v, (int, float)) else None) for i, v in enumerate(values)}
|
||||
|
||||
# [수정 구간] 2. XAU/KRW 계산 및 프리미엄용 전일종가 매핑
|
||||
# 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"
|
||||
try:
|
||||
# [A] 기본 Asset 테이블 업데이트
|
||||
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.execute(
|
||||
update(Asset)
|
||||
.where(Asset.symbol == symbol)
|
||||
.values(current_price=price, price_state=state, last_updated=datetime.now())
|
||||
)
|
||||
)
|
||||
|
||||
db.commit()
|
||||
|
||||
# [B] XAU/KRW 프리미엄용 특수 로직 (KRX 가격을 전일종가로 세팅)
|
||||
if raw_results.get("KRX/GLD"):
|
||||
try:
|
||||
from .models import UserAsset, Asset as ModelAsset # 로컬 임포트 유지
|
||||
except ImportError:
|
||||
from models import UserAsset, Asset as ModelAsset
|
||||
|
||||
target_asset = db.query(ModelAsset).filter(ModelAsset.symbol == "XAU/KRW").first()
|
||||
if target_asset:
|
||||
db.execute(
|
||||
update(UserAsset)
|
||||
.where(UserAsset.asset_id == target_asset.id)
|
||||
.values(previous_close=raw_results["KRX/GLD"])
|
||||
)
|
||||
# 메모리 기준가도 함께 갱신하여 UI 상태 화살표 유지
|
||||
self.daily_closing_prices["XAU/KRW"] = raw_results["KRX/GLD"]
|
||||
|
||||
db.commit()
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
print(f"❌ DB 저장 중 오류 발생: {e}")
|
||||
|
||||
print(f"✅ [{datetime.now().strftime('%H:%M:%S')}] 수집 및 DB 저장 완료 ({time.time()-start_time:.2f}s)")
|
||||
return raw_results
|
||||
|
||||
# 싱글톤 인스턴스 생성
|
||||
# 싱글톤 인스턴스
|
||||
fetcher = DataFetcher()
|
||||
@@ -4,13 +4,15 @@ import asyncio
|
||||
import httpx
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict
|
||||
from contextlib import asynccontextmanager # lifespan용 추가
|
||||
|
||||
from fastapi import FastAPI, Depends, HTTPException, Request
|
||||
from fastapi.responses import HTMLResponse, StreamingResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
# [변경] 비동기용 스케줄러로 교체
|
||||
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import update, or_
|
||||
from pydantic import BaseModel
|
||||
from dotenv import load_dotenv
|
||||
|
||||
@@ -24,36 +26,6 @@ load_dotenv()
|
||||
# 데이터베이스 테이블 생성
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI(title="Asset Pilot - Orange Pi Edition", version="1.2.0")
|
||||
# 1. 현재 main.py 파일의 절대 경로를 가져옵니다.
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
# 2. Static 파일 경로 설정 (절대 경로 사용)
|
||||
static_path = os.path.join(BASE_DIR, "static")
|
||||
if os.path.exists(static_path):
|
||||
app.mount("/static", StaticFiles(directory=static_path), name="static")
|
||||
print(f"✅ Static 마운트 성공: {static_path}")
|
||||
else:
|
||||
print(f"❌ Static 폴더를 찾을 수 없습니다: {static_path}")
|
||||
|
||||
# 3. 템플릿 설정 (절대 경로 사용)
|
||||
templates_path = os.path.join(BASE_DIR, "templates")
|
||||
templates = Jinja2Templates(directory=templates_path)
|
||||
|
||||
# [변경] 비동기 스케줄러 설정
|
||||
scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
|
||||
|
||||
# 전역 상태 관리
|
||||
connected_clients = 0
|
||||
clients_lock = asyncio.Lock()
|
||||
last_alert_time = {}
|
||||
|
||||
# [신규] 시스템 상태 모니터링 변수 (Heartbeat)
|
||||
system_status = {
|
||||
"last_fetch_time": None,
|
||||
"status": "initializing"
|
||||
}
|
||||
|
||||
# ==================== Pydantic 모델 ====================
|
||||
class UserAssetUpdate(BaseModel):
|
||||
symbol: str
|
||||
@@ -64,6 +36,19 @@ class UserAssetUpdate(BaseModel):
|
||||
class AlertSettingUpdate(BaseModel):
|
||||
settings: Dict
|
||||
|
||||
# ==================== 전역 변수 및 설정 ====================
|
||||
# [변경] 비동기 스케줄러 설정
|
||||
scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
|
||||
|
||||
connected_clients = 0
|
||||
clients_lock = asyncio.Lock()
|
||||
last_alert_time = {}
|
||||
|
||||
system_status = {
|
||||
"last_fetch_time": None,
|
||||
"status": "initializing"
|
||||
}
|
||||
|
||||
# ==================== 유틸리티 함수 ====================
|
||||
async def send_telegram_msg_async(text: str):
|
||||
"""비동기 방식으로 텔레그램 메시지 전송"""
|
||||
@@ -80,7 +65,6 @@ async def send_telegram_msg_async(text: str):
|
||||
if resp.status_code != 200: print(f"❌ 텔레그램 실패: {resp.text}")
|
||||
except Exception as e: print(f"❌ 텔레그램 오류: {e}")
|
||||
|
||||
# ==================== DB 초기화 ====================
|
||||
def init_db_data():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
@@ -113,7 +97,7 @@ def init_db_data():
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
# ==================== 백그라운드 태스크 (Watchdog & 알림 통합) ====================
|
||||
# ==================== 백그라운드 태스크 ====================
|
||||
async def background_fetch():
|
||||
"""비동기 수집 루프: DB 업데이트 + Heartbeat + 알림"""
|
||||
while True:
|
||||
@@ -123,14 +107,10 @@ async def background_fetch():
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
# 1. 수집 및 DB 업데이트
|
||||
current_data = await fetcher.update_realtime_prices(db)
|
||||
|
||||
# [성공] Heartbeat 기록
|
||||
system_status["last_fetch_time"] = datetime.now()
|
||||
system_status["status"] = "healthy"
|
||||
|
||||
# 2. 알림 로직
|
||||
settings_raw = db.query(AlertSetting).all()
|
||||
sets = {s.setting_key: json.loads(s.setting_value) for s in settings_raw}
|
||||
user_assets = db.query(Asset, UserAsset).join(UserAsset).all()
|
||||
@@ -145,7 +125,6 @@ async def background_fetch():
|
||||
prev_c = float(ua.previous_close) if ua.previous_close else 0
|
||||
avg_p = float(ua.average_price) if ua.average_price else 0
|
||||
|
||||
# 급등락 체크
|
||||
if sets.get("급등락_감지") and prev_c > 0:
|
||||
change = ((price - prev_c) / prev_c) * 100
|
||||
if abs(change) >= float(sets.get("급등락_임계값", 3.0)):
|
||||
@@ -154,7 +133,6 @@ async def background_fetch():
|
||||
await send_telegram_msg_async(f"<b>[{icon}] {symbol}</b>\n현재가: {price:,.2f}\n변동률: {change:+.2f}%")
|
||||
last_alert_time[f"{symbol}_vol"] = now_ts
|
||||
|
||||
# 수익률 체크
|
||||
if sets.get("목표수익률_감지") and avg_p > 0:
|
||||
profit = ((price - avg_p) / avg_p) * 100
|
||||
if profit >= float(sets.get("목표수익률", 10.0)):
|
||||
@@ -162,7 +140,6 @@ async def background_fetch():
|
||||
await send_telegram_msg_async(f"<b>💰 수익 목표달성! ({symbol})</b>\n수익률: {profit:+.2f}%\n현재가: {price:,.2f}")
|
||||
last_alert_time[f"{symbol}_profit"] = now_ts
|
||||
|
||||
# 특정가격 감지
|
||||
if sets.get("특정가격_감지"):
|
||||
if symbol == "KRX/GLD" and price >= float(sets.get("금_목표가격", 0)):
|
||||
if now_ts - last_alert_time.get("gold_hit", 0) > 43200:
|
||||
@@ -172,7 +149,6 @@ async def background_fetch():
|
||||
if now_ts - last_alert_time.get("btc_hit", 0) > 43200:
|
||||
await send_telegram_msg_async(f"<b>₿ BTC 목표가 돌파!</b>\n현재가: {price:,.0f}원")
|
||||
last_alert_time["btc_hit"] = now_ts
|
||||
|
||||
finally:
|
||||
db.close()
|
||||
except Exception as e:
|
||||
@@ -181,28 +157,73 @@ async def background_fetch():
|
||||
|
||||
await asyncio.sleep(interval)
|
||||
|
||||
# ==================== 앱 생명주기 (AsyncIOScheduler 적용) ====================
|
||||
@app.on_event("startup")
|
||||
async def startup_event():
|
||||
# ==================== [신규] Lifespan 핸들러 ====================
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
# [Startup] 앱 시작 시 실행
|
||||
init_db_data()
|
||||
|
||||
# [변경] 7시 10분 비동기 전용 스케줄러 작업
|
||||
async def daily_job():
|
||||
print(f"🌅 [기준가 업데이트 시작] {datetime.now()}")
|
||||
# 서버 재시작 시 DB의 종가를 메모리로 복구 (fetcher에 해당 함수가 있다고 가정)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
# DB에서 기존 종가를 불러와 메모리 fetcher.daily_closing_prices 채우기
|
||||
assets = db.query(Asset, UserAsset).join(UserAsset).all()
|
||||
for a, ua in assets:
|
||||
if ua.previous_close:
|
||||
fetcher.daily_closing_prices[a.symbol] = float(ua.previous_close)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
# 1. 비트코인 전용 자정 스케줄러 (00:00:05)
|
||||
async def btc_daily_job():
|
||||
print(f"🕛 [BTC 자정 스냅샷] {datetime.now()}")
|
||||
db = SessionLocal()
|
||||
try:
|
||||
await fetcher.update_closing_prices(db)
|
||||
finally:
|
||||
db.close()
|
||||
assets = db.query(Asset).filter(Asset.symbol.like("BTC/%")).all()
|
||||
for asset in assets:
|
||||
if asset.current_price:
|
||||
db.execute(update(UserAsset).where(UserAsset.asset_id == asset.id).values(previous_close=asset.current_price))
|
||||
fetcher.daily_closing_prices[asset.symbol] = asset.current_price
|
||||
db.commit()
|
||||
finally: db.close()
|
||||
|
||||
scheduler.add_job(daily_job, 'cron', hour=7, minute=10, id='daily_snapshot')
|
||||
# 2. 전통 자산용 아침 스케줄러 (07:10:00)
|
||||
async def legacy_daily_job():
|
||||
print(f"🌅 [전통자산 아침 스냅샷] {datetime.now()}")
|
||||
db = SessionLocal()
|
||||
try:
|
||||
assets = db.query(Asset).filter(~Asset.symbol.like("BTC/%")).all()
|
||||
for asset in assets:
|
||||
if asset.current_price:
|
||||
db.execute(update(UserAsset).where(UserAsset.asset_id == asset.id).values(previous_close=asset.current_price))
|
||||
fetcher.daily_closing_prices[asset.symbol] = asset.current_price
|
||||
db.commit()
|
||||
finally: db.close()
|
||||
|
||||
scheduler.add_job(btc_daily_job, 'cron', hour=0, minute=0, second=5, id='btc_snapshot')
|
||||
scheduler.add_job(legacy_daily_job, 'cron', hour=7, minute=10, id='legacy_snapshot')
|
||||
scheduler.start()
|
||||
|
||||
asyncio.create_task(background_fetch())
|
||||
fetch_task = asyncio.create_task(background_fetch())
|
||||
|
||||
yield # 서버 운영 중
|
||||
|
||||
@app.on_event("shutdown")
|
||||
def stop_scheduler():
|
||||
# [Shutdown] 서버 종료 시 실행
|
||||
scheduler.shutdown()
|
||||
fetch_task.cancel()
|
||||
print("👋 서버가 종료되었습니다.")
|
||||
|
||||
# ==================== FastAPI 앱 선언 ====================
|
||||
app = FastAPI(title="Asset Pilot - Orange Pi Edition", version="1.2.0", lifespan=lifespan)
|
||||
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
static_path = os.path.join(BASE_DIR, "static")
|
||||
if os.path.exists(static_path):
|
||||
app.mount("/static", StaticFiles(directory=static_path), name="static")
|
||||
print(f"✅ Static 마운트 성공: {static_path}")
|
||||
|
||||
templates_path = os.path.join(BASE_DIR, "templates")
|
||||
templates = Jinja2Templates(directory=templates_path)
|
||||
|
||||
# ==================== API 엔드포인트 ====================
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
@@ -211,10 +232,7 @@ async def read_root(request: Request):
|
||||
|
||||
@app.get("/api/prices")
|
||||
async def get_prices(db: Session = Depends(get_db)):
|
||||
"""[개선] 데이터 신선도 상태와 서버 시각을 포함하여 반환"""
|
||||
assets = db.query(Asset).all()
|
||||
|
||||
# 지연 판별 (마지막 성공 후 60초 경과 시 stale)
|
||||
is_stale = False
|
||||
if system_status["last_fetch_time"]:
|
||||
if datetime.now() - system_status["last_fetch_time"] > timedelta(seconds=60):
|
||||
@@ -300,8 +318,4 @@ async def update_alert_settings(data: AlertSettingUpdate, db: Session = Depends(
|
||||
|
||||
@app.get("/health")
|
||||
async def health_check():
|
||||
return {"status": "healthy", "last_fetch": system_status["last_fetch_time"]}
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# import uvicorn
|
||||
# uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=False)
|
||||
return {"status": "healthy", "last_fetch": system_status["last_fetch_time"]}
|
||||
162
asset_pilot_docker/updateRecord.md
Normal file
162
asset_pilot_docker/updateRecord.md
Normal file
@@ -0,0 +1,162 @@
|
||||
2026-02-14 (08:46)
|
||||
|
||||
선임님, VSCode에서 멤버명(변수나 함수 이름)에 **가운데 줄(취소선, Strikethrough)**이 그어져 있는 건 **"Deprecated(사용 권장되지 않음)"**라는 뜻입니다.
|
||||
|
||||
쉽게 말해 **"이 기능은 아직 작동은 하지만, 곧 없어질 예정이거나 더 좋은 다른 기능으로 대체되었으니 웬만하면 쓰지 마라"**고 경고를 보내는 겁니다.
|
||||
|
||||
🧐 왜 가운데 줄이 생겼을까요?
|
||||
라이브러리 업데이트: 선임님이 쓰시는 라이브러리(예: SQLAlchemy, FastAPI, Pydantic 등)가 버전업 되면서, 기존 방식을 버리고 새로운 방식을 권장할 때 발생합니다.
|
||||
|
||||
보안/성능 이슈: 해당 멤버를 계속 쓰면 보안에 취약하거나 성능이 떨어질 수 있다고 판단될 때 개발자가 표시를 해둔 겁니다.
|
||||
|
||||
🛠️ 어떻게 해결하나요?
|
||||
가장 쉬운 확인 방법은 그 가운데 줄이 쳐진 이름 위에 마우스 커서를 가만히 올려보는 겁니다.
|
||||
|
||||
그러면 "@deprecated: Use 'NewFunctionName' instead" 같은 가이드 팝업이 뜰 거예요.
|
||||
|
||||
거기서 추천하는 새로운 멤버명으로 이름을 바꿔주기만 하면 줄이 사라지고 코드도 최신 트렌드에 맞게 정리가 됩니다.
|
||||
|
||||
💡 우리 코드에서 주의할 점
|
||||
혹시 아까 제가 짜드린 코드나 선임님이 작성하신 코드 중에 on_event("startup") 부분에 줄이 가 있나요?
|
||||
최신 FastAPI에서는 @app.on_event("startup") 대신 **lifespan**이라는 방식을 권장하기 때문에 취소선이 생길 수 있습니다.
|
||||
|
||||
ython
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
# [최신 방식] 앱 실행 전후를 한 번에 관리하는 lifespan 함수
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
# --- Startup (앱 시작 시 실행) ---
|
||||
init_db_data()
|
||||
|
||||
# 이원화 스케줄러 등록
|
||||
async def btc_daily_job():
|
||||
# ... (비트코인 자정 스냅샷 로직 동일) ...
|
||||
pass
|
||||
|
||||
async def legacy_daily_job():
|
||||
# ... (전통자산 아침 스냅샷 로직 동일) ...
|
||||
pass
|
||||
|
||||
scheduler.add_job(btc_daily_job, 'cron', hour=0, minute=0, second=5, id='btc_snapshot')
|
||||
scheduler.add_job(legacy_daily_job, 'cron', hour=7, minute=10, id='legacy_snapshot')
|
||||
scheduler.start()
|
||||
|
||||
fetch_task = asyncio.create_task(background_fetch())
|
||||
|
||||
yield # 앱이 실행되는 동안 여기서 대기합니다.
|
||||
|
||||
# --- Shutdown (앱 종료 시 실행) ---
|
||||
scheduler.shutdown()
|
||||
fetch_task.cancel()
|
||||
print("👋 서버가 안전하게 종료되었습니다.")
|
||||
|
||||
# FastAPI 선언 시 lifespan을 넣어줍니다.
|
||||
app = FastAPI(title="Asset Pilot", lifespan=lifespan)
|
||||
🧐 왜 이걸로 갈아타야 할까요? (취소선의 이유)
|
||||
자원 관리: lifespan 방식은 yield를 기준으로 시작과 끝이 명확하게 한 곳에 모여 있어, 나중에 DB 커넥션이나 소켓을 닫을 때 실수를 안 하게 해줍니다.
|
||||
|
||||
비동기 최적화: 이전 방식보다 비동기 태스크들의 생명주기를 더 정확하게 제어할 수 있습니다.
|
||||
|
||||
VSCode 클린업: 이제 그 지긋지긋한 **가운데 줄(취소선)**이 사라지고, 선임님의 VSCode가 다시 깨끗해질 겁니다.
|
||||
|
||||
|
||||
Gemini said
|
||||
선임님, fetcher.py 소스 코드 잘 봤습니다. 코드를 뜯어보니 데이터 수집은 되는데 왜 현재가가 안 변했는지 확실한 범인이 잡혔습니다.
|
||||
|
||||
범인은 바로 db.commit()의 위치와 XAU/KRW의 특수 로직 때문입니다.
|
||||
|
||||
🕵️ fetcher.py의 문제점 분석
|
||||
1. XAU/KRW 프리미엄 로직의 '월권' (핵심 원인)
|
||||
코드 중간에 보면 이런 로직이 있습니다:
|
||||
|
||||
Python
|
||||
# [수정 구간] 2. XAU/KRW 계산 및 프리미엄용 전일종가 매핑
|
||||
if raw_results.get("KRX/GLD"):
|
||||
# ... 중략 ...
|
||||
db.execute(update(UserAsset).where(UserAsset.asset_id == asset_xau.id).values(previous_close=raw_results["KRX/GLD"]))
|
||||
여기서 db.execute를 실행했는데, 이 작업이 완료되기도 전에 밑에서 다른 update(Asset) 루프가 돌고 있습니다. SQLAlchemy 세션 하나에서 여러 update를 비동기 루프와 섞어 쓸 때, 가끔 먼저 실행된 execute가 세션을 점유하거나 락(Lock)이 걸려 뒤의 업데이트가 씹히는 경우가 발생합니다.
|
||||
|
||||
2. db.commit()이 너무 뒤에 있음
|
||||
현재 모든 자산의 가격을 루프로 돌면서 db.execute를 예약만 해두고, **맨 마지막에 한 번만 db.commit()**을 합니다. 만약 루프 도중 단 하나의 자산이라도 DB 제약 조건에 걸리거나 에러가 나면, 그전까지 업데이트한 모든 가격이 롤백되어 반영되지 않습니다. #### 3. fetch_google_finance의 정규식 오류 가능성
|
||||
현재 구글 파이낸스 정규식이 data-last-price="([\d,.]+)"로 되어 있는데, 최근 구글 파이낸스 HTML 구조가 바뀌어서 이 속성이 안 나타날 때가 있습니다. 만약 환율(USD/KRW) 값이 None이 되면, 이를 기반으로 계산하는 XAU/KRW도 줄줄이 사탕으로 업데이트가 안 됩니다.
|
||||
|
||||
🛠️ 수정한 fetcher.py (핵심 구간)
|
||||
선임님, 아래 부분을 특히 주의해서 봐주세요. 안정성을 위해 업데이트 방식을 더 직관적으로 고쳤습니다.
|
||||
|
||||
Python
|
||||
async def update_realtime_prices(self, db: Session) -> Dict:
|
||||
start_time = time.time()
|
||||
|
||||
async with httpx.AsyncClient(headers=self.headers, follow_redirects=True) as client:
|
||||
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 = {keys[i]: (v if isinstance(v, (int, float)) else None) for i, v in enumerate(values)}
|
||||
|
||||
# [개선] 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)
|
||||
|
||||
# [개선] DB 업데이트 로직 분리 및 안정화
|
||||
try:
|
||||
for symbol, price in raw_results.items():
|
||||
if price is not None:
|
||||
# 1. 상태(Up/Down) 계산
|
||||
state = "stable"
|
||||
closing = self.daily_closing_prices.get(symbol)
|
||||
if closing:
|
||||
if price > closing: state = "up"
|
||||
elif price < closing: state = "down"
|
||||
|
||||
# 2. Asset 테이블 업데이트 (현재가)
|
||||
db.query(Asset).filter(Asset.symbol == symbol).update({
|
||||
"current_price": price,
|
||||
"price_state": state,
|
||||
"last_updated": datetime.now()
|
||||
})
|
||||
|
||||
# 3. 특수 로직 (XAU/KRW 프리미엄 계산용 기준가 변경)
|
||||
if raw_results.get("KRX/GLD"):
|
||||
asset_xau = db.query(Asset).filter(Asset.symbol == "XAU/KRW").first()
|
||||
if asset_xau:
|
||||
# UserAsset 테이블의 previous_close를 KRX 현물가로 강제 동기화
|
||||
db.query(UserAsset).filter(UserAsset.asset_id == asset_xau.id).update({
|
||||
"previous_close": raw_results["KRX/GLD"]
|
||||
})
|
||||
self.daily_closing_prices["XAU/KRW"] = raw_results["KRX/GLD"]
|
||||
|
||||
db.commit() # 모든 업데이트 완료 후 커밋
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
print(f"❌ DB 업데이트 중 롤백 발생: {e}")
|
||||
|
||||
print(f"✅ [{datetime.now().strftime('%H:%M:%S')}] 수집 및 DB 저장 완료 ({time.time()-start_time:.2f}s)")
|
||||
return raw_results
|
||||
💡 선임님을 위한 조언
|
||||
XAU/KRW 프리미엄 수식: 지금 선임님 코드는 국제 시세와 국내 시세의 차이를 보여주기 위해 previous_close 값을 실시간으로 계속 덮어쓰고 계시네요. 이 방식은 등락률 계산 로직을 우회하는 아주 영리한 방법이지만, DB에 계속 쓰기 작업을 발생시킵니다. (오렌지파이 eMMC 수명에는 조금 가혹할 수 있지만, 지금 데이터 양으로는 큰 문제 없습니다!)
|
||||
|
||||
구글 환율 체크: 만약 로그에 USD/KRW가 None으로 찍힌다면, 구글 파이낸스 정규식을 r'class="fxKb7e".*?>([\d,.]+)<' 같은 클래스 기반으로 바꿔야 할 수도 있습니다.
|
||||
|
||||
|
||||
|
||||
🔍 빠진 게 없는지 대조 결과
|
||||
패턴 매칭: patterns 리스트에 4가지 정규식을 모두 유지했습니다.
|
||||
|
||||
병렬 수집: asyncio.gather를 통해 모든 자산(8종)을 동시에 긁어옵니다.
|
||||
|
||||
XAU/KRW 로직: 환율과 국제 금값을 곱해 원화 금값을 계산하고, 국내 현물가(KRX/GLD)와 비교할 수 있도록 previous_close에 강제로 주입하는 핵심 로직 살아있습니다.
|
||||
|
||||
안전 장치: try-except-rollback 구조를 넣어 데이터 정합성을 지켰습니다.
|
||||
Reference in New Issue
Block a user