diff --git a/scripts/weekly_report.py b/scripts/weekly_report.py index b05e0ba..108f6c2 100644 --- a/scripts/weekly_report.py +++ b/scripts/weekly_report.py @@ -11,6 +11,7 @@ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) +import re import subprocess from datetime import date, timedelta @@ -67,3 +68,54 @@ def run_backtest( ) wf = WalkForwardBacktester(cfg) return wf.run() + + +# ── 로그 파싱 패턴 ──────────────────────────────────────────────── +_RE_ENTRY = re.compile( + r"\[(\w+)\]\s+(LONG|SHORT)\s+진입:\s+가격=([\d.]+),\s+수량=([\d.]+),\s+SL=([\d.]+),\s+TP=([\d.]+)" +) +_RE_CLOSE = re.compile( + r"\[(\w+)\]\s+청산 감지\((\w+)\):\s+exit=([\d.]+),\s+rp=([\d.-]+),\s+commission=([\d.]+),\s+net_pnl=([\d.-]+)" +) +_RE_TIMESTAMP = re.compile(r"^(\d{4}-\d{2}-\d{2})\s") + + +def parse_live_trades(log_path: str, days: int = 7) -> list[dict]: + """봇 로그에서 최근 N일간의 진입/청산 기록을 파싱한다.""" + path = Path(log_path) + if not path.exists(): + return [] + + cutoff = (date.today() - timedelta(days=days)).isoformat() + open_trades: dict[str, dict] = {} + closed_trades: list[dict] = [] + + for line in path.read_text().splitlines(): + m_ts = _RE_TIMESTAMP.match(line) + if m_ts and m_ts.group(1) < cutoff: + continue + + m = _RE_ENTRY.search(line) + if m: + sym, side, price, qty, sl, tp = m.groups() + open_trades[sym] = { + "symbol": sym, "side": side, + "entry_price": float(price), "quantity": float(qty), + "sl": float(sl), "tp": float(tp), + "entry_time": m_ts.group(1) if m_ts else "", + } + continue + + m = _RE_CLOSE.search(line) + if m: + sym, reason, exit_price, rp, commission, net_pnl = m.groups() + trade = open_trades.pop(sym, {"symbol": sym, "side": "UNKNOWN"}) + trade.update({ + "close_reason": reason, "exit_price": float(exit_price), + "expected_pnl": float(rp), "commission": float(commission), + "net_pnl": float(net_pnl), + "exit_time": m_ts.group(1) if m_ts else "", + }) + closed_trades.append(trade) + + return closed_trades diff --git a/tests/test_weekly_report.py b/tests/test_weekly_report.py index bb4693c..429bbfb 100644 --- a/tests/test_weekly_report.py +++ b/tests/test_weekly_report.py @@ -46,3 +46,30 @@ def test_run_backtest_returns_summary(): assert result["summary"]["profit_factor"] == 1.57 assert result["summary"]["total_trades"] == 27 + + +def test_parse_live_trades_extracts_entries(tmp_path): + """봇 로그에서 진입/청산 패턴을 파싱하여 트레이드 리스트를 반환.""" + from scripts.weekly_report import parse_live_trades + + log_content = """2026-03-01 10:00:00.000 | INFO | src.bot:process_candle:42 - [XRPUSDT] LONG 진입: 가격=2.5000, 수량=100.0, SL=2.4000, TP=2.7000 +2026-03-01 10:15:00.000 | INFO | src.bot:process_candle:42 - [XRPUSDT] 신호: HOLD | 현재가: 2.5500 USDT +2026-03-01 12:00:00.000 | INFO | src.user_data_stream:_handle_order:80 - [XRPUSDT] 청산 감지(TAKE_PROFIT): exit=2.7000, rp=20.0000, commission=0.2160, net_pnl=19.5680 +""" + log_file = tmp_path / "bot.log" + log_file.write_text(log_content) + + trades = parse_live_trades(str(log_file), days=7) + assert len(trades) == 1 + assert trades[0]["symbol"] == "XRPUSDT" + assert trades[0]["side"] == "LONG" + assert trades[0]["net_pnl"] == pytest.approx(19.568) + assert trades[0]["close_reason"] == "TAKE_PROFIT" + + +def test_parse_live_trades_empty_log(tmp_path): + """로그 파일이 없으면 빈 리스트 반환.""" + from scripts.weekly_report import parse_live_trades + + trades = parse_live_trades(str(tmp_path / "nonexistent.log"), days=7) + assert trades == []