2025.11 자산 앱 프로젝트 진행 현황

작성자:

지난 11월은 자산 앱의 핵심 기능 개발에 집중한 한 달이었습니다. 이번 달 개발 범위는 특히 실시간 자산 시세 데이터 수집 및 화면 반영 안정화에 초점을 맞췄습니다. 국내·해외 주식부터 가상자산까지 실제 금융 시장의 변동 데이터를 끊김 없이 취합하고 화면에 제공하기 위해 전체 데이터 처리 흐름을 재설계하고 구현했습니다.

프로젝트 핵심 요약

구분주요 진행 사항
실시간 주식/해외주식 시세 연동한국투자증권 Open API 기반 REST 조회 방식 적용
실시간 가상자산 시세 연동Upbit Open API 연동 및 병렬 처리 구조 적용
시세 배치 운영 방식 변경1분 주기 반복 수집 + DB 최신가 갱신 구조 확정
UI 연동 구조 안정화외부 API 직접 호출 → 내부 DB 조회 방식으로 전환

전체 데이터 처리 흐름 구조

본 프로젝트의 핵심 데이터 파이프라인은 아래와 같습니다.

[국내, 해외주식, 가산자산 시세 조회*]
                 ↓
[1분 단위 Batch 수집 및 가격 정규화]
                 ↓
         [DB asset 갱신]
                 ↓
          [UI 화면 조회]
  • 국내 및 해외주식 시세: 한국투자증권 Open API
  • 가상자산 시세: Upbit Open API

이 구조는 실시간 API 호출의 불안정성, API rate-limit 초과, 트래픽 증가에 따른 장애 가능성을 최소화하며, 사용자는 언제든 접속만 하면 최신 가격을 확인하는 방식입니다.

실시간 주식·해외주식 시세 연동 (한국투자증권 Open API)

연동 목적

  • 국내 및 미국/해외 시장 주식 데이터를 안정적으로 수집하기 위함입니다.
  • 기존 테스트 단계에서는 단일 API 조회 방식이었으나, 11월부터는 배치 자동화 및 티커 리스트 관리 방식을 적용하여 처리량을 크게 개선했습니다.

주요 구현 포인트

  • 토큰 인증(OAuth2) 자동 재발급 및 캐싱
  • 종목 단위 시세 조회 실패 시 Exponential Backoff 재시도 적용
  • rate-limit 초과 방지(0.05~0.1s 단위 지연 로직 삽입)

💡 UI에서 실시간으로 보이는 가격은 Live API 직접 호출이 아닌 최신 DB asset 테이블 기반입니다.

# 단일 종목 현재가 조회
def _get_single_price(self, symbol: str, token: str) -> Optional[Dict[str, Any]]:
 
        url = f"{self.BASE_URL}/uapi/domestic-stock/v1/quotations/inquire-price"
        
        headers = {
            "Content-Type": "application/json; charset=UTF-8",
            "authorization": f"Bearer {token}",
            "appkey": self.app_key,
            "appsecret": self.app_secret,
            "tr_id": "FHKST01010100",
            "custtype": "P"
        }
        
        params = {
            "FID_COND_MRKT_DIV_CODE": "J",
            "FID_INPUT_ISCD": symbol
        }
        
        response = self._retry_request("GET", url, headers=headers, params=params)
        
        if response and response.get("output"):
            output = response["output"]
            
            def safe_int(value, default=0):
                try:
                    return int(value) if value else default
                except (ValueError, TypeError):
                    return default
            
            def safe_float(value, default=0.0):
                try:
                    return float(value) if value else default
                except (ValueError, TypeError):
                    return default
            
            return {
                "code": symbol,
                "price": safe_int(output.get("stck_prpr", "0")),
                "change": safe_int(output.get("prdy_vrss", "0")),
                "change_rate": safe_float(output.get("prdy_ctrt", "0.0")),
                "timestamp": datetime.now()
            }
        
        return None
    
# 단일 해외 종목 현재가 조회
def _get_single_price(self, symbol: str, exchange: str, token: str) -> Optional[Dict[str, Any]]:
        url = f"{self.BASE_URL}/uapi/overseas-price/v1/quotations/price-detail"
        
        # 거래소 코드 - 원본 그대로 사용 (매핑 불필요)
        excd = exchange
        
        headers = {
            "Content-Type": "application/json; charset=UTF-8",
            "authorization": f"Bearer {token}",
            "appkey": self.app_key,
            "appsecret": self.app_secret,
            "tr_id": "HHDFS76200200",
            "custtype": "P"
        }
        
        params = {
            "AUTH": "",
            "EXCD": excd,
            "SYMB": symbol
        }
        
        response = self._retry_request("GET", url, headers=headers, params=params)
        
        if response and response.get("output"):
            output = response["output"]
            
            def safe_float(value, default=0.0):
                try:
                    return float(value) if value else default
                except (ValueError, TypeError):
                    return default
            
            # 현재가
            current_price = safe_float(output.get("last", "0"))
            # 전일종가
            prev_close = safe_float(output.get("base", "0"))
            # 전일대비
            price_diff = current_price - prev_close if current_price and prev_close else 0
            # 전일대비율
            price_diff_rate = (price_diff / prev_close * 100) if prev_close else 0
            
            # 원환산 현재가
            krw_current_price = safe_float(output.get("t_xprc", "0"))
            
            # 원환산 전일대비 부호 (1=상한, 2=상승, 3=보합, 4=하한, 5=하락)
            krw_sign = output.get("t_xsgn", "3")
            
            # 원환산 전일대비 (절댓값)
            krw_price_diff_abs = safe_float(output.get("t_xdif", "0"))
            # 부호 적용
            if krw_sign in ["4", "5"]:
                krw_price_diff = -krw_price_diff_abs
            else:
                krw_price_diff = krw_price_diff_abs
            
            # 원환산 전일대비율
            krw_price_diff_rate = safe_float(output.get("t_xrat", "0"))
            
            return {
                "symbol": symbol,
                "exchange": exchange,
                "current_price": current_price,
                "price_diff": price_diff,
                "price_diff_rate": price_diff_rate,
                "krw_current_price": krw_current_price,
                "krw_price_diff": krw_price_diff,
                "krw_price_diff_rate": krw_price_diff_rate,
                "timestamp": datetime.now()
            }
        
        return None

실시간 가상자산 시세 연동 (Upbit Open API)

기술 적용 사항

  • KRW 기준 마켓(KRW-BTC, KRW-ETH) 일괄 호출
  • 단일 티커 외 복수 마켓 동시 조회 방식 적용
  • API 응답 실패/데이터 누락 대비 리트라이 로직 내장

적용 효과

항목개선 내용
API 비용 절감다중 종목 단일 호출 기반
성능 최적화REST 요청 최소화 + 1분 배치 고정
데이터 신뢰도 향상빈 응답/오류 시 저장 생략 및 로그 축적
def get_prices(self, tickers: List[str]) -> Dict[str, Dict[str, Any]]:
        BASE_URL = "https://api.upbit.com/v1"

        if not tickers:
            return {}
        
        # Upbit API는 "KRW-BTC" 형식 요구
        # 티커에 "KRW-" prefix가 없으면 추가
        markets = [t if t.startswith("KRW-") else f"KRW-{t}" for t in tickers]
        
        # Upbit은 한 번에 여러 종목 조회 가능
        url = f"{self.BASE_URL}/ticker"
        params = {"markets": ",".join(markets)}
        logger.info(f"[Upbit] Requesting prices: url={url}, params={params}")
        
        response = self._retry_request("GET", url, params=params)
        
        results = {}
        if response and isinstance(response, list):
            for item in response:
                ticker = item.get("market")
                if ticker:
                    results[ticker] = {
                        "ticker": ticker,
                        "price": float(item.get("trade_price", 0)),
                        "change": float(item.get("signed_change_price", 0)),
                        "change_rate": float(item.get("signed_change_rate", 0)),
                        "timestamp": datetime.now()
                    }
        
        return results

1분 단위 시세 배치 구조 설계 및 구현

자산 앱의 목표는 완전한 실시간 운영이 아닌, 실시간에 가장 가까운 안정적 시세 제공입니다. 이를 위해 전체 시세 수집 주기를 1분으로 고정했습니다.

배치 주기

  • 매 60초 단위
  • 주식 + 해외주식 + 가상자산 동시 수집
  • 수집 후 DB(asset)에 가격 업데이트

장점

특징설명
장애 격리외부 API 지연/오류 발생 시 UI는 DB 기준으로 표시
서버 부하 최소화사용자 조회량 증가와 무관한 데이터 처리
확장 용이환율/지수/ETF 등 향후 추가 데이터 편리 연결

UI 반영 방식: 외부 호출 차단 및 DB 기반 렌더링

- 기존 방식: 사용자 화면 조회 시 외부 API 직접 호출
+ 개선 방식: 화면은 asset 테이블에서 최신 값만 조회

효과

  • 외부 API 장애 시에도 UI는 정상 동작
  • 가격 표시 속도 체감 개선
  • 다중 사용자가 접속해도 퍼포먼스 저하 없음
포트폴리오 화면 자산의 실시간 시세

향후 개선 계획

  • 차트 기반 자산 히스토리 분석 화면 추가 예정
  • 히스토리 분석을 위핸 일자별 스냅샷 기능 구현

참고

댓글 남기기