231 lines
8.5 KiB
Python
231 lines
8.5 KiB
Python
"""
|
|
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 collections import Counter
|
|
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: 초기자본 대비 누적 낙폭 (%)
|
|
STARTING_CAPITAL = 10_000_000
|
|
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) / STARTING_CAPITAL * 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)
|
|
exit_reason_counts = Counter(t.get("exit_reason") or "UNKNOWN" for t in closed)
|
|
forced_count = exit_reason_counts.get("FORCE", 0)
|
|
|
|
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),
|
|
"exit_reason_counts": dict(exit_reason_counts),
|
|
"force_exit_ratio": round(forced_count / len(closed) * 100, 1) if closed else 0,
|
|
"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)
|