import datetime as dtm
from typing import Optional, Union
import pandas as pd
from pyqqq.brokerage.kis.domestic_futureoption import KISDomesticFutureOption
from pyqqq.brokerage.kis.oauth import KISAuth
from pyqqq.datatypes import FutureOptionMarket
from pyqqq.utils.logger import get_logger
from pyqqq.utils.market_schedule import get_market_schedule
# 파생 주간장 = 주식 정규장 개장 -15분 ~ 마감 +15분
_FUTURES_HOURS_OFFSET = dtm.timedelta(minutes=15)
# 지수 상품(코스피200, 미니코스피200, 코스닥150)의 단축코드 프리픽스 뒤 2자리
_INDEX_PRODUCT_CODES = {"01", "05", "06"}
def _infer_market_div_code(asset_code: str) -> str:
"""
단축코드로 시장분류코드를 추론합니다.
단축코드 첫 글자는 상품 종류를 나타냅니다: 선물 1(구코드)/A(신코드), 콜옵션 2/B, 풋옵션 3/C.
지수선물(6자리)과 지수옵션(9자리)만 추론하며, 그 외(주식선물, 상품선물 등)는 코드만으로
구분할 수 없으므로 market_div_code를 명시적으로 전달해야 합니다.
"""
if len(asset_code) == 6 and asset_code[0] in ("1", "A") and asset_code[1:3] in _INDEX_PRODUCT_CODES:
return "F"
if len(asset_code) == 9 and asset_code[0] in ("2", "3", "B", "C"):
return "O"
raise ValueError(f"cannot infer market_div_code from asset_code {asset_code!r}. pass market_div_code explicitly")
# 주간 시장코드 → 야간 시장코드. 야간시장은 지수선물/지수옵션만 존재한다
_NIGHT_MARKET_DIV_CODES = {"F": "CM", "O": "EU"}
def _resolve_market_div_code(asset_code: str, market_div_code: Optional[str], session: Union[str, FutureOptionMarket]) -> tuple:
"""
session을 검증하고, market_div_code 미지정 시 asset_code로 추론한 뒤 야간 시장이면 야간 시장코드로 변환합니다.
Returns:
tuple[FutureOptionMarket, str]: (검증된 session, 최종 시장분류코드)
"""
session = FutureOptionMarket.validate(session)
if market_div_code is None:
market_div_code = _infer_market_div_code(asset_code)
assert market_div_code in ["F", "O", "JF", "JO", "CF"], 'market_div_code must be one of "F", "O", "JF", "JO", or "CF"'
if session == FutureOptionMarket.NIGHT:
assert market_div_code in _NIGHT_MARKET_DIV_CODES, "야간세션은 지수선물(F)/지수옵션(O)만 지원합니다"
market_div_code = _NIGHT_MARKET_DIV_CODES[market_div_code]
return session, market_div_code
[docs]
class KISSimpleDomesticFutureOption:
"""
한국투자증권 국내 선물옵션 API를 사용하여 시세를 조회하기 위한 클래스입니다.
기존 KISDomesticFutureOption 클래스를 감싸고, 간단한 조회 기능을 제공합니다.
Attributes:
auth (KISAuth): 인증 정보
corp_data (Optional[dict]): 기업 고객의 경우 추가로 필요한 정보를 담고 있는 객체
"""
[docs]
def __init__(
self,
auth: KISAuth,
corp_data: Optional[dict] = None,
):
self.fuop_api = KISDomesticFutureOption(auth, corp_data)
self.logger = get_logger(__name__ + ".KISSimpleDomesticFutureOption")
[docs]
def get_historical_daily_data(
self,
asset_code: str,
first_date: dtm.date,
last_date: dtm.date,
session: Union[str, FutureOptionMarket] = FutureOptionMarket.DAY,
market_div_code: Optional[str] = None,
) -> pd.DataFrame:
"""
선물옵션 일봉 데이터 검색
Note:
- 조회 기간 중 휴장일은 포함되지 않습니다.
- 한 번의 API 호출에 최대 100건까지 조회되며, 초과하는 기간은 나누어 호출한 후 병합합니다.
- 종목코드는 조회 기간 당시 유효한 코드 체계를 사용해야 합니다. (2025-12-11 이전은 구코드 ex. 101W09, 이후는 신코드 ex. A01609)
- 야간세션(session="night") 봉의 date는 거래일 기준입니다. (거래일 18:00 ~ 익일 06:00 세션)
Args:
asset_code(str): 종목코드 (지수선물 6자리, 지수옵션 9자리)
first_date(datetime.date): 조회 시작일자
last_date(datetime.date): 조회 종료일자
session(str | FutureOptionMarket): 조회할 세션 - day:정규(주간) night:야간. 기본값은 day. 야간세션은 지수선물/지수옵션만 지원합니다.
market_div_code(str, optional): 시장분류코드 - F:지수선물 O:지수옵션.
기본값은 None이며 asset_code로 추론합니다. 추론할 수 없는 상품은 명시적으로 전달해야 합니다.
Returns:
pd.DataFrame: 일봉 데이터. date 인덱스에 open/high/low/close(float), volume/value(int) 컬럼을 가지며 날짜 내림차순으로 정렬됩니다.
Raises:
ValueError: market_div_code가 None이고 asset_code로 시장분류코드를 추론할 수 없는 경우. 지원하지 않는 session 값이 전달된 경우.
"""
assert first_date <= last_date, "last_date는 first_date와 같거나, 이후 날짜여야 합니다"
assert last_date <= dtm.date.today(), "last_date는 오늘과 같거나 이전이어야 합니다."
session, market_div_code = _resolve_market_div_code(asset_code, market_div_code, session)
# 한 번의 호출에 최대 100건이 반환되므로 100건을 넘지 않는 날짜 창으로 나누어 조회
max_days_per_request = 100
total_days = (last_date - first_date).days
result = []
for i in range(0, total_days + 1, max_days_per_request + 1):
search_start = first_date + dtm.timedelta(days=i)
search_end = min(first_date + dtm.timedelta(days=i + max_days_per_request), last_date)
r = self.fuop_api.inquire_daily_fuopchartprice(
asset_code,
search_start,
search_end,
fid_period_div_code="D",
fid_cond_mrkt_div_code=market_div_code,
)
for item in r["output2"]:
result.append(
{
"date": item["stck_bsop_date"],
"open": float(item["futs_oprc"]),
"high": float(item["futs_hgpr"]),
"low": float(item["futs_lwpr"]),
"close": float(item["futs_prpr"]),
"volume": item["acml_vol"],
"value": item["acml_tr_pbmn"],
}
)
df = pd.DataFrame(result, columns=["date", "open", "high", "low", "close", "volume", "value"])
df["date"] = pd.to_datetime(df["date"])
df.set_index("date", inplace=True)
df = df.sort_index(ascending=False)
return df
[docs]
def get_today_minute_data(
self,
asset_code: str,
session: Union[str, FutureOptionMarket] = FutureOptionMarket.DAY,
market_div_code: Optional[str] = None,
) -> pd.DataFrame:
"""
당일 1분봉 데이터 검색
Note:
- 실전계좌에서만 사용 가능합니다.
- 휴장일이나 장 시작 전에는 빈 DataFrame을 반환합니다.
- 한 번의 API 호출에 최대 102건까지 조회되므로, 장 시작까지 역방향으로 반복 호출하여 병합합니다.
- 야간세션(session="night")은 거래일 18:00 ~ 익일 06:00이며, 진행 중이면 진행 중인 세션을, 아니면 가장 최근 야간세션을 조회합니다.
자정 이후 봉의 time은 달력 기준(거래일 익일)으로 표기됩니다.
Args:
asset_code(str): 종목코드 (지수선물 6자리, 지수옵션 9자리)
session(str | FutureOptionMarket): 조회할 세션 - day:정규(주간) night:야간. 기본값은 day. 야간세션은 지수선물/지수옵션만 지원합니다.
market_div_code(str, optional): 시장분류코드 - F:지수선물 O:지수옵션 JF:주식선물 JO:주식옵션 CF:상품선물.
기본값은 None이며 asset_code로 추론합니다. 지수선물/지수옵션 외 상품은 명시적으로 전달해야 합니다.
Returns:
pd.DataFrame: 분봉 데이터 (시간의 역순). time 인덱스에 open/high/low/close(float), volume/value/cum_volume/cum_value(int) 컬럼을 가집니다.
Raises:
ValueError: market_div_code가 None이고 asset_code로 시장분류코드를 추론할 수 없는 경우. 지원하지 않는 session 값이 전달된 경우.
"""
session, market_div_code = _resolve_market_div_code(asset_code, market_div_code, session)
def _create_minute_dataframe(data: list = None) -> pd.DataFrame:
df = pd.DataFrame(data or [], columns=["time", "open", "high", "low", "close", "volume", "value", "cum_volume", "cum_value"])
df["time"] = pd.to_datetime(df["time"])
df.set_index("time", inplace=True)
return df
now = dtm.datetime.now().replace(second=0, microsecond=0)
if session == FutureOptionMarket.DAY:
trading_date = now.date()
schedule = get_market_schedule(trading_date)
if schedule.full_day_closed:
return _create_minute_dataframe()
# 수능일 등 지연 개장일에도 주식 정규장 시간에서 파생 마감 시각을 도출한다
session_close = dtm.datetime.combine(trading_date, schedule.close_time) + _FUTURES_HOURS_OFFSET
else:
# 18:00 이전이면 전일 거래일의 야간세션 (자정~06:00 진행 중이거나, 06:00에 종료된 직전 세션)
trading_date = now.date() if now.time() >= dtm.time(18, 0) else now.date() - dtm.timedelta(days=1)
schedule = get_market_schedule(trading_date)
if schedule.full_day_closed:
return _create_minute_dataframe()
# 야간세션 시간은 개장 지연과 무관하게 고정 (18:00 ~ 익일 06:00)
session_close = dtm.datetime.combine(trading_date + dtm.timedelta(days=1), dtm.time(6, 0))
request_time = min(now, session_close)
result = []
max_requests = 20 # 야간세션(18:00~익일 06:00) 전체도 호출 8회 이내로 충분. 예상 밖 응답으로 인한 무한 루프 방지용
for _ in range(max_requests):
# 야간세션의 자정 이후(거래일 익일) 시간은 +24시간 HHMMSS 문자열로 전달
if request_time.date() > trading_date:
api_hour = f"{request_time.hour + 24:02d}{request_time.minute:02d}{request_time.second:02d}"
else:
api_hour = request_time.time()
r = self.fuop_api.inquire_time_fuopchartprice(
asset_code,
trading_date,
api_hour,
fid_cond_mrkt_div_code=market_div_code,
)
output = [item for item in r["output2"] if item]
session_output = [item for item in output if item["stck_bsop_date"] == trading_date]
for item in session_output:
bar_date = item["stck_bsop_date"]
if session == FutureOptionMarket.NIGHT and item["stck_cntg_hour"] < dtm.time(18, 0):
bar_date = bar_date + dtm.timedelta(days=1)
result.append(
{
"time": dtm.datetime.combine(bar_date, item["stck_cntg_hour"]),
"open": float(item["futs_oprc"]),
"high": float(item["futs_hgpr"]),
"low": float(item["futs_lwpr"]),
"close": float(item["futs_prpr"]),
"volume": item["cntg_vol"],
"cum_value": item["acml_tr_pbmn"],
}
)
# 응답이 없거나 이전 세션 봉이 나타나면 장 시작까지 도달한 것
if len(session_output) == 0 or len(session_output) < len(output):
break
request_time = result[-1]["time"] - dtm.timedelta(minutes=1)
# 거래대금, 누적거래량 계산 (시간 역순으로 수집되었으므로 뒤에서부터)
prev_cum_value = None
prev_cum_volume = None
for i in range(len(result)):
idx = len(result) - i - 1
curr = result[idx]
if prev_cum_value is None:
curr["value"] = curr["cum_value"]
else:
curr["value"] = curr["cum_value"] - prev_cum_value
if prev_cum_volume is None:
curr["cum_volume"] = curr["volume"]
else:
curr["cum_volume"] = curr["volume"] + prev_cum_volume
prev_cum_value = curr["cum_value"]
prev_cum_volume = curr["cum_volume"]
return _create_minute_dataframe(result)