gahusb
7ea1a21487
Atomic mv of root V1 assets (main_server.py + modules/ + data/ + tests/ + entry scripts + docs + logs) into signal_v1/ subdirectory. load_dotenv() updated to load web-ai/.env explicitly via Path. Adds web-ai/CLAUDE.md (workspace guide) and web-ai/start.bat (signal_v1 entry wrapper). Prepares for signal_v2/ Phase 2. Tests: signal_v1/tests/unit baseline preserved (no regression). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
212 lines
8.6 KiB
Python
212 lines
8.6 KiB
Python
"""
|
|
성과 데이터 영구 저장 - 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()
|
|
}
|