AI 퀀트 통합 운용 대시보드 구축: 전략 신호부터 포트폴리오 리스크까지 한 화면에서 관리하기

AI 퀀트 통합 운용 대시보드 구축: 전략 신호부터 포트폴리오 리스크까지 한 화면에서 관리하기

퀀트 투자를 공부하면서 가장 먼저 만들었던 것은 매매 전략이었다.

이동평균선을 계산하고, 조건이 맞으면 BUY, 아니면 SELL을 출력하는 아주 단순한 코드였다. 처음 신호가 제대로 찍혔을 때는 꽤 신기했다. 마치 내가 작은 펀드매니저라도 된 것 같았다.

그런데 전략이 두 개, 세 개로 늘어나기 시작하니 문제가 생겼다.

지금 어떤 전략이 실제 포지션을 들고 있는지, 전체 계좌의 손익은 얼마인지, 특정 종목 비중이 지나치게 커지지는 않았는지, 손실이 커졌을 때 어떤 전략을 멈춰야 하는지 한눈에 알 수 없었다.

결국 매일 여러 CSV 파일과 로그를 번갈아 열어보게 됐다.

이때 깨달았다.

퀀트 시스템에서 정말 어려운 것은 매수 신호 하나를 만드는 일이 아니라, 여러 전략과 계좌를 안전하게 통합 운용하는 일이다.

이번 글에서는 Python과 Streamlit을 이용해 다음 정보를 하나의 화면에 표시하는 AI 퀀트 통합 운용 대시보드를 만들어본다.

• AI 전략 신호
• 누적 수익률
• 벤치마크 비교
• 연환산 수익률과 변동성
• 샤프지수
• 최대 낙폭
• 포트폴리오 현재 비중
• 목표 비중과 리밸런싱 차이
• 손절 및 비중 초과 경보
• 주문 검토 후보

Streamlit은 Python 코드만으로 데이터 애플리케이션을 만들 수 있는 오픈소스 프레임워크다. 별도의 React 프런트엔드나 API 서버를 먼저 만들지 않아도 대시보드 프로토타입을 빠르게 구현할 수 있다.

---

1. 우리가 만들 대시보드의 구조

이번 프로젝트의 데이터 흐름은 다음과 같다.

시장 데이터
    ↓
전략 지표 계산
    ↓
AI·퀀트 신호 생성
    ↓
포지션 결정
    ↓
전략 수익률 계산
    ↓
포트폴리오 평가
    ↓
리스크 감시
    ↓
주문 후보 생성
    ↓
Streamlit 대시보드 표시

처음부터 실제 증권사 API를 연결하지는 않는다.

실제 주문 기능까지 한 번에 구현하면 데이터 오류, 인증서 관리, 주문 중복, 체결 지연, 장 운영시간 처리까지 동시에 고민해야 한다. 그러면 대시보드의 핵심 구조를 이해하기가 오히려 어려워진다.

그래서 이번 예제에서는 고정된 난수 시드로 모의 OHLCV 데이터를 생성한다.

고정 시드를 사용하면 프로그램을 다시 실행해도 같은 데이터를 얻을 수 있다. 테스트 결과가 계속 달라지지 않으니 전략 계산과 리스크 로직을 검증하기 좋다.

---

2. 개발 환경 구성

프로젝트 폴더를 하나 만든다.

mkdir ai-quant-dashboard
cd ai-quant-dashboard

가상환경을 생성한다.

Windows

python -m venv .venv
.venv\Scripts\activate

macOS·Linux

python3 -m venv .venv
source .venv/bin/activate

requirements.txt 파일을 만든다.

streamlit
pandas
numpy
plotly

라이브러리를 설치한다.

pip install -r requirements.txt

Streamlit 애플리케이션은 일반적으로 다음 명령으로 실행한다.

streamlit run app.py

Streamlit 공식 치트시트에서도 pip install streamlit과 streamlit run 방식으로 설치 및 실행하는 예제를 제공한다.

프로젝트 구조는 아주 단순하다.

ai-quant-dashboard/
├── app.py
└── requirements.txt

서비스가 커지면 데이터 수집, 전략, 포트폴리오, 주문 모듈을 분리해야 하지만, 첫 프로젝트에서는 전체 흐름을 한 파일에서 확인하는 편이 이해하기 쉽다.

---

3. 실행 가능한 전체 코드

app.py 파일을 만들고 아래 코드를 입력한다.

from __future__ import annotations

from dataclasses import dataclass
from typing import Final

import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import streamlit as st


TRADING_DAYS: Final[int] = 252
RISK_FREE_RATE: Final[float] = 0.03


@dataclass(frozen=True)
class PortfolioMetrics:
    total_return: float
    annual_return: float
    annual_volatility: float
    sharpe_ratio: float
    max_drawdown: float
    win_rate: float


def create_market_data(seed: int = 42, periods: int = 260) -> pd.DataFrame:
    """재현 가능한 예제 OHLCV와 전략 신호 데이터를 생성한다."""
    rng = np.random.default_rng(seed)

    dates = pd.bdate_range(
        end=pd.Timestamp.today().normalize(),
        periods=periods,
    )

    returns = rng.normal(
        loc=0.00045,
        scale=0.012,
        size=periods,
    )

    close = 100 * np.cumprod(1 + returns)
    open_price = close * (1 + rng.normal(0, 0.003, periods))

    high = np.maximum(open_price, close) * (
        1 + rng.uniform(0.001, 0.012, periods)
    )

    low = np.minimum(open_price, close) * (
        1 - rng.uniform(0.001, 0.012, periods)
    )

    volume = rng.integers(
        150_000,
        1_200_000,
        periods,
    )

    df = pd.DataFrame(
        {
            "Open": open_price,
            "High": high,
            "Low": low,
            "Close": close,
            "Volume": volume,
        },
        index=dates,
    )

    df["Return"] = df["Close"].pct_change().fillna(0.0)

    df["MA20"] = (
        df["Close"]
        .rolling(20, min_periods=1)
        .mean()
    )

    df["MA60"] = (
        df["Close"]
        .rolling(60, min_periods=1)
        .mean()
    )

    df["Signal"] = np.where(
        df["MA20"] > df["MA60"],
        1,
        -1,
    )

    df["Position"] = (
        pd.Series(df["Signal"], index=df.index)
        .shift(1)
        .fillna(0)
    )

    df["StrategyReturn"] = (
        df["Position"] * df["Return"]
    )

    df["Equity"] = (
        1 + df["StrategyReturn"]
    ).cumprod()

    df["Benchmark"] = (
        1 + df["Return"]
    ).cumprod()

    df["Peak"] = df["Equity"].cummax()

    df["Drawdown"] = (
        df["Equity"] / df["Peak"] - 1
    )

    return df


def calculate_metrics(
    strategy_returns: pd.Series,
) -> PortfolioMetrics:
    """전략 수익률을 이용해 주요 운용 지표를 계산한다."""
    clean = strategy_returns.dropna()

    if clean.empty:
        return PortfolioMetrics(
            total_return=0,
            annual_return=0,
            annual_volatility=0,
            sharpe_ratio=0,
            max_drawdown=0,
            win_rate=0,
        )

    equity = (1 + clean).cumprod()

    total_return = float(
        equity.iloc[-1] - 1
    )

    years = max(
        len(clean) / TRADING_DAYS,
        1 / TRADING_DAYS,
    )

    annual_return = float(
        equity.iloc[-1] ** (1 / years) - 1
    )

    annual_volatility = float(
        clean.std(ddof=1)
        * np.sqrt(TRADING_DAYS)
    )

    if annual_volatility > 0:
        sharpe_ratio = float(
            (
                annual_return
                - RISK_FREE_RATE
            )
            / annual_volatility
        )
    else:
        sharpe_ratio = 0.0

    peak = equity.cummax()

    max_drawdown = float(
        (equity / peak - 1).min()
    )

    active_returns = clean[clean != 0]

    if active_returns.empty:
        win_rate = 0.0
    else:
        win_rate = float(
            (active_returns > 0).mean()
        )

    return PortfolioMetrics(
        total_return=total_return,
        annual_return=annual_return,
        annual_volatility=annual_volatility,
        sharpe_ratio=sharpe_ratio,
        max_drawdown=max_drawdown,
        win_rate=win_rate,
    )


def create_positions(
    latest_close: float,
) -> pd.DataFrame:
    """대시보드 표시용 모의 포트폴리오를 생성한다."""
    positions = pd.DataFrame(
        {
            "자산": [
                "AI-MOMENTUM",
                "KOSPI200",
                "NASDAQ100",
                "USD-CASH",
            ],
            "보유수량": [
                120,
                35,
                18,
                1,
            ],
            "평균단가": [
                latest_close * 0.94,
                352.4,
                487.2,
                1_380_000,
            ],
            "현재가": [
                latest_close,
                361.8,
                503.5,
                1_380_000,
            ],
            "목표비중": [
                0.40,
                0.25,
                0.20,
                0.15,
            ],
            "현재비중": [
                0.43,
                0.22,
                0.21,
                0.14,
            ],
            "AI신호": [
                "매수 유지",
                "중립",
                "매수",
                "대기",
            ],
        }
    )

    positions["평가손익률"] = (
        positions["현재가"]
        / positions["평균단가"]
        - 1
    )

    positions["리밸런싱차이"] = (
        positions["목표비중"]
        - positions["현재비중"]
    )

    return positions


def main() -> None:
    st.set_page_config(
        page_title="AI 퀀트 통합 운용 대시보드",
        page_icon="📊",
        layout="wide",
    )

    st.title("📊 AI 퀀트 통합 운용 대시보드")

    st.caption(
        "전략 신호, 포트폴리오, 손익, 리스크, "
        "주문 후보를 한 화면에서 점검합니다."
    )

    with st.sidebar:
        st.header("운용 설정")

        initial_capital = st.number_input(
            "초기 운용자금(원)",
            min_value=1_000_000,
            value=100_000_000,
            step=1_000_000,
        )

        max_position = (
            st.slider(
                "종목당 최대 비중",
                5,
                50,
                25,
                5,
            )
            / 100
        )

        stop_loss = (
            st.slider(
                "손절 기준",
                -30,
                -1,
                -8,
                1,
            )
            / 100
        )

        seed = st.number_input(
            "데이터 시드",
            min_value=1,
            value=42,
            step=1,
        )

        st.info(
            "이 예제는 학습용 모의 데이터이며 "
            "실제 주문을 전송하지 않습니다."
        )

    df = create_market_data(
        seed=int(seed)
    )

    metrics = calculate_metrics(
        df["StrategyReturn"]
    )

    positions = create_positions(
        float(df["Close"].iloc[-1])
    )

    current_equity = (
        initial_capital
        * (1 + metrics.total_return)
    )

    col1, col2, col3, col4, col5 = (
        st.columns(5)
    )

    col1.metric(
        "현재 평가금액",
        f"{current_equity:,.0f}원",
        f"{metrics.total_return:.2%}",
    )

    col2.metric(
        "연환산 수익률",
        f"{metrics.annual_return:.2%}",
    )

    col3.metric(
        "연환산 변동성",
        f"{metrics.annual_volatility:.2%}",
    )

    col4.metric(
        "샤프지수",
        f"{metrics.sharpe_ratio:.2f}",
    )

    col5.metric(
        "최대 낙폭",
        f"{metrics.max_drawdown:.2%}",
    )

    tab1, tab2, tab3, tab4 = st.tabs(
        [
            "통합 현황",
            "전략 분석",
            "포트폴리오",
            "주문·리스크",
        ]
    )

    with tab1:
        left, right = st.columns([2, 1])

        with left:
            equity_df = df[
                ["Equity", "Benchmark"]
            ].rename(
                columns={
                    "Equity": "AI 전략",
                    "Benchmark": "벤치마크",
                }
            )

            equity_figure = px.line(
                equity_df,
                title="누적 수익 곡선",
            )

            equity_figure.update_layout(
                yaxis_title="누적 가치",
                xaxis_title="날짜",
                legend_title="구분",
            )

            st.plotly_chart(
                equity_figure,
                use_container_width=True,
            )

        with right:
            latest_signal = (
                "매수"
                if int(df["Signal"].iloc[-1]) == 1
                else "매도/회피"
            )

            st.subheader("오늘의 AI 판단")

            st.metric(
                "최종 신호",
                latest_signal,
            )

            st.metric(
                "승률",
                f"{metrics.win_rate:.2%}",
            )

            st.metric(
                "현재 포지션 한도",
                f"{max_position:.0%}",
            )

            st.metric(
                "손절 기준",
                f"{stop_loss:.0%}",
            )

            st.write(
                "단기 이동평균과 장기 이동평균의 "
                "관계를 이용한 기본 추세 전략입니다. "
                "실전에서는 거래비용, 슬리피지, "
                "체결 실패를 반드시 반영해야 합니다."
            )

    with tab2:
        candle = go.Figure(
            data=[
                go.Candlestick(
                    x=df.index,
                    open=df["Open"],
                    high=df["High"],
                    low=df["Low"],
                    close=df["Close"],
                    name="OHLC",
                )
            ]
        )

        candle.add_trace(
            go.Scatter(
                x=df.index,
                y=df["MA20"],
                name="MA20",
            )
        )

        candle.add_trace(
            go.Scatter(
                x=df.index,
                y=df["MA60"],
                name="MA60",
            )
        )

        candle.update_layout(
            title="가격·이동평균·전략 신호",
            xaxis_rangeslider_visible=False,
        )

        st.plotly_chart(
            candle,
            use_container_width=True,
        )

        signal_log = df[
            [
                "Close",
                "MA20",
                "MA60",
                "Signal",
                "Position",
            ]
        ].tail(20).copy()

        signal_log["Signal"] = (
            signal_log["Signal"].map(
                {
                    1: "BUY",
                    -1: "SELL",
                }
            )
        )

        st.dataframe(
            signal_log.sort_index(
                ascending=False
            ),
            use_container_width=True,
        )

    with tab3:
        display_positions = (
            positions.copy()
        )

        display_positions["평가손익률"] = (
            display_positions[
                "평가손익률"
            ].map(
                lambda value: f"{value:.2%}"
            )
        )

        display_positions["목표비중"] = (
            display_positions[
                "목표비중"
            ].map(
                lambda value: f"{value:.1%}"
            )
        )

        display_positions["현재비중"] = (
            display_positions[
                "현재비중"
            ].map(
                lambda value: f"{value:.1%}"
            )
        )

        display_positions[
            "리밸런싱차이"
        ] = display_positions[
            "리밸런싱차이"
        ].map(
            lambda value: f"{value:+.1%}"
        )

        st.dataframe(
            display_positions,
            use_container_width=True,
            hide_index=True,
        )

        allocation = (
            positions
            .set_index("자산")[
                ["현재비중", "목표비중"]
            ]
        )

        allocation_figure = px.bar(
            allocation,
            barmode="group",
            title="현재 비중과 목표 비중 비교",
            labels={
                "value": "비중",
                "variable": "구분",
            },
        )

        st.plotly_chart(
            allocation_figure,
            use_container_width=True,
        )

    with tab4:
        risk_column, order_column = (
            st.columns(2)
        )

        with risk_column:
            st.subheader("리스크 경보")

            alerts: list[str] = []

            if (
                metrics.max_drawdown
                <= stop_loss
            ):
                alerts.append(
                    "최대 낙폭이 손절 기준을 "
                    "초과했습니다."
                )

            if (
                positions["현재비중"].max()
                > max_position
            ):
                alerts.append(
                    "종목당 최대 비중을 초과한 "
                    "포지션이 있습니다."
                )

            if (
                metrics.annual_volatility
                > 0.25
            ):
                alerts.append(
                    "연환산 변동성이 25%를 "
                    "초과했습니다."
                )

            if alerts:
                for alert in alerts:
                    st.error(alert)
            else:
                st.success(
                    "현재 설정 기준으로 즉시 "
                    "대응이 필요한 경보가 없습니다."
                )

            drawdown_figure = px.area(
                df,
                y="Drawdown",
                title="전략 낙폭 추이",
            )

            drawdown_figure.update_layout(
                yaxis_tickformat=".1%",
                yaxis_title="낙폭",
            )

            st.plotly_chart(
                drawdown_figure,
                use_container_width=True,
            )

        with order_column:
            st.subheader(
                "리밸런싱 주문 후보"
            )

            orders = positions.loc[
                positions[
                    "리밸런싱차이"
                ].abs()
                >= 0.02,
                [
                    "자산",
                    "현재비중",
                    "목표비중",
                    "리밸런싱차이",
                    "AI신호",
                ],
            ].copy()

            orders["주문방향"] = np.where(
                orders[
                    "리밸런싱차이"
                ]
                > 0,
                "매수 검토",
                "매도 검토",
            )

            st.dataframe(
                orders,
                use_container_width=True,
                hide_index=True,
            )

            st.warning(
                "주문 후보는 자동 주문이 아닙니다. "
                "최종 주문 전 가격, 호가, "
                "거래비용을 검증하세요."
            )


if __name__ == "__main__":
    main()

실행한다.

streamlit run app.py

브라우저가 자동으로 열리지 않는다면 터미널에 표시되는 로컬 주소로 접속한다.

일반적으로 다음과 같은 형태다.

http://localhost:8501

---

4. 시장 데이터 생성 구조 이해하기

반응형

실제 퀀트 시스템에서는 증권사 API, 거래소 데이터, 데이터 공급 업체 또는 자체 데이터베이스에서 시세를 가져온다.

하지만 이번 프로젝트에서는 다음 코드로 일별 수익률을 생성한다.

returns = rng.normal(
    loc=0.00045,
    scale=0.012,
    size=periods,
)

loc은 모의 수익률의 평균이고 scale은 변동성 수준이다.

가격은 일별 수익률을 누적해서 만든다.

close = 100 * np.cumprod(
    1 + returns
)

예를 들어 전일 종가가 100이고 오늘 수익률이 1%라면 새로운 가격은 101이 된다.

그다음 시가, 고가, 저가, 거래량을 생성해 OHLCV 데이터 구조를 완성한다.

df = pd.DataFrame(
    {
        "Open": open_price,
        "High": high,
        "Low": low,
        "Close": close,
        "Volume": volume,
    },
    index=dates,
)

실제 데이터 공급처를 연결할 때도 최종적으로는 이와 비슷한 DataFrame 구조로 표준화하는 편이 좋다.

데이터 공급처마다 컬럼 이름이나 날짜 형식이 다르기 때문이다.

---

5. 전략 신호 계산하기

이번 예제에서는 가장 기본적인 이동평균 교차 전략을 사용한다.

df["MA20"] = (
    df["Close"]
    .rolling(20, min_periods=1)
    .mean()
)

df["MA60"] = (
    df["Close"]
    .rolling(60, min_periods=1)
    .mean()
)

Pandas의 rolling()은 지정한 구간을 이동하면서 평균, 합계, 표준편차 같은 통계를 계산할 때 사용한다.

단기 이동평균이 장기 이동평균보다 높으면 상승 추세로 판단한다.

df["Signal"] = np.where(
    df["MA20"] > df["MA60"],
    1,
    -1,
)

여기서 중요한 코드는 포지션을 하루 뒤로 이동하는 부분이다.

df["Position"] = (
    pd.Series(
        df["Signal"],
        index=df.index,
    )
    .shift(1)
    .fillna(0)
)

오늘 종가를 보고 생성한 신호로 오늘 종가 수익률을 얻었다고 계산하면 미래 정보를 미리 사용한 셈이 된다.

이런 오류를 룩어헤드 바이어스, 즉 미래 참조 편향이라고 한다.

그래서 오늘 계산된 신호는 다음 거래일부터 포지션에 반영한다.

df["StrategyReturn"] = (
    df["Position"]
    * df["Return"]
)

수익률은 pct_change()로 계산한다.

df["Return"] = (
    df["Close"]
    .pct_change()
    .fillna(0.0)
)

Pandas 공식 문서에 따르면 pct_change()는 기본적으로 직전 행과 현재 행 사이의 변화 비율을 계산한다.

---

6. 누적 수익률 계산하기

전략 수익률은 매일 발생한다.

이를 누적하면 자산 곡선을 만들 수 있다.

df["Equity"] = (
    1 + df["StrategyReturn"]
).cumprod()

예를 들어 첫째 날 2%, 둘째 날 -1%라면 누적 가치는 단순히 1%가 아니다.

1.02 × 0.99 = 1.0098

따라서 누적 수익률은 약 0.98%다.

복리 수익률을 다룰 때는 수익률을 단순히 더하지 않고 cumprod()로 누적 곱을 계산해야 한다.

벤치마크도 동일한 방식으로 만든다.

df["Benchmark"] = (
    1 + df["Return"]
).cumprod()

여기서 벤치마크는 동일한 모의 자산을 계속 보유한 결과다.

대시보드에서 AI 전략과 벤치마크를 함께 표시하는 이유는 단순하다.

전략 수익률이 플러스라는 사실만으로는 좋은 전략이라고 말하기 어렵기 때문이다.

시장이 30% 상승했는데 전략은 5% 수익에 그쳤다면 전략을 사용하지 않고 그냥 보유하는 편이 나았을 수 있다.

---

7. 연환산 수익률과 변동성

대시보드 상단에는 다음 지표를 표시한다.

현재 평가금액
연환산 수익률
연환산 변동성
샤프지수
최대 낙폭

연환산 수익률은 서로 다른 기간의 전략을 비교하기 위해 필요하다.

years = max(
    len(clean) / TRADING_DAYS,
    1 / TRADING_DAYS,
)

annual_return = (
    equity.iloc[-1]
    ** (1 / years)
    - 1
)

예를 들어 6개월 동안 10%의 수익을 기록한 전략과 2년 동안 10%의 수익을 기록한 전략은 같은 성과라고 보기 어렵다.

연환산 변동성은 일별 수익률의 표준편차에 거래일 수의 제곱근을 곱해 계산한다.

annual_volatility = (
    clean.std(ddof=1)
    * np.sqrt(TRADING_DAYS)
)

수익률만 높고 변동성이 지나치게 큰 전략은 실제 운용 과정에서 버티기 어려울 수 있다.

백테스트 화면에서는 아름다운 우상향 그래프로 보이더라도, 중간에 30~40%의 손실을 견뎌야 했다면 현실적으로 운용 가능한 전략인지 다시 생각해야 한다.

---

8. 샤프지수 계산하기

샤프지수는 수익률을 위험으로 나누어 전략의 위험 대비 성과를 살펴보는 지표다.

이번 코드에서는 다음과 같이 계산한다.

sharpe_ratio = (
    annual_return
    - RISK_FREE_RATE
) / annual_volatility

다만 이 예제의 샤프지수는 학습용으로 단순화한 계산이다.

실무에서는 다음 내용을 더 세밀하게 결정해야 한다.

• 무위험수익률의 기준
• 일별 또는 월별 수익률 사용 여부
• 수익률 분포의 비대칭성
• 거래가 없는 날의 처리
• 레버리지 반영 방식
• 운용 수수료
• 성과보수
• 환율 변동

숫자 하나만 보고 전략을 판단해서는 안 된다.

샤프지수가 좋더라도 최대 낙폭이 크거나 특정 시장 국면에 수익이 집중되어 있을 수 있다.

---

9. 최대 낙폭 계산하기

내가 실제 운용 대시보드에서 가장 자주 보는 지표는 최대 낙폭이다.

수익률이 좋은지는 나중 문제일 때가 많다. 손실이 감당 가능한 범위인지부터 확인해야 한다.

현재까지의 최고 자산가치를 계산한다.

df["Peak"] = (
    df["Equity"].cummax()
)

현재 자산가치가 최고점에서 얼마나 하락했는지 계산한다.

df["Drawdown"] = (
    df["Equity"]
    / df["Peak"]
    - 1
)

예를 들어 자산이 1억 원에서 1억 2천만 원까지 증가한 뒤 9천만 원으로 하락했다면, 단순히 최초 자금 대비 10% 손실만 보면 안 된다.

고점 1억 2천만 원을 기준으로 보면 낙폭은 25%다.

9,000만 원 ÷ 1억 2,000만 원 - 1
= -25%

투자자가 실제로 느끼는 고통은 최초 투자금보다 최근 고점에서의 손실과 더 가까울 때가 많다.

---

10. 포트폴리오 목표 비중과 현재 비중

전략 신호가 좋아도 특정 자산에 지나치게 많은 자금이 몰리면 계좌 전체가 위험해질 수 있다.

그래서 대시보드에서는 현재 비중과 목표 비중을 함께 관리한다.

positions["리밸런싱차이"] = (
    positions["목표비중"]
    - positions["현재비중"]
)

계산 결과가 양수라면 현재 비중이 목표보다 낮다는 뜻이다.

목표 비중 25%
현재 비중 22%
리밸런싱 차이 +3%

매수 검토 대상이다.

반대로 결과가 음수라면 현재 비중이 목표보다 높다.

목표 비중 40%
현재 비중 43%
리밸런싱 차이 -3%

매도 검토 대상이다.

이번 예제에서는 목표와 현재 비중의 차이가 2%포인트 이상인 자산만 주문 후보로 표시한다.

orders = positions.loc[
    positions[
        "리밸런싱차이"
    ].abs()
    >= 0.02
]

차이가 발생할 때마다 주문하면 소액 주문이 지나치게 자주 생성될 수 있다.

그 결과 거래 수수료와 슬리피지가 증가한다.

따라서 실제 시스템에서는 일정 허용 범위를 두는 리밸런싱 밴드 방식이 유용하다.

---

11. 리스크 경보 만들기

자동매매 시스템에서 매수 신호보다 먼저 구현해야 하는 것이 리스크 경보다.

이번 코드에서는 세 가지 조건을 검사한다.

최대 낙폭 초과

if metrics.max_drawdown <= stop_loss:
    alerts.append(
        "최대 낙폭이 손절 기준을 초과했습니다."
    )

사이드바에서 손절 기준을 -8%로 설정하면 현재 전략의 낙폭이 -8% 이하로 내려갔을 때 경보가 발생한다.

종목당 최대 비중 초과

if (
    positions["현재비중"].max()
    > max_position
):
    alerts.append(
        "종목당 최대 비중을 초과한 "
        "포지션이 있습니다."
    )

목표 비중을 관리하더라도 가격 급등락이나 부분 체결로 현재 비중이 한도를 넘어갈 수 있다.

변동성 초과

if (
    metrics.annual_volatility
    > 0.25
):
    alerts.append(
        "연환산 변동성이 25%를 "
        "초과했습니다."
    )

실전에서는 이 기준을 숫자로 고정하기보다 전략 종류와 자산군에 따라 다르게 설정해야 한다.

예를 들어 채권 전략과 암호화폐 전략에 동일한 변동성 기준을 적용하는 것은 적절하지 않을 수 있다.

---

12. Plotly로 금융 차트 만들기

가격 차트는 Plotly의 Candlestick 객체로 만든다.

go.Candlestick(
    x=df.index,
    open=df["Open"],
    high=df["High"],
    low=df["Low"],
    close=df["Close"],
)

캔들스틱 차트는 일정 시점의 시가, 고가, 저가, 종가를 표현하는 금융 차트다.

이동평균선은 Scatter 객체로 추가한다.

candle.add_trace(
    go.Scatter(
        x=df.index,
        y=df["MA20"],
        name="MA20",
    )
)

완성된 Plotly 차트는 Streamlit의 st.plotly_chart()로 출력한다.

st.plotly_chart(
    candle,
    use_container_width=True,
)

Streamlit은 Plotly 차트를 애플리케이션 화면에 직접 표시할 수 있는 API를 제공한다.

---

13. 이 대시보드에서 AI는 어디에 들어가는가

현재 예제의 전략은 이동평균 교차 방식이다.

그런데 화면에는 AI 신호라는 표현이 들어가 있다.

조금 과장처럼 느껴질 수 있다. 맞다. 아직 이 코드 자체에는 머신러닝 모델이나 LLM이 포함되어 있지 않다.

여기서 중요한 것은 대시보드의 인터페이스다.

AI 모델이든 전통적인 퀀트 전략이든 최종적으로 운용 시스템에 전달해야 하는 값은 어느 정도 표준화할 수 있다.

{
    "symbol": "NASDAQ100",
    "signal": "BUY",
    "confidence": 0.81,
    "target_weight": 0.20,
    "risk_score": 0.34,
    "generated_at": "2026-07-11T09:00:00+09:00"
}

예를 들어 머신러닝 모델의 예측 결과를 다음처럼 변환할 수 있다.

def convert_prediction_to_signal(
    probability: float,
) -> str:
    if probability >= 0.65:
        return "BUY"

    if probability <= 0.35:
        return "SELL"

    return "HOLD"

하지만 예측 확률만으로 주문을 실행해서는 안 된다.

def approve_signal(
    signal: str,
    confidence: float,
    current_weight: float,
    maximum_weight: float,
) -> bool:
    if signal != "BUY":
        return False

    if confidence < 0.65:
        return False

    if current_weight >= maximum_weight:
        return False

    return True

AI 모델은 가격 방향을 예측할 수 있지만, 계좌의 전체 위험을 책임지지는 않는다.

그래서 시스템 역할을 분리해야 한다.

AI 모델
→ 예상 방향과 확률 제공

전략 엔진
→ 진입·청산 조건 결정

포트폴리오 엔진
→ 목표 비중 결정

리스크 엔진
→ 주문 허용 여부 판단

주문 엔진
→ 증권사 API로 주문 전달

AI가 BUY라고 말해도 리스크 엔진이 주문을 거부할 수 있어야 한다.

이 구조가 없으면 AI 예측 모델의 오류가 바로 실제 손실로 연결된다.

---

14. 실전 API 연결 전 반드시 확인할 항목

이 예제를 실제 자동매매 시스템으로 확장하려면 최소한 다음 항목이 필요하다.

데이터 검증

가격이 0 또는 음수가 아닌가?
시간 순서가 뒤섞이지 않았는가?
중복 데이터가 존재하지 않는가?
장 마감 데이터가 확정되었는가?
수정주가가 반영되었는가?
환율 기준 시점이 일치하는가?

주문 중복 방지

order_key = (
    f"{account_id}:"
    f"{symbol}:"
    f"{signal_date}:"
    f"{strategy_id}"
)

같은 주문 키가 이미 처리되었다면 다시 주문하지 않도록 해야 한다.

거래비용 반영

net_return = (
    gross_return
    - commission
    - tax
    - slippage
)

백테스트에서 수익이 조금 나는 전략은 거래비용을 넣는 순간 손실 전략이 될 수 있다.

체결 상태 관리

CREATED
SUBMITTED
PARTIALLY_FILLED
FILLED
CANCELED
REJECTED

주문을 전송했다는 사실과 주문이 체결됐다는 사실은 전혀 다르다.

비상 정지

TRADING_ENABLED = False

운영자가 모든 신규 주문을 즉시 막을 수 있는 킬 스위치가 필요하다.

---

15. 프로젝트를 모듈화한다면

대시보드가 커지면 다음 구조로 분리하는 것이 좋다.

ai-quant-dashboard/
├── app.py
├── config.py
├── requirements.txt
│
├── data/
│   ├── market_data.py
│   └── repository.py
│
├── strategies/
│   ├── base.py
│   ├── moving_average.py
│   └── ai_signal.py
│
├── portfolio/
│   ├── allocator.py
│   ├── metrics.py
│   └── rebalance.py
│
├── risk/
│   ├── limits.py
│   └── monitor.py
│
├── execution/
│   ├── broker.py
│   └── order_manager.py
│
└── tests/
    ├── test_metrics.py
    ├── test_strategy.py
    └── test_risk.py

특히 전략과 주문 코드는 반드시 분리하는 편이 좋다.

전략 함수가 증권사 주문 API를 직접 호출하기 시작하면 백테스트와 실거래의 경계가 모호해진다.

좋은 구조는 전략이 주문을 직접 실행하지 않고 주문 의도를 반환하는 구조다.

{
    "symbol": "005930",
    "side": "BUY",
    "target_weight": 0.15,
    "reason": "MA20 crossed above MA60",
}

이 주문 의도를 리스크 엔진이 검증하고, 승인된 요청만 주문 엔진으로 전달한다.

---

16. 테스트 코드 추가하기

퀀트 시스템은 화면보다 계산 정확도가 더 중요하다.

test_app.py 파일을 만들어 핵심 계산을 검사할 수 있다.

from app import (
    calculate_metrics,
    create_market_data,
    create_positions,
)


def test_market_data_columns() -> None:
    df = create_market_data(
        seed=42,
        periods=260,
    )

    required_columns = {
        "Open",
        "High",
        "Low",
        "Close",
        "Volume",
        "Return",
        "MA20",
        "MA60",
        "Signal",
        "Position",
        "StrategyReturn",
        "Equity",
        "Benchmark",
        "Drawdown",
    }

    assert len(df) == 260

    assert required_columns.issubset(
        set(df.columns)
    )


def test_max_drawdown_range() -> None:
    df = create_market_data(
        seed=42,
        periods=260,
    )

    metrics = calculate_metrics(
        df["StrategyReturn"]
    )

    assert -1 <= metrics.max_drawdown <= 0


def test_portfolio_weight_sum() -> None:
    positions = create_positions(
        latest_close=100
    )

    total_weight = (
        positions["현재비중"].sum()
    )

    assert abs(total_weight - 1.0) < 1e-9

pytest를 설치한다.

pip install pytest

테스트를 실행한다.

pytest -q

이번 글의 전체 app.py 코드는 다음 검증을 통과하도록 확인했다.

Python 문법 컴파일 검사
AST 구문 분석
시장 데이터 260개 생성
필수 계산 컬럼 생성 확인
최대 낙폭 범위 확인
포트폴리오 현재 비중 합계 확인
핵심 지표 계산 함수 실행

---

17. 직접 만들어보고 나서 느낀 점

처음에는 화면에 차트가 많을수록 좋은 대시보드라고 생각했다.

그래서 RSI, MACD, 볼린저밴드, 거래량, 뉴스 점수, 공포탐욕지수까지 한 화면에 넣어본 적도 있다. 그런데 막상 장이 흔들릴 때는 아무것도 눈에 들어오지 않았다.

실제 운용 중 필요한 질문은 의외로 단순했다.

지금 전체 계좌는 얼마인가?
오늘 어떤 전략이 포지션을 바꿨는가?
손실 한도를 넘은 전략이 있는가?
비중이 지나치게 큰 종목이 있는가?
오늘 주문해야 할 자산은 무엇인가?
그 주문은 왜 필요한가?

좋은 운용 대시보드는 데이터를 많이 보여주는 화면이 아니다.

운영자가 잘못된 결정을 내리기 전에 이상을 발견하게 해주는 화면이다.

AI 퀀트 시스템에서도 마찬가지다.

모델의 예측 정확도를 1% 높이는 일보다 주문 중복을 막고, 최대 손실을 제한하고, 포지션을 정확하게 계산하는 일이 훨씬 중요할 때가 많다.

조금 재미없는 이야기일 수 있다.

하지만 이런 기본기가 있어야 실제 돈이 들어간 시스템을 오래 운영할 수 있다.

---

마무리

이번 프로젝트에서는 Python, Pandas, Plotly, Streamlit을 이용해 AI 퀀트 통합 운용 대시보드를 만들었다.

구현한 핵심 기능은 다음과 같다.

재현 가능한 모의 OHLCV 생성
이동평균 기반 전략 신호
미래 참조 편향 방지를 위한 포지션 지연
전략 및 벤치마크 누적 수익률
연환산 수익률
연환산 변동성
샤프지수
최대 낙폭
승률
포트폴리오 목표 비중
리밸런싱 주문 후보
손절·비중·변동성 경보
캔들스틱 금융 차트

이 대시보드는 아직 실제 주문을 보내지 않는다.

그게 오히려 중요하다.

먼저 데이터와 전략, 포트폴리오와 리스크 계산이 정확한지 확인해야 한다. 이후 실제 증권사 API를 연결해도 늦지 않다.

자동매매에서 가장 위험한 순간은 시스템이 멈췄을 때가 아니다.

잘못된 계산을 정상적인 판단처럼 보여주면서 계속 주문할 때다.

그러니 화려한 AI 모델을 붙이기 전에, 계좌의 현재 상태를 정확하게 보여주는 운용 대시보드부터 만들어보자. 생각보다 배울 것이 정말 많다.

참고 자료

• Streamlit 공식 문서: Python 기반 데이터 애플리케이션 구성과 API 사용법
• Streamlit Plotly 차트 공식 문서
• Pandas pct_change() 공식 문서
• Pandas rolling() 공식 문서
• Plotly 캔들스틱 차트 공식 문서

AI퀀트, 퀀트투자, 퀀트대시보드, 자동매매시스템, 알고리즘트레이딩, Python퀀트, Streamlit대시보드, Pandas시계열, Plotly금융차트, 포트폴리오관리, 리스크관리, 최대낙폭, 샤프지수, 연환산수익률, 리밸런싱, AI투자, 퀀트개발, 데이터대시보드, 투자자동화, 파이썬프로젝트