YesXYZ

아딸이 함께 만드는 세상

한국투자증권 모의투자/실전투자 트레이딩 유틸리티 구성 방법

by · 2024년 09월 11일

한국투자증권의 Open API와 Python을 사용해서 모의투자/실적투자 트레이딩 유틸리티를 구성하는 절차를 알아보겠습니다. (Ⅰ단계) Python 코딩 환경 설정와 (Ⅱ단계) 문자메세지 수신 환경 설정은 별도의 글로 작성하였으며, 다른 방법을 사용해도 됩니다.

Ⅰ. Python 코딩 환경 설정: VS Code 사용하기
Ⅱ. 문자메세지 수신 환경 설정: Discord 웹후크 사용하기
Ⅲ. 한국투자증권 모의투자 ID 및 계좌번호 생성
Ⅳ. 한국투자증권 Open API 신청: 모의투자/실전투자
Ⅴ. Python 프로그램 등록/수정 및 모의투자 테스트

 

Ⅲ. 한국투자증권 모의투자 ID 및 계좌번호 생성
  1. 한국투자증권 홈페이지에 로그인합니다.
    – ID와 증권계좌가 없다면 신규로 가입합니다.
  2. 모의투자를 신청합니다.
    • “트레이딩 > 모의투자 > 주식/선물옵션 모의투자 > 모의투자안내”로 이동
      한국투자증권 자동 트레이딩
    • 모의투자 “신청/재도전” 탭으로 이동한 후 참여하고자 하는 리그구분을 선택(여기서는 국내주식 5억원, 3개월 선택)하고 “신청” 버튼을 클릭
      한국투자증권 자동 트레이딩
    • 신청이 완료되면 등록 완료 메세지가 표시되며 “나의계좌” 탭으로 이동하여 모의투자를 위한 계좌번호를 확인
      – 모의투자 계좌번호는 복사한 후 메모장 등에 저장해 둡니다.
      한국투자증권 자동 트레이딩한국투자증권 자동 트레이딩

 

Ⅳ. 한국투자증권 Open API 신청: 모의투자/실전투자
  1. Python으로 트레이딩을 하려면 Open API가 필요합니다.
    • “트레이딩 > Open API > KIS Developers > KIS Developers 서비스 신청하기” 메뉴로 이동
      한국투자증권 자동 트레이딩
    • 간편인증, 스마트폰인증,공인인증서 등을 통해 로그인한 후 휴대폰 인증 진행
      한국투자증권 자동 트레이딩
  2. 실전투자할지 모의투자할지를 선택합니다.
    • 여기서는 “모의투자계좌”를 선택하고, 3단계 모의투자 신청 시 생성된 모의투자 계좌번호를 입력하고 화면 아래의 “신청” 버튼 클릭
      – 실전투자를 하기 위해서 나중에 추가신청을 통해 “실전투자계좌”를 선택/인증한 후 진행합니다.
      한국투자증권 자동 트레이딩
    • 한국투자증권 스마트폰 앱의 “인증 > PC인증“으로 이동하여 스마트폰 인증 진행
      한국투자증권 자동 트레이딩
  3. APP Key 및 APP Secret를 복사하여 모의투자 계좌번호와 함께 메모장 등에 저장
    한국투자증권 자동 트레이딩

 

Ⅴ. Python 프로그램 등록/수정 및 모의투자 테스트
  1. 환경설정 파일 및 Python 실행파일을 동일한 폴더에 저장합니다.
    • 환경설정 파일(config.yaml)
      – APP Key, APP Secret, 모의투자 계좌번호, Discord 웹후크 URL을 당신의 것으로 수정합니다.
      – 실전투자를 하려면 APP Key, APP Secret, 계좌번호를 등록하고, 실전투자 URL_BASE를 uncomment 처리하고 모의투자 URL_BASE를 comment 처리합니다. 
      #홈페이지에서 API서비스 신청시 받은 Appkey, Appsecret 값 설정
APP_KEY: "당신의 APP Key"
APP_SECRET: "당신의 APP Secret"

#계좌번호 앞 8자리
CANO: "당신의 모의투자 계좌번호"
#계좌번호 뒤 2자리
ACNT_PRDT_CD: "01"

#실전투자
#URL_BASE: "https://openapi.koreainvestment.com:9443"
#모의투자
URL_BASE: "https://openapivts.koreainvestment.com:29443"

#디스코드 웹훅 URL
DISCORD_WEBHOOK_URL: "당신의 Discord 웹후크 URL"
    • 모의투자 실행파일(mock_investment.py)
      – 변동성 돌파 전략으로 주식을 매수/매도하는 알고리즘으로 작성되었습니다.
      – 한국투자증권 KIS Developers > API 문서 페이지를 참고하면 다양한 Open API를 사용해서 나만의 매수/매도 전략을 구현할 수 있겠지요! 
      import requests
import json
import datetime
import time
import yaml

# 환경설정 파일 불러오기(Open AIP, 계좌정보, Webhook 등)
with open('config.yaml', encoding='UTF-8') as f:
    _cfg = yaml.load(f, Loader=yaml.FullLoader)
APP_KEY = _cfg['APP_KEY']
APP_SECRET = _cfg['APP_SECRET']
ACCESS_TOKEN = ""
CANO = _cfg['CANO']
ACNT_PRDT_CD = _cfg['ACNT_PRDT_CD']
DISCORD_WEBHOOK_URL = _cfg['DISCORD_WEBHOOK_URL']
URL_BASE = _cfg['URL_BASE']

# 디스코드 메시지 발송 함수
def send_message(msg):
    """디스코드 메세지 전송"""
    now = datetime.datetime.now()
    message = {"content": f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] {str(msg)}"}
    requests.post(DISCORD_WEBHOOK_URL, data=message)
    print(message)

# 한국투자 API에서 토큰 발급받는 함수
def get_access_token():
    """토큰 발급"""
    headers = {"content-type":"application/json"}
    body = {"grant_type":"client_credentials",
    "appkey":APP_KEY, 
    "appsecret":APP_SECRET}
    PATH = "oauth2/tokenP"
    URL = f"{URL_BASE}/{PATH}"
    res = requests.post(URL, headers=headers, data=json.dumps(body))
    ACCESS_TOKEN = res.json()["access_token"]
    print(ACCESS_TOKEN)
    return ACCESS_TOKEN

# datas를 암호화하는 함수
def hashkey(datas):
    """암호화"""
    PATH = "uapi/hashkey"
    URL = f"{URL_BASE}/{PATH}"
    headers = {
    'content-Type' : 'application/json',
    'appKey' : APP_KEY,
    'appSecret' : APP_SECRET,
    }
    res = requests.post(URL, headers=headers, data=json.dumps(datas))
    hashkey = res.json()["HASH"]
    return hashkey

# 특정 종목 실시간 주가 조회
def get_current_price(code):
    """현재가 조회"""
    PATH = "uapi/domestic-stock/v1/quotations/inquire-price"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
            "authorization": f"Bearer {ACCESS_TOKEN}",
            "appKey":APP_KEY,
            "appSecret":APP_SECRET,
            "tr_id":"FHKST01010100"}       
    params = {
    "fid_cond_mrkt_div_code":"J",
    "fid_input_iscd":code,
    }
    res = requests.get(URL, headers=headers, params=params)
    return int(res.json()['output']['stck_prpr'])

# 변동성 돌파 전략
def get_target_price(code):
    """변동성 돌파 전략으로 매수 목표가 조회"""
    PATH = "uapi/domestic-stock/v1/quotations/inquire-daily-price"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
        "authorization": f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"FHKST01010400"}                              
    params = {
    "fid_cond_mrkt_div_code":"J",
    "fid_input_iscd":code,
    # 1 : 원시가격
    "fid_org_adj_prc":"1",
    # D : 일일 데이터 요청
    "fid_period_div_code":"D"
    }
    res = requests.get(URL, headers=headers, params=params)
    stck_oprc = int(res.json()['output'][0]['stck_oprc']) #오늘 시가
    stck_hgpr = int(res.json()['output'][1]['stck_hgpr']) #전일 고가
    stck_lwpr = int(res.json()['output'][1]['stck_lwpr']) #전일 저가
    target_price = stck_oprc + (stck_hgpr - stck_lwpr) * 0.5
    return target_price

# 실시간 주식 잔고조회
def get_stock_balance():
    """주식 잔고조회"""
    PATH = "uapi/domestic-stock/v1/trading/inquire-balance"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"VTTC8434R",                        
        "custtype":"P",
    }
    params = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "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": "01",
        "CTX_AREA_FK100": "",
        "CTX_AREA_NK100": ""
    }
    res = requests.get(URL, headers=headers, params=params)
    stock_list = res.json()['output1']
    evaluation = res.json()['output2']
    stock_dict = {}
    send_message(f"====주식 보유잔고====")
    for stock in stock_list:
        if int(stock['hldg_qty']) > 0:
            stock_dict[stock['pdno']] = stock['hldg_qty']
            send_message(f"{stock['prdt_name']}({stock['pdno']}): {stock['hldg_qty']}주")
            time.sleep(0.1)
    # 디스코드로 메시지 전송
    send_message(f"주식 평가 금액: {evaluation[0]['scts_evlu_amt']}원")
    time.sleep(0.1)
    send_message(f"평가 손익 합계: {evaluation[0]['evlu_pfls_smtl_amt']}원")
    time.sleep(0.1)
    send_message(f"총 평가 금액: {evaluation[0]['tot_evlu_amt']}원")
    time.sleep(0.1)
    send_message(f"=================")
    return stock_dict

# 실시간 현금 잔고조회
def get_balance():
    """현금 잔고조회"""
    PATH = "uapi/domestic-stock/v1/trading/inquire-psbl-order"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"VTTC8908R",                                
        "custtype":"P",
    }
    params = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "PDNO": "005930",
        "ORD_UNPR": "65500",
        "ORD_DVSN": "01",
        "CMA_EVLU_AMT_ICLD_YN": "Y",
        "OVRS_ICLD_YN": "Y"
    }
    res = requests.get(URL, headers=headers, params=params)
    cash = res.json()['output']['ord_psbl_cash']
    send_message(f"주문 가능 현금 잔고: {cash}원")
    return int(cash)

# 매수 함수
def buy(code="005930", qty="1"):
    """주식 시장가 매수"""  
    PATH = "uapi/domestic-stock/v1/trading/order-cash"
    URL = f"{URL_BASE}/{PATH}"
    data = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "PDNO": code,
        "ORD_DVSN": "01",
        "ORD_QTY": str(int(qty)),
        "ORD_UNPR": "0",
    }
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"VTTC0802U",                
        "custtype":"P",
        "hashkey" : hashkey(data)
    }
    res = requests.post(URL, headers=headers, data=json.dumps(data))
    if res.json()['rt_cd'] == '0':
        send_message(f"[매수 성공]{str(res.json())}")
        return True
    else:
        send_message(f"[매수 실패]{str(res.json())}")
        return False

# 매도 함수
def sell(code="005930", qty="1"):
    """주식 시장가 매도"""
    PATH = "uapi/domestic-stock/v1/trading/order-cash"
    URL = f"{URL_BASE}/{PATH}"
    data = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "PDNO": code,
        "ORD_DVSN": "01",
        "ORD_QTY": qty,
        "ORD_UNPR": "0",
    }
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"VTTC0801U",                
        "custtype":"P",
        "hashkey" : hashkey(data)
    }
    res = requests.post(URL, headers=headers, data=json.dumps(data))
    if res.json()['rt_cd'] == '0':
        send_message(f"[매도 성공]{str(res.json())}")
        return True
    else:
        send_message(f"[매도 실패]{str(res.json())}")
        return False

# 자동매매 시작
try:
    ACCESS_TOKEN = get_access_token()
    symbol_list = ["005930","035720"]                                       # 매수 희망 종목 리스트
    bought_list = []                                                        # 매수 완료된 종목 리스트
    total_cash = get_balance()                                              # 보유 현금 조회
    stock_dict = get_stock_balance()                                        # 보유 주식 조회
    for sym in stock_dict.keys():
        bought_list.append(sym)
    target_buy_count = int(len(symbol_list))                                # 매수할 종목 수
    buy_percent = 1 / int(len(symbol_list))                                 # 종목당 매수 금액 비율
    buy_amount = total_cash * buy_percent                                   # 종목별 주문 금액 계산
    soldout = False
    
    for sym in symbol_list:
        target_price2 = get_target_price(sym)
        print("목표가 계산", target_price2)
        current_price2 = get_current_price(sym)
        print("현재가 계산", current_price2)      
        time.sleep(2)

    send_message("===국내 주식 자동매매 프로그램을 시작합니다===")
    while True:
        t_now = datetime.datetime.now()
        t_9 = t_now.replace(hour=9, minute=0, second=0, microsecond=0)      # 시장개장(09:00)
        t_start = t_now.replace(hour=9, minute=5, second=0, microsecond=0)  # 매입시작(09:05)
        t_sell = t_now.replace(hour=15, minute=15, second=0, microsecond=0) # 매도시작(15:15)
        t_exit = t_now.replace(hour=15, minute=20, second=0,microsecond=0)  # 프로그램 종료(15:20)
        today = datetime.datetime.today().weekday()                         # 요일계산
        if today == 5 or today == 6:                                        # 토요일이나 일요일이면 자동 종료
            send_message("주말이므로 프로그램을 종료합니다.")
            break
        if t_9 < t_now < t_start and soldout == False:                      # 잔여 수량 매도(09:00 ~ 09:05, 보유중인 주식 모두 매도) 
            for sym, qty in stock_dict.items():
                sell(sym, qty)
            soldout == True
            bought_list = []
            stock_dict = get_stock_balance()
        if t_start < t_now < t_sell :                                       # 매매시작(09:05 ~ 15:15)
            for sym in symbol_list:
                if len(bought_list) < target_buy_count:
                    if sym in bought_list:
                        continue
                    target_price = get_target_price(sym)                    # 해당 종목의 매수 목표가를 계산
                    current_price = get_current_price(sym)                  # 현재 종목의 현재가
                    if target_price < current_price:                        # 현재 가격이 목표가를 초과하면, 해당 종목을 매수 시작
                        buy_qty = 0                                         # 매수할 수량 초기화
                        buy_qty = int(buy_amount // current_price)
                        if buy_qty > 0:
                            send_message(f"{sym} 목표가 달성({target_price} < {current_price}) 매수를 시도합니다.")
                            result = buy(sym, buy_qty)
                            if result:
                                soldout = False
                                bought_list.append(sym)
                                get_stock_balance()
                    time.sleep(1)
            time.sleep(1)
            if t_now.minute == 30 and t_now.second <= 5:                    # 실시간 주식 잔고조회(매시30분)
                get_stock_balance()
                time.sleep(5)
        if t_sell < t_now < t_exit:                                         # 15:15 ~ 15:20 : 일괄 매도
            if soldout == False:
                stock_dict = get_stock_balance()
                for sym, qty in stock_dict.items():
                    sell(sym, qty)
                soldout = True
                bought_list = []
                time.sleep(1)
        if t_exit < t_now:                                                  # 15:20 : 프로그램 종료
            send_message("프로그램을 종료합니다.")
            break
except Exception as e:
    send_message(f"[오류 발생]{e}")
    time.sleep(1)
    • 실전투자 실행파일(stock_trade.py)
      - 모의투자 실행파일과의 차이점은 tr_id입니다. 한국투자증권 KIS Developers > API 문서 페이지를 참고하세요. 
      import requests
import json
import datetime
import time
import yaml

# 개인 계좌정보 불러오기
with open('config.yaml', encoding='UTF-8') as f:
    _cfg = yaml.load(f, Loader=yaml.FullLoader)
APP_KEY = _cfg['APP_KEY']
APP_SECRET = _cfg['APP_SECRET']
ACCESS_TOKEN = ""
CANO = _cfg['CANO']
ACNT_PRDT_CD = _cfg['ACNT_PRDT_CD']
DISCORD_WEBHOOK_URL = _cfg['DISCORD_WEBHOOK_URL']
URL_BASE = _cfg['URL_BASE']

# 디스코드 메시지 발송 함수
def send_message(msg):
    """디스코드 메세지 전송"""
    now = datetime.datetime.now()
    # 현재 날짜 + 메시지 전달 기능
    message = {"content": f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] {str(msg)}"}
    # HTTP POST 방식으로 config.yaml 에서 불러온 디스코드 특정 채널에 메시지 전송
    requests.post(DISCORD_WEBHOOK_URL, data=message)
    print(message)

# 한국투자 API에서 토큰 발급받는 함수
def get_access_token():
    """토큰 발급"""
    # 토큰 발급을 위한 headers, body, path 준비
    headers = {"content-type":"application/json"}
    body = {"grant_type":"client_credentials",
    "appkey":APP_KEY, 
    "appsecret":APP_SECRET}
    PATH = "oauth2/tokenP"
    URL = f"{URL_BASE}/{PATH}"
    # 토큰 발급 요청
    res = requests.post(URL, headers=headers, data=json.dumps(body))
    # 발급된 토큰 ACCESS_TOKEN에 저장
    ACCESS_TOKEN = res.json()["access_token"]
    return ACCESS_TOKEN

# datas를 암호화하는 함수
# 매수, 매도시 사용
def hashkey(datas):
    """암호화"""
    PATH = "uapi/hashkey"
    URL = f"{URL_BASE}/{PATH}"
    headers = {
    'content-Type' : 'application/json',
    'appKey' : APP_KEY,
    'appSecret' : APP_SECRET,
    }
    res = requests.post(URL, headers=headers, data=json.dumps(datas))
    hashkey = res.json()["HASH"]
    return hashkey

# 특정 종목 실시간 주가 조회
# 005930 : 삼성전자
def get_current_price(code="005930"):
    """현재가 조회"""
    PATH = "uapi/domestic-stock/v1/quotations/inquire-price"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
            "authorization": f"Bearer {ACCESS_TOKEN}",
            "appKey":APP_KEY,
            "appSecret":APP_SECRET,
            "tr_id":"FHKST01010100"}
    params = {
    # J : 한국주식시장
    "fid_cond_mrkt_div_code":"J",
    "fid_input_iscd":code,
    }
    # URL, HEADER, PARAMS 넣어 GET 방식으로 실시간 주가 조회
    res = requests.get(URL, headers=headers, params=params)
    # JSON 형식으로 파싱하고, 'output' 키 아래의 'stck_prpr' 값을 추출
    return int(res.json()['output']['stck_prpr'])

# 변동성 돌파 전략
# 변동성 돌파 전략 : 주식의 가격이 일정한 변동성을 보일 때, 가격이 어느 방향으로 크게 움직일 가능성이 높다고 가정하여 매매를 결정
# 005930 : 삼성전자
def get_target_price(code="005930"):
    """변동성 돌파 전략으로 매수 목표가 조회"""
    PATH = "uapi/domestic-stock/v1/quotations/inquire-daily-price"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
        "authorization": f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"FHKST01010400"}
    params = {
    "fid_cond_mrkt_div_code":"J",
    "fid_input_iscd":code,
    # 1 : 원시가격
    "fid_org_adj_prc":"1",
    # D : 일일 데이터 요청
    "fid_period_div_code":"D"
    }
    res = requests.get(URL, headers=headers, params=params)
    stck_oprc = int(res.json()['output'][0]['stck_oprc']) #오늘 시가
    stck_hgpr = int(res.json()['output'][1]['stck_hgpr']) #전일 고가
    stck_lwpr = int(res.json()['output'][1]['stck_lwpr']) #전일 저가
    # 매수 목표가 계산
    # 변동성 돌파 전략에 따라 매수 목표가는 오늘 시가에 변동성의 절반(전일 고가와 전일 저가의 차이의 50%)을 추가하여 계산
    target_price = stck_oprc + (stck_hgpr - stck_lwpr) * 0.5
    # target_price : 계산된 매수 목표가
    return target_price

# 실시간 주식 잔고조회
def get_stock_balance():
    """주식 잔고조회"""
    PATH = "uapi/domestic-stock/v1/trading/inquire-balance"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"TTTC8434R",
        "custtype":"P",
    }
    params = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "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": "01",
        "CTX_AREA_FK100": "",
        "CTX_AREA_NK100": ""
    }
    res = requests.get(URL, headers=headers, params=params)
    # 보유 주식 리스트
    stock_list = res.json()['output1']
    # 주식 평가금액 리스트
    evaluation = res.json()['output2']
    stock_dict = {}
    send_message(f"====주식 보유잔고====")
    # 보유한 주식 중에서 보유 수량이 0 초과인 주식
    # prdt_name : 종목명, pdno : 종목코드, hldg_qty : 수량
    for stock in stock_list:
        if int(stock['hldg_qty']) > 0:
            stock_dict[stock['pdno']] = stock['hldg_qty']
            send_message(f"{stock['prdt_name']}({stock['pdno']}): {stock['hldg_qty']}주")
            time.sleep(0.1)
    # 디스코드로 메시지 전송
    send_message(f"주식 평가 금액: {evaluation[0]['scts_evlu_amt']}원")
    time.sleep(0.1)
    send_message(f"평가 손익 합계: {evaluation[0]['evlu_pfls_smtl_amt']}원")
    time.sleep(0.1)
    send_message(f"총 평가 금액: {evaluation[0]['tot_evlu_amt']}원")
    time.sleep(0.1)
    send_message(f"=================")
    return stock_dict

# 실시간 현금 잔고조회
def get_balance():
    """현금 잔고조회"""
    PATH = "uapi/domestic-stock/v1/trading/inquire-psbl-order"
    URL = f"{URL_BASE}/{PATH}"
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"TTTC8908R",
        "custtype":"P",
    }
    # 현금잔고 조회하는데 해당 PARAMS를 전송하는 이유 모름...
    params = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "PDNO": "005930",
        "ORD_UNPR": "65500",
        "ORD_DVSN": "01",
        "CMA_EVLU_AMT_ICLD_YN": "Y",
        "OVRS_ICLD_YN": "Y"
    }
    res = requests.get(URL, headers=headers, params=params)
    cash = res.json()['output']['ord_psbl_cash']
    send_message(f"주문 가능 현금 잔고: {cash}원")
    return int(cash)

# 매수 함수
def buy(code="005930", qty="1"):
    """주식 시장가 매수"""  
    PATH = "uapi/domestic-stock/v1/trading/order-cash"
    URL = f"{URL_BASE}/{PATH}"
    data = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "PDNO": code,
        "ORD_DVSN": "01",
        "ORD_QTY": str(int(qty)),
        "ORD_UNPR": "0",
    }
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"TTTC0802U",
        "custtype":"P",
        "hashkey" : hashkey(data)
    }
    res = requests.post(URL, headers=headers, data=json.dumps(data))
    if res.json()['rt_cd'] == '0':
        send_message(f"[매수 성공]{str(res.json())}")
        return True
    else:
        send_message(f"[매수 실패]{str(res.json())}")
        return False

# 매도 함수
def sell(code="005930", qty="1"):
    """주식 시장가 매도"""
    PATH = "uapi/domestic-stock/v1/trading/order-cash"
    URL = f"{URL_BASE}/{PATH}"
    data = {
        "CANO": CANO,
        "ACNT_PRDT_CD": ACNT_PRDT_CD,
        "PDNO": code,
        "ORD_DVSN": "01",
        "ORD_QTY": qty,
        "ORD_UNPR": "0",
    }
    headers = {"Content-Type":"application/json", 
        "authorization":f"Bearer {ACCESS_TOKEN}",
        "appKey":APP_KEY,
        "appSecret":APP_SECRET,
        "tr_id":"TTTC0801U",
        "custtype":"P",
        "hashkey" : hashkey(data)
    }
    res = requests.post(URL, headers=headers, data=json.dumps(data))
    if res.json()['rt_cd'] == '0':
        send_message(f"[매도 성공]{str(res.json())}")
        return True
    else:
        send_message(f"[매도 실패]{str(res.json())}")
        return False

# 자동매매 시작
# 005930 : 삼성전자, 035720 : 카카오, 000660 : SK하이닉스, 069500 : KODEX200
try:
    ACCESS_TOKEN = get_access_token()
    symbol_list = ["005930","035720","000660","069500"]                     # 매수 희망 종목 리스트
    bought_list = []                                                        # 매수 완료된 종목 리스트
    total_cash = get_balance()                                              # 보유 현금 조회
    stock_dict = get_stock_balance()                                        # 보유 주식 조회
    for sym in stock_dict.keys():
        bought_list.append(sym)
    target_buy_count = 3                                                    # 매수할 종목 수
    buy_percent = 0.33                                                      # 종목당 매수 금액 비율
    buy_amount = total_cash * buy_percent                                   # 종목별 주문 금액 계산
    soldout = False

    send_message("===국내 주식 자동매매 프로그램을 시작합니다===")
    while True:
        t_now = datetime.datetime.now()
        t_9 = t_now.replace(hour=9, minute=0, second=0, microsecond=0)      # 시장개장(09:00)
        t_start = t_now.replace(hour=9, minute=5, second=0, microsecond=0)  # 매입시작(09:05)
        t_sell = t_now.replace(hour=15, minute=15, second=0, microsecond=0) # 매도시작(15:15)
        t_exit = t_now.replace(hour=15, minute=20, second=0,microsecond=0)  # 프로그램 종료(15:20)
        today = datetime.datetime.today().weekday()                         # 요일계산
        if today == 5 or today == 6:                                        # 토요일이나 일요일이면 자동 종료
            send_message("주말이므로 프로그램을 종료합니다.")
            break
        if t_9 < t_now < t_start and soldout == False:                      # 잔여 수량 매도(09:00 ~ 09:05, 보유중인 주식 모두 매도) 
            for sym, qty in stock_dict.items():
                sell(sym, qty)
            soldout == True
            bought_list = []
            stock_dict = get_stock_balance()
        if t_start < t_now < t_sell :                                       # 매매시작(09:05 ~ 15:15)
            for sym in symbol_list:
                if len(bought_list) < target_buy_count:
                    if sym in bought_list:
                        continue
                    target_price = get_target_price(sym)                    # 해당 종목의 매수 목표가를 계산
                    current_price = get_current_price(sym)                  # 현재 종목의 현재가
                    if target_price < current_price:                        # 현재 가격이 목표가를 초과하면, 해당 종목을 매수 시작
                        buy_qty = 0                                         # 매수할 수량 초기화
                        buy_qty = int(buy_amount // current_price)
                        if buy_qty > 0:
                            send_message(f"{sym} 목표가 달성({target_price} < {current_price}) 매수를 시도합니다.")
                            result = buy(sym, buy_qty)
                            if result:
                                soldout = False
                                bought_list.append(sym)
                                get_stock_balance()
                    time.sleep(1)
            time.sleep(1)
            if t_now.minute == 30 and t_now.second <= 5:                    # 실시간 주식 잔고조회(매시30분)
                get_stock_balance()
                time.sleep(5)
        if t_sell < t_now < t_exit:                                         # 15:15 ~ 15:20 : 일괄 매도
            if soldout == False:
                stock_dict = get_stock_balance()
                for sym, qty in stock_dict.items():
                    sell(sym, qty)
                soldout = True
                bought_list = []
                time.sleep(1)
        if t_exit < t_now:                                                  # 15:20 : 프로그램 종료
            send_message("프로그램을 종료합니다.")
            break
except Exception as e:
    send_message(f"[오류 발생]{e}")
    time.sleep(1)
  2. VS Code에서 Python 프로그램을 불러들입니다.
    • Python 프로그램 파일들이 저장된 폴더 선택
      한국투자증권 자동 트레이딩
    • 환경설정 파일(config.yaml)에 모의투자를 위해 아래 값들을 확인/등록
      - Open API APP Key
      - Open API Secret
      - 모의투자 계좌번호
      - 모의투자 URL_BASE uncomment(실전투자를 하려면 실전투자 URL_BASE uncomment)
      - Discord 웹후크 URL
      한국투자증권 자동 트레이딩
  3. 모의투자 프로그램(mock_investment.py) 파일을 선택하여 필요한 패키지를 설치합니다.
    • requests 패키지와 yaml 패키지가 설치되지 않아 물결무늬 밑줄이 표시됨
    • "Terminal > New Terminal" 메뉴를 선택하고 CLI에 pip install requests pyyaml 입력 후 Enter
      한국투자증권 자동 트레이딩
  4. 모의투자 프로그램(mock_investment.py)를 수정/실행시켜 결과를 확인합니다.
    • 228번째 열에 모의투자 희망 종목 수정 가능(KRX 정보데이터시스템)
      - 현재 "005930/삼성전자","035720/카카오"가 등록되어 있음 
          symbol_list = ["005930","035720"]                                       # 매수 희망 종목 리스트
    • CLI에 python .\mock_investment.py 입력하고 실행
      한국투자증권 자동 트레이딩
    • 실행이 완료되면 메세지 출력
      한국투자증권 자동 트레이딩
    • Discord 앱으로도 메세지 도착
      한국투자증권 자동 트레이딩

Tags:

You may also like...

답글 남기기

이메일 주소는 공개되지 않습니다. 필수 필드는 *로 표시됩니다