#!/usr/bin/env python3 """2단계: GT v3 타점 · 멀티 TF 피처 상관 분석.""" from __future__ import annotations import argparse import logging import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from bithumb.config import load_settings from bithumb.evaluation.mtf_report import ( build_mtf_correlation_report, render_mtf_html, save_mtf_report, ) from bithumb.mtf.extractor import MtfFeatureExtractor from bithumb.mtf.rules import derive_rules_from_report, save_mtf_rules from bithumb.mtf.store import MultiTimeframeStore from bithumb.techniques.runner import load_ground_truth def _configure_logging(verbose: bool) -> None: """로깅 레벨을 설정한다.""" level = logging.DEBUG if verbose else logging.INFO logging.basicConfig( level=level, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) def _resolve_gt_path(settings, gt_file: str | None) -> Path: """GT JSON 경로를 결정한다.""" if gt_file: path = Path(gt_file) if not path.is_absolute(): path = ROOT / path return path return settings.ground_truth_file def main() -> int: """CLI 진입점.""" parser = argparse.ArgumentParser(description="2단계: GT v3 MTF 피처 상관 분석") parser.add_argument("--gt-file", type=str, default=None, help="GT JSON (기본 v3)") parser.add_argument( "--days", type=int, default=None, help="분석 구간(일). 기본 GT_SIM_LOOKBACK_DAYS", ) parser.add_argument( "--negative-samples", type=int, default=2000, help="음성 샘플 3분봉 수", ) parser.add_argument( "--exclude-bars", type=int, default=60, help="GT 주변 제외 3분봉 수", ) parser.add_argument("--seed", type=int, default=42, help="음성 샘플 RNG seed") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args() _configure_logging(args.verbose) settings = load_settings() gt_path = _resolve_gt_path(settings, args.gt_file) lookback_days = args.days or settings.gt_sim_lookback_days logging.info( "MTF 상관 분석: %s, 최근 %d일, 음성 %d건", gt_path.name, lookback_days, args.negative_samples, ) gt_result = load_ground_truth(gt_path) store = MultiTimeframeStore( db_path=settings.db_path, symbol=settings.symbol, lookback_days=lookback_days + 120, zigzag_reversal_pct=settings.gt_zigzag_reversal_pct, ) extractor = MtfFeatureExtractor( store=store, base_interval_min=settings.gt_interval_min, ) report = build_mtf_correlation_report( gt_result=gt_result, extractor=extractor, lookback_days=lookback_days, negative_sample_count=args.negative_samples, exclude_bars=args.exclude_bars, seed=args.seed, ) json_path = save_mtf_report(report, settings.mtf_report_json) html_path = render_mtf_html(report, settings.mtf_report_html) rule_set = derive_rules_from_report(report) rules_path = save_mtf_rules(rule_set, settings.mtf_rules_json) gt = report.get("gt", {}) top = (report.get("global_feature_ranking") or [])[:5] print("\n=== GT v3 MTF 상관 분석 ===") print(f"구간: {report['analysis']['period_from']} ~ {report['analysis']['period_to']}") print( f"GT 신호 {gt.get('signals_in_period', 0)}건 · " f"스냅샷 {gt.get('snapshots_extracted', 0)}건 · " f"음성 {report['analysis']['negative_sample_count']}건" ) print("\n상위 피처 (|Cohen's d|):") for row in top: print( f" {row['signal_label']} | {row['timeframe']} | {row['feature']} | " f"d={row['cohens_d']}" ) print(f"\nJSON: {json_path}") print(f"HTML: {html_path}") print(f"MTF rules: {rules_path}") return 0 if __name__ == "__main__": raise SystemExit(main())