Skip to content

Instantly share code, notes, and snippets.

@youtube-jocoding
Last active May 12, 2022 00:14
Show Gist options
  • Save youtube-jocoding/c78c266be801a34a3891901b50d293ba to your computer and use it in GitHub Desktop.
Save youtube-jocoding/c78c266be801a34a3891901b50d293ba to your computer and use it in GitHub Desktop.
KoreaStockAutoTrade.py
#홈페이지에서 API서비스 신청시 받은 Appkey, Appsecret 값 설정
APP_KEY: "***"
APP_SECRET: "***"
#계좌번호 앞 8자리
CANO: "***"
#계좌번호 뒤 2자리
ACNT_PRDT_CD: "**"
#실전투자
URL_BASE: "https://openapi.koreainvestment.com:9443"
#모의투자
# URL_BASE: "https://openapivts.koreainvestment.com:29443"
#디스코드 웹훅 URL
DISCORD_WEBHOOK_URL: "***"
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)}"}
requests.post(DISCORD_WEBHOOK_URL, data=message)
print(message)
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"]
return ACCESS_TOKEN
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="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 = {
"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="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,
"fid_org_adj_prc":"1",
"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":"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"====주식 보유잔고====")
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 = {
"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
# 자동매매 시작
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)
t_start = t_now.replace(hour=9, minute=5, second=0, microsecond=0)
t_sell = t_now.replace(hour=15, minute=15, second=0, microsecond=0)
t_exit = t_now.replace(hour=15, minute=20, second=0,microsecond=0)
today = datetime.datetime.today().weekday()
if today == 5 or today == 6: # 토요일이나 일요일이면 자동 종료
send_message("주말이므로 프로그램을 종료합니다.")
break
if t_9 < t_now < t_start and soldout == False: # 잔여 수량 매도
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 : # AM 09:05 ~ PM 03: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:
get_stock_balance()
time.sleep(5)
if t_sell < t_now < t_exit: # PM 03:15 ~ PM 03: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: # PM 03:20 ~ :프로그램 종료
send_message("프로그램을 종료합니다.")
break
except Exception as e:
send_message(f"[오류 발생]{e}")
time.sleep(1)
import requests
import json
import datetime
from pytz import timezone
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)}"}
requests.post(DISCORD_WEBHOOK_URL, data=message)
print(message)
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"]
return ACCESS_TOKEN
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(market="NAS", code="AAPL"):
"""현재가 조회"""
PATH = "uapi/overseas-price/v1/quotations/price"
URL = f"{URL_BASE}/{PATH}"
headers = {"Content-Type":"application/json",
"authorization": f"Bearer {ACCESS_TOKEN}",
"appKey":APP_KEY,
"appSecret":APP_SECRET,
"tr_id":"HHDFS00000300"}
params = {
"AUTH": "",
"EXCD":market,
"SYMB":code,
}
res = requests.get(URL, headers=headers, params=params)
return float(res.json()['output']['last'])
def get_target_price(market="NAS", code="AAPL"):
"""변동성 돌파 전략으로 매수 목표가 조회"""
PATH = "uapi/overseas-price/v1/quotations/dailyprice"
URL = f"{URL_BASE}/{PATH}"
headers = {"Content-Type":"application/json",
"authorization": f"Bearer {ACCESS_TOKEN}",
"appKey":APP_KEY,
"appSecret":APP_SECRET,
"tr_id":"HHDFS76240000"}
params = {
"AUTH":"",
"EXCD":market,
"SYMB":code,
"GUBN":"0",
"BYMD":"",
"MODP":"0"
}
res = requests.get(URL, headers=headers, params=params)
stck_oprc = float(res.json()['output2'][0]['open']) #오늘 시가
stck_hgpr = float(res.json()['output2'][1]['high']) #전일 고가
stck_lwpr = float(res.json()['output2'][1]['low']) #전일 저가
target_price = stck_oprc + (stck_hgpr - stck_lwpr) * 0.5
return target_price
def get_stock_balance():
"""주식 잔고조회"""
PATH = "uapi/overseas-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":"JTTT3012R",
"custtype":"P"
}
params = {
"CANO": CANO,
"ACNT_PRDT_CD": ACNT_PRDT_CD,
"OVRS_EXCG_CD": "NASD",
"TR_CRCY_CD": "USD",
"CTX_AREA_FK200": "",
"CTX_AREA_NK200": ""
}
res = requests.get(URL, headers=headers, params=params)
stock_list = res.json()['output1']
evaluation = res.json()['output2']
# params['OVRS_EXCG_CD'] = "NYSE"
# res = requests.get(URL, headers=headers, params=params)
# stock_list = stock_list + res.json()['output1']
# params['OVRS_EXCG_CD'] = "AMEX"
# res = requests.get(URL, headers=headers, params=params)
# stock_list = stock_list + res.json()['output1']
stock_dict = {}
send_message(f"====주식 보유잔고====")
for stock in stock_list:
if int(stock['ovrs_cblc_qty']) > 0:
stock_dict[stock['ovrs_pdno']] = stock['ovrs_cblc_qty']
send_message(f"{stock['ovrs_item_name']}({stock['ovrs_pdno']}): {stock['ovrs_cblc_qty']}주")
time.sleep(0.1)
send_message(f"주식 평가 금액: {evaluation['tot_evlu_pfls_amt']}원")
time.sleep(0.1)
send_message(f"평가 손익 합계: {evaluation['ovrs_tot_pfls']}원")
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 = {
"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(market="NASD", code="AAPL", qty="1", price="0"):
"""미국 주식 지정가 매수"""
PATH = "uapi/overseas-stock/v1/trading/order"
URL = f"{URL_BASE}/{PATH}"
data = {
"CANO": CANO,
"ACNT_PRDT_CD": ACNT_PRDT_CD,
"OVRS_EXCG_CD": market,
"PDNO": code,
"ORD_DVSN": "00",
"ORD_QTY": str(int(qty)),
"OVRS_ORD_UNPR": f"{round(price,2)}",
"ORD_SVR_DVSN_CD": "0"
}
headers = {"Content-Type":"application/json",
"authorization":f"Bearer {ACCESS_TOKEN}",
"appKey":APP_KEY,
"appSecret":APP_SECRET,
"tr_id":"JTTT1002U",
"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(market="NASD", code="AAPL", qty="1", price="0"):
"""미국 주식 지정가 매도"""
PATH = "uapi/overseas-stock/v1/trading/order"
URL = f"{URL_BASE}/{PATH}"
data = {
"CANO": CANO,
"ACNT_PRDT_CD": ACNT_PRDT_CD,
"OVRS_EXCG_CD": market,
"PDNO": code,
"ORD_DVSN": "00",
"ORD_QTY": str(int(qty)),
"OVRS_ORD_UNPR": f"{round(price,2)}",
"ORD_SVR_DVSN_CD": "0"
}
headers = {"Content-Type":"application/json",
"authorization":f"Bearer {ACCESS_TOKEN}",
"appKey":APP_KEY,
"appSecret":APP_SECRET,
"tr_id":"JTTT1006U",
"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 get_exchange_rate():
"""환율 조회"""
PATH = "uapi/overseas-stock/v1/trading/inquire-present-balance"
URL = f"{URL_BASE}/{PATH}"
headers = {"Content-Type":"application/json",
"authorization": f"Bearer {ACCESS_TOKEN}",
"appKey":APP_KEY,
"appSecret":APP_SECRET,
"tr_id":"CTRP6504R"}
params = {
"CANO": CANO,
"ACNT_PRDT_CD": ACNT_PRDT_CD,
"OVRS_EXCG_CD": "NASD",
"WCRC_FRCR_DVSN_CD": "01",
"NATN_CD": "840",
"TR_MKET_CD": "01",
"INQR_DVSN_CD": "00"
}
res = requests.get(URL, headers=headers, params=params)
return float(res.json()['output2'][0]['frst_bltn_exrt'])
# 자동매매 시작
try:
ACCESS_TOKEN = get_access_token()
nasd_symbol_list = ["AAPL"] # 매수 희망 종목 리스트 (NASD)
nyse_symbol_list = ["KO"] # 매수 희망 종목 리스트 (NYSE)
amex_symbol_list = ["LIT"] # 매수 희망 종목 리스트 (AMEX)
symbol_list = nasd_symbol_list + nyse_symbol_list + amex_symbol_list
bought_list = [] # 매수 완료된 종목 리스트
total_cash = get_balance() # 보유 현금 조회
exchange_rat = get_exchange_rate() # 환율 조회
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 / exchange_rat # 종목별 주문 금액 계산 (달러)
print(buy_amount)
soldout = False
send_message("===해외 주식 자동매매 프로그램을 시작합니다===")
while True:
t_now = datetime.datetime.now(timezone('America/New_York')) # 뉴욕 기준 현재 시간
t_9 = t_now.replace(hour=9, minute=30, second=0, microsecond=0)
t_start = t_now.replace(hour=9, minute=35, second=0, microsecond=0)
t_sell = t_now.replace(hour=15, minute=45, second=0, microsecond=0)
t_exit = t_now.replace(hour=15, minute=50, second=0,microsecond=0)
today = datetime.datetime.today().weekday()
print(t_now, t_9, t_start, t_sell, t_exit)
if today == 5 or today == 6: # 토요일이나 일요일이면 자동 종료
send_message("주말이므로 프로그램을 종료합니다.")
break
if t_9 < t_now < t_start and soldout == False: # 잔여 수량 매도
for sym, qty in stock_dict.items():
market1 = "NASD"
market2 = "NAS"
if sym in nyse_symbol_list:
market1 = "NYSE"
market2 = "NYS"
if sym in amex_symbol_list:
market1 = "AMEX"
market2 = "AMS"
sell(market=market1, code=sym, qty=qty, price=get_current_price(market=market2, code=sym))
soldout == True
bought_list = []
time.sleep(1)
stock_dict = get_stock_balance()
if t_start < t_now < t_sell : # AM 09:35 ~ PM 03:45 : 매수
for sym in symbol_list:
if len(bought_list) < target_buy_count:
if sym in bought_list:
continue
market1 = "NASD"
market2 = "NAS"
if sym in nyse_symbol_list:
market1 = "NYSE"
market2 = "NYS"
if sym in amex_symbol_list:
market1 = "AMEX"
market2 = "AMS"
target_price = get_target_price(market2, sym)
current_price = get_current_price(market2, 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}) 매수를 시도합니다.")
market = "NASD"
if sym in nyse_symbol_list:
market = "NYSE"
if sym in amex_symbol_list:
market = "AMEX"
result = buy(market=market1, code=sym, qty=buy_qty, price=get_current_price(market=market2, code=sym))
time.sleep(1)
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:
get_stock_balance()
time.sleep(5)
if t_sell < t_now < t_exit: # PM 03:45 ~ PM 03:50 : 일괄 매도
if soldout == False:
stock_dict = get_stock_balance()
for sym, qty in stock_dict.items():
market1 = "NASD"
market2 = "NAS"
if sym in nyse_symbol_list:
market1 = "NYSE"
market2 = "NYS"
if sym in amex_symbol_list:
market1 = "AMEX"
market2 = "AMS"
sell(market=market1, code=sym, qty=qty, price=get_current_price(market=market2, code=sym))
soldout = True
bought_list = []
time.sleep(1)
if t_exit < t_now: # PM 03:50 ~ :프로그램 종료
send_message("프로그램을 종료합니다.")
break
except Exception as e:
send_message(f"[오류 발생]{e}")
time.sleep(1)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment