자동 매매 시스템 구현

Posted Jan 10, 2026

By humkyung

22 min read

한국투자증권 API 연동 가이드

개요

본 문서는 ‘자동 주식 매매 프로그램 개발’ 프로젝트의 핵심 기능인 기능 3: 자동 매매 시스템을 구현하기 위한 한국투자증권(이하 한투)의 API 연동 절차 및 Python 구현 가이드를 제공합니다. Streamlit 기반 웹 서비스의 백엔드 모듈인 trader.py를 중심으로 매수/매도 주문 로직을 상세히 설명합니다.

1. API 연동 개요 및 준비 사항

자동 매매를 위해서는 한국투자증권의 Open API (eFriend Expert) 접근 권한이 필수적입니다.

1.1. 필수 준비 사항

리소스	내용	비고
API 키	앱 키(App Key) 및 앱 시크릿(App Secret)	한투 개발자 센터에서 발급 필요 한국투자증권 KIS Developers 접속 로그인 후 [서비스 신청] -> [Open API] 신청 앱 키(App Key) 와 앱 시크릿(App Secret) 발급받기 (초보자 추천) 모의투자 신청을 먼저 하시는 것을 권장합니다. (실전 투자와 API 주소가 다릅니다.)
계좌 정보	계좌번호 및 계좌 비밀번호	매매 주문 시 사용
API 연동 문서	File한국투자증권 API 연동 상세 문서	개발자 센터에서 다운로드 가능
보안 환경 설정	`.env` 파일에 API 키 저장	절대 코드에 직접 노출 금지

1.2. API 통신 방식

한투 API는 크게 두 가지 통신 방식을 사용합니다.

REST API (HTTP 통신): 잔고 조회, 계좌 정보 조회, 매수/매도 주문 실행 등 1회성 요청 및 응답 처리에 사용됩니다.
WebSocket (실시간 통신): 실시간 체결 알림, 실시간 주가 변동 수신 등 지속적인 데이터 스트리밍에 사용됩니다.

2. Python (trader.py) 모듈 구성

자동 매매 로직은 modules/trader.py 파일에 구현되며, API 인증, 주문 실행, 잔고 관리를 담당합니다.

2.1. 필수 라이브러리 설치

pip install requests

2.2. `.env` 파일 설정 (보안)

프로젝트 루트의 .env 파일을 열고(없으면 생성), 발급받은 정보를 입력하세요. (이 파일은 에 따라 Git에 업로드하면 안 됩니다.)

  
# .env 파일 내용
# 실전투자: PROD, 모의투자: VIRTUAL
KIS_MODE=VIRTUAL # 캔들/시세 조회용 (모의/실전 키가 다를 수 있음) 

KIS_APP_KEY=발급받은_APP_KEY_붙여넣기
KIS_APP_SECRET=발급받은_APP_SECRET_붙여넣기

# 계좌 정보 (계좌번호 8자리 + 2자리)
KIS_ACCOUNT_NO=12345678
KIS_ACCOUNT_CODE=01

2.2. Trader 클래스 구조

한투의 경우 접근 토큰은 1분당 1회만 생성 가능합니다.
단타처럼 1분 내에 여러 건을 거래할 때는 거래할 때마다 토큰을 생성할 수 없으므로 생성한 토큰을 저장하고 재 사용해야 합니다.

  
# modules/trader.py
import os
import requests
import json
from datetime import datetime, timedelta
import streamlit as st
from dotenv import load_dotenv

TOKEN_FILE = "token.json"
TOKEN_SAFETY_MIN = 5  # 만료 5분 전이면 갱신
TOKEN_EXPIRE_HOURS = 23  # 24시간보다 조금 짧게

# .env 파일 로드
load_dotenv()

class KisTrader:
    """
    한국투자증권(KIS) REST API 연동 클래스
    """
    def __init__(self):
        self.mode = os.getenv("KIS_MODE", "VIRTUAL")
        self.app_key = os.getenv("KIS_APP_KEY")
        self.app_secret = os.getenv("KIS_APP_SECRET")
        self.account_no = os.getenv("KIS_ACCOUNT_NO") # 계좌번호 앞 8자리
        self.account_code = os.getenv("KIS_ACCOUNT_CODE", "01") # 계좌번호 뒤 2자리
        
        # 모의투자 vs 실전투자 URL 설정
        if self.mode == "PROD":
            self.base_url = "https://openapi.koreainvestment.com:9443"
        else:
            self.base_url = "https://openapivts.koreainvestment.com:29443"
            
        self.access_token = None
        self._auth() # 초기화 시 바로 인증 토큰 발급 시도

    def _auth(self):
        """
        접근 토큰(Access Token) 발급 (1일 1회 갱신 필요)
        """

# 1️⃣ 캐시된 토큰 먼저 확인
        cached_token = self._load_cached_token()
        if cached_token:
    self.access_token = cached_token
    self.token_issued_at = issued_at # issue 날짜 저장
    print("♻️ 캐시된 토큰 사용")
           return

        url = f"{self.base_url}/oauth2/tokenP"
        headers = {"content-type": "application/json"}
        body = {
            "grant_type": "client_credentials",
            "appkey": self.app_key,
            "appsecret": self.app_secret
        }
        
        try:
            res = requests.post(url, headers=headers, data=json.dumps(body))
            if res.status_code == 200:
                self.access_token = res.json()["access_token"]
                self.token_issued_at = datetime.now()
   self._save_token(self.access_token) # 토큰 저장
                print("✅ 한국투자증권 토큰 발급 성공")
            else:
                print(f"❌ 토큰 발급 실패: {res.text}")
                self.access_token = None
                self.token_issued_at = None
        except Exception as e:
            print(f"❌ 인증 중 오류 발생: {e}")
            self.access_token = None
            self.token_issued_at = None

    def _load_cached_token(self):
        if not os.path.exists(TOKEN_FILE):
            return None, None

        try:
            with open(TOKEN_FILE, "r", encoding="utf-8") as f:
                data = json.load(f)

            token = data.get("access_token")
            issued_at = datetime.fromisoformat(data["issued_at"])

            if not token or not issued_at:
                return None, None

            issued_at = datetime.fromisoformat(data["issued_at"])
            if datetime.now() - issued_at < timedelta(hours=TOKEN_EXPIRE_HOURS):
                return token, issued_at

        except Exception:
            return None, None

        return None, None

    def _save_token(self, token):
        with open(TOKEN_FILE, "w", encoding="utf-8") as f:
            json.dump(
                {
                    "access_token": token,
                    "issued_at": datetime.now().isoformat(),
                },
                f,
                ensure_ascii=False,
                indent=2,
            )

    def _ensure_token(self):
        if not self.access_token or not self.token_issued_at:
            self._auth()
            return

        # issued_at 기준 23시간 초과 시 재발급
        if datetime.now() - self.token_issued_at >= timedelta(hours=TOKEN_EXPIRE_HOURS):
            print("🔄 토큰 23시간 초과 → 재발급")
            self._auth()

    def _get_common_headers(self, tr_id):
        """
        API 호출에 필요한 공통 헤더 생성
        """
        self._ensure_token()
        return {
            "content-type": "application/json; charset=utf-8",
            "authorization": f"Bearer {self.access_token}",
            "appkey": self.app_key,
            "appsecret": self.app_secret,
            "tr_id": tr_id
        }

    def get_balance(self):
        """
        주식 잔고 조회 (TTTC8434R: 주식잔고조회_실전 / VTTC8434R: 주식잔고조회_모의)
        """
        url = f"{self.base_url}/uapi/domestic-stock/v1/trading/inquire-balance"
        
        # 모의투자와 실전투자 TR ID가 다름
        tr_id = "VTTC8434R" if self.mode == "VIRTUAL" else "TTTC8434R"
        
        headers = self._get_common_headers(tr_id)
        
        params = {
            "CANO": self.account_no,
            "ACNT_PRDT_CD": self.account_code,
            "AFHR_FLPR_YN": "N",
            "OFL_YN": "",
            "INQR_DVSN": "02",
            "UNPR_DVSN": "01",
            "FUND_STTL_ICLD_YN": "N",
            "FNCG_AMT_AUTO_RDPT_YN": "N",
            "PRCS_DVSN": "00",
            "CTX_AREA_FK100": "",
            "CTX_AREA_NK100": ""
        }
        
        try:
            res = requests.get(url, headers=headers, params=params)
            data = res.json()
            if res.status_code == 200 and data['rt_cd'] == '0':
                return data['output1'] # 보유 종목 리스트
            else:
                print(f"잔고 조회 실패: {data.get('msg1')}")
                return []
        except Exception as e:
            print(f"잔고 조회 에러: {e}")
            return []

    def send_order(self, ticker, quantity, price, order_type="buy"):
        """
        주문 실행 (지정가 기준)
        Args:
            ticker: 종목코드 (6자리)
            quantity: 수량
            price: 가격 (0이면 시장가)
            order_type: 'buy' (매수) or 'sell' (매도)
        """
        url = f"{self.base_url}/uapi/domestic-stock/v1/trading/order-cash"
        
        # TR ID 설정 (매수/매도 구분)
        if self.mode == "VIRTUAL":
            tr_id = "VTTC0802U" if order_type == "buy" else "VTTC0801U"
        else:
            tr_id = "TTTC0802U" if order_type == "buy" else "TTTC0801U"

        headers = self._get_common_headers(tr_id)
        
        data = {
            "CANO": self.account_no,
            "ACNT_PRDT_CD": self.account_code,
            "PDNO": ticker,
            "ORD_DVSN": "01", # 00:지정가, 01:시장가
            "ORD_QTY": str(quantity),
            "ORD_UNPR": str(price) if price > 0 else "0",
        }
        
        # 지정가일 경우 '00'으로 변경
        if price > 0:
            data["ORD_DVSN"] = "00"

        try:
            res = requests.post(url, headers=headers, data=json.dumps(data))
            result = res.json()
            if result['rt_cd'] == '0':
                print(f"✅ {order_type} 주문 성공: {result['msg1']}")
                return True
            else:
                print(f"❌ 주문 실패: {result['msg1']}")
                return False
        except Exception as e:
            print(f"❌ 주문 중 에러: {e}")
            return False

2.3. 연결 테스트 (`test_trader.py`)

제대로 설정되었는지 확인하기 위해 테스트 파일을 생성하고 실행해 보세요.

  
# test_trader.py
from modules.trader import KisTrader

print("--- 한국투자증권 API 테스트 ---")

# 1. 객체 생성 및 로그인(토큰발급)
bot = KisTrader()

if bot.access_token:
    # 2. 잔고 조회 테스트
    print("\n[잔고 조회]")
    balance = bot.get_balance()
    if balance:
        for stock in balance:
            print(f"종목: {stock['prdt_name']}, 수량: {stock['hldg_qty']}, 수익률: {stock['evlu_pfls_rt']}%")
    else:
        print("보유 종목이 없거나 조회 실패")

    # 3. (주의) 매수 주문 테스트 - 모의투자일 경우만 주석 해제하세요
    # print("\n[매수 테스트]")
    # bot.send_order("005930", 1, 0, "buy") # 삼성전자 1주 시장가 매수
else:
    print("API 연결 실패. .env 파일을 확인하세요.")

토요일 오전 8시 30분 경에 테스트해보니 아래와 같은 메시지가 표시됩니다.

3. Streamlit과의 연동 및 자동 매매 로직

Streamlit 앱 (main.py)에서는 사용자 설정값(목표 매수/매도 가격)을 trader.py의 함수에 전달하고, 실시간 주가와 비교하여 매매 조건을 실행합니다.

Streamlit에서 자동 매매 버튼(UI)과 trader.py(로직)를 연결하려면 Streamlit의 실행 방식(Session State) 을 이해해야 합니다.

Streamlit은 버튼을 누르거나 입력 값이 바뀌면 스크립트 전체가 다시 실행되는 구조입니다. 따라서, 자동 매매가 켜져 있는 동안 주기적으로 시세를 확인하고 매매 로직을 실행하는 “루프(Loop)”를 main.py에 구현해야 합니다.

또한, 중복 주문 방지를 위해 “이미 매수했는지”를 기억하는 상태 관리(st.session_state)가 필수적입니다.

3.1. `modules/trader.py` 확인 (싱글톤 패턴 적용)

매번 리프레시될 때마다 API 토큰을 새로 발급 받으면 비효율적입니다. Streamlit의 캐싱 기능을 활용하여 KisTrader 객체를 한 번만 생성하도록 합니다.

기존 modules/trader.py 파일 하단에 아래 코드를 추가하거나, main.py에서 이 방식으로 호출할 것입니다.

# modules/trader.py 파일 내 클래스 정의 아래에 추가할 필요는 없지만,
# main.py에서 @st.cache_resource를 사용할 예정입니다.

3.2. 자동 매매 실행 함수 (main.py) 업데이트

main.py를 대폭 수정하여, 사이드바의 설정 값(config)을 받아 실제 매매 로직을 수행하도록 연결합니다.

주요 변경점:

@st.cache_resource: KisTrader를 한 번만 로그인하고 계속 재사용합니다.
st.session_state: 매수/매도 상태를 저장하여 중복 주문을 막습니다.
자동 리프레시: 자동 매매가 켜져 있으면 time.sleep() 후 st.rerun()을 호출하여 계속 감시합니다.

  
# main.py
import time
import streamlit as st
import pandas as pd
from modules.scraper import StockScraper, fetch_stock_history, fetch_stock_info
from modules.trader import KisTrader
from ui.sidebar import render_sidebar
from ui.dashboard import render_dashboard

# 페이지 기본 설정
st.set_page_config(page_title="AutoTrade Pro", page_icon="📈", layout="wide")

# 1. Trader 객체 캐싱 (앱 실행 중 1회만 로그인)
@st.cache_resource
def get_trader():
    return KisTrader()

# 2. 세션 상태 초기화 (중복 주문 방지용)
if 'bought_status' not in st.session_state:
    st.session_state['bought_status'] = {} # {ticker: True/False}

def main():
    # 사이드바 렌더링
    config = render_sidebar()
    ticker = config['ticker']
    period = config['period']
    is_auto = config['is_auto']
    
    # 목표가 설정값 가져오기 (sidebar.py에서 반환값에 추가되어야 함)
    # ※ ui/sidebar.py의 return 딕셔너리에 target_buy, target_sell을 추가했다고 가정
    # 수정된 sidebar.py 코드는 아래 '참고' 섹션 확인
    target_buy = st.session_state.get('target_buy', 0) 
    target_sell = st.session_state.get('target_sell', 0)

    st.title("📈 AI Stock Trading Dashboard")

    # 데이터 수집
    if ticker:
        with st.spinner('데이터 수집 중...'):
            df = fetch_stock_history(ticker, period)
            info = fetch_stock_info(ticker)
            scraper = StockScraper(ticker)
            news = scraper.get_news()

        # 대시보드 표시
        if info and not df.empty:
            render_dashboard(df, info, news)
            current_price = df['Close'].iloc[-1]
        else:
            st.error("데이터 오류")
            return

        # ---------------------------------------------------------
        # [핵심] 자동 매매 로직 연결
        # ---------------------------------------------------------
        if is_auto:
            st.divider()
            st.subheader("🤖 자동 매매 모니터링")
            
            status_cols = st.columns(4)
            status_cols[0].metric("현재가", f"{current_price:,.0f}")
            status_cols[1].metric("목표 매수가", f"{config['target_buy']:,.0f}")
            status_cols[2].metric("목표 매도가", f"{config['target_sell']:,.0f}")
            
            # 로그 창 (컨테이너)
            log_container = st.empty()

            trader = get_trader()
            
            # 매수 로직
            # 1. 목표가가 설정되어 있고
            # 2. 현재가가 목표가보다 낮거나 같으며
            # 3. 아직 매수하지 않은 상태일 때
            if config['target_buy'] > 0 and current_price <= config['target_buy']:
                if not st.session_state['bought_status'].get(ticker, False):
                    log_container.warning(f"⚡ 매수 조건 충족! ({current_price} <= {config['target_buy']}) 주문 실행 중...")
                    
                    # API 주문 실행 (수량 1주로 고정 예시)
                    success = trader.send_order(ticker, 1, 0, "buy") 
                    
                    if success:
                        st.session_state['bought_status'][ticker] = True
                        st.success(f"✅ {ticker} 1주 매수 완료!")
                        time.sleep(1) # 메시지 확인용 대기
                    else:
                        st.error("❌ 매수 주문 실패")
                else:
                    status_cols[3].info("상태: 이미 매수함")

            # 매도 로직
            elif config['target_sell'] > 0 and current_price >= config['target_sell']:
                if st.session_state['bought_status'].get(ticker, False):
                    log_container.warning(f"⚡ 매도 조건 충족! ({current_price} >= {config['target_sell']}) 주문 실행 중...")
                    
                    success = trader.send_order(ticker, 1, 0, "sell")
                    
                    if success:
                        st.session_state['bought_status'][ticker] = False # 매도했으므로 상태 초기화
                        st.success(f"✅ {ticker} 1주 매도 완료!")
                        time.sleep(1)
                    else:
                        st.error("❌ 매도 주문 실패")
                else:
                    status_cols[3].info("상태: 보유 주식 없음")
            
            else:
                log_container.info("⏳ 조건 감시 중... (특이사항 없음)")

            # 자동 리프레시 (3초마다 재실행하여 실시간 감시 효과)
            time.sleep(3)
            st.rerun()

if __name__ == "__main__":
    main()

3.3. `ui/sidebar.py` 수정 (필수)

app.py에서 목표 가격을 읽을 수 있도록, render_sidebar 함수가 반환하는 딕셔너리에 가격 정보를 포함해야 합니다.

  
# ui/sidebar.py 수정 부분
def render_sidebar():
    # ... (이전 코드 동일) ...
    
    # 3. 자동 매매 조건
    st.sidebar.subheader("2. 자동 매매 조건")
    # key를 지정하여 session_state에 자동 저장되게 함
    target_buy_price = st.sidebar.number_input("목표 매수가 ($)", min_value=0.0, value=0.0, step=1.0, key="target_buy")
    target_sell_price = st.sidebar.number_input("목표 매도가 ($)", min_value=0.0, value=0.0, step=1.0, key="target_sell")
    
    is_auto_trading = st.sidebar.toggle("🤖 자동 매매 활성화")

    # ... (버튼 코드 등) ...

    return {
        "ticker": ticker.upper(),
        "period": period,
        "run_btn": run_btn,
        "is_auto": is_auto_trading,
        "target_buy": target_buy_price,  # 추가됨
        "target_sell": target_sell_price # 추가됨
    }

2026-01-10 오전 8시 30분 현재 애플을 $260로 자동 매매를 실행해봤습니다.

매매불가 종목으로 나오네요.

작동 방식 설명

사용자 설정: 사이드바에서 ‘목표 매수가’를 설정하고 ‘자동 매매 활성화’ 토글을 켭니다.
무한 반복(Loop) 효과: main.py의 if is_auto: 블록이 실행됩니다.
조건 비교: current_price와 config['target_buy']를 비교합니다.
주문 실행: 조건이 맞으면 trader.send_order()를 호출하여 한국투자증권 API로 주문을 전송합니다.
상태 저장: st.session_state['bought_status']를 True로 변경하여, 다음 루프 때 또 매수 주문이 나가는 것을 방지합니다.
리프레시: time.sleep(3) 후 st.rerun()이 실행되어 화면을 새로고침하고 최신 주가를 다시 가져옵니다.

테스트 팁

반드시 .env 파일의 KIS_MODE가 VIRTUAL(모의투자)인지 확인하세요.
장 운영 시간(09:00~15:30)에 테스트하거나, 모의투자의 경우 장 운영 시간에만 주문이 접수될 수 있습니다.
target_buy를 현재가보다 약간 높게 설정하면 즉시 매수 로직이 발동되어 테스트하기 좋습니다.

4. 개발 참고 사항 및 위험 요소

항목	상세 내용
보안	API 키, 비밀번호는 `.env` 파일로 관리하고, 절대 Git 저장소에 업로드하지 않습니다(.gitignore 필수).
슬리피지	실시간 주가 조회 후 주문이 서버에 도달하기까지 주가가 변동될 수 있습니다. 시장가 주문 또는 호가 조회 후 지정가 주문을 고려해야 합니다.
API 제한	한투 API는 초당/일일 요청 횟수 제한이 있을 수 있습니다. `time.sleep()`을 적절히 사용하여 과도한 요청을 방지해야 합니다.
테스트 환경	실제 자산 손실을 방지하기 위해, 개발 초기에 반드시 한투의 모의 투자 서버 (VTS) URL을 사용하여 테스트해야 합니다. (`api_base_url` 변경 필요)

5. 연락처 및 다음 일정

구현 관련 상세 문의 사항은 백흠경 에게 연락 주시기 바랍니다.

증권사 API 연동 완료 보고 및 통합 테스트 계획을 논의하기 위한 미팅은 Calendar event 일정에 진행될 예정입니다.

자동 주식 매매 프로그램

한국투자증권