이제 실시간 가격변동을 볼수 있는 기초공사가 마무리 되었습니다. 그러나 웹상에 표시 할려면 프론트엔드에서도 어떤 자산이 현재 얼마의 가격인지를 알아야 됩니다.
데이터가 외부 거래소에서 출발해 최종적으로 사용자의 프론트엔드 화면에서 예쁘게 깜빡거리려면, 이 흩어진 조각들이 어떤 순서로 맞물려 돌아가는지 ‘전체 조감도’를 머릿속에 명확히 그려야 합니다.
지금까지 만든 기능들이 어떻게 연결되는지 데이터의 경로를 한눈에 살펴보겠습니다. 이 전체 그림을 이해하면 앞으로 바이브 코딩을 할 때 내가 지금 어느 부분을 만들고 있는지 이해가 될 것입니다.
실시간 데이터 파이프 4단계
실시간 데이터 파이프라인은 다음과 같은 4번의 바통 터치를 거쳐 완성됩니다.
[외부 거래소] ──▶ [컨슈머] ──▶ [화이트보드] ──▶ [브로드캐스터] ──▶ [프론트엔드]
Finnhub 등 BaseConsumer Python dict ConnectionManager 사용자 화면
지금 1, 2번이 완성되었고 3, 4번이 필요한 단계입니다. 왜 이렇게 복잡하게 구조를 만들어야 되는지에 대해 설명하겠습니다.
내부 화이트보드 (파이썬 임시 메모리)
FastAPI 서버 안에 만들어둔 아주 단순한 ‘파이썬 전역 딕셔너리(Dictionary)’ 변수입니다. 컨슈머 기자가 0.5초나 1초마다 최신 비트코인 가격을 이 화이트보드에 덮어씁니다.
브로드캐스터 (사내 아나운서)
현재 우리 웹사이트에 접속해 있는 100명의 유저(웹소켓 파이프)를 관리합니다. 그리고 주기적으로 ‘화이트보드’를 쳐다보다가 숫자가 바뀌면, 마이크를 켜고 연결된 100명의 유저에게 동시에 쫙 뿌려줍니다(Broadcast).
“왜 컨슈머가 프론트엔드에 직접 주면 안 되나요?”
란 질문이 바로 떠오릅니다. 그리고 자연스럽게
“어차피 둘 다 백엔드(FastAPI) 안에 있는데, 왜 굳이 화이트보드랑 브로드캐스터를 거치나요?”
라고 생각할수도 있습니다.
만약 방송을 중계하는 기자가 (컨슈머)가 유저 100명에게 일일이 알람을 보내서 가격을 알려줘야 한다고 상상해 보세요.
현장 기자는 외부 거래소 눈치 보랴, 끊기면 재연결하랴 정신이 없는데 유저 응대까지 하다가 결국 과부하로 서버가 쓰러지고 맙니다.
그래서 ‘데이터를 수집하는 역할(컨슈머)’과 ‘유저에게 뿌려주는 역할(브로드캐스터)’을 철저하게 분리하고, 그 사이에 ‘화이트보드’를 두는 것입니다. 현장 기자가 잠시 기절(연결 끊김)하더라도, 브로드캐스터는 그냥 화이트보드에 마지막으로 적힌 가격을 띄워두면 되니까 웹사이트가 에러를 뿜으며 터지는 대참사를 막을 수 있습니다.
💡 작가의 팁: 나중에 서비스가 커지고 방문자가 많아지면, 이 단순한 화이트보드 대신 ‘Redis(레디스)’ 라는 초고속 전문 메모장을 도입하게 됩니다.
🤖 실전! 실시간 가격 임시저장 및 관리(2-14)
🧠 이번 실습의 목적
개별적인 컨슈머에서 받은 자산별 실시간 가격을 임시로 저장하는 일명 “화이트 보드” 즉 ‘파이썬 전역 딕셔너리(Dictionary)’ 를 만들어 컨슈머가 이상이 있어도 재시도 할 여유가 있도록 임시 저장 공간을 만드는 과정입니다.
✨ 사용자의 의도(Vibe)
핀허브, 바이낸스 컨슈머에서 받은 실시간 가격을 임시로 저장하고 빈상자(딕셔너리)를 만들고 이를 관리하고 방송하여 프론트엔드와 연결하는 브로드캐스터를 만드는 계획을 수립하고 실행을 승인후 해줘
파일명
💻 AI가 생성한 코드 (Syntax)
- 계획 확인후 승인하고 테스트 요청합니다.
- backend/app/core/state.py
from typing import Dict, Any# 시장 데이터를 보관하는 전역 화이트보드 (딕셔너리)# 구조: {"BINANCE:BTCUSDT": {"price": 65000.5, "timestamp": 123456789}}_global_market_state: Dict[str, Any] = {}
def update_state(key: str, data: Any): """ 특정 자산(key)의 실시간 데이터를 업데이트합니다. """ _global_market_state[key] = data
def get_all_state() -> Dict[str, Any]: """ 현재 보관중인 모든 시장 데이터 스냅샷을 반환합니다. (브로드캐스터가 호출하여 클라이언트에 전송 시 사용) """
return _global_market_state.copy()
- backend/app/services/websocket/manage.py
import asyncioimport loggingfrom fastapi import WebSocketfrom typing import Listfrom app.core import statelogger = logging.getLogger(__name__)class ConnectionManager:이하 생략 except asyncio.CancelledError: logger.info("브로드캐스팅 태스크가 취소/종료되었습니다.") except Exception as e: logger.error(f"브로드캐스팅 태스크 오류: {str(e)}")
- backend/app/api/v1/endpoints/ws.py(자동생성)
from fastapi import APIRouter, WebSocket, WebSocketDisconnect생략async
def websocket_endpoint(websocket: WebSocket): """ 프론트엔드 실시간 연동을 위한 메인 WebSocket 엔드포인트입니다. """ await manager.connect(websocket)생략 except WebSocketDisconnect: manager.disconnect(websocket)