# [국내주식] 순위분석 - 국내주식 시간외거래량순위 # Generated by KIS API Generator (Single API Mode) # -*- coding: utf-8 -*- """ Created on 2025-06-17 """ import logging import sys import time from typing import Optional, Tuple import pandas as pd sys.path.extend(['../..', '.']) import kis_auth as ka # 로깅 설정 logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s') logger = logging.getLogger(__name__) ############################################################################################## # [국내주식] 국내주식 > 국내주식 시간외거래량순위[국내주식-139] ############################################################################################## # 상수 정의 API_URL = "/uapi/domestic-stock/v1/ranking/overtime-volume" def overtime_volume( fid_cond_mrkt_div_code: str, # 조건 시장 분류 코드 fid_cond_scr_div_code: str, # 조건 화면 분류 코드 fid_input_iscd: str, # 입력 종목코드 fid_rank_sort_cls_code: str, # 순위 정렬 구분 코드 fid_input_price_1: str, # 입력 가격1 fid_input_price_2: str, # 입력 가격2 fid_vol_cnt: str, # 거래량 수 fid_trgt_cls_code: str, # 대상 구분 코드 fid_trgt_exls_cls_code: str, # 대상 제외 구분 코드 dataframe1: Optional[pd.DataFrame] = None, # 누적 데이터프레임 (output1) dataframe2: Optional[pd.DataFrame] = None, # 누적 데이터프레임 (output2) tr_cont: str = "", depth: int = 0, max_depth: int = 10 ) -> Tuple[pd.DataFrame, pd.DataFrame]: """ [국내주식] 순위분석 국내주식 시간외거래량순위[국내주식-139] 국내주식 시간외거래량순위 API를 호출하여 DataFrame으로 반환합니다. Args: fid_cond_mrkt_div_code (str): 시장구분코드 (J: 주식) fid_cond_scr_div_code (str): Unique key(20235) fid_input_iscd (str): 0000(전체), 0001(코스피), 1001(코스닥) fid_rank_sort_cls_code (str): 0(매수잔량), 1(매도잔량), 2(거래량) fid_input_price_1 (str): 가격 ~ fid_input_price_2 (str): ~ 가격 fid_vol_cnt (str): 거래량 ~ fid_trgt_cls_code (str): 공백 fid_trgt_exls_cls_code (str): 공백 dataframe1 (Optional[pd.DataFrame]): 누적 데이터프레임 (output1) dataframe2 (Optional[pd.DataFrame]): 누적 데이터프레임 (output2) tr_cont (str): 연속 거래 여부 depth (int): 현재 재귀 깊이 max_depth (int): 최대 재귀 깊이 (기본값: 10) Returns: Tuple[pd.DataFrame, pd.DataFrame]: 국내주식 시간외거래량순위 데이터 Example: >>> df1, df2 = overtime_volume( fid_cond_mrkt_div_code='J', fid_cond_scr_div_code='20235', fid_input_iscd='0000', fid_rank_sort_cls_code='2', fid_input_price_1='', fid_input_price_2='', fid_vol_cnt='', fid_trgt_cls_code='', fid_trgt_exls_cls_code='' ) >>> print(df1) >>> print(df2) """ # 필수 파라미터 검증 if not fid_cond_mrkt_div_code: logger.error("fid_cond_mrkt_div_code is required. (e.g. 'J')") raise ValueError("fid_cond_mrkt_div_code is required. (e.g. 'J')") if not fid_cond_scr_div_code: logger.error("fid_cond_scr_div_code is required. (e.g. '20235')") raise ValueError("fid_cond_scr_div_code is required. (e.g. '20235')") if not fid_input_iscd: logger.error("fid_input_iscd is required. (e.g. '0000')") raise ValueError("fid_input_iscd is required. (e.g. '0000')") if not fid_rank_sort_cls_code: logger.error("fid_rank_sort_cls_code is required. (e.g. '2')") raise ValueError("fid_rank_sort_cls_code is required. (e.g. '2')") # 최대 재귀 깊이 체크 if depth >= max_depth: logger.warning("Maximum recursion depth (%d) reached. Stopping further requests.", max_depth) return dataframe1 if dataframe1 is not None else pd.DataFrame(), dataframe2 if dataframe2 is not None else pd.DataFrame() tr_id = "FHPST02350000" params = { "FID_COND_MRKT_DIV_CODE": fid_cond_mrkt_div_code, "FID_COND_SCR_DIV_CODE": fid_cond_scr_div_code, "FID_INPUT_ISCD": fid_input_iscd, "FID_RANK_SORT_CLS_CODE": fid_rank_sort_cls_code, "FID_INPUT_PRICE_1": fid_input_price_1, "FID_INPUT_PRICE_2": fid_input_price_2, "FID_VOL_CNT": fid_vol_cnt, "FID_TRGT_CLS_CODE": fid_trgt_cls_code, "FID_TRGT_EXLS_CLS_CODE": fid_trgt_exls_cls_code, } res = ka._url_fetch(API_URL, tr_id, tr_cont, params) if res.isOK(): # output1 처리 if hasattr(res.getBody(), 'output1'): output_data = res.getBody().output1 if output_data: current_data1 = pd.DataFrame(output_data if isinstance(output_data, list) else [output_data]) dataframe1 = pd.concat([dataframe1, current_data1], ignore_index=True) if dataframe1 is not None else current_data1 else: dataframe1 = dataframe1 if dataframe1 is not None else pd.DataFrame() else: dataframe1 = dataframe1 if dataframe1 is not None else pd.DataFrame() # output2 처리 if hasattr(res.getBody(), 'output2'): output_data = res.getBody().output2 if output_data: current_data2 = pd.DataFrame(output_data if isinstance(output_data, list) else [output_data]) dataframe2 = pd.concat([dataframe2, current_data2], ignore_index=True) if dataframe2 is not None else current_data2 else: dataframe2 = dataframe2 if dataframe2 is not None else pd.DataFrame() else: dataframe2 = dataframe2 if dataframe2 is not None else pd.DataFrame() tr_cont = res.getHeader().tr_cont if tr_cont in ["M", "F"]: logger.info("Calling next page...") ka.smart_sleep() return overtime_volume( fid_cond_mrkt_div_code, fid_cond_scr_div_code, fid_input_iscd, fid_rank_sort_cls_code, fid_input_price_1, fid_input_price_2, fid_vol_cnt, fid_trgt_cls_code, fid_trgt_exls_cls_code, "N", dataframe1, dataframe2, depth + 1, max_depth ) else: logger.info("Data fetch complete.") return dataframe1, dataframe2 else: logger.error("API call failed: %s - %s", res.getErrorCode(), res.getErrorMessage()) res.printError(API_URL) return pd.DataFrame(), pd.DataFrame()