예제 소스

build_market_extreme_labels.py

"""
Market.get_data()로 DB 캔들을 불러온 뒤, Market.py 의 AdvancedAnomalyDetector 로
극단·이상 구간 라벨을 붙여 저장합니다.

출력: 기본 cache/market_extreme_labels.csv
(경로는 아래 OUTPUT_CSV 상수만 바꿔서 조정)

주의:
- IsolationForest 는 전체 시계열로 한 번 학습됩니다(Market.add_enhanced_anomaly_detection 과 동일).
  실시간 순행 시뮬레이션용 라벨이면 pattern_anomaly 해석에 누수가 있을 수 있습니다.
- 규칙 기반 crash / surge / price·volume spike 는 각 시점 t 의 과거 창만 사용합니다.
"""

from __future__ import annotations

import json
import logging.config
from pathlib import Path

import numpy as np
import pandas as pd

from Market import AdvancedAnomalyDetector, Market

OUTPUT_CSV = "cache/market_extreme_labels.csv"
# True 이면 bar_index + 아래 컬럼(있는 것만) + ext_* 만 저장. False 이면 get_data() 전체 컬럼 + ext_*.
SAVE_SLIM_COLUMNS = True
SLIM_SOURCE_COLUMNS = (
    "open",
    "high",
    "low",
    "close",
    "volume",
    "market_status",
    "macd",
    "macd_degrees",
    "rsi",
    "ma20",
    "ma60",
)


def _setup_logging() -> None:
    cfg = Path(__file__).resolve().parent / "config" / "logging.json"
    if cfg.exists():
        with open(cfg, "r", encoding="utf-8") as f:
            logging.config.dictConfig(json.load(f))


def _regime_id(row: pd.Series) -> int:
    """극단 상황 단순 분류 (약한 라벨)."""
    c = bool(row.get("ext_crash_signal"))
    s = bool(row.get("ext_surge_signal"))
    if c and s:
        return 3
    if c:
        return 1
    if s:
        return 2
    if bool(row.get("ext_price_spike")):
        return 4
    if bool(row.get("ext_volume_spike")):
        return 5
    if bool(row.get("ext_pattern_anomaly")):
        return 6
    return 0


def build_extreme_label_frame(df: pd.DataFrame) -> pd.DataFrame:
    """각 bar 에 대해 detect_anomalies 결과를 행으로 펼친다."""
    detector = AdvancedAnomalyDetector(df)
    rows: list[dict] = []
    n = len(df)
    for i in range(n):
        rows.append(detector.detect_anomalies(i))
    ext = pd.DataFrame(rows)
    ext.columns = [f"ext_{c}" for c in ext.columns]
    ext["ext_regime_id"] = ext.apply(_regime_id, axis=1)
    ext["ext_regime_name"] = ext["ext_regime_id"].map(
        {
            0: "normal",
            1: "crash",
            2: "surge",
            3: "crash_surge_conflict",
            4: "price_spike",
            5: "volume_spike",
            6: "pattern_anomaly",
        }
    )
    base = df.reset_index(drop=True).copy()
    base.insert(0, "bar_index", np.arange(len(base), dtype=np.int64))
    out = pd.concat([base, ext], axis=1)
    return out


def main() -> None:
    _setup_logging()
    out_path = Path(OUTPUT_CSV)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    market = Market()
    df = market.get_data()
    if df is None or len(df) < 60:
        raise SystemExit("Market.get_data() 결과가 비었거나 너무 짧습니다.")

    labeled = build_extreme_label_frame(df)
    ext_cols = [c for c in labeled.columns if c.startswith("ext_")]

    if SAVE_SLIM_COLUMNS:
        cols = ["bar_index"]
        for c in SLIM_SOURCE_COLUMNS:
            if c in labeled.columns:
                cols.append(c)
        labeled = labeled[cols + ext_cols]
    else:
        pass  # 전체 컬럼 + ext_* (중복 없음: ext_ 는 원본에 없음)

    labeled.to_csv(out_path, index=False, encoding="utf-8-sig")
    print(f"[OK] 저장: {out_path.resolve()}  rows={len(labeled)}", flush=True)

    vc = labeled["ext_regime_name"].value_counts()
    print("ext_regime_name 분포:\n", vc.to_string(), flush=True)


if __name__ == "__main__":
    main()