CBOE_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
def fetch_cboe_options(symbol: str) -> dict:
"""CBOE CDN 지연 시세 API로부터 특정 심볼의 전체 옵션 체인 JSON 수신"""
url = f"https://cdn.cboe.com/api/global/delayed_quotes/options/{symbol}.json"
resp = requests.get(url, headers=CBOE_HEADERS, timeout=15)
resp.raise_for_status()
return resp.json()
def parse_option_symbol(opt_symbol: str):
"""OSI 옵션 심볼 규격 파싱 (만기일, Call/Put, Strike 도출)"""
match = re.match(r"^([A-Z_]+)(\d{2})(\d{2})(\d{2})([CP])(\d{8})$", opt_symbol)
if not match:
return None, None, None, None
base_sym, yy, mm, dd, opt_type, strike_str = match.groups()
exp_date = f"20{yy}-{mm}-{dd}"
strike = int(strike_str) / 1000.0
return exp_date, opt_type, strike
def calculate_max_pain(pivot_oi: pd.DataFrame) -> float:
"""행사가격별 총 옵션 매수자 내재가치를 계산하여 최소 손실 행사가 산출"""
strikes = pivot_oi.index.values
call_oi = pivot_oi["CALL"].values
put_oi = pivot_oi["PUT"].values
total_losses = []
for s in strikes:
call_loss = np.sum(np.maximum(0, s - strikes) * call_oi)
put_loss = np.sum(np.maximum(0, strikes - s) * put_oi)
total_losses.append(call_loss + put_loss)
return strikes[np.argmin(total_losses)]
def analyze_cboe_options(symbol: str):
"""단일 심볼에 대한 옵션 체인 정제, 주요 PCR 및 구조 레벨 산출 총괄"""
raw = fetch_cboe_options(symbol)
d = raw.get("data", {})
current_price = d.get("current_price", 0.0)
prev_close = d.get("prev_day_close", 0.0)
iv30 = d.get("iv30", 0.0)
raw_options = d.get("options", [])
parsed_rows = []
for opt in raw_options:
opt_sym = opt.get("option", "")
exp_date, opt_type, strike = parse_option_symbol(opt_sym)
if not opt_type:
continue
vol = float(opt.get("volume", 0.0) or 0.0)
oi = float(opt.get("open_interest", 0.0) or 0.0)
iv = float(opt.get("iv", 0.0) or 0.0)
parsed_rows.append({
"symbol": symbol,
"option": opt_sym,
"expiration": exp_date,
"type": "CALL" if opt_type == "C" else "PUT",
"strike": strike,
"volume": vol,
"open_interest": oi,
"iv": iv
})
df_opts = pd.DataFrame(parsed_rows)
call_vol = df_opts[df_opts["type"] == "CALL"]["volume"].sum()
put_vol = df_opts[df_opts["type"] == "PUT"]["volume"].sum()
call_oi = df_opts[df_opts["type"] == "CALL"]["open_interest"].sum()
put_oi = df_opts[df_opts["type"] == "PUT"]["open_interest"].sum()
pcr_vol = (put_vol / call_vol) if call_vol > 0 else np.nan
pcr_oi = (put_oi / call_oi) if call_oi > 0 else np.nan
# 행사가격별 미결제약정 피벗
pivot_oi = df_opts.groupby(["strike", "type"])["open_interest"].sum().unstack(fill_value=0)
if "CALL" not in pivot_oi: pivot_oi["CALL"] = 0
if "PUT" not in pivot_oi: pivot_oi["PUT"] = 0
call_wall = pivot_oi["CALL"].idxmax() if len(pivot_oi) > 0 else np.nan
put_wall = pivot_oi["PUT"].idxmax() if len(pivot_oi) > 0 else np.nan
max_pain = calculate_max_pain(pivot_oi) if len(pivot_oi) > 0 else np.nan
summary = {
"Symbol": symbol,
"Price": current_price,
"IV30": iv30,
"Put Wall": put_wall,
"Max Pain": max_pain,
"Call Wall": call_wall,
"PCR (Volume)": round(pcr_vol, 4),
"PCR (OI)": round(pcr_oi, 4),
"Call Volume": int(call_vol),
"Put Volume": int(put_vol),
"Call OI": int(call_oi),
"Put OI": int(put_oi),
"Total Contracts": len(df_opts)
}
return summary, df_opts, pivot_oi
print("Option analytics engine defined successfully.")