1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903 | #!/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)
|