Python 실시간 데이터 분석

Python 실시간 데이터 분석

안녕하세요, 'Deep Python Studio'에 오신 것을 환영합니다! 지난 포스팅에서는 Python의 다양한 고급 주제들을 실제 프로젝트에 활용하는 방법을 알아보았습니다. 이번 포스팅에서는 이 주제들을 하나로 묶어 통합 프로젝트를 만들어보겠습니다. 오늘 다룰 프로젝트는 실시간 데이터 분석 애플리케이션입니다. 이 프로젝트는 비동기 프로그래밍, REST API, 데이터 시각화, 대시보드를 모두 사용하여 실제 응용 프로그램을 만드는 방법을 보여줍니다.

프로젝트 개요

우리의 목표는 주식 가격 데이터를 실시간으로 수집하고, 이를 분석하여 대시보드를 통해 시각화하는 것입니다. 이를 통해 실시간 데이터를 기반으로 한 분석 시스템을 구축하고, 다양한 Python 기술을 활용하여 이를 구현하는 과정을 배워보겠습니다.

사용 기술

• 비동기 프로그래밍 (asyncio, aiohttp): 실시간 데이터를 빠르게 수집하기 위함
• REST API (FastAPI): 데이터를 외부 애플리케이션과 주고받기 위함
• 데이터베이스 (SQLAlchemy): 데이터를 저장하고 관리하기 위함
• 데이터 시각화 및 대시보드 (Dash, Plotly): 데이터를 직관적으로 시각화하여 사용자에게 제공하기 위함

1. 실시간 데이터 수집

첫 번째 단계는 주식 데이터를 비동기적으로 수집하는 것입니다. 이를 위해 aiohttp를 사용하여 여러 주식의 데이터를 동시에 수집합니다.

import aiohttp
import asyncio
import yfinance as yf
import json

async def fetch_stock_data(symbol):
    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))

asyncio.run(main())

위 코드는 여러 주식 데이터를 비동기적으로 수집하는 간단한 예제입니다. 주식 데이터 API에서 실시간 정보를 받아오는 동시에 여러 요청을 처리할 수 있습니다.

2. REST API 서버 구축

실시간으로 수집한 데이터를 REST API를 통해 제공하려면 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"}

위 예제에서는 /stocks 엔드포인트에서 최신 주식 데이터를 GET 요청으로 제공하고, /update 엔드포인트에서 POST 요청으로 최신 데이터를 업데이트합니다. 이를 통해 실시간으로 데이터를 외부 클라이언트와 주고받을 수 있습니다.

3. 데이터베이스에 저장

수집된 데이터를 데이터베이스에 저장하여 분석 및 기록으로 활용할 수 있습니다. 이를 위해 SQLAlchemy를 사용하여 데이터베이스와 연동합니다.

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./stocks.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class Stock(Base):
    __tablename__ = "stocks"
    id = Column(Integer, primary_key=True, index=True)
    symbol = Column(String, index=True)
    price = Column(Float, index=True)

Base.metadata.create_all(bind=engine)

위 코드에서는 Stock이라는 테이블을 정의하고, 주식의 심볼(symbol)과 가격(price)을 저장합니다. 이를 통해 수집된 데이터를 데이터베이스에 영구적으로 저장할 수 있습니다.

4. 데이터 시각화 및 대시보드 구현

마지막으로, 수집된 데이터를 시각화하여 사용자에게 제공하기 위해 Dash와 Plotly를 사용하여 대시보드를 구축합니다. 이를 통해 사용자는 실시간으로 주식 데이터를 확인하고, 분석할 수 있습니다.

import dash
from dash import dcc, html
import pandas as pd
import plotly.express as px

app = dash.Dash(__name__)

# 가상 주식 데이터 생성
data = {
    'Symbol': ['AAPL', 'GOOGL', 'AMZN'],
    'Price': [150.25, 2800.50, 3450.00]
}
df = pd.DataFrame(data)

fig = px.bar(df, x='Symbol', y='Price', title='Real-time Stock Prices')

app.layout = html.Div(children=[
    html.H1(children='Stock Price Dashboard'),
    dcc.Graph(
        id='stock-price-graph',
        figure=fig
    )
])

if __name__ == '__main__':
    app.run_server(debug=True)

위 예제에서는 Dash를 사용하여 주식 데이터를 바 차트로 시각화하고, 대시보드 형태로 제공합니다. 이 대시보드를 통해 사용자들은 실시간으로 주식 가격을 확인할 수 있습니다.

통합 작업

위에서 설명한 각 요소를 하나의 프로젝트로 통합하면 다음과 같은 흐름으로 동작합니다:

1. 데이터 수집: 비동기적으로 여러 주식의 실시간 데이터를 수집합니다.
2. REST API 서버: 수집한 데이터를 외부에서 요청할 수 있도록 REST API 서버를 통해 제공하고 업데이트합니다.
3. 데이터베이스 저장: 수집된 데이터를 데이터베이스에 저장하여 기록 및 분석에 활용합니다.
4. 데이터 시각화: Dash를 사용하여 수집된 데이터를 대시보드 형태로 시각화하고 사용자에게 제공합니다.

이러한 통합 프로젝트를 통해 Python의 강력한 기능들을 활용하여 실시간 데이터 분석 시스템을 구축할 수 있습니다.

마무리

오늘은 Python의 다양한 고급 주제들을 하나로 통합하여 실시간 데이터 분석 애플리케이션을 만드는 방법을 살펴보았습니다. 비동기 프로그래밍, REST API, 데이터베이스 연동, 데이터 시각화와 같은 주제들을 모두 하나의 프로젝트에 적용하여 실제 응용 프로그램을 만들 수 있음을 확인했습니다. 이 프로젝트를 통해 Python의 강력한 기능을 실무에 적용하는 방법을 배워보셨길 바랍니다.