1f0dcb1ddb
- Frontend: Next.js 14 with TypeScript - Backend: FastAPI with SQLAlchemy - Agent: Carmodoo sync agent - Deployment: Docker Compose based staging/production setup - Scripts: Automated deployment with rollback support 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
306 lines
9.5 KiB
Python
306 lines
9.5 KiB
Python
"""
|
|
Exchange Rate Service - 한국수출입은행 API 연동
|
|
API 문서: https://www.koreaexim.go.kr/ir/HPHKIR020M01?apino=2&viewtype=C
|
|
"""
|
|
import httpx
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, List
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models.exchange_rate import ExchangeRate, ExchangeRateHistory
|
|
|
|
|
|
# 한국수출입은행 API 설정
|
|
KOREAEXIM_API_URL = "https://oapi.koreaexim.go.kr/site/program/financial/exchangeJSON"
|
|
KOREAEXIM_API_KEY = os.getenv("KOREAEXIM_API_KEY", "rOzKaATDEinF9luHla1wVTosjWribjKL")
|
|
|
|
# 지원 통화 목록
|
|
SUPPORTED_CURRENCIES = {
|
|
"USD": {"name_ko": "미국 달러", "name_en": "US Dollar", "symbol": "$"},
|
|
"MNT": {"name_ko": "몽골 투그릭", "name_en": "Mongolian Tugrik", "symbol": "₮"},
|
|
"RUB": {"name_ko": "러시아 루블", "name_en": "Russian Ruble", "symbol": "₽"},
|
|
"CNY": {"name_ko": "중국 위안", "name_en": "Chinese Yuan", "symbol": "¥"},
|
|
"JPY": {"name_ko": "일본 엔", "name_en": "Japanese Yen", "symbol": "¥"},
|
|
"EUR": {"name_ko": "유로", "name_en": "Euro", "symbol": "€"},
|
|
}
|
|
|
|
# 기본 환율 (API 실패 시 사용, 2024년 12월 기준)
|
|
DEFAULT_RATES = {
|
|
"USD": 1450.0,
|
|
"MNT": 0.42, # 1 MNT = 0.42 KRW
|
|
"RUB": 14.0,
|
|
"CNY": 198.0,
|
|
"JPY": 9.5, # 100엔 기준이면 950
|
|
"EUR": 1510.0,
|
|
}
|
|
|
|
|
|
async def fetch_rates_from_koreaexim(search_date: Optional[str] = None) -> Optional[List[Dict]]:
|
|
"""
|
|
한국수출입은행 API에서 환율 정보 조회
|
|
|
|
Args:
|
|
search_date: 조회일자 (YYYYMMDD 형식), 없으면 오늘
|
|
|
|
Returns:
|
|
환율 데이터 리스트 또는 None
|
|
"""
|
|
if not KOREAEXIM_API_KEY:
|
|
print("Warning: KOREAEXIM_API_KEY not set, using fallback rates")
|
|
return None
|
|
|
|
if not search_date:
|
|
search_date = datetime.now().strftime("%Y%m%d")
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
KOREAEXIM_API_URL,
|
|
params={
|
|
"authkey": KOREAEXIM_API_KEY,
|
|
"searchdate": search_date,
|
|
"data": "AP01" # 환율 데이터
|
|
},
|
|
timeout=15.0
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
# API 결과 코드 확인
|
|
if isinstance(data, list) and len(data) > 0:
|
|
return data
|
|
else:
|
|
print(f"Korea Exim API returned empty data for date {search_date}")
|
|
# 주말/공휴일이면 이전 영업일 데이터 조회
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f"Failed to fetch from Korea Exim API: {e}")
|
|
|
|
return None
|
|
|
|
|
|
def parse_koreaexim_response(data: List[Dict]) -> Dict[str, Dict]:
|
|
"""
|
|
한국수출입은행 API 응답 파싱
|
|
|
|
Response format:
|
|
{
|
|
"result": 1,
|
|
"cur_unit": "USD",
|
|
"cur_nm": "미국 달러",
|
|
"ttb": "1,438.71", # 전신환(송금) 받을때
|
|
"tts": "1,467.28", # 전신환(송금) 보낼때
|
|
"deal_bas_r": "1,452.99", # 매매 기준율
|
|
"bkpr": "1,452", # 장부가격
|
|
...
|
|
}
|
|
"""
|
|
parsed = {}
|
|
|
|
for item in data:
|
|
try:
|
|
cur_unit = item.get("cur_unit", "").replace("(100)", "").strip()
|
|
|
|
if cur_unit not in SUPPORTED_CURRENCIES:
|
|
continue
|
|
|
|
# 쉼표 제거 후 숫자 변환
|
|
deal_base_rate = float(item.get("deal_bas_r", "0").replace(",", ""))
|
|
ttb_rate = float(item.get("ttb", "0").replace(",", ""))
|
|
tts_rate = float(item.get("tts", "0").replace(",", ""))
|
|
|
|
# 100엔 단위인 경우 (JPY(100))
|
|
if "(100)" in item.get("cur_unit", ""):
|
|
deal_base_rate /= 100
|
|
ttb_rate /= 100
|
|
tts_rate /= 100
|
|
|
|
parsed[cur_unit] = {
|
|
"currency_code": cur_unit,
|
|
"currency_name": item.get("cur_nm", SUPPORTED_CURRENCIES[cur_unit]["name_ko"]),
|
|
"deal_base_rate": deal_base_rate,
|
|
"ttb_rate": ttb_rate,
|
|
"tts_rate": tts_rate,
|
|
}
|
|
|
|
except (ValueError, KeyError) as e:
|
|
print(f"Error parsing currency {item.get('cur_unit')}: {e}")
|
|
continue
|
|
|
|
return parsed
|
|
|
|
|
|
async def update_exchange_rates(db: Session, force: bool = False) -> Dict:
|
|
"""
|
|
환율 정보 업데이트
|
|
|
|
Args:
|
|
db: DB 세션
|
|
force: 강제 업데이트 여부
|
|
|
|
Returns:
|
|
업데이트 결과
|
|
"""
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
|
|
# 오늘 이미 업데이트했는지 확인 (force가 아닌 경우)
|
|
if not force:
|
|
existing = db.query(ExchangeRate).filter(
|
|
ExchangeRate.source_date == today
|
|
).first()
|
|
|
|
if existing:
|
|
return {
|
|
"status": "skipped",
|
|
"message": f"Already updated for {today}",
|
|
"source_date": today
|
|
}
|
|
|
|
# API 호출 (오늘 데이터 시도)
|
|
api_data = await fetch_rates_from_koreaexim(today)
|
|
source_date = today
|
|
|
|
# 오늘 데이터 없으면 어제 시도 (주말/공휴일 대응)
|
|
if not api_data:
|
|
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
|
|
api_data = await fetch_rates_from_koreaexim(yesterday)
|
|
source_date = yesterday
|
|
|
|
# 그래도 없으면 기본값 사용
|
|
if not api_data:
|
|
print("Using fallback rates")
|
|
rates_data = {
|
|
code: {
|
|
"currency_code": code,
|
|
"currency_name": info["name_ko"],
|
|
"deal_base_rate": DEFAULT_RATES.get(code, 1.0),
|
|
"ttb_rate": DEFAULT_RATES.get(code, 1.0) * 0.98,
|
|
"tts_rate": DEFAULT_RATES.get(code, 1.0) * 1.02,
|
|
}
|
|
for code, info in SUPPORTED_CURRENCIES.items()
|
|
}
|
|
source = "fallback"
|
|
else:
|
|
rates_data = parse_koreaexim_response(api_data)
|
|
source = "koreaexim"
|
|
|
|
# DB에 저장/업데이트
|
|
updated_currencies = []
|
|
for code, rate_info in rates_data.items():
|
|
existing = db.query(ExchangeRate).filter(
|
|
ExchangeRate.currency_code == code
|
|
).first()
|
|
|
|
if existing:
|
|
# 기존 데이터 업데이트
|
|
old_rate = existing.deal_base_rate
|
|
existing.currency_name = rate_info["currency_name"]
|
|
existing.deal_base_rate = rate_info["deal_base_rate"]
|
|
existing.ttb_rate = rate_info["ttb_rate"]
|
|
existing.tts_rate = rate_info["tts_rate"]
|
|
existing.adjusted_rate = rate_info["deal_base_rate"] * (1 + existing.weight_percent / 100)
|
|
existing.source_date = source_date
|
|
|
|
# 변동이 있으면 히스토리 저장
|
|
if old_rate != rate_info["deal_base_rate"]:
|
|
history = ExchangeRateHistory(
|
|
currency_code=code,
|
|
deal_base_rate=rate_info["deal_base_rate"],
|
|
source_date=source_date
|
|
)
|
|
db.add(history)
|
|
else:
|
|
# 신규 데이터 추가
|
|
new_rate = ExchangeRate(
|
|
currency_code=code,
|
|
currency_name=rate_info["currency_name"],
|
|
deal_base_rate=rate_info["deal_base_rate"],
|
|
ttb_rate=rate_info["ttb_rate"],
|
|
tts_rate=rate_info["tts_rate"],
|
|
weight_percent=0.0,
|
|
adjusted_rate=rate_info["deal_base_rate"],
|
|
source_date=source_date,
|
|
is_active=True
|
|
)
|
|
db.add(new_rate)
|
|
|
|
# 히스토리 저장
|
|
history = ExchangeRateHistory(
|
|
currency_code=code,
|
|
deal_base_rate=rate_info["deal_base_rate"],
|
|
source_date=source_date
|
|
)
|
|
db.add(history)
|
|
|
|
updated_currencies.append(code)
|
|
|
|
db.commit()
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": f"Updated {len(updated_currencies)} currencies",
|
|
"currencies": updated_currencies,
|
|
"source": source,
|
|
"source_date": source_date
|
|
}
|
|
|
|
|
|
def get_exchange_rate(db: Session, currency_code: str) -> Optional[ExchangeRate]:
|
|
"""특정 통화 환율 조회"""
|
|
return db.query(ExchangeRate).filter(
|
|
ExchangeRate.currency_code == currency_code,
|
|
ExchangeRate.is_active == True
|
|
).first()
|
|
|
|
|
|
def get_all_exchange_rates(db: Session) -> List[ExchangeRate]:
|
|
"""모든 환율 조회"""
|
|
return db.query(ExchangeRate).filter(
|
|
ExchangeRate.is_active == True
|
|
).all()
|
|
|
|
|
|
def convert_krw_to_currency(db: Session, krw_amount: float, currency_code: str) -> Optional[float]:
|
|
"""
|
|
KRW를 다른 통화로 변환
|
|
|
|
Args:
|
|
db: DB 세션
|
|
krw_amount: 원화 금액
|
|
currency_code: 대상 통화 코드 (USD, MNT, RUB, CNY)
|
|
|
|
Returns:
|
|
변환된 금액 또는 None
|
|
"""
|
|
rate = get_exchange_rate(db, currency_code)
|
|
|
|
if not rate or rate.adjusted_rate <= 0:
|
|
return None
|
|
|
|
# KRW / 환율 = 외화
|
|
return krw_amount / rate.adjusted_rate
|
|
|
|
|
|
def convert_currency_to_krw(db: Session, amount: float, currency_code: str) -> Optional[float]:
|
|
"""
|
|
다른 통화를 KRW로 변환
|
|
|
|
Args:
|
|
db: DB 세션
|
|
amount: 외화 금액
|
|
currency_code: 원화 통화 코드
|
|
|
|
Returns:
|
|
KRW 금액 또는 None
|
|
"""
|
|
rate = get_exchange_rate(db, currency_code)
|
|
|
|
if not rate:
|
|
return None
|
|
|
|
# 외화 * 환율 = KRW
|
|
return amount * rate.adjusted_rate
|