본문 바로가기
시스템 트레이딩 소개/데이터 수집

나만의 주식 DB 만들기 (대신증권 편)

by 흔한트리이더 2026. 2. 10.
반응형
이 글의 목표
1. 대신증권 API(Cybos Plus)를 활용해 안정적으로 주식 데이터를 수집합니다.
2. yfinance의 한계를 넘어, 수정 주가(Adjusted Price)가 반영된 깨끗한 데이터를 확보합니다.
3. 1종목 1테이블 구조로 나만의 주식 DB(SQLite)를 구축하는 코드를 완성합니다.

시스템 트레이딩을 시작하려고 마음먹었을 때, 가장 먼저 부딪히는 벽이 바로 데이터입니다. 무료로 제공되는 yfinance는 정말 훌륭하지만, 한국 주식시장의 액면분할이나 배당락 같은 이벤트를 제대로 반영하지 못할 때가 많거든요.

그래서 오늘은 여러분의 컴퓨터 안에 나만의 증권사 서버를 구축해 볼 거예요. 첫 번째 순서는 바로 대신증권(Cybos Plus)입니다.

많은 고수분들이 시스템 트레이딩용으로 대신증권을 추천하시는데요. 그 이유는 키움증권보다 코드가 훨씬 간결하고, 서버 연결이 아주 안정적이기 때문이에요. 저도 복잡한 로직은 대신증권으로 돌리고 있답니다.


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

대신증권 API도 키움증권과 마찬가지로 32비트(x86) 환경에서만 작동해요. 요즘 컴퓨터는 대부분 64비트라서, 그냥 실행하면 에러가 날 거예요.

아나콘다(Anaconda)를 사용해서 32비트 전용 공간을 하나 만들어주세요.

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

설치가 끝났다면, 대신증권 Cybos Plus 프로그램을 관리자 권한으로 실행해서 로그인을 해주세요. 작업표시줄 트레이에 아이콘이 보이면 준비 끝입니다.


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

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

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

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

Step 1. DB 관리자 (DBManager)

SQLite를 다루는 부분이에요. 종목 코드를 주면 알아서 테이블을 만들고 데이터를 저장해 줍니다. 복잡한 SQL 문법은 몰라도 괜찮아요. 이 코드가 다 알아서 해주니까요.


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 클래스 (CreonAPI)

대신증권 API의 핵심은 win32com 라이브러리를 사용한다는 점이에요. 엑셀 매크로를 다뤄보신 분이라면 익숙하실 거예요.

여기서 가장 중요한 건 수정주가 요청입니다. SetInputValue(9, ord('1')) 이 한 줄이 없으면, 액면분할한 종목의 차트가 엉망으로 나오니 꼭 챙겨주세요.


import win32com.client

class CreonAPI:
    def __init__(self):
        # 대신증권 차트 객체를 불러옵니다
        self.obj_stock_chart = win32com.client.Dispatch("CpSysDib.StockChart")

    def get_ohlcv(self, code, start_date=None):
        # 대신증권 종목코드는 'A'로 시작해야 해요 (예: A005930)
        if not code.startswith('A'):
            cybos_code = 'A' + code
        else:
            cybos_code = code

        self.obj_stock_chart.SetInputValue(0, cybos_code)
        self.obj_stock_chart.SetInputValue(1, ord('1'))  # 기간으로 요청할게요
        self.obj_stock_chart.SetInputValue(2, 0) # 종료일 (오늘)
        self.obj_stock_chart.SetInputValue(3, start_date if start_date else '19900101') # 시작일
        self.obj_stock_chart.SetInputValue(4, 0) 
        self.obj_stock_chart.SetInputValue(5, [0, 2, 3, 4, 5, 8]) # 날짜, 시가, 고가, 저가, 종가, 거래량
        self.obj_stock_chart.SetInputValue(6, ord('D'))  # 일봉
        
        # [매우 중요] 수정주가(Adjusted Price)를 적용해주세요!
        self.obj_stock_chart.SetInputValue(9, ord('1'))

        # 서버에 요청하고 기다립니다 (Blocking)
        # 키움증권과 달리 여기서 코드가 잠시 멈추고 응답을 기다려줘요. 훨씬 편하죠?
        self.obj_stock_chart.BlockRequest()

        count = self.obj_stock_chart.GetHeaderValue(3)
        data_list = []

        for i in range(count):
            data_list.append({
                'date': str(self.obj_stock_chart.GetDataValue(0, i)),
                'open': self.obj_stock_chart.GetDataValue(1, i),
                'high': self.obj_stock_chart.GetDataValue(2, i),
                'low': self.obj_stock_chart.GetDataValue(3, i),
                'close': self.obj_stock_chart.GetDataValue(4, i),
                'volume': self.obj_stock_chart.GetDataValue(5, i),
            })
        
        # API는 최신순으로 데이터를 줘요. 우리가 보기 편하게 과거->현재 순으로 뒤집을게요.
        df = pd.DataFrame(data_list)
        if not df.empty:
            df = df.sort_values('date').reset_index(drop=True)
            
        return df

Step 3. 메인 실행 파일

target_list.txt 파일에 원하는 종목 코드를 적고 실행하면 끝입니다. 대신증권은 별도의 이벤트 루프 관리가 필요 없어서 메인 코드가 아주 깔끔해요.


# target_list.txt 예시:
# 005930
# 000660

import time
from datetime import datetime, timedelta

def main():
    db = DBManager()
    api = CreonAPI()
    today = datetime.now().strftime("%Y%m%d")

    # 파일에서 종목 리스트를 읽어옵니다
    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:
        # 1. DB에 저장된 마지막 날짜를 확인해요
        last_date = db.get_last_date(code)
        
        start_date = None
        if last_date:
            # 마지막 날짜 다음 날부터 요청해요 (이어받기)
            last_dt = datetime.strptime(last_date, "%Y%m%d")
            next_dt = last_dt + timedelta(days=1)
            start_date = next_dt.strftime("%Y%m%d")
            
            if start_date > today:
                print(f"[{code}] 이미 최신 데이터입니다.")
                continue
            print(f"[{code}] 업데이트 시작 (기준일: {last_date})")
        else:
            print(f"[{code}] 신규 다운로드 (전체 기간)")
            start_date = '20000101'

        # 2. 데이터를 요청하고 저장해요
        df = api.get_ohlcv(code, start_date=start_date)
        if not df.empty:
            db.save_data(df, code)
        else:
            print(f"[{code}] 가져올 데이터가 없어요.")

        # 너무 빨리 요청하면 차단당할 수 있으니 0.3초 쉬어줍니다
        time.sleep(0.3)

    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 sqlite3
import pandas as pd
import win32com.client
import time
import os
from datetime import datetime, timedelta

# 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 get_last_date(self, code):
        table_name = f"stock_{code}"
        try:
            query = f"SELECT MAX(date) FROM {table_name}"
            self.cursor.execute(query)
            result = self.cursor.fetchone()
            return result[0] if result[0] else None
        except sqlite3.OperationalError:
            return 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'])
            ))
        query = f"INSERT OR REPLACE INTO {table_name} VALUES (?, ?, ?, ?, ?, ?)"
        self.cursor.executemany(query, data)
        self.conn.commit()
        print(f"[{code}] {len(data)}건 저장")

# 2. Creon API (대신증권 통신 담당)
class CreonAPI:
    def __init__(self):
        self.obj_stock_chart = win32com.client.Dispatch("CpSysDib.StockChart")

    def get_ohlcv(self, code, start_date='19900101'):
        if not code.startswith('A'):
            cybos_code = 'A' + code
        else:
            cybos_code = code

        self.obj_stock_chart.SetInputValue(0, cybos_code)
        self.obj_stock_chart.SetInputValue(1, ord('1'))
        self.obj_stock_chart.SetInputValue(2, 0)
        self.obj_stock_chart.SetInputValue(3, start_date)
        self.obj_stock_chart.SetInputValue(4, 0)
        self.obj_stock_chart.SetInputValue(5, [0, 2, 3, 4, 5, 8])
        self.obj_stock_chart.SetInputValue(6, ord('D'))
        self.obj_stock_chart.SetInputValue(9, ord('1'))

        self.obj_stock_chart.BlockRequest()

        count = self.obj_stock_chart.GetHeaderValue(3)
        data_list = []
        for i in range(count):
            data_list.append({
                'date': str(self.obj_stock_chart.GetDataValue(0, i)),
                'open': self.obj_stock_chart.GetDataValue(1, i),
                'high': self.obj_stock_chart.GetDataValue(2, i),
                'low': self.obj_stock_chart.GetDataValue(3, i),
                'close': self.obj_stock_chart.GetDataValue(4, i),
                'volume': self.obj_stock_chart.GetDataValue(5, i),
            })
        
        df = pd.DataFrame(data_list)
        if not df.empty:
            df = df.sort_values('date').reset_index(drop=True)
        return df

# 3. Main (실행)
def main():
    if not os.path.exists('target_list.txt'):
        print("target_list.txt 파일이 없습니다.")
        return

    db = DBManager()
    api = CreonAPI()
    today = datetime.now().strftime("%Y%m%d")

    with open('target_list.txt', 'r') as f:
        targets = [line.strip() for line in f if line.strip()]

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

    for code in targets:
        last_date = db.get_last_date(code)
        start_date = '20000101'
        
        if last_date:
            last_dt = datetime.strptime(last_date, "%Y%m%d")
            next_dt = last_dt + timedelta(days=1)
            start_date = next_dt.strftime("%Y%m%d")
            
            if start_date > today:
                print(f"[{code}] 최신 데이터 보유 중")
                continue
            print(f"[{code}] 업데이트 중... (기준: {last_date})")
        else:
            print(f"[{code}] 신규 다운로드 중...")

        try:
            df = api.get_ohlcv(code, start_date)
            if not df.empty:
                db.save_data(df, code)
            else:
                print(f"[{code}] 데이터 없음")
        except Exception as e:
            print(f"[{code}] 오류 발생: {e}")
        
        time.sleep(0.3)

    print("작업 완료")

if __name__ == "__main__":
    main()

 

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

'시스템 트레이딩 소개 > 데이터 수집' 카테고리의 다른 글

나만의 주식 DB 만들기 (LS증권 편)  (0) 2026.02.10
나만의 주식 DB 만들기 (키움증권 편)  (0) 2026.02.10
기술적 분석의 첫 단추, 데이터 무결성 확보하기 (yfinance)  (0) 2026.02.07

관련글

티스토리툴바