feat(weekly-report): add trend tracking from previous reports

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
21in7
2026-03-07 00:40:39 +09:00
parent 35177bf345
commit 3b0335f57e
2 changed files with 75 additions and 0 deletions

View File

@@ -11,6 +11,7 @@ import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
import json
import re
import subprocess
from datetime import date, timedelta
@@ -119,3 +120,40 @@ def parse_live_trades(log_path: str, days: int = 7) -> list[dict]:
closed_trades.append(trade)
return closed_trades
# ── 추이 추적 ────────────────────────────────────────────────────
WEEKLY_DIR = Path("results/weekly")
def load_trend(report_dir: str, weeks: int = 4) -> dict:
"""이전 주간 리포트에서 PF/승률/MDD 추이를 로드한다."""
rdir = Path(report_dir)
if not rdir.exists():
return {"pf": [], "win_rate": [], "mdd": [], "pf_declining_3w": False}
reports = sorted(rdir.glob("report_*.json"))
recent = reports[-weeks:] if len(reports) >= weeks else reports
pf_list, wr_list, mdd_list = [], [], []
for rpath in recent:
try:
data = json.loads(rpath.read_text())
s = data["backtest"]["summary"]
pf_list.append(s["profit_factor"])
wr_list.append(s["win_rate"])
mdd_list.append(s["max_drawdown_pct"])
except (json.JSONDecodeError, KeyError):
continue
declining = False
if len(pf_list) >= 3:
last3 = pf_list[-3:]
declining = last3[0] > last3[1] > last3[2]
return {
"pf": pf_list,
"win_rate": wr_list,
"mdd": mdd_list,
"pf_declining_3w": declining,
}

View File

@@ -73,3 +73,40 @@ def test_parse_live_trades_empty_log(tmp_path):
trades = parse_live_trades(str(tmp_path / "nonexistent.log"), days=7)
assert trades == []
import json
from datetime import date, timedelta
def test_load_trend_reads_previous_reports(tmp_path):
"""이전 주간 리포트를 읽어 PF/승률/MDD 추이를 반환."""
from scripts.weekly_report import load_trend
for i, (pf, wr, mdd) in enumerate([
(1.31, 48.0, 9.0), (1.24, 45.0, 11.0),
(1.20, 44.0, 12.0), (1.18, 43.0, 14.0),
]):
d = date(2026, 3, 7) - timedelta(weeks=3 - i)
report = {
"date": d.isoformat(),
"backtest": {"summary": {
"profit_factor": pf, "win_rate": wr, "max_drawdown_pct": mdd,
"total_trades": 20,
}},
}
(tmp_path / f"report_{d.isoformat()}.json").write_text(json.dumps(report))
trend = load_trend(str(tmp_path), weeks=4)
assert len(trend["pf"]) == 4
assert trend["pf"] == [1.31, 1.24, 1.20, 1.18]
assert trend["pf_declining_3w"] is True
def test_load_trend_empty_dir(tmp_path):
"""리포트가 없으면 빈 추이 반환."""
from scripts.weekly_report import load_trend
trend = load_trend(str(tmp_path), weeks=4)
assert trend["pf"] == []
assert trend["pf_declining_3w"] is False