import datetime
from typing import Dict, Optional, Union
import numpy as np
import pandas as pd
import pytz
import pyqqq.config as c
from pyqqq.datatypes import FutureOptionMarket
from pyqqq.utils.api_client import raise_for_status, send_request
from pyqqq.utils.local_cache import DiskCacheManager
from pyqqq.utils.logger import get_logger
logger = get_logger(__name__)
futuresCache = DiskCacheManager("futures_minute_cache")
[docs]
@futuresCache.memoize()
def get_futures_all_day_data(
date: datetime.date,
codes: Union[list, str],
session: Optional[Union[str, FutureOptionMarket]] = FutureOptionMarket.DAY,
ascending: bool = True,
) -> Union[Dict[str, pd.DataFrame], pd.DataFrame]:
"""
지정된 거래일에 대해 하나 이상의 선물 종목의 1분봉 OHLCV 데이터를 반환합니다.
지수선물(코스피200/미니코스피200/코스닥150) 최근월물 데이터가 제공됩니다. 2025년 7월 21일 데이터 부터 조회 가능합니다.
- date는 거래일 기준입니다. 야간세션(night)은 거래일 18:00 ~ 익일 06:00이며, 자정 이후 봉의 time은 익일 날짜로 표기됩니다.
- session=None 지정 시 주간+야간 세션을 모두 반환합니다. (전일 야간세션의 새벽 봉은 포함되지 않습니다)
이때 cum_volume/cum_value는 세션별로 각각 누적되므로 (주간/야간 시장이 별개), 누적값을 사용할 때는 session 컬럼으로 나누어 사용해야 합니다.
- 거래가 없던 분은 직전 종가로 채워진 허봉(volume/value 0)이 포함되어 있습니다.
Args:
date (datetime.date): 조회할 거래일.
codes (list[str] | str): 조회할 선물 단축코드 리스트 또는 단일 코드. 최대 20개까지 지정할 수 있습니다.
session (str | FutureOptionMarket, optional): 조회할 세션. 'day'(정규/주간) 또는 'night'(야간)를 지정할 수 있습니다.
기본값은 day이며, None 지정 시 주간+야간을 모두 반환합니다.
ascending (bool): 시간 오름차순 여부. 기본값은 True.
Returns:
dict[str, pd.DataFrame] | pd.DataFrame: 종목코드를 키로 하는 딕셔너리. codes를 단일 코드(str)로 지정한 경우 해당 종목의 DataFrame만 반환합니다.
각 DataFrame은 'time' 열이 인덱스로 설정되어 있습니다.
DataFrame의 열은 다음과 같습니다:
- time (datetime.datetime): 시간 (인덱스)
- product (str): 상품 구분
- session (str): 세션 구분 ('day' | 'night')
- open (float): 시가
- high (float): 고가
- low (float): 저가
- close (float): 종가
- volume (int): 거래량
- value (int): 거래대금
- cum_volume (int): 누적거래량
- cum_value (int): 누적거래대금
Raises:
requests.exceptions.RequestException: PYQQQ API로부터 데이터를 검색하는 과정에서 오류가 발생한 경우.
Examples:
>>> df = get_futures_all_day_data(datetime.date(2026, 7, 29), "A01609", session="day")
>>> print(df)
product session open high low close volume value cum_volume cum_value
time
2026-07-29 08:45:00 KOSPI200선물 day 428.75 429.10 428.60 429.00 1234 132456789000 1234 132456789000
... ... ... ... ... ... ... ... ... ... ...
"""
assert type(date) is datetime.date, "date must be a datetime.date object"
assert isinstance(codes, (list, str)), "codes must be a list of strings or single code"
if isinstance(codes, list):
assert all(isinstance(code, str) for code in codes), "codes must be a list of strings"
assert len(codes) > 0, "codes must not be empty"
assert len(codes) <= 20, "codes must not exceed 20"
if session is not None:
session = FutureOptionMarket.validate(session)
tz = pytz.timezone("Asia/Seoul")
target_codes = codes if isinstance(codes, list) else [codes]
url = f"{c.PYQQQ_API_URL}/derivatives/futures/ohlcv/minutes/{date}"
params = {
"codes": ",".join(target_codes),
"current_date": datetime.date.today(),
}
if session is not None:
params["session"] = session.value
r = send_request("GET", url, params=params)
if r.status_code != 200 and r.status_code != 201:
logger.error(f"Failed to get futures day data: {r.text}")
r.raise_for_status()
result = {}
for code in target_codes:
result[code] = pd.DataFrame()
entries = r.json()
cols = entries["cols"]
if len(cols) == 0:
if isinstance(codes, str):
return result[codes]
return result
time_index = cols.index("time")
multirows = entries["rows"]
for code in multirows.keys():
rows = multirows[code]
for row in rows:
time = row[time_index].replace("Z", "+00:00")
time = datetime.datetime.fromisoformat(time).astimezone(tz).replace(tzinfo=None)
row[time_index] = time
rows.reverse()
df = pd.DataFrame(rows, columns=cols)
if not df.empty:
dtypes = df.dtypes
for k in ["volume", "value", "cum_volume", "cum_value"]:
if k in dtypes:
dtypes[k] = np.dtype("int64")
df = df.astype(dtypes)
df = df[[col for col in ["time", "product", "session", "open", "high", "low", "close", "volume", "value", "cum_volume", "cum_value"] if col in df.columns]]
df.set_index("time", inplace=True)
df.sort_index(ascending=ascending, inplace=True)
result[code] = df
if isinstance(codes, str):
return result[codes]
else:
return result
[docs]
def get_investor_net_purchases(
date: Optional[datetime.date] = None,
session: Optional[Union[str, FutureOptionMarket]] = None,
detail: bool = False,
) -> pd.DataFrame:
"""
지수선물 투자자별 순매수 거래량(투자자별 매매동향)을 조회합니다.
지수선물(코스피200/미니코스피200/코스닥150)의 투자자별 순매수 거래량이 제공됩니다. 만기일의 경우 단축코드(code)가 부정확할 수 있습니다.
2025년 7월 21일 데이터 부터 조회 가능합니다.
Args:
date (datetime.date, optional): 조회할 거래일. 기본값은 None (가장 최근 데이터)
session (str | FutureOptionMarket, optional): 조회할 세션. 'day'(정규/주간) 또는 'night'(야간)를 지정할 수 있습니다.
기본값은 None이며, 주간+야간 합계(all)를 반환합니다.
detail (bool): True 지정 시 기관 합계(institutional) 대신 기관 세부 주체별 순매수량 컬럼을 반환합니다. 기본값은 False.
Returns:
pd.DataFrame: 상품별 투자자 순매수 거래량. 해당일에 저장된 데이터가 없으면 빈 DataFrame이 반환됩니다.
- code (str, index): 해당 일자의 근월물 단축코드 (예: "A01609"). 데이터는 상품 단위(전 월물 합산) 통계이며 근월물 코드는 대표 코드입니다.
- product (str): 상품 구분 (코스피200, 미니코스피200, 코스닥150)
- date (str): 거래일 (YYYYMMDD)
- session (str): 세션 구분 (all | day | night)
- individual (int): 개인 순매수량
- foreign (int): 외국인 순매수량
- institutional (int): 기관 순매수량 (detail=False인 경우)
- others (int): 기타법인 순매수량
detail=True인 경우 institutional 대신 다음 컬럼이 포함됩니다:
- financial_investment (int): 금융투자 순매수량
- insurance (int): 보험 순매수량
- investment_trust (int): 투자신탁 순매수량
- bank (int): 은행 순매수량
- other_financial (int): 기타금융 순매수량
- pension_fund (int): 연기금 등 순매수량
Raises:
requests.exceptions.RequestException: PYQQQ API로부터 데이터를 검색하는 과정에서 오류가 발생한 경우.
Examples:
>>> df = get_investor_net_purchases(datetime.date(2026, 8, 5), session="night")
>>> print(df)
product date session individual foreign institutional others
code
A01609 코스피200 20260805 night 1024 -1957 646 287
A05608 미니코스피200 20260805 night 345 -456 123 -12
A06609 코스닥150 20260805 night -30 -45 70 5
"""
assert date is None or type(date) is datetime.date, "date must be a datetime.date object"
if session is not None:
session = FutureOptionMarket.validate(session)
url = f"{c.PYQQQ_API_URL}/derivatives/futures/investor/all"
params = {}
if date:
params["date"] = date
if session is not None:
params["session"] = session.value
r = send_request("GET", url, params=params)
raise_for_status(r)
rows = r.json()
df = pd.DataFrame(rows)
if not df.empty:
if detail:
detail_cols = ["financial_investment", "insurance", "investment_trust", "bank", "other_financial", "pension_fund"]
details = df["institutional_detail"] if "institutional_detail" in df.columns else pd.Series([{}] * len(df), index=df.index)
detail_df = pd.DataFrame([d if isinstance(d, dict) else {} for d in details], index=df.index)
df = pd.concat([df.drop(columns=["institutional_detail"], errors="ignore"), detail_df], axis=1)
investor_cols = ["individual", "foreign", *detail_cols, "others"]
else:
investor_cols = ["individual", "foreign", "institutional", "others"]
df = df.reindex(columns=["code", "product", "date", "session", *investor_cols])
# 아직 수집되지 않은 세션의 값은 0으로 채운다
df[investor_cols] = df[investor_cols].fillna(0).astype(np.int64)
df.set_index("code", inplace=True)
return df