#!/usr/bin/env python3 """ Fear & Greed 일별 데이터 수집 → fg_daily.csv에 1행 추가 cron: 매일 18:30 KST (장 마감 후) 수집 항목: - 코스피 / 코스닥: KRX Open API 일별 지수 종가 - 코스피 200 변동성지수: investing.com exact instrument (원본 Quantiwise 값과 일치) - 5년/10년 국채선물 추종 지수: KRX 선물 종가 수익률로 기존 지수 연장 - 콜/풋 거래량(raw): KRX 코스피200 옵션 정규장 거래량 """ from __future__ import annotations import argparse import csv import json import logging import unicodedata from datetime import date, datetime, timedelta from pathlib import Path import numpy as np import pandas as pd import requests # ── 자기치유(self-healing) 백필 윈도우 ── # KRX Open API 가 T+1 지연으로 0 을 리턴할 경우 그 시점 collector 는 row 를 skip 한다. # 이후 실행에서 마지막 LOOKBACK_BUSINESS_DAYS 영업일을 항상 재요청하여 누락분/오염분을 # 정상값으로 upsert 한다. 너무 짧으면 늦게 들어온 데이터 회복 못 하고, 너무 길면 # 매일 무의미한 호출이 늘어나므로 10영업일을 기본값으로. LOOKBACK_BUSINESS_DAYS = 10 # 5MA 재계산용 raw 컬럼 매핑. RAW_TO_MA = { "콜 거래량(raw)": "최근월물 CALL ATM", "풋 거래량(raw)": "최근월물 PUT ATM", } # 같은 날짜 row upsert 시 변동률이 이 임계값 초과면 WARNING 로그 (이상 시그널 감지). DRIFT_WARN_PCT = 0.5 # ── 경로 ── BASE_DIR = Path(__file__).resolve().parent CSV_PATH = BASE_DIR / "fg_daily.csv" STATUS_PATH = BASE_DIR / "fg_collection_status.json" LOG_PATH = Path.home() / "Library" / "Logs" / "fg_daily.log" # 캐노니컬 엑셀 후보: 태버지 폴더 → Downloads. "피어앤그리드*.xlsx" 패턴 매칭 후 # 그 중 가장 최근 mtime 을 채택 — 사용자가 새 다운로드를 드롭하면 자동으로 반영됨. def _pick_latest_excel() -> Path | None: search_dirs = [BASE_DIR, Path.home() / "Downloads"] candidates: list[Path] = [] for d in search_dirs: if not d.exists(): continue for p in d.glob("*.xlsx"): if not unicodedata.normalize("NFC", p.name).startswith("피어앤그리드"): continue if p.name.startswith("~$") or ".bak" in p.name: continue candidates.append(p) if not candidates: return None return max(candidates, key=lambda p: p.stat().st_mtime) EXCEL_PATH = _pick_latest_excel() # ── 로깅 ── logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[ logging.FileHandler(LOG_PATH, encoding="utf-8"), logging.StreamHandler(), ], ) log = logging.getLogger(__name__) # ── investing.com ── INVESTING_HEADERS = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)", "domain-id": "www", } # ── KRX Open API (승인 후 활성화) ── def _load_krx_api_key() -> str: import os key = os.environ.get("KRX_API_KEY", "") if not key: env_path = Path.home() / ".env" if env_path.exists(): for line in env_path.read_text().splitlines(): if line.strip().startswith("KRX_API_KEY="): key = line.split("=", 1)[1].strip().strip("'\"") break if not key: raise RuntimeError("KRX_API_KEY가 없습니다. ~/.env 에 KRX_API_KEY=... 를 추가하세요.") return key KRX_API_KEY = _load_krx_api_key() KRX_BASE = "https://data-dbg.krx.co.kr/svc/apis" VKOSPI_INSTRUMENT_ID = 956761 # Investing exact KOSPI Volatility index def fetch_investing(pair_id: int, name: str, date_str: str) -> float | None: """investing.com에서 특정 날짜의 종가를 가져온다.""" url = f"https://api.investing.com/api/financialdata/historical/{pair_id}" params = { "start-date": date_str, "end-date": date_str, "time-frame": "Daily", "add-missing-rows": "false", } try: r = requests.get(url, params=params, headers=INVESTING_HEADERS, timeout=15) if r.status_code != 200: log.warning("%s: HTTP %s", name, r.status_code) return None rows = r.json().get("data", []) for row in rows: if str(row.get("rowDateTimestamp", ""))[:10] == date_str: val = float(str(row.get("last_close", "")).replace(",", "")) return val if np.isfinite(val) and val > 0 else None except Exception as e: log.warning("%s: %s", name, e) return None def fetch_index_close(symbol: str, date_str: str) -> float | None: """캐시가 아닌 KRX 원천에서 요청 날짜·대표지수를 정확히 선택한다.""" endpoint, name = { "KS11": ("/idx/kospi_dd_trd", "코스피"), "KQ11": ("/idx/kosdaq_dd_trd", "코스닥"), }[symbol] rows = fetch_krx(endpoint, date_str.replace("-", "")) for row in rows: if row.get("IDX_NM") == name: value = float(str(row.get("CLSPRC_IDX", "")).replace(",", "")) if not np.isfinite(value) or value <= 0: raise ValueError(f"{date_str} {name}: 유효하지 않은 종가") return value if rows: raise ValueError(f"{date_str} {name}: 응답에 대표지수 종가 없음") return None def compute_5ma(csv_path: Path, col: str, new_val: float) -> float: """원본 엑셀 + CSV raw 거래량을 이어서 5일 이동평균 계산.""" raw_col = "콜 거래량(raw)" if "CALL" in col else "풋 거래량(raw)" sheet_name = "코스피200옵션 콜매수 거래량" if "CALL" in col else "코스피200옵션 풋매수 거래량" history_frames: list[pd.DataFrame] = [] if EXCEL_PATH is not None: try: excel_df = pd.read_excel(EXCEL_PATH, sheet_name=sheet_name) excel_history = excel_df[["일자", "전체"]].copy() excel_history.columns = ["Date", raw_col] history_frames.append(excel_history) except Exception as e: log.warning("원본 옵션 히스토리 로드 실패 (%s): %s", sheet_name, e) if csv_path.exists(): csv_df = pd.read_csv(csv_path) if {"Date", raw_col}.issubset(csv_df.columns): csv_history = csv_df[["Date", raw_col]].copy() history_frames.append(csv_history) if not history_frames: return float(new_val) history = pd.concat(history_frames, ignore_index=True).copy() history.loc[:, "Date"] = pd.to_datetime(history["Date"], errors="coerce") history.loc[:, raw_col] = pd.to_numeric(history[raw_col], errors="coerce") history = history.dropna(subset=["Date", raw_col]) history = history[history[raw_col] > 0] history = history.sort_values("Date").drop_duplicates(subset=["Date"], keep="last") recent = history[raw_col].tail(4).tolist() recent.append(float(new_val)) return float(np.mean(recent)) def has_complete_fg_inputs( vkospi: float | None, bond_5y: float | None, bond_10y: float | None, raw_call: float, raw_put: float, ) -> bool: values = [vkospi, bond_5y, bond_10y, raw_call, raw_put] return all(v is not None and np.isfinite(v) and v > 0 for v in values) def fetch_krx(endpoint: str, bas_dd: str) -> list[dict]: """KRX Open API 호출.""" url = f"{KRX_BASE}{endpoint}?basDd={bas_dd}" headers = {"Content-Type": "application/json", "AUTH_KEY": KRX_API_KEY} r = requests.get(url, headers=headers, timeout=30) # 인증·통신·스키마 오류를 빈 거래일로 삼키지 않는다. 예외에는 API key를 넣지 않는다. if r.status_code != 200: raise RuntimeError(f"KRX {endpoint}: HTTP {r.status_code}") rows = r.json().get("OutBlock_1") if not isinstance(rows, list): raise ValueError(f"KRX {endpoint}: 일별 데이터 응답 형식 오류") if any(str(row.get("BAS_DD", "")) != bas_dd for row in rows): raise ValueError(f"KRX {endpoint}: 요청일 {bas_dd}와 응답 기준일 불일치") return rows _EXCEL_KOSPI_CACHE: pd.DataFrame | None = None # 원본 엑셀 KOSPI 시트의 컬럼 → CSV 컬럼 매핑. # 엑셀에 그 날짜 row 가 있으면 이 컬럼들은 엑셀 값을 신뢰 source 로 사용 (override). _EXCEL_OVERRIDE_COLS = { "코스피 200 변동성지수": "코스피 200 변동성지수", "5년 국채선물 추종 지수": "5년 국채선물 추종 지수", "10년국채선물지수": "10년국채선물지수", "코스피": "코스피", } def _load_excel_kospi_lookup() -> pd.DataFrame: """원본 엑셀 KOSPI 시트를 한 번만 로드하여 날짜→row 룩업으로 캐시. 원본 엑셀은 Quantiwise 가 검증한 캐노니컬 데이터 — 우리 collector 의 KRX 재구성은 엑셀에 row 가 없을 때만 (가장 최근 며칠) 추정으로 사용. 엑셀이 갱신되어 그날 row 가 들어오면 다음 collector 실행에서 자동으로 엑셀 값으로 정정됨. """ global _EXCEL_KOSPI_CACHE if _EXCEL_KOSPI_CACHE is not None: return _EXCEL_KOSPI_CACHE if EXCEL_PATH is None: _EXCEL_KOSPI_CACHE = pd.DataFrame() return _EXCEL_KOSPI_CACHE try: df = pd.read_excel(EXCEL_PATH, sheet_name="KOSPI") df["Date"] = pd.to_datetime(df["Date"], errors="coerce") for c in _EXCEL_OVERRIDE_COLS.values(): if c in df.columns: df[c] = pd.to_numeric(df[c], errors="coerce") df = df.dropna(subset=["Date"]).set_index("Date") _EXCEL_KOSPI_CACHE = df except Exception as e: log.warning("원본 엑셀 KOSPI 시트 로드 실패 (override 비활성): %s", e) _EXCEL_KOSPI_CACHE = pd.DataFrame() return _EXCEL_KOSPI_CACHE def _apply_excel_override(target_date, row: dict) -> dict: """원본 엑셀에 그 날짜 row 가 있으면 권위 있는 컬럼들을 엑셀 값으로 덮어쓴다.""" excel_df = _load_excel_kospi_lookup() if excel_df.empty: return row ts = pd.Timestamp(target_date) if ts not in excel_df.index: return row excel_row = excel_df.loc[ts] if isinstance(excel_row, pd.DataFrame): # 같은 날짜 중복 — 마지막 행 사용 excel_row = excel_row.iloc[-1] overridden = [] for excel_col, csv_col in _EXCEL_OVERRIDE_COLS.items(): if excel_col not in excel_df.columns: continue v = excel_row.get(excel_col) if pd.isna(v): continue v = float(v) cur = row.get(csv_col) if cur is None or (isinstance(cur, float) and pd.isna(cur)) or abs(float(cur) - v) > 1e-6: row[csv_col] = v overridden.append(csv_col) if overridden: log.info("원본 엑셀 override (%s): %s", target_date, ", ".join(overridden)) return row def load_existing_fg_history() -> pd.DataFrame: """원본 엑셀 + 누적 CSV에서 마지막 완전한 F&G 히스토리를 읽는다.""" history_cols = [ "Date", "5년 국채선물 추종 지수", "10년국채선물지수", "코스피 200 변동성지수", ] frames: list[pd.DataFrame] = [] if EXCEL_PATH is not None: try: excel_df = pd.read_excel(EXCEL_PATH, sheet_name="KOSPI") frames.append(excel_df[history_cols].copy()) except Exception as e: log.warning("원본 KOSPI 히스토리 로드 실패: %s", e) if CSV_PATH.exists(): csv_df = pd.read_csv(CSV_PATH) if set(history_cols).issubset(csv_df.columns): frames.append(csv_df[history_cols].copy()) if not frames: return pd.DataFrame(columns=history_cols) history = pd.concat(frames, ignore_index=True).copy() history.loc[:, "Date"] = pd.to_datetime(history["Date"], errors="coerce") for col in history_cols[1:]: history.loc[:, col] = pd.to_numeric(history[col], errors="coerce") history = history.dropna(subset=["Date"]) history = history.sort_values("Date").drop_duplicates(subset=["Date"], keep="last") return history.reset_index(drop=True) def fetch_krx_futures(bas_dd: str) -> tuple[float | None, float | None, float | None]: """KRX 선물 일별 데이터에서 변동성지수 선물/국채선물 종가를 읽는다.""" vkospi_future = None bond_5y_raw = None bond_10y_raw = None fut_rows = fetch_krx("/drv/fut_bydd_trd", bas_dd) for row in fut_rows: prod = row.get("PROD_NM", "") isu = row.get("ISU_NM", "") close = str(row.get("TDD_CLSPRC", "")).replace(",", "").strip() if not close: continue try: val = float(close) except ValueError: continue # ISU_NM 에 "주간" 표기가 없는 종목(5년국채 등)도 잡을 수 있도록 # "주간 in isu" → "야간 not in isu" 로 완화한다. SP (스프레드) 는 계속 배제. if "변동성" in prod and "SP" not in isu and "야간" not in isu and vkospi_future is None: vkospi_future = val if "5년국채" in prod and "야간" not in isu and "SP" not in isu and bond_5y_raw is None: bond_5y_raw = val if "10년국채" in prod and "야간" not in isu and "SP" not in isu and bond_10y_raw is None: bond_10y_raw = val if bond_5y_raw is None: for row in fut_rows: prod = row.get("PROD_NM", "") isu = row.get("ISU_NM", "") close = str(row.get("TDD_CLSPRC", "")).replace(",", "").strip() if "3년국채" in prod and "야간" not in isu and "SP" not in isu and close: try: bond_5y_raw = float(close) break except ValueError: pass return vkospi_future, bond_5y_raw, bond_10y_raw def fetch_bond_tracking_indices(bas_dd: str) -> tuple[float | None, float | None]: """직전 원본 지수값에 KRX 선물 수익률을 이어서 채권 추종 지수를 연장한다. bas_dd: 타겟일(YYYYMMDD). history 에서 strict 하게 그 이전 날짜의 row 를 base 로 사용해야 백필 시 미래 날짜를 base 로 잡아 KRX 가 미래 데이터를 못 줘서 row skip 되는 버그를 막을 수 있다. """ history = load_existing_fg_history() history = history.dropna(subset=["5년 국채선물 추종 지수", "10년국채선물지수"]) if history.empty: log.warning("채권 추종 지수 히스토리가 없어 연장 불가") return None, None target_ts = pd.Timestamp(datetime.strptime(bas_dd, "%Y%m%d").date()) history = history[history["Date"] < target_ts] if history.empty: log.warning("채권 추종 지수: %s 이전 base row 없음 — 연장 불가", bas_dd) return None, None last_row = history.iloc[-1] prev_date = last_row["Date"].strftime("%Y%m%d") _, prev_5y_raw, prev_10y_raw = fetch_krx_futures(prev_date) _, curr_5y_raw, curr_10y_raw = fetch_krx_futures(bas_dd) # 5Y/10Y 못 구하면 row 자체를 skip — 다른 듀레이션(3Y) 으로 대체하면 인덱스가 영구히 # 어긋나서 fear&greed feature 가 원본과 안 맞게 됨. 다음 실행에서 KRX 가 채워줄 때까지 기다림. if any(pd.isna(v) or v is None for v in [prev_5y_raw, prev_10y_raw, curr_5y_raw, curr_10y_raw]): log.warning( "채권 선물 원시값 누락: prev(%s)=[%s, %s], curr(%s)=[%s, %s] — row skip", prev_date, prev_5y_raw, prev_10y_raw, bas_dd, curr_5y_raw, curr_10y_raw, ) return None, None bond_5y = float(last_row["5년 국채선물 추종 지수"]) * float(curr_5y_raw) / float(prev_5y_raw) bond_10y = float(last_row["10년국채선물지수"]) * float(curr_10y_raw) / float(prev_10y_raw) log.info( "채권 추종지수 연장: 5Y %.2f→%.2f (raw %.2f→%.2f), 10Y %.2f→%.2f (raw %.2f→%.2f)", float(last_row["5년 국채선물 추종 지수"]), bond_5y, float(prev_5y_raw), float(curr_5y_raw), float(last_row["10년국채선물지수"]), bond_10y, float(prev_10y_raw), float(curr_10y_raw), ) return bond_5y, bond_10y def fetch_pykrx_bond_futures(bas_dd: str) -> tuple[float | None, float | None, float | None]: """pykrx 로 국채선물 + V-KOSPI 선물 종가 조회. KRX Open API T+1 지연 우회. Returns (vkospi_future, bond_3y_or_5y, bond_10y). BM3 (3년) 는 '5년 국채선물 추종 지수' 의 prev-day factor 로 사용됨. """ try: from pykrx import stock # type: ignore except ImportError: log.warning("pykrx 미설치 — bond futures fallback 불가") return None, None, None def _get_main(prod: str) -> float | None: """주간 최근월물 종가 (없으면 야간, 없으면 현물가).""" try: df = stock.get_future_ohlcv_by_ticker(bas_dd, prod) except Exception as e: log.warning("pykrx %s @%s 실패: %s", prod, bas_dd, e) return None if df is None or df.empty: return None # 우선순위: 주간 > 야간 > 현물가 names = df['종목명'].astype(str) mask_day = names.str.contains("주간") & ~names.str.contains("SP") & (df['종가'] > 0) if mask_day.any(): return float(df.loc[mask_day, '종가'].iloc[0]) # fallback: 모든 non-SP 중 종가 > 0 mask_any = ~names.str.contains("SP") & (df['종가'] > 0) if mask_any.any(): return float(df.loc[mask_any, '종가'].iloc[0]) # final: 현물가 mask_spot = ~names.str.contains("SP") & (df['현물가'] > 0) if mask_spot.any(): return float(df.loc[mask_spot, '현물가'].iloc[0]) return None vkospi = _get_main("KRDRVFUVKI") bond_3y = _get_main("KRDRVFUBM3") # 5Y slot 대체 후보 (collector 가 3년으로 5년 fallback) bond_10y = _get_main("KRDRVFUBMA") return vkospi, bond_3y, bond_10y def fetch_pykrx_option_volumes(bas_dd: str) -> tuple[float, float]: """pykrx 로 KOSPI200 정규 월간옵션 CALL/PUT 총 거래량 (원본 KRX 방식과 일치).""" try: from pykrx import stock # type: ignore except ImportError: log.warning("pykrx 미설치 — option fallback 불가") return 0.0, 0.0 try: df = stock.get_future_ohlcv_by_ticker(bas_dd, "KRDRVOPK2I") except Exception as e: log.warning("pykrx option fetch 실패: %s", e) return 0.0, 0.0 if df is None or df.empty: return 0.0, 0.0 calls = df[df['종목명'].str.contains(' C ', regex=False, na=False)] puts = df[df['종목명'].str.contains(' P ', regex=False, na=False)] call_vol = float(calls['거래량'].sum()) if '거래량' in df.columns else 0.0 put_vol = float(puts['거래량'].sum()) if '거래량' in df.columns else 0.0 pc_ratio = put_vol / call_vol if call_vol else 0 log.info("pykrx 옵션 (%s KRDRVOPK2I): CALL=%s PUT=%s P/C=%.3f", bas_dd, f"{int(call_vol):,}", f"{int(put_vol):,}", pc_ratio) return call_vol, put_vol def fetch_kis_option_volumes(mtrt_yyyymm: str | None = None) -> tuple[float, float]: """KIS REST API 로 코스피200 옵션 콜/풋 총 거래량 합계. mtrt_yyyymm=None 이면 오늘 기준 '현재 월물' (다음달) 자동 선택. KRX T+1 지연 우회 — 실시간/당일 데이터 즉시 제공. """ import sys as _sys _sys.path.insert(0, "/Users/loonshots/program-trading-predictor") from data.kis_client import KISProgramClient # type: ignore if mtrt_yyyymm is None: today = date.today() # 옵션 최근월물 = 현재월(만기 전) 또는 다음월 m = today.month + 1 if today.day > 14 else today.month y = today.year + (m > 12) m = m if m <= 12 else 1 mtrt_yyyymm = f"{y:04d}{m:02d}" kis = KISProgramClient() token = kis.authenticate() url = "https://openapi.koreainvestment.com:9443/uapi/domestic-futureoption/v1/quotations/display-board-callput" hdr = { "authorization": f"Bearer {token}", "appkey": kis.app_key, "appsecret": kis.app_secret, "tr_id": "FHPIF05030100", "Content-Type": "application/json", } r = requests.get(url, headers=hdr, params={ "FID_COND_MRKT_DIV_CODE": "O", "FID_COND_SCR_DIV_CODE": "20503", "FID_MRKT_CLS_CODE": "CO", "FID_MTRT_CNT": mtrt_yyyymm, "FID_MRKT_CLS_CODE1": "PO", "FID_COND_MRKT_CLS_CODE": "", }, timeout=15) d = r.json() if d.get("rt_cd") != "0": log.warning("KIS option API error: %s", d.get("msg1")) return 0.0, 0.0 calls = d.get("output1") or [] puts = d.get("output2") or [] call_vol = sum(int(c.get("acml_vol", "0") or 0) for c in calls) put_vol = sum(int(p.get("acml_vol", "0") or 0) for p in puts) log.info("KIS 옵션 (%s): CALL_vol=%,d PUT_vol=%,d P/C=%.3f", mtrt_yyyymm, call_vol, put_vol, put_vol / call_vol if call_vol else 0) return float(call_vol), float(put_vol) def fetch_krx_option_volumes(bas_dd: str) -> tuple[float, float]: """KRX 코스피200 옵션 일별 전체 거래량(야간 포함)을 수집한다.""" opt_rows = fetch_krx("/drv/opt_bydd_trd", bas_dd) call_vol = 0.0 put_vol = 0.0 for row in opt_rows: prod = row.get("PROD_NM", "") rght = row.get("RGHT_TP_NM", "") vol_str = str(row.get("ACC_TRDVOL", "0")).replace(",", "") if prod != "코스피200 옵션": continue try: vol = float(vol_str) except ValueError: continue if rght == "CALL": call_vol += vol elif rght == "PUT": put_vol += vol log.info("K200 옵션 거래량(raw): CALL=%.0f, PUT=%.0f", call_vol, put_vol) return call_vol, put_vol def collect_for_date(target_date) -> dict | None: """지정 날짜의 F&G 데이터 수집. None 이면 해당 날짜 데이터 없음.""" target_str = target_date.strftime("%Y-%m-%d") target_compact = target_date.strftime("%Y%m%d") log.info("=== F&G 일별 수집 시작: %s ===", target_str) # 1. 지수 — 휴장일 감지용으로도 쓰인다 kospi = fetch_index_close("KS11", target_str) kosdaq = fetch_index_close("KQ11", target_str) if kospi is None or kosdaq is None: if kospi is not None or kosdaq is not None: raise ValueError(f"{target_str}: KOSPI/KOSDAQ 중 한 지수 누락") log.info("KRX 양 시장 공표 자료 없음 (%s): 휴장 또는 미공표", target_str) return None # 2. 변동성지수: 원본 Quantiwise와 동일한 investing instrument 사용 vkospi = fetch_investing(VKOSPI_INSTRUMENT_ID, "KOSPI Volatility", target_str) # 3. 채권 추종지수: 기존 지수에 KRX 선물 수익률을 이어 붙인다. bond_5y, bond_10y = fetch_bond_tracking_indices(target_compact) # 4. 옵션 거래량(raw): KRX Open API only. # pykrx(KRDRVOPK2I)/KIS REST 는 정의가 다른 통계량 (월간만 / 100 strike 만) 이므로 # fallback 으로 섞으면 원본과 영구히 안 맞는다. 0 이면 row skip → 다음 실행에서 자가치유. raw_call, raw_put = fetch_krx_option_volumes(target_compact) if not has_complete_fg_inputs(vkospi, bond_5y, bond_10y, raw_call, raw_put): log.warning( "핵심 F&G 입력 누락으로 CSV 저장 스킵 (%s): VKOSPI=%s, 5Y=%s, 10Y=%s, CALL(raw)=%s, PUT(raw)=%s", target_str, vkospi, bond_5y, bond_10y, raw_call, raw_put, ) raise ValueError(f"{target_str}: 거래일의 핵심 F&G 입력 누락") # 5MA 컬럼은 raw 만 저장하고 NaN 으로 둠 — 수집 단계에서는 그 시점 history 에 의존하므로 # 이후 실행에서 raw 가 자기치유로 바뀌어도 5MA 가 stuck 됨. 모든 row 수집 끝난 뒤 # recompute_5ma_from_raw() 가 전체 CSV 를 재스캔해서 raw 의 rolling(5).mean() 으로 일관되게 채움. row = { "Date": target_str, "코스피": kospi, "코스닥": kosdaq, "코스피 200 변동성지수": vkospi, "5년 국채선물 추종 지수": bond_5y, "10년국채선물지수": bond_10y, "콜 거래량(raw)": raw_call, "풋 거래량(raw)": raw_put, "최근월물 CALL ATM": np.nan, "최근월물 PUT ATM": np.nan, } # 원본 엑셀이 그 날짜 row 를 갖고 있으면 KRX 재구성 대신 엑셀 값을 신뢰 source 로 사용. # (특히 채권 추종지수처럼 chained product 라 base 가 한 번 어긋나면 영구 어긋남 — 엑셀로 리셋.) row = _apply_excel_override(target_date, row) # 전일 row 대비 변동량 + stuck 탐지 (target 이전 날짜 row 와 비교) prev = _load_prev_row(exclude_date=target_str) delta_parts = [] stuck_cols = [] def _fmt_delta(name: str, cur: float, key: str) -> str: pv = prev.get(key) if prev else None if pv is None or cur is None: return f"{name}={cur}" try: pv_f = float(pv) d = cur - pv_f pct = (d / pv_f * 100) if pv_f else 0.0 if abs(d) < 1e-9: stuck_cols.append(name) return f"{name}={cur:.2f} (Δ{d:+.2f}/{pct:+.2f}%)" except (TypeError, ValueError): return f"{name}={cur}" pc_ratio = (raw_put / raw_call) if raw_call else float("nan") bond_diff = (bond_10y - bond_5y) if (bond_5y is not None and bond_10y is not None) else None log.info( "수집 완료 (%s) | %s | %s | %s | %s | %s | CALL_raw=%.0f PUT_raw=%.0f | P/C=%.3f | BondDiff=%s (5MA 는 후처리에서 재계산)", target_str, _fmt_delta("KOSPI", kospi, "코스피"), _fmt_delta("KOSDAQ", kosdaq, "코스닥"), _fmt_delta("VKOSPI", vkospi, "코스피 200 변동성지수"), _fmt_delta("5Y", bond_5y, "5년 국채선물 추종 지수"), _fmt_delta("10Y", bond_10y, "10년국채선물지수"), raw_call, raw_put, pc_ratio, f"{bond_diff:+.2f}" if bond_diff is not None else "?", ) if stuck_cols: log.warning( "⚠️ 전일 대비 변동 없음 (stuck 가능성) — %s (%s)", ", ".join(stuck_cols), target_str, ) return row def _load_prev_row(exclude_date: str | None = None) -> dict: """CSV 에서 exclude_date 를 제외한 가장 최근 row 반환.""" if not CSV_PATH.exists(): return {} try: with open(CSV_PATH, "r", encoding="utf-8") as f: rows = list(csv.DictReader(f)) for r in reversed(rows): if exclude_date and r.get("Date", "") >= exclude_date: continue return r return {} except Exception: return {} def append_to_csv(row: dict) -> None: """CSV에 일자 기준 upsert. 파일 없으면 생성. 같은 날짜의 기존 row 와 비교해 drift 감지.""" file_exists = CSV_PATH.exists() fieldnames = list(row.keys()) existing_rows: list[dict] = [] prev_row: dict | None = None if file_exists: with open(CSV_PATH, "r", encoding="utf-8") as f: reader = csv.DictReader(f) if reader.fieldnames: fieldnames = list(reader.fieldnames) for k in row: if k not in fieldnames: fieldnames.append(k) for existing_row in reader: if existing_row.get("Date") == row["Date"]: prev_row = existing_row continue existing_rows.append(existing_row) with open(CSV_PATH, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for existing_row in existing_rows: writer.writerow(existing_row) writer.writerow(row) if prev_row is not None: # drift 감지: 같은 날짜 row 가 이미 있었는데 값이 크게 바뀌었다면 로깅 # (자가치유 정상 보정이 대부분이지만 50% 초과 변화는 데이터 소스 이상 시그널일 수 있음) for k, new_v in row.items(): if k == "Date": continue try: prev_v = float(prev_row.get(k) or "nan") cur_v = float(new_v) if new_v is not None and not (isinstance(new_v, float) and pd.isna(new_v)) else float("nan") except (TypeError, ValueError): continue if pd.isna(prev_v) or pd.isna(cur_v) or prev_v == 0: continue ratio = abs(cur_v - prev_v) / abs(prev_v) if ratio >= DRIFT_WARN_PCT: log.warning( "DRIFT (%s) %s: %.4f → %.4f (Δ%+.1f%%)", row["Date"], k, prev_v, cur_v, (cur_v - prev_v) / prev_v * 100, ) elif ratio > 0.001: log.info( "보정 (%s) %s: %.4f → %.4f (Δ%+.2f%%)", row["Date"], k, prev_v, cur_v, (cur_v - prev_v) / prev_v * 100, ) action = "갱신" else: action = "추가" log.info("CSV 저장(%s): %s (→ %s)", action, CSV_PATH.name, row["Date"]) def recompute_5ma_from_raw() -> None: """전체 CSV 를 재스캔하여 raw 컬럼의 5일 rolling mean 으로 5MA 컬럼을 일관되게 채운다. 수집 단계에서 5MA 를 inline 으로 계산하면 그 시점의 history 가 굳어서 이후 raw 가 자기치유로 갱신돼도 stuck 됨. 모든 raw upsert 가 끝난 뒤 한 번에 재계산하면 raw 컬럼만이 single source of truth 가 되고 5MA 는 항상 그 파생값이 됨. """ if not CSV_PATH.exists(): return df = pd.read_csv(CSV_PATH) if df.empty or "Date" not in df.columns: return df["Date"] = pd.to_datetime(df["Date"], errors="coerce") df = df.dropna(subset=["Date"]).sort_values("Date").reset_index(drop=True) changed = 0 for raw_col, ma_col in RAW_TO_MA.items(): if raw_col not in df.columns: continue if ma_col not in df.columns: df[ma_col] = np.nan raw = pd.to_numeric(df[raw_col], errors="coerce") # 5일 rolling mean — raw 가 NaN 인 날은 결과도 NaN 이 되도록 min_periods=5 new_ma = raw.rolling(window=5, min_periods=5).mean().round(2) old_ma = pd.to_numeric(df[ma_col], errors="coerce") # 변화량 카운트 (drift 표시용) diff_mask = (~old_ma.isna() & ~new_ma.isna() & ((old_ma - new_ma).abs() > 0.5)) changed += int(diff_mask.sum()) df[ma_col] = new_ma # Date 를 다시 문자열로 (원본 포맷 유지) df["Date"] = df["Date"].dt.strftime("%Y-%m-%d") df.to_csv(CSV_PATH, index=False) log.info("5MA 재계산 완료: %d개 셀 변경 (raw rolling(5).mean)", changed) def _last_csv_date() -> date | None: """CSV 의 마지막 저장된 날짜. 없거나 파싱 실패 시 None.""" if not CSV_PATH.exists(): return None try: df = pd.read_csv(CSV_PATH) except Exception as e: log.warning("CSV 마지막 날짜 조회 실패: %s", e) return None if "Date" not in df.columns or df.empty: return None dates = pd.to_datetime(df["Date"], errors="coerce").dropna() if dates.empty: return None return dates.max().date() def _previous_business_day(d: date) -> date: """d 이전의 가장 최근 weekday (Mon-Fri). 공휴일은 고려하지 않음 — API 가 0 row 를 반환하면 자연스럽게 skip 된다.""" d = d - timedelta(days=1) while d.weekday() >= 5: # 5=Sat, 6=Sun d -= timedelta(days=1) return d def _shift_business_days(d: date, n: int) -> date: """d 로부터 n 영업일 이전 날짜. n>=0.""" out = d moved = 0 while moved < n: out -= timedelta(days=1) if out.weekday() < 5: moved += 1 return out def main(argv: list[str] | None = None) -> dict: """수집 + 자기치유 백필 + 5MA 재계산. KRX `data-dbg` Open API 는 일별 선물/옵션 데이터를 T+1 패턴으로 반영한다 (다음 영업일 오후 1시 경부터). 따라서 오늘 데이터는 조회 자체가 불가능하고, 어제(직전 영업일)까지만 수집 대상이다. 수집 범위: - 기본: max(CSV 마지막 저장일 - LOOKBACK_BUSINESS_DAYS, 시작일) ~ 어제 → 마지막 N영업일을 항상 재요청 → fallback 으로 채워졌던 row 가 정상값으로 upsert. - --backfill-from YYYY-MM-DD: 그 날짜부터 ~ 어제 (1회성 대규모 정정용). - --backfill-days N: 마지막 N영업일 재수집 (--backfill-from 없을 때 LOOKBACK 대체). 수집 끝나면 recompute_5ma_from_raw() 가 전체 CSV 의 raw 컬럼에서 5MA 를 재계산해 raw 갱신과 5MA 가 항상 동기화되도록 한다. """ parser = argparse.ArgumentParser(description="Fear&Greed daily collector") parser.add_argument("--backfill-from", type=str, default=None, help="이 날짜(YYYY-MM-DD)부터 어제까지 재수집. 1회성 대규모 정정용.") parser.add_argument("--backfill-days", type=int, default=None, help="마지막 N영업일을 재수집 (기본: LOOKBACK_BUSINESS_DAYS=%d)" % LOOKBACK_BUSINESS_DAYS) parser.add_argument("--skip-recompute", action="store_true", help="5MA 재계산 단계 스킵 (raw 만 업데이트).") # 노트북은 main([])으로 호출해 ipykernel의 -f 등을 전달하지 않는다. args = parser.parse_args(argv) today = datetime.now().date() target_end = _previous_business_day(today) if args.backfill_from: try: start = datetime.strptime(args.backfill_from, "%Y-%m-%d").date() except ValueError: parser.error("--backfill-from: YYYY-MM-DD 형식이 필요합니다.") log.info("백필 모드: %s ~ %s (1회성 정정)", start, target_end) else: lookback = args.backfill_days if args.backfill_days is not None else LOOKBACK_BUSINESS_DAYS last = _last_csv_date() # 마지막 N영업일 재수집 + 신규 누락분 — 둘 중 더 이른 날짜 채택 lookback_start = _shift_business_days(target_end, lookback) if last is None: start = lookback_start else: new_start = last + timedelta(days=1) start = min(lookback_start, new_start) log.info( "정상 모드: 자기치유 lookback=%d영업일, 수집 범위 %s ~ %s (CSV 마지막=%s)", lookback, start, target_end, last, ) # 이전 실행에서 실패한 날짜가 lookback 밖으로 밀려나도 재시도한다. if STATUS_PATH.exists(): try: previous = json.loads(STATUS_PATH.read_text(encoding="utf-8")) for failed_date in previous.get("failed_dates", {}): start = min(start, date.fromisoformat(failed_date)) except (ValueError, OSError, TypeError): pass current = start collected = 0 skipped = 0 failed_dates = {} no_data_dates = [] while current <= target_end: if current.weekday() < 5: # Mon-Fri 만 시도 try: row = collect_for_date(current) if row is not None: append_to_csv(row) collected += 1 else: no_data_dates.append(current.isoformat()) skipped += 1 except Exception as exc: failed_dates[current.isoformat()] = f"{type(exc).__name__}: {exc}" log.error("수집 실패 (%s): %s", current, failed_dates[current.isoformat()]) current += timedelta(days=1) log.info("수집 종료: 저장/갱신=%d, 스킵=%d", collected, skipped) if collected == 0: log.warning( "수집 데이터 없음 — 스킵 (%s ~ %s). KRX Open API 는 T+1 반영이므로 " "너무 이른 시각이면 내일 오후 14:00 이후 재실행.", start, target_end, ) if not args.skip_recompute: recompute_5ma_from_raw() last = _last_csv_date() status = { "checked_at": datetime.now().isoformat(timespec="seconds"), "requested_start": start.isoformat(), "requested_through": target_end.isoformat(), "as_of": last.isoformat() if last else None, "status": "failed" if failed_dates else "ok" if last == target_end and collected else "pending", "collected": collected, "no_data_dates": no_data_dates, "failed_dates": failed_dates, "source": "KRX Open API + Investing VKOSPI + 원본 Excel", } tmp = STATUS_PATH.with_suffix(".json.tmp") tmp.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(STATUS_PATH) return status if __name__ == "__main__": result = main() if result["status"] != "ok": raise SystemExit(1)