""" 성과 데이터 영구 저장 - PerformanceDB 데이터 파일: data/performance/daily_snapshots.json - 일별 자산 스냅샷 data/performance/trade_records.json - 강화 매매 기록 (영구 보관) """ import os import json from datetime import datetime, timedelta from modules.config import Config PERF_DIR = os.path.join(Config.DATA_DIR, "performance") SNAPSHOTS_FILE = os.path.join(PERF_DIR, "daily_snapshots.json") TRADES_FILE = os.path.join(PERF_DIR, "trade_records.json") class PerformanceDB: def __init__(self): os.makedirs(PERF_DIR, exist_ok=True) self._snapshots = self._load_json(SNAPSHOTS_FILE, []) self._trades = self._load_json(TRADES_FILE, []) def _load_json(self, path, default): if os.path.exists(path): try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: print(f"[PerformanceDB] Load failed {path}: {e}") return default return default def _save_json(self, path, data): try: with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) except Exception as e: print(f"[PerformanceDB] Save failed {path}: {e}") # ───────────────────────────────────────── # 일별 스냅샷 # ───────────────────────────────────────── def save_daily_snapshot(self, total_eval, deposit, holdings_count, benchmark_close=None): """일별 자산 스냅샷 저장 (하루 1회 호출 권장). Args: total_eval (int): 총 평가액 (원) deposit (int): 예수금 (원) holdings_count (int): 보유 종목 수 benchmark_close (float|None): KOSPI 현재가 (벤치마크 비교용) """ today = datetime.now().strftime("%Y-%m-%d") # 오늘 이미 저장된 스냅샷이 있으면 업데이트 for snap in self._snapshots: if snap.get("date") == today: snap["total_eval"] = total_eval snap["deposit"] = deposit snap["holdings_count"] = holdings_count if benchmark_close is not None: snap["benchmark_kospi_close"] = benchmark_close self._save_json(SNAPSHOTS_FILE, self._snapshots) return # 일별/누적 수익률 계산 daily_return_pct = 0.0 cumulative_return_pct = 0.0 if self._snapshots: prev_eval = self._snapshots[-1].get("total_eval", 0) if prev_eval > 0: daily_return_pct = (total_eval - prev_eval) / prev_eval * 100 initial_capital = self.get_initial_capital() if initial_capital and initial_capital > 0: cumulative_return_pct = (total_eval - initial_capital) / initial_capital * 100 snap = { "date": today, "total_eval": total_eval, "deposit": deposit, "holdings_count": holdings_count, "benchmark_kospi_close": benchmark_close, "daily_return_pct": round(daily_return_pct, 4), "cumulative_return_pct": round(cumulative_return_pct, 4) } self._snapshots.append(snap) self._save_json(SNAPSHOTS_FILE, self._snapshots) print(f"[PerformanceDB] Snapshot saved: {today} " f"total={total_eval:,}원 daily={daily_return_pct:+.2f}%") # ───────────────────────────────────────── # 매매 기록 # ───────────────────────────────────────── def save_trade_record(self, action, ticker, name, qty, price, scores_dict=None, reason="", macro_state="SAFE"): """매수/매도 기록 저장. Args: action (str): "BUY" | "SELL" ticker (str): 종목 코드 name (str): 종목명 qty (int): 수량 price (float): 체결가 scores_dict (dict|None): 분석 점수 딕셔너리 {tech, sentiment, lstm_score, score, ai_confidence, prediction_change} reason (str): 매매 사유 macro_state (str): 매크로 상태 ("SAFE"/"CAUTION"/"DANGER") """ sd = scores_dict or {} now_iso = datetime.now().isoformat() trade = { "id": f"{ticker}_{now_iso}", "action": action, "ticker": ticker, "name": name, "qty": qty, "price": price, "timestamp": now_iso, "reason": reason, "macro_state": macro_state, # 점수 (BUY 시에만 의미 있음) "tech_score": float(sd.get("tech", 0.0)), "sentiment_score": float(sd.get("sentiment", 0.0)), "lstm_score": float(sd.get("lstm_score", 0.0)), "total_score": float(sd.get("score", 0.0)), "ai_confidence": float(sd.get("ai_confidence", 0.5)), "ai_prediction_change": float(sd.get("prediction_change", 0.0)), # 매도 후 채워지는 결과 필드 "outcome_return_pct": None, "holding_days": None, "closed_at": None } self._trades.append(trade) self._save_json(TRADES_FILE, self._trades) def close_trade(self, ticker, sell_price, sell_yield_pct=None): """가장 최근 미체결 BUY를 찾아 매도 결과를 기록. Args: ticker (str): 종목 코드 sell_price (float): 매도 체결가 sell_yield_pct (float|None): KIS에서 받은 수익률 (보조용) """ for trade in reversed(self._trades): if (trade.get("ticker") == ticker and trade.get("action") == "BUY" and trade.get("outcome_return_pct") is None): buy_price = trade.get("price", 0) if buy_price and buy_price > 0: outcome_return_pct = (sell_price - buy_price) / buy_price * 100 elif sell_yield_pct is not None: outcome_return_pct = sell_yield_pct else: outcome_return_pct = 0.0 # 보유일 계산 holding_days = 0 buy_ts = trade.get("timestamp", "") if buy_ts: try: buy_dt = datetime.fromisoformat(buy_ts) holding_days = (datetime.now() - buy_dt).days except Exception: pass trade["outcome_return_pct"] = round(outcome_return_pct, 4) trade["holding_days"] = holding_days trade["closed_at"] = datetime.now().isoformat() self._save_json(TRADES_FILE, self._trades) print(f"[PerformanceDB] Trade closed: {ticker} " f"return={outcome_return_pct:.2f}% holding={holding_days}d") return print(f"[PerformanceDB] No open BUY found for {ticker}") # ───────────────────────────────────────── # 조회 # ───────────────────────────────────────── def load_snapshots(self, days=90): """최근 N일 스냅샷 반환.""" cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") return [s for s in self._snapshots if s.get("date", "") >= cutoff] def load_trades(self, days=90): """최근 N일 매매 기록 반환.""" cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") return [t for t in self._trades if t.get("timestamp", "")[:10] >= cutoff] def get_initial_capital(self): """첫 스냅샷 기준 초기 자본 반환.""" if self._snapshots: return self._snapshots[0].get("total_eval", 0) return 0 def get_summary(self): """간단한 현황 딕셔너리 반환 (디버깅용).""" return { "total_snapshots": len(self._snapshots), "total_trades": len(self._trades), "closed_trades": sum(1 for t in self._trades if t.get("outcome_return_pct") is not None), "initial_capital": self.get_initial_capital() }