dsyoon
b52d61b777
로고스/루트 레거시를 제거하고 deepcoin 패키지·scripts 01~05 CLI·docs/reference로 데이터·GT·분석·매칭·운영 단계를 정리했다. config와 .env 기반 설정, trade_anaysis.html 동기화 포함. Co-authored-by: Cursor <cursoragent@cursor.com>
101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
"""
|
||
general_analysis ground truth 타점 MTF 스냅샷 생성.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pandas as pd
|
||
|
||
from deepcoin.analysis.general_analysis_align import general_analysis_mtf_scores
|
||
from deepcoin.analysis.general_analysis_config import (
|
||
DEFAULT_OUTPUT_CSV,
|
||
DEFAULT_TRADES_FILE,
|
||
GENERAL_ANALYSIS_INTERVALS,
|
||
)
|
||
from deepcoin.analysis.general_analysis_core import interval_tf_prefix
|
||
from deepcoin.analysis.general_analysis_pipeline import general_analysis_enrich_bars, general_analysis_snapshot_at_bar
|
||
from deepcoin.ground_truth.ground_truth import load_ground_truth
|
||
|
||
|
||
def _prefixed_snap(snap: dict[str, Any], interval: int) -> dict[str, Any]:
|
||
p = interval_tf_prefix(interval)
|
||
return {f"{p}_{k}": v for k, v in snap.items()}
|
||
|
||
|
||
def build_trade_mtf_snapshots(
|
||
frames: dict[int, pd.DataFrame],
|
||
trades: list[dict[str, Any]],
|
||
) -> pd.DataFrame:
|
||
"""
|
||
모든 타점에 대해 8개 간격 general_analysis 스냅샷.
|
||
|
||
Args:
|
||
frames: interval → OHLCV.
|
||
trades: ground_truth trades.
|
||
|
||
Returns:
|
||
wide DataFrame (1 row per trade).
|
||
"""
|
||
enriched: dict[int, pd.DataFrame] = {}
|
||
for iv in GENERAL_ANALYSIS_INTERVALS:
|
||
raw = frames.get(iv)
|
||
if raw is None or raw.empty:
|
||
continue
|
||
print(f" [GA] {interval_tf_prefix(iv)} 봉 지표 계산 ({len(raw)}봉)...")
|
||
enriched[iv] = general_analysis_enrich_bars(raw, iv, full_context=True)
|
||
|
||
rows: list[dict[str, Any]] = []
|
||
for i, t in enumerate(sorted(trades, key=lambda x: x["dt"])):
|
||
ts = pd.Timestamp(t["dt"])
|
||
row: dict[str, Any] = {
|
||
"trade_idx": i,
|
||
"dt": t["dt"],
|
||
"action": t["action"],
|
||
"price": t["price"],
|
||
"weight": t.get("weight", 1.0),
|
||
"leg_id": t.get("leg_id", 0),
|
||
"memo": t.get("memo", ""),
|
||
}
|
||
flat: dict[str, Any] = {}
|
||
for iv in GENERAL_ANALYSIS_INTERVALS:
|
||
ef = enriched.get(iv)
|
||
if ef is None:
|
||
continue
|
||
snap = general_analysis_snapshot_at_bar(ef, ts, iv)
|
||
flat.update(_prefixed_snap(snap, iv))
|
||
row.update(flat)
|
||
row.update(general_analysis_mtf_scores(flat))
|
||
rows.append(row)
|
||
if (i + 1) % 50 == 0:
|
||
print(f" 타점 스냅샷 {i + 1}/{len(trades)}")
|
||
|
||
return pd.DataFrame(rows)
|
||
|
||
|
||
def export_trade_snapshots(
|
||
frames: dict[int, pd.DataFrame],
|
||
trades_path: Path | str = DEFAULT_TRADES_FILE,
|
||
output_csv: Path | str = DEFAULT_OUTPUT_CSV,
|
||
) -> Path:
|
||
"""
|
||
CSV로 타점 MTF 스냅샷 저장.
|
||
|
||
Returns:
|
||
저장 경로.
|
||
"""
|
||
data = load_ground_truth(Path(trades_path))
|
||
if not data:
|
||
raise FileNotFoundError(f"정답 파일 없음: {trades_path}")
|
||
trades = data.get("trades") or []
|
||
print(f"타점 {len(trades)}건 × {len(GENERAL_ANALYSIS_INTERVALS)} TF general_analysis")
|
||
df = build_trade_mtf_snapshots(frames, trades)
|
||
out = Path(output_csv)
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
df.to_csv(out, index=False, encoding="utf-8-sig")
|
||
print(f"저장: {out} ({len(df)}행 × {len(df.columns)}열)")
|
||
return out
|