나만의 주식 DB 만들기 (키움증권 편)

이 글의 목표
1. 초보자에게 가장 큰 장벽인 키움증권 OpenAPI와 PyQt5를 아주 쉽게 이해합니다.
2. 이벤트 루프(Event Loop)가 무엇인지, 식당 진동벨 비유를 통해 확실하게 잡고 갑니다.
3. 1종목 1테이블 구조로 나만의 주식 데이터를 차곡차곡 쌓는 코드를 완성합니다.

지난번에는 대신증권으로 데이터를 모으는 방법을 알아봤는데요. 사실 우리 개미 투자자분들이 가장 많이 쓰는 건 역시 키움증권이죠.

그런데 솔직히 말씀드리면, 시스템 트레이딩 입문할 때 가장 먼저 포기하게 되는 구간이 바로 여기예요. 키움증권 API는 OCX니 비동기니 하는 낯선 용어들이 튀어나와서 정말 머리가 아프거든요. 저도 처음에 이것 때문에 며칠을 고생했는지 몰라요.

하지만 걱정하지 마세요. 어려운 개념은 제가 우리 실생활에 빗대어 아주 쉽게 설명해 드릴게요. 원리만 알면 코드는 복사해서 쓰시면 그만이니까요. 오늘도 수정 주가가 반영된 깨끗한 데이터를 모으러 가볼까요?

0. 사전 지식: 왜 굳이 PyQt5를 써야 하나요?

많은 분들이 물어보세요. "나는 차트 데이터만 필요한데, 왜 그래픽 창을 만드는 PyQt5를 배워야 해요?"

이유는 간단해요. 키움증권 API가 아주 예전 기술(OCX)로 만들어졌기 때문이에요. 이 친구는 혼자서는 작동을 못 하고, 반드시 어떤 창(Window) 위에 얹혀 있어야만 움직일 수 있어요. 그래서 우리는 눈에 보이지는 않더라도, 키움 API가 활동할 수 있는 가상의 창을 만들어줘야 하는데, 그 역할을 PyQt5가 해주는 거예요.

[핵심 개념] 이벤트 루프 (Event Loop) 쉽게 이해하기

키움증권 API는 맛집 진동벨과 똑같아요.

1. 여러분이 카운터에 주문을 넣습니다. (데이터 요청)
2. 직원이 진동벨을 줍니다. 여러분은 자리에 앉아서 기다리죠. (이벤트 루프 대기)
3. 요리가 나오면 진동벨이 울립니다. (이벤트 발생)
4. 그때 음식을 받아옵니다. (데이터 수신)

우리가 작성할 코드에도 이 기다리는 시간(Loop)을 만들어주지 않으면, 프로그램은 주문만 넣고 음식도 안 받고 그냥 집에 가버릴 거예요.

1. 준비물: 32비트 환경이 필수예요

키움증권은 32비트 환경에서만 작동해요. 혹시 64비트 파이썬을 쓰고 계신다면, 아나콘다를 이용해서 32비트 전용 방을 하나 만들어주세요.

[아나콘다 터미널 명령어]
set CONDA_FORCE_32BIT=1
conda create -n kiwoom32 python=3.8
conda activate kiwoom32
pip install pyqt5 pandas

설치가 다 되었다면, OpenAPI+ 모듈 설치와 자동 로그인 설정까지 잊지 말고 챙겨주세요.

2. 데이터베이스 설계 (Schema)

나중에 증권사를 바꾸더라도 데이터는 그대로 쓸 수 있어야겠죠? 그래서 지난번 대신증권 편과 똑같은 구조로 만들 거예요.

• 파일명: stock.db
• 테이블: stock_005930 (종목별로 방을 따로 만듭니다)
• 컬럼: date (PK), open, high, low, close, volume

3. 파이썬 구현: 차근차근 따라 해봐요

Step 1. DB 관리자 (DBManager)

SQLite를 다루는 부분이에요. 종목 코드를 주면 알아서 테이블을 만들고 데이터를 저장해 주는 든든한 친구죠. 지난번 코드와 완전히 똑같으니 편하게 복사하세요.

import sqlite3
import pandas as pd

class DBManager:
    def __init__(self, db_name="stock.db"):
        self.conn = sqlite3.connect(db_name)
        self.cursor = self.conn.cursor()

    def create_table(self, code):
        table_name = f"stock_{code}"
        query = f"""
        CREATE TABLE IF NOT EXISTS {table_name} (
            date TEXT PRIMARY KEY,
            open INTEGER,
            high INTEGER,
            low INTEGER,
            close INTEGER,
            volume INTEGER
        )
        """
        self.cursor.execute(query)
        self.conn.commit()

    def get_last_date(self, code):
        table_name = f"stock_{code}"
        # 테이블이 있는지 먼저 확인해요
        check_query = f"SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{table_name}'"
        self.cursor.execute(check_query)
        if self.cursor.fetchone()[0] == 0:
            return None
            
        # 가장 최근 날짜를 가져와요
        query = f"SELECT MAX(date) FROM {table_name}"
        self.cursor.execute(query)
        result = self.cursor.fetchone()
        return result[0] if result[0] else None

    def save_data(self, df, code):
        self.create_table(code)
        table_name = f"stock_{code}"
        
        data = []
        for _, row in df.iterrows():
            data.append((
                row['date'],
                int(row['open']), int(row['high']),
                int(row['low']), int(row['close']),
                int(row['volume'])
            ))
        
        # 날짜가 겹치면 덮어쓰기(Update)를 해요
        query = f"INSERT OR REPLACE INTO {table_name} VALUES (?, ?, ?, ?, ?, ?)"
        self.cursor.executemany(query, data)
        self.conn.commit()
        print(f"[{code}] {len(data)}일치 데이터 저장 완료")
    

Step 2. 키움증권 API 클래스 (KiwoomAPI)

여기가 오늘 수업의 핵심이에요. QAxWidget이라는 도구를 써서 키움증권 기능을 불러올 거예요.

코드 중간에 QEventLoop가 보일 텐데, 이게 바로 아까 말씀드린 진동벨 대기 기능이에요. 그리고 수정주가구분을 1로 설정하는 것도 절대 잊으면 안 돼요!

from PyQt5.QAxContainer import QAxWidget
from PyQt5.QtCore import QEventLoop
import time

class KiwoomAPI(QAxWidget):
    def __init__(self):
        super().__init__()
        # 키움증권 엔진을 장착합니다
        self.setControl("KHOPENAPI.KHOpenAPICtrl.1")
        
        self.login_loop = None
        self.request_loop = None 
        self.data = [] 

        # 이벤트(신호)가 오면 특정 함수를 실행하라고 연결해둡니다
        self.OnEventConnect.connect(self._on_login) # 로그인 결과가 오면
        self.OnReceiveTrData.connect(self._on_receive_tr_data) # 데이터가 오면

    def login(self):
        self.login_loop = QEventLoop()
        self.CommConnect() # 로그인 창 띄워주세요!
        self.login_loop.exec_() # 로그인 다 될 때까지 여기서 멈춰서 기다릴게요

    def _on_login(self, err_code):
        if err_code == 0:
            print("키움증권 서버 연결 성공")
        else:
            print("연결 실패")
        self.login_loop.exit() # 대기 끝, 다음으로 넘어갑니다

    def get_ohlcv(self, code, date=None):
        self.data = [] # 담을 그릇을 비워요
        
        # 입력값을 설정합니다
        self.set_input_value("종목코드", code)
        self.set_input_value("기준일자", date if date else time.strftime("%Y%m%d"))
        self.set_input_value("수정주가구분", "1") # [중요] 1:수정주가, 0:단순주가

        # 데이터를 요청합니다 (TR번호: opt10081)
        self.comm_rq_data("opt10081", "opt10081", 0, "0101")
        
        # 서버 응답을 기다립니다 (진동벨 받고 대기)
        self.request_loop = QEventLoop()
        self.request_loop.exec_()
        
        # 받은 데이터를 보기 좋게 DataFrame으로 만듭니다
        df = pd.DataFrame(self.data, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
        if not df.empty:
            df = df.sort_values('date').reset_index(drop=True)
            
        return df

    # 편의를 위해 만든 함수들
    def set_input_value(self, id, value):
        self.dynamicCall("SetInputValue(QString, QString)", id, value)

    def comm_rq_data(self, rqname, trcode, next, screen_no):
        self.dynamicCall("CommRqData(QString, QString, int, QString)", rqname, trcode, next, screen_no)

    # 데이터가 도착하면 실행되는 함수
    def _on_receive_tr_data(self, screen_no, rqname, trcode, record_name, next, unused1, unused2, unused3, unused4):
        if rqname == "opt10081":
            count = self.dynamicCall("GetRepeatCnt(QString, QString)", trcode, rqname)
            
            for i in range(count):
                date = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "일자").strip()
                open = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "시가").strip()
                high = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "고가").strip()
                low = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "저가").strip()
                close = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "현재가").strip()
                volume = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "거래량").strip()
                
                # 키움은 가격이 떨어지면 '-1000' 처럼 음수로 줘요. 그래서 절대값(abs) 처리가 필수예요.
                self.data.append([date, abs(int(open)), abs(int(high)), abs(int(low)), abs(int(close)), abs(int(volume))])
            
            # 대기 중인 루프를 종료합니다 (진동벨 울림)
            self.request_loop.exit()
    

Step 3. 메인 실행 파일

키움증권을 쓰려면 반드시 QApplication이라는 관리자가 필요해요. 코드는 길어 보이지만, 사실 "로그인하고 -> 리스트 읽어서 -> 데이터 저장한다"는 흐름은 단순해요.

from PyQt5.QtWidgets import QApplication
import sys
import time

def main():
    # PyQt5 관리자를 실행합니다 (필수!)
    app = QApplication(sys.argv)
    
    # API와 DB를 준비합니다
    kiwoom = KiwoomAPI()
    kiwoom.login() # 로그인 창이 뜰 거예요
    db = DBManager()

    # 타겟 리스트를 읽어옵니다
    try:
        with open('target_list.txt', 'r') as f:
            targets = [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        print("target_list.txt 파일이 없어요. 메모장에 종목코드를 적어서 저장해주세요.")
        return

    print(f"총 {len(targets)}개 종목 작업을 시작합니다.")

    for code in targets:
        print(f"[{code}] 데이터 요청 중...")
        
        try:
            df = kiwoom.get_ohlcv(code)
            if not df.empty:
                db.save_data(df, code)
        except Exception as e:
            print(f"오류가 났어요: {e}")
        
        # 키움은 1초에 5번까지만 요청할 수 있어요. 안전하게 0.5초 쉽니다.
        time.sleep(0.5)

    print("모든 작업이 끝났습니다. 고생하셨어요!")

if __name__ == "__main__":
    main()
    

혹시 데이터가 꼬였나요? (초기화)

가끔 데이터를 받다가 멈추거나 꼬일 때가 있어요. 그럴 땐 고민하지 말고 해당 종목을 싹 지우고 다시 받는 게 정신 건강에 좋아요.

import sqlite3

def reset_stock(code):
    conn = sqlite3.connect("stock.db")
    # 테이블을 아예 삭제해버립니다
    conn.execute(f"DROP TABLE IF EXISTS stock_{code}")
    conn.commit()
    conn.close()
    print(f"[{code}] 초기화 완료! 다시 받으세요.")
    

전체 소스 코드

위에서 설명한 코드들을 하나로 합친 완성본이에요. 아래 코드를 복사해서 main.py로 저장하고 실행해 보세요.

전체 소스 코드 보기 (클릭)

import sys
import time
import sqlite3
import pandas as pd
from PyQt5.QtWidgets import QApplication
from PyQt5.QAxContainer import QAxWidget
from PyQt5.QtCore import QEventLoop

# 1. DB Manager (데이터 저장 담당)
class DBManager:
    def __init__(self, db_name="stock.db"):
        self.conn = sqlite3.connect(db_name)
        self.cursor = self.conn.cursor()

    def create_table(self, code):
        table_name = f"stock_{code}"
        query = f"""
        CREATE TABLE IF NOT EXISTS {table_name} (
            date TEXT PRIMARY KEY,
            open INTEGER,
            high INTEGER,
            low INTEGER,
            close INTEGER,
            volume INTEGER
        )
        """
        self.cursor.execute(query)
        self.conn.commit()

    def save_data(self, df, code):
        self.create_table(code)
        table_name = f"stock_{code}"
        data = []
        for _, row in df.iterrows():
            data.append((
                row['date'], int(row['open']), int(row['high']),
                int(row['low']), int(row['close']), int(row['volume'])
            ))
        query = f"INSERT OR REPLACE INTO {table_name} VALUES (?, ?, ?, ?, ?, ?)"
        self.cursor.executemany(query, data)
        self.conn.commit()
        print(f"[{code}] {len(data)}건 저장")

# 2. Kiwoom API (데이터 수집 담당)
class KiwoomAPI(QAxWidget):
    def __init__(self):
        super().__init__()
        self.setControl("KHOPENAPI.KHOpenAPICtrl.1")
        self.login_loop = None
        self.request_loop = None 
        self.data = []

        self.OnEventConnect.connect(self._on_login)
        self.OnReceiveTrData.connect(self._on_receive_tr_data)

    def login(self):
        self.login_loop = QEventLoop()
        self.CommConnect()
        self.login_loop.exec_()

    def _on_login(self, err_code):
        if err_code == 0:
            print("로그인 성공")
        else:
            print("로그인 실패")
        self.login_loop.exit()

    def get_ohlcv(self, code):
        self.data = []
        self.set_input_value("종목코드", code)
        self.set_input_value("기준일자", time.strftime("%Y%m%d"))
        self.set_input_value("수정주가구분", "1") # 1:수정주가

        self.comm_rq_data("opt10081", "opt10081", 0, "0101")
        
        self.request_loop = QEventLoop()
        self.request_loop.exec_()
        
        df = pd.DataFrame(self.data, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
        if not df.empty:
            df = df.sort_values('date').reset_index(drop=True)
        return df

    def set_input_value(self, id, value):
        self.dynamicCall("SetInputValue(QString, QString)", id, value)

    def comm_rq_data(self, rqname, trcode, next, screen_no):
        self.dynamicCall("CommRqData(QString, QString, int, QString)", rqname, trcode, next, screen_no)

    def _on_receive_tr_data(self, screen_no, rqname, trcode, record_name, next, unused1, unused2, unused3, unused4):
        if rqname == "opt10081":
            count = self.dynamicCall("GetRepeatCnt(QString, QString)", trcode, rqname)
            for i in range(count):
                date = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "일자").strip()
                open = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "시가").strip()
                high = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "고가").strip()
                low = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "저가").strip()
                close = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "현재가").strip()
                volume = self.dynamicCall("GetCommData(QString, QString, int, QString)", trcode, rqname, i, "거래량").strip()
                
                self.data.append([date, abs(int(open)), abs(int(high)), abs(int(low)), abs(int(close)), abs(int(volume))])
            
            self.request_loop.exit()

# 3. Main (실행)
def main():
    app = QApplication(sys.argv)
    
    kiwoom = KiwoomAPI()
    kiwoom.login()
    db = DBManager()

    try:
        with open('target_list.txt', 'r') as f:
            targets = [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        print("target_list.txt 파일이 없습니다.")
        return

    print(f"총 {len(targets)}개 종목 작업 시작")

    for code in targets:
        print(f"[{code}] 요청 중...")
        try:
            df = kiwoom.get_ohlcv(code)
            if not df.empty:
                db.save_data(df, code)
        except Exception as e:
            print(f"오류: {e}")
        
        time.sleep(0.5) 

    print("완료되었습니다.")

if __name__ == "__main__":
    main()
            

자, 이제 키움증권에서도 원하는 종목의 데이터를 마음껏 가져올 수 있게 되었어요. 어렵게만 느껴지던 PyQt5와 이벤트 루프도 직접 써보니 별것 아니죠? 그냥 진동벨 받고 기다리는 거랑 똑같아요.

다음 편에서는 시스템 트레이딩 고수들이 숨겨놓고 쓴다는 LS증권(구 이베스트) 편으로 찾아올게요. 그때까지 DB에 데이터 꽉 채워두세요!

*본 포스팅은 실무 경험을 바탕으로 작성된 기술 참고 자료입니다. 제공된 코드와 전략은 수익을 보장하지 않으며, 모든 투자에 대한 최종 결정과 책임은 투자자 본인에게 있습니다. 실제 매매에 적용하기 전 반드시 충분한 검증과 모의 투자를 거치시기 바랍니다.*