90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
"""build_operational_t20_outcome_ledger_v1.py — ALPHA_FEEDBACK_LOOP_V2
|
|
|
|
매수/매도 결정 20거래일 후의 실제 수익률을 추적하여 레저(Ledger)를 구축한다.
|
|
성과 인텔리전스(Phase 4)의 핵심 데이터 소스로 활용됨.
|
|
|
|
로직:
|
|
1. alpha_history 시트에서 과거 결정(Decision) 데이터를 읽음.
|
|
2. 현재 가격(Close)을 T+20 가격으로 가정하여 실현 수익률 계산.
|
|
3. outcome_ledger.json 생성.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
TEMP = ROOT / "Temp"
|
|
DEFAULT_JSON = ROOT / "GatherTradingData.json"
|
|
DEFAULT_OUT = TEMP / "outcome_ledger_v1.json"
|
|
|
|
def _load(path: Path) -> Any:
|
|
if not path.exists(): return {}
|
|
try: return json.loads(path.read_text(encoding="utf-8"))
|
|
except: return {}
|
|
|
|
def _f(v: Any, default: float = 0.0) -> float:
|
|
try: return float(v)
|
|
except: return default
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--json", default=str(DEFAULT_JSON))
|
|
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
|
args = ap.parse_args()
|
|
|
|
payload = _load(Path(args.json))
|
|
data = payload.get("data", {})
|
|
|
|
# alpha_history: 과거 예측 데이터
|
|
alpha_history = data.get("alpha_history", [])
|
|
# data_feed: 현재 가격 데이터 (T+20 프록시)
|
|
df_rows = data.get("data_feed", [])
|
|
current_prices = {str(r.get("Ticker")): _f(r.get("Close")) for r in df_rows if r.get("Ticker")}
|
|
|
|
ledger_rows = []
|
|
for h in alpha_history:
|
|
ticker = str(h.get("ticker"))
|
|
entry_price = _f(h.get("close_at_record"))
|
|
current_price = current_prices.get(ticker, 0.0)
|
|
|
|
if entry_price > 0 and current_price > 0:
|
|
return_pct = round((current_price - entry_price) / entry_price * 100, 2)
|
|
verdict = h.get("synthesis_verdict")
|
|
|
|
# 예측 적중 여부 (간단 로직: BUY면 +, EXIT면 -)
|
|
is_correct = False
|
|
if "BUY" in str(verdict) and return_pct > 0: is_correct = True
|
|
if "EXIT" in str(verdict) and return_pct < 0: is_correct = True
|
|
|
|
ledger_rows.append({
|
|
"date": h.get("date"),
|
|
"ticker": ticker,
|
|
"verdict": verdict,
|
|
"entry_price": entry_price,
|
|
"exit_price": current_price,
|
|
"return_pct": return_pct,
|
|
"is_correct": is_correct
|
|
})
|
|
|
|
win_rate = round(sum(1 for r in ledger_rows if r["is_correct"]) / len(ledger_rows) * 100, 2) if ledger_rows else 0
|
|
|
|
result = {
|
|
"formula_id": "ALPHA_FEEDBACK_LOOP_V2",
|
|
"as_of_date": datetime.now().strftime("%Y-%m-%d"),
|
|
"total_cases": len(ledger_rows),
|
|
"win_rate_pct": win_rate,
|
|
"ledger": ledger_rows
|
|
}
|
|
|
|
Path(args.out).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(args.out).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"Outcome Ledger Built. Total Cases: {len(ledger_rows)}, Win Rate: {win_rate}%")
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
main()
|