안녕하세요, '소프트웨어 공장'에 오신 것을 환영합니다!
지난 포스팅에서는 Python의 다양한 고급 주제들을 이론과 짧은 예제로 알아보았는데요. "코드는 눈으로 볼 때보다 직접 하나의 서비스로 엮어볼 때 진짜 내 것이 된다"는 말이 있죠. 그래서 이번에는 그동안 배운 핵심 기술들을 하나로 묶어, 실제로 동작하는 '실시간 데이터 분석 애플리케이션'을 함께 만들어보려고 합니다.
실시간 데이터를 긁어오고(Asyncio), 이를 서버로 유통하며(FastAPI), 안정적으로 저장하고(SQLAlchemy), 멋진 대시보드로 시각화(Dash)하는 전 과정을 한눈에 이해하실 수 있도록 준비했습니다. 실무 파이프라인이 어떻게 구성되는지 궁금하셨다면 이번 글을 끝까지 주목해 주세요!
📌 핵심 요약 3줄
- 기술 통합: Asyncio, FastAPI, SQLAlchemy, Dash를 연동하여 데이터 수집부터 시각화까지의 전 과정을 구현합니다.
- 비동기 최적화: aiohttp와 asyncio를 활용해 대량의 주식 데이터를 병목 현상 없이 실시간으로 수집합니다.
- 모듈화 구조: 데이터 수집(수집기), 서버(API), 저장소(DB), 화면(대시보드)으로 역할을 나누어 실무형 아키텍처를 배웁니다.
🛠️ 프로젝트 개요 및 사용 기술
우리의 목표는 주식 가격 데이터를 실시간으로 수집하고, 이를 분석하여 대시보드를 통해 시각화하는 것입니다. 각 단계에서 사용하는 기술과 그 역할은 다음과 같습니다. 업무에서 실시간 데이터를 다룰 때 가장 표준적으로 쓰이는 조합이기도 합니다.
| 단계 | 사용 기술 | 핵심 역할 | 선택 이유 |
| 1. 데이터 수집 | asyncio, aiohttp | 실시간 주식 데이터 비동기 수집 | 여러 개의 API 요청을 대기 시간 없이 동시에 처리하기 위함 |
| 2. API 서버 | FastAPI | 데이터 수집 및 제공을 위한 REST API 구축 | 빠른 처리 속도와 간결한 코드로 실시간 데이터 유통에 적합 |
| 3. 데이터베이스 | SQLAlchemy | 수집된 주가 데이터 저장 및 관리 | ORM을 통해 소규모부터 대규모 데이터까지 안정적으로 적재 |
| 4. 대시보드 | Dash, Plotly | 실시간 데이터 시각화 및 웹 UI 화면 구성 | 파이썬 코드만으로 동적이고 직관적인 웹 대시보드 구현 가능 |
1. 실시간 데이터 수집 (Asyncio & aiohttp)
첫 번째 단계는 주식 데이터를 비동기적으로 수집하는 것입니다. 서버에 무리를 주지 않으면서 여러 주식의 데이터를 동시에, 그리고 빠르게 가져오기 위해 aiohttp와 asyncio.gather를 사용합니다.
import asyncio
import json
import aiohttp
import yfinance as yf
async def fetch_stock_data(symbol):
# 실제 환경에서는 비동기 HTTP 클라이언트를 사용해 API를 호출합니다.
# 예시를 위해 yfinance 라이브러리로 종가(Close) 데이터를 가져옵니다.
ticker = yf.Ticker(symbol)
data = ticker.history(period="1d")
return {
"symbol": symbol,
"close": data["Close"].iloc[-1] if not data.empty else None,
}
async def main():
symbols = ["AAPL", "GOOGL", "AMZN"]
# 여러 심볼의 데이터를 동시에 비동기적으로 요청합니다.
responses = await asyncio.gather(
*(fetch_stock_data(symbol) for symbol in symbols)
)
for response in responses:
print(json.dumps(response, indent=2))
if __name__ == "__main__":
asyncio.run(main())
위 코드는 여러 주식 데이터를 비동기적으로 수집하는 예제입니다. 주식 데이터 API에서 실시간 정보를 받아올 때 수십, 수백 개의 요청이 겹쳐도 동기 방식처럼 멈추지 않고 효율적으로 처리할 수 있습니다.
2. REST API 서버 구축 (FastAPI)
실시간으로 수집한 데이터를 대시보드나 외부 애플리케이션에 매끄럽게 전달하려면 중심축 역할을 하는 서버가 필요합니다. FastAPI를 사용해 뚝딱 만들어보겠습니다.
from fastapi import FastAPI
app = FastAPI()
# 메모리에 최신 주가 데이터를 임시 저장하는 공간입니다.
latest_stock_data = []
@app.get("/stocks")
async def get_stocks():
"""클라이언트(대시보드 등)가 최신 데이터를 조회할 때 사용합니다."""
return latest_stock_data
@app.post("/update")
async def update_stocks(data: list):
"""데이터 수집기가 새로운 데이터를 서버에 밀어넣을 때 사용합니다."""
global latest_stock_data
latest_stock_data = data
return {"status": "updated", "count": len(data)}
이 서버는 아주 명확한 두 가지 역할을 합니다. /update 엔드포인트로 들어온 최신 데이터를 메모리에 얹고, /stocks 엔드포인트를 요청하는 대시보드 화면에 그 데이터를 그대로 던져줍니다. 데이터 유통의 허브가 되는 셈이죠.
3. 데이터베이스에 안정적으로 저장하기 (SQLAlchemy)
조회용 데이터는 메모리에 두더라도, 과거 데이터 분석이나 통계를 위해서는 데이터를 디스크에 안전하게 쌓아야 합니다. 파이썬의 대표적인 ORM인 SQLAlchemy를 활용해 가벼운 SQLite 데이터베이스에 연결해보겠습니다.
from sqlalchemy import Column, Float, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./stocks.db"
engine = create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# 데이터베이스에 생성될 주식 테이블 스키마 정의
class StockRecord(Base):
__tablename__ = "stocks"
id = Column(Integer, primary_key=True, index=True)
symbol = Column(String, index=True)
price = Column(Float)
# 스크립트 실행 시 테이블이 없으면 자동으로 생성해줍니다.
Base.metadata.create_all(bind=engine)
이렇게 StockRecord라는 클래스를 정의해두면 파이썬 객체를 다루듯 편하게 데이터를 넣고 뺄 수 있습니다. 수집된 주가 데이터를 이 테이블에 차곡차곡 쌓으면 나중에 일주일 치, 한 달 치 이동평균선을 그리기도 쉬워집니다.
4. 데이터 시각화 및 웹 대시보드 구현 (Dash & Plotly)
이제 마지막 하이라이트입니다. 서버에서 데이터를 받아와 사용자가 직관적으로 볼 수 있게 멋진 그래프로 띄워볼 시간입니다. 프론트엔드 지식이 부족해도 Dash와 Plotly를 쓰면 파이썬 코드 몇 줄로 모던한 대시보드를 만들 수 있습니다.
import dash
import pandas as pd
import plotly.express as px
from dash import dcc, html
app = dash.Dash(__name__)
# 시각화를 위한 샘플 데이터 플레임 생성 (실제로는 API 서버나 DB에서 가져옵니다)
raw_data = {
"Symbol": ["AAPL", "GOOGL", "AMZN"],
"Price": [150.25, 2800.50, 3450.00],
}
df = pd.DataFrame(raw_data)
# Plotly를 이용한 바 차트 생성
fig = px.bar(
df,
x="Symbol",
y="Price",
title="Real-time Stock Prices",
color="Symbol",
template="plotly_dark",
)
# 대시보드 레이아웃 구성
app.layout = html.Div(
style={"fontFamily": "sans-serif", "padding": "20px"},
children=[
html.H1(
children="실시간 주가 분석 대시보드",
style={"textAlign": "center", "color": "#2c3e50"},
),
html.P(
"이 대시보드는 Asyncio와 FastAPI를 거쳐 수집된 실시간 데이터를 보여줍니다.",
style={"textAlign": "center"},
),
dcc.Graph(id="stock-price-graph", figure=fig),
],
)
if __name__ == "__main__":
app.run_server(debug=True)
Dash는 HTML 태그를 파이썬 클래스(html.Div, html.H1) 형태로 제공하므로, 레이아웃을 짜기가 정말 편합니다. Plotly의 강력한 인터랙티브 기능 덕분에 마우스를 올리면 상세 수치가 나오는 동적 그래프가 기본으로 완성됩니다.
🛠️ 개발을 위한 팁 (Tips for Developers)
- 로그 기록의 습관화: 실시간 파이프라인은 어디 한 곳에서 병목이나 에러가 나면 전체 시스템이 꼬이기 쉽습니다. print() 대신 파이썬 내장 logging 라이브러리를 사용해 컴포넌트별 상태를 기록해두면 디버깅이 훨씬 편해집니다.
- Uvicorn과 Dash의 포트 충돌 방지: FastAPI 서버와 Dash 대시보드를 로컬 환경에서 동시에 띄울 때 포트 번호가 겹치지 않도록 주의하세요. 예를 들어 FastAPI는 8000번, Dash는 8050번 포트로 명확히 분리해 실행해야 합니다.
- DB 커넥션 관리: SQLAlchemy 세션을 열었다면 작업이 끝난 뒤 반드시 session.close()를 호출하거나 with 문을 사용해 커넥션 풀이 마르는 현상을 예방하세요.
⚠️ 흔히 하는 실수 (Common Mistakes)
- 비동기 루프 안에서 동기 코드 실행하기: async def 내부에서 외부 API를 호출할 때 requests 같은 동기 라이브러리를 그냥 쓰면 비동기의 이점이 전부 사라지고 프로그램이 멈추게(Blocking) 됩니다. 비동기 환경에서는 반드시 aiohttp나 httpx 같은 비동기 클라이언트를 써야 합니다.
- Global 변수 남용에 따른 동시성 이슈: FastAPI 예제에서 데이터 임시 저장을 위해 global 변수를 사용했습니다. 테스트용으로는 괜찮지만, 실제 운영 환경에서 여러 사용자가 동시에 접속하면 데이터가 꼬이거나 유실될 수 있습니다. 실제 서비스에서는 Redis 같은 인메모리 DB를 공유 저장소로 쓰는 것이 좋습니다.
🏁 마치며
오늘은 파이썬의 핵심 고급 기술들을 따로따로 보지 않고, 하나의 유기적인 실시간 데이터 분석 시스템으로 엮는 방법을 알아봤습니다.
처음에는 비동기 처리나 ORM, API 서버라는 단어들이 각각 어렵게 느껴졌을지 몰라도, 이렇게 하나의 파이프라인으로 연결해보니 각각의 기술이 왜 필요한지 뼈저리게 느껴지셨을 겁니다. 이번에 다룬 기본 구조에 여러분만의 아이디어(예: 알림 기능 추가, 예측 AI 모델 연동)를 더해 한 단계 더 발전된 토이 프로젝트를 만들어보시는 건 어떨까요?
오늘 진행하시면서 막히는 부분이나 궁금한 점이 있다면 언제든 댓글로 남겨주세요. 여러분의 즐거운 코딩을 응원합니다!