예제 소스
한국투자증권 API와 LLM 주식 응용 프로그램 · 파일 19개
main.py
# -*- coding: utf-8 -*-
"""
차트 — 한국투자증권 Open API(KIS Developers) + PyQt5 캔들 차트 대시보드.
화면 구성
┌───────────────────────────────────────┬─────────────────────┐
│ 차트 (현재 bar까지) │ LLM 판단 │
│ - 캔들 + 이동평균(MA5/20/60) ├─────────────────────┤
│ - 보조지표 (RSI, 빨간 채움 영역) │ 판단 이유 (근거) │
│ - 거래량 (빨강 막대) ├─────────────────────┤
│ │ 이력 (자동 시 누적·스크롤)│
├───────────────────────────────────────┴─────────────────────┤
│ 모의 포트폴리오 · 수익 │ 수익 히스토리 │
└─────────────────────────────────────────────────────────────┘
LLM 판단 / 포트폴리오 영역은 현재 placeholder UI 입니다.
차트는 KIS API 의 일/주/월봉 데이터를 실시간 조회하여 표시합니다.
기본 종목은 005930(예시)이며 상단 입력란에서 자유롭게 변경할 수 있습니다.
환경 변수(.env.local 권장):
APP_KEY, APP_SECRET — KIS 앱키/앱시크릿
KIS_PAPER=1 — 모의투자 도메인 사용(선택)
"""
from __future__ import annotations
import platform
import sys
import numpy as np
import pandas as pd
from kis_client import KISClient, candles_to_dataframe
import matplotlib
matplotlib.use("Qt5Agg") # PyQt5 임베드용 백엔드 (import 순서 중요)
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure
from matplotlib.patches import Rectangle
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QFont
from PyQt5.QtWidgets import (
QApplication,
QComboBox,
QGridLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QListWidget,
QMainWindow,
QMessageBox,
QProgressBar,
QPushButton,
QSplitter,
QTextEdit,
QVBoxLayout,
QWidget,
)
# ──────────────────────────────────────────────────────────────────────────────
# matplotlib 한글 폰트 설정 (라벨 깨짐 방지)
# ──────────────────────────────────────────────────────────────────────────────
def _setup_matplotlib_korean() -> None:
sysname = platform.system()
if sysname == "Darwin":
matplotlib.rc("font", family="AppleGothic")
elif sysname == "Windows":
matplotlib.rc("font", family="Malgun Gothic")
else:
for name in ("NanumGothic", "Noto Sans CJK KR", "DejaVu Sans"):
try:
matplotlib.rc("font", family=name)
break
except Exception:
continue
matplotlib.rcParams["axes.unicode_minus"] = False
_setup_matplotlib_korean()
def compute_rsi(close: pd.Series, period: int = 14) -> pd.Series:
delta = close.diff()
gain = delta.where(delta > 0, 0.0)
loss = -delta.where(delta < 0, 0.0)
avg_gain = gain.rolling(window=period).mean()
avg_loss = loss.rolling(window=period).mean()
rs = avg_gain / avg_loss.replace(0, np.nan)
rsi = 100 - (100 / (1 + rs))
return rsi.fillna(50.0)
# ──────────────────────────────────────────────────────────────────────────────
# 차트 캔버스 — 3단(캔들+MA / RSI / Volume), 동일 X축 공유
# ──────────────────────────────────────────────────────────────────────────────
class DashboardCanvas(FigureCanvas):
def __init__(self, parent: QWidget | None = None) -> None:
self.fig = Figure(figsize=(10, 8), facecolor="white")
super().__init__(self.fig)
self.setParent(parent)
gs = self.fig.add_gridspec(3, 1, height_ratios=[3.0, 1.1, 1.1], hspace=0.06)
self.ax_price = self.fig.add_subplot(gs[0])
self.ax_ind = self.fig.add_subplot(gs[1], sharex=self.ax_price)
self.ax_vol = self.fig.add_subplot(gs[2], sharex=self.ax_price)
self.fig.subplots_adjust(left=0.07, right=0.99, top=0.98, bottom=0.07)
def plot(self, df: pd.DataFrame) -> None:
for ax in (self.ax_price, self.ax_ind, self.ax_vol):
ax.clear()
if df.empty:
for ax in (self.ax_price, self.ax_ind, self.ax_vol):
ax.text(0.5, 0.5, "no data", ha="center", va="center", transform=ax.transAxes)
self.draw()
return
n = len(df)
x = np.arange(n)
# ── 1) 캔들 + 이동평균 ─────────────────────────────────────────────
body_w = 0.6
opens = df["Open"].values
closes = df["Close"].values
highs = df["High"].values
lows = df["Low"].values
for i in range(n):
up = closes[i] >= opens[i]
color = "red" if up else "blue"
bottom = min(opens[i], closes[i])
height = max(abs(closes[i] - opens[i]), 1.0)
self.ax_price.add_patch(
Rectangle(
(i - body_w / 2, bottom),
body_w,
height,
facecolor=color,
edgecolor=color,
linewidth=0.5,
)
)
self.ax_price.plot([i, i], [lows[i], highs[i]], color=color, linewidth=0.8)
if n >= 5:
self.ax_price.plot(x, df["Close"].rolling(5).mean().values,
color="red", linewidth=1.0, label="MA5")
if n >= 20:
self.ax_price.plot(x, df["Close"].rolling(20).mean().values,
color="blue", linewidth=1.0, label="MA20")
if n >= 60:
self.ax_price.plot(x, df["Close"].rolling(60).mean().values,
color="black", linewidth=1.0, label="MA60")
self.ax_price.set_xlim(-1, n)
self.ax_price.grid(True, alpha=0.3)
self.ax_price.tick_params(labelbottom=False)
# ── 2) RSI(14) – 빨강 채움 영역 ────────────────────────────────────
rsi = compute_rsi(df["Close"], 14).values
self.ax_ind.fill_between(x, 0, rsi, color="red", alpha=0.30)
self.ax_ind.plot(x, rsi, color="red", linewidth=1.0)
self.ax_ind.set_ylim(0, 100)
self.ax_ind.grid(True, alpha=0.3)
self.ax_ind.tick_params(labelbottom=False)
# ── 3) 거래량 (스크린샷처럼 빨강 막대) ───────────────────────────────
self.ax_vol.bar(x, df["Volume"].values, width=0.6, color="red")
self.ax_vol.grid(True, alpha=0.3)
# X 라벨 (날짜)
idxs = np.linspace(0, n - 1, 6).astype(int)
self.ax_vol.set_xticks(idxs)
self.ax_vol.set_xticklabels(
[df.index[i].strftime("%Y-%m-%d") for i in idxs],
rotation=0,
fontsize=8,
)
self.draw()
# ──────────────────────────────────────────────────────────────────────────────
# 우측 패널: LLM 판단 / 판단 이유 / 이력
# ──────────────────────────────────────────────────────────────────────────────
class JudgmentPanel(QWidget):
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(8)
# LLM 판단 (게이지 형태)
gb1 = QGroupBox("LLM 판단")
v1 = QVBoxLayout(gb1)
self.judgment_bar = QProgressBar()
self.judgment_bar.setMinimum(-100)
self.judgment_bar.setMaximum(100)
self.judgment_bar.setValue(0)
self.judgment_bar.setTextVisible(False)
self.judgment_bar.setFixedHeight(18)
self.judgment_bar.setStyleSheet(
"QProgressBar { background: #eef2f7; border: 1px solid #d3dbe6; border-radius: 4px; }"
"QProgressBar::chunk { background: #4a90e2; }"
)
v1.addWidget(self.judgment_bar)
self.judgment_label = QLabel("—")
self.judgment_label.setAlignment(Qt.AlignCenter)
v1.addWidget(self.judgment_label)
gb1.setMaximumHeight(96)
# 판단 이유 (근거)
gb2 = QGroupBox("판단 이유 (근거)")
v2 = QVBoxLayout(gb2)
self.reason_text = QTextEdit()
self.reason_text.setReadOnly(True)
self.reason_text.setMinimumHeight(110)
self.reason_text.setMaximumHeight(150)
v2.addWidget(self.reason_text)
# 이력 (자동 시 누적·스크롤)
gb3 = QGroupBox("이력 (자동 시 누적·스크롤)")
v3 = QVBoxLayout(gb3)
self.history_list = QListWidget()
v3.addWidget(self.history_list)
layout.addWidget(gb1)
layout.addWidget(gb2)
layout.addWidget(gb3, 1)
# ──────────────────────────────────────────────────────────────────────────────
# 하단 패널: 모의 포트폴리오 · 수익 / 수익 히스토리
# ──────────────────────────────────────────────────────────────────────────────
class PortfolioPanel(QWidget):
METRIC_LAYOUT = [
# 4-column rows: (key, label)
[("init_cash", "초기 자금"), ("cash", "현금 잔고"),
("hold", "보유 (수량)"), ("eval", "평가액 (원)")],
[("ret", "누적 수익률"), ("sells", "매도 체결"),
("winrate", "매도 승률"), ("price", "현재가 (종가)")],
[("buy_p", "매수 가격"), ("buy_q", "매수 수량"),
("sell_p", "매도 가격"), ("sell_pnl", "매도 수익")],
[("reward", "학습 리워드(최근)")],
]
DEFAULTS = {
"init_cash": "5,000,000 원",
"cash": "5,000,000 원",
"hold": "0",
"eval": "5,000,000 원",
"ret": "+0.00 %",
"sells": "0 회",
"winrate": "0.0 %",
"price": "—",
"buy_p": "—",
"buy_q": "—",
"sell_p": "—",
"sell_pnl": "—",
"reward": "—",
}
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
outer = QHBoxLayout(self)
outer.setContentsMargins(0, 0, 0, 0)
outer.setSpacing(8)
gb_left = QGroupBox("모의 포트폴리오 · 수익")
grid = QGridLayout(gb_left)
grid.setHorizontalSpacing(28)
grid.setVerticalSpacing(2)
grid.setContentsMargins(12, 14, 12, 10)
self._labels: dict[str, QLabel] = {}
row = 0
for line in self.METRIC_LAYOUT:
for col, (k, label) in enumerate(line):
cap = QLabel(label)
cap.setStyleSheet("color: #6a7280; font-size: 9pt;")
val = QLabel(self.DEFAULTS.get(k, "—"))
if k == "ret":
val.setStyleSheet("font-weight: 700; font-size: 13pt; color: #16a34a;")
else:
val.setStyleSheet("font-weight: 700; font-size: 13pt; color: #111827;")
self._labels[k] = val
grid.addWidget(cap, row, col)
grid.addWidget(val, row + 1, col)
row += 2
gb_right = QGroupBox("수익 히스토리")
v = QVBoxLayout(gb_right)
self.profit_list = QListWidget()
v.addWidget(self.profit_list)
outer.addWidget(gb_left, 3)
outer.addWidget(gb_right, 1)
def set_metric(self, key: str, text: str) -> None:
if key in self._labels:
self._labels[key].setText(text)
# ──────────────────────────────────────────────────────────────────────────────
# 메인 윈도우
# ──────────────────────────────────────────────────────────────────────────────
PERIOD_CHOICES = [
("일봉 (D)", "D"),
("주봉 (W)", "W"),
("월봉 (M)", "M"),
]
class MainWindow(QMainWindow):
def __init__(self) -> None:
super().__init__()
self.setWindowTitle("차트")
self.resize(1500, 920)
self._client: KISClient | None = None
# ── 상단 컨트롤 바 ────────────────────────────────────────────────
self.code_input = QLineEdit("005930")
self.code_input.setFixedWidth(110)
self.period_box = QComboBox()
for label, code in PERIOD_CHOICES:
self.period_box.addItem(label, code)
self.refresh_btn = QPushButton("조회 / 새로고침")
self.refresh_btn.clicked.connect(self.on_refresh)
self.status_label = QLabel("준비됨")
self.status_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
top_bar = QHBoxLayout()
top_bar.setContentsMargins(2, 2, 2, 2)
top_bar.addWidget(QLabel("종목코드:"))
top_bar.addWidget(self.code_input)
top_bar.addSpacing(8)
top_bar.addWidget(QLabel("주기:"))
top_bar.addWidget(self.period_box)
top_bar.addSpacing(8)
top_bar.addWidget(self.refresh_btn)
top_bar.addStretch(1)
top_bar.addWidget(self.status_label)
# ── 좌측: 차트 (라벨 + 캔버스) ─────────────────────────────────────
chart_title = QLabel("차트 (현재 bar까지)")
chart_title.setStyleSheet("font-weight: 700; font-size: 11pt;")
self.canvas = DashboardCanvas(self)
chart_box = QWidget()
chart_v = QVBoxLayout(chart_box)
chart_v.setContentsMargins(4, 0, 0, 0)
chart_v.setSpacing(4)
chart_v.addWidget(chart_title)
chart_v.addWidget(self.canvas, 1)
# ── 우측: 판단 패널 ───────────────────────────────────────────────
self.judgment = JudgmentPanel(self)
upper = QSplitter(Qt.Horizontal)
upper.addWidget(chart_box)
upper.addWidget(self.judgment)
upper.setStretchFactor(0, 3)
upper.setStretchFactor(1, 1)
upper.setSizes([1050, 450])
# ── 하단: 포트폴리오 ──────────────────────────────────────────────
self.portfolio = PortfolioPanel(self)
self.portfolio.setMinimumHeight(190)
main_split = QSplitter(Qt.Vertical)
main_split.addWidget(upper)
main_split.addWidget(self.portfolio)
main_split.setStretchFactor(0, 4)
main_split.setStretchFactor(1, 1)
main_split.setSizes([720, 200])
# ── 중앙 위젯 ─────────────────────────────────────────────────────
central = QWidget(self)
layout = QVBoxLayout(central)
layout.setContentsMargins(8, 8, 8, 8)
layout.addLayout(top_bar)
layout.addWidget(main_split, 1)
self.setCentralWidget(central)
def _ensure_client(self) -> KISClient:
if self._client is None:
self._client = KISClient()
self._client.issue_access_token()
return self._client
def on_refresh(self) -> None:
code = self.code_input.text().strip() or "005930"
period = self.period_box.currentData() or "D"
self.status_label.setText("조회 중...")
QApplication.processEvents()
try:
client = self._ensure_client()
body = client.inquire_daily_itemchart(fid_input_iscd=code, period=period)
rt = body.get("rt_cd")
msg = body.get("msg1", "")
if rt != "0":
raise RuntimeError(f"API 오류({rt}): {msg}")
df = candles_to_dataframe(body)
self.canvas.plot(df)
if not df.empty:
last = float(df["Close"].iloc[-1])
self.portfolio.set_metric("price", f"{int(last):,} 원")
self.status_label.setText(f"조회 완료 ({len(df)} bars)")
except Exception as exc: # noqa: BLE001
self.status_label.setText("오류")
QMessageBox.critical(self, "조회 실패", str(exc))
def main() -> None:
app = QApplication(sys.argv)
app.setFont(QFont("", 9))
win = MainWindow()
win.show()
win.on_refresh() # 시작 시 삼성전자 일봉 1회 조회
sys.exit(app.exec_())
if __name__ == "__main__":
main()