""" app/ai/evening.py 장 후 데이터 준비 스크립트 (claude_evening 헬퍼) Claude Code headless가 이 스크립트를 실행해 당일 매매 결과와 최근 30일 통계를 수집한 뒤, 직접 분석해 리포트를 작성한다. 실행: python app/ai/evening.py # 요약 출력 python app/ai/evening.py --print # Claude 분석용 전체 데이터 JSON 출력 """ import json import math import os import sqlite3 import sys from datetime import datetime, timedelta from pathlib import Path def _load_env(): env_path = Path(".env") if not env_path.exists(): return with open(env_path, encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") k = k.strip() v = v.strip() if " #" in v: v = v[: v.index(" #")] v = v.strip().strip('"').strip("'") if k and v and k not in os.environ: os.environ[k] = v _load_env() import sys ROOT = Path(__file__).parent.parent.parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) DB_PATH = os.getenv("DB_PATH", "data/stockbot.db") TODAY = datetime.now().strftime("%Y-%m-%d") def _conn(): return sqlite3.connect(DB_PATH) # ── 오늘 매매 내역 ──────────────────────────────────────────────────────────── def get_today_trades() -> list[dict]: if not Path(DB_PATH).exists(): return [] with _conn() as c: rows = c.execute( "SELECT ticker, name, entry_time, exit_time, entry_price, exit_price, " "quantity, exit_reason, pnl, fee, ai_boosted " "FROM trades WHERE date=? ORDER BY entry_time", (TODAY,), ).fetchall() cols = ["ticker", "name", "entry_time", "exit_time", "entry_price", "exit_price", "quantity", "exit_reason", "pnl", "fee", "ai_boosted"] return [dict(zip(cols, r)) for r in rows] # ── 최근 N일 일별 요약 ──────────────────────────────────────────────────────── def get_daily_summaries(days: int = 30) -> list[dict]: if not Path(DB_PATH).exists(): return [] since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") with _conn() as c: rows = c.execute( "SELECT date, total_trades, win_trades, lose_trades, " "net_pnl, max_drawdown, trading_stopped " "FROM daily_summary WHERE date >= ? ORDER BY date", (since,), ).fetchall() cols = ["date", "total_trades", "win_trades", "lose_trades", "net_pnl", "max_drawdown", "trading_stopped"] return [dict(zip(cols, r)) for r in rows] # ── 실전 전환 조건 계산 ────────────────────────────────────────────────────── def calc_live_ready(summaries: list[dict]) -> dict: """5가지 실전 전환 조건 계산""" trading_days = len(summaries) if trading_days == 0: return { "trading_days": 0, "win_rate": 0, "mdd_pct": 0, "sharpe": 0, "l3_count": 0, "all_pass": False, "details": [] } total_trades = sum(s["total_trades"] for s in summaries) win_trades = sum(s["win_trades"] for s in summaries) win_rate = win_trades / total_trades * 100 if total_trades else 0 # MDD: 누적 손익 기준 최대 낙폭 cum = 0.0 peak = 0.0 mdd = 0.0 for s in summaries: cum += s["net_pnl"] or 0 if cum > peak: peak = cum drawdown = (peak - cum) / max(abs(peak), 1) * 100 if drawdown > mdd: mdd = drawdown # Sharpe: 일별 손익의 mean/std (연환산) daily_pnl = [s["net_pnl"] or 0 for s in summaries] if len(daily_pnl) >= 2: mean = sum(daily_pnl) / len(daily_pnl) variance = sum((x - mean) ** 2 for x in daily_pnl) / len(daily_pnl) std = math.sqrt(variance) if variance > 0 else 1 sharpe = (mean / std) * math.sqrt(252) else: sharpe = 0.0 # L3 발동 횟수 (trading_stopped=1인 날) l3_count = sum(1 for s in summaries if s.get("trading_stopped")) checks = [ ("누적 운영 30거래일", trading_days >= 30, f"{trading_days}일"), ("승률 > 48%", win_rate > 48, f"{win_rate:.1f}%"), ("MDD < -10%", mdd < 10, f"-{mdd:.1f}%"), ("샤프 > 1.0", sharpe > 1.0, f"{sharpe:.2f}"), ("L3 월 2회 이하", l3_count <= 2, f"{l3_count}회"), ] return { "trading_days": trading_days, "win_rate": round(win_rate, 1), "mdd_pct": round(-mdd, 1), "sharpe": round(sharpe, 2), "l3_count": l3_count, "all_pass": all(c[1] for c in checks), "details": [{"name": c[0], "pass": c[1], "value": c[2]} for c in checks], } # ── 최근 로그 ───────────────────────────────────────────────────────────────── def get_today_log_tail(lines: int = 50) -> str: log_path = Path("logs/stockbot.log") if not log_path.exists(): return "" all_lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines() today_lines = [l for l in all_lines if TODAY in l] return "\n".join(today_lines[-lines:]) # ── 최근 30일 리포트 목록 ───────────────────────────────────────────────────── def get_recent_reports(days: int = 5) -> list[str]: reports_dir = Path("reports/daily") if not reports_dir.exists(): return [] files = sorted(reports_dir.glob("*.md"), reverse=True)[:days] return [f.name for f in files] # ── 메인 ───────────────────────────────────────────────────────────────────── def main(print_mode: bool = False): trades = get_today_trades() summaries = get_daily_summaries(30) live = calc_live_ready(summaries) log_tail = get_today_log_tail(30) reports = get_recent_reports(5) # 오늘 요약 통계 closed = [t for t in trades if t["exit_time"]] wins = [t for t in closed if (t["pnl"] or 0) > 0] losses = [t for t in closed if (t["pnl"] or 0) <= 0] net_pnl = sum(t["pnl"] or 0 for t in closed) summary = { "date": TODAY, "today": { "total_trades": len(closed), "wins": len(wins), "losses": len(losses), "win_rate": round(len(wins) / len(closed) * 100, 1) if closed else 0, "net_pnl": round(net_pnl), "trades": trades, }, "last_30_days": { "trading_days": live["trading_days"], "win_rate": live["win_rate"], "mdd_pct": live["mdd_pct"], "sharpe": live["sharpe"], "l3_count": live["l3_count"], "daily_summaries": summaries[-10:], # 최근 10일만 (토큰 절약) }, "live_ready": live, "recent_reports": reports, "log_tail": log_tail, } if print_mode: print(json.dumps(summary, ensure_ascii=False, indent=2)) else: # 콘솔 요약 출력 print(f"\n[{TODAY}] 장 후 결산") print(f" 매매: {len(closed)}회 / 승 {len(wins)} 패 {len(losses)} " f"({summary['today']['win_rate']}%) / {net_pnl:+,.0f}원") print(f" 실전 전환: {'✅ ALL PASS' if live['all_pass'] else '❌ 미충족'}") for d in live["details"]: mark = "✅" if d["pass"] else "❌" print(f" {mark} {d['name']}: {d['value']}") if __name__ == "__main__": main(print_mode="--print" in sys.argv)