feat(backtest): 매집 점수 신호 검증 + 포트폴리오 커버 이미지

.gitignore

@@ -7,6 +7,8 @@ data/
 *.csv
 *.png
 collect.log
+# README 커버 이미지는 추적 (재현: scripts/make_figures.py)
+!docs/images/*.png
 
 # Python
 __pycache__/

README.md

@@ -6,6 +6,26 @@
 
 ---
 
+## 결과 미리보기
+
+**전 종목 스크리닝 → 매집 후보 랭킹**
+
+![매집 후보 상위 종목](docs/images/ranking.png)
+
+**신호 검증 — 매집 점수가 후속 수익률과 정렬되는가**
+
+형성구간(12거래일)에서 점수화한 후보를, 보유구간(이후 거래일)의 실제 수익률로 평가했습니다. 점수 상위 분위(Q1)일수록 평균 수익률이 높고 하위(Q5)는 음(–)으로, 점수가 단조적으로 수익률과 정렬됩니다.
+
+![매집 점수 vs 후속 수익률](docs/images/backtest.png)
+
+**종목 수급 차트 — 횡보 속 외국인·기관 누적 순매수**
+
+![종목별 투자자 수급](docs/images/candidate.png)
+
+> 위 그림은 2026-05-15~06-12 수집분(19거래일)으로 생성한 **in-sample 예시**입니다. 단일 짧은 윈도우라 롤링 아웃오브샘플·거래비용·생존편향을 통제한 정식 백테스트가 아니며, 스크리너가 신호를 담고 있음을 보이는 용도입니다. 재현은 [개발](#개발) 참고.
+
+---
+
 ## 핵심 기능
 
 - **전 종목 수급 수집** — 코스피·코스닥 보통주(~2,600개)의 최근 N일 투자주체별 순매수를 SQLite DB로 적재 (재실행 안전, 이어받기 지원)
@@ -53,7 +73,10 @@ kq-collect --limit 5                           # 동작 확인
 kq-screen --top 30
 kq-screen --max-range 0.10 --csv candidates.csv
 
-# 3) 종목 수급 차트
+# 3) 신호 검증 (형성구간 스크리닝 → 보유구간 수익률)
+kq-backtest --formation-days 12
+
+# 4) 종목 수급 차트
 kq-chart --code 005930
 ```
 
@@ -79,8 +102,11 @@ kq-chart --code 005930
 ## 개발
 
 ```bash
-uv run pytest        # 네트워크 없이 통과 (storage/screener 로직)
+uv run pytest        # 네트워크 없이 통과 (storage/screener/backtest 로직)
 uv run ruff check .
+
+# README 커버 이미지 재생성 (DB 수집 후)
+python scripts/make_figures.py   # → docs/images/{ranking,backtest,candidate}.png
 ```
 
 ## 라이선스

pyproject.toml

@@ -24,6 +24,7 @@ dev = ["pytest>=8.0", "ruff>=0.6", "matplotlib>=3.8"]
 [project.scripts]
 kq-collect = "kr_quant.collectors.supply_demand:main"
 kq-screen = "kr_quant.strategies.accumulation:main"
+kq-backtest = "kr_quant.strategies.backtest:main"
 kq-chart = "kr_quant.viz.supply_demand_chart:main"
 
 [project.urls]

scripts/make_figures.py

@@ -0,0 +1,57 @@
+"""Generate the README cover figures from the collected DB.
+
+Produces three PNGs under ``docs/images/``:
+  * ``ranking.png``   — top accumulation candidates by score
+  * ``backtest.png``  — accumulation score vs forward return (signal validation)
+  * ``candidate.png`` — supply/demand chart of the #1 ranked candidate
+
+Run after ``kq-collect``::
+
+    .venv/bin/python scripts/make_figures.py
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from kr_quant.storage import connect, default_db_path  # noqa: E402
+from kr_quant.strategies.accumulation import load_frame, screen  # noqa: E402
+from kr_quant.strategies.backtest import backtest  # noqa: E402
+from kr_quant.viz.portfolio import plot_ranking, plot_score_vs_return  # noqa: E402
+from kr_quant.viz.supply_demand_chart import build_chart  # noqa: E402
+
+OUT_DIR = Path(__file__).resolve().parents[1] / "docs" / "images"
+
+
+def main() -> int:
+    con = connect(str(default_db_path()))
+    df = load_frame(con)
+
+    result = screen(df, min_days=8, max_range_pct=0.15)
+    if result.empty:
+        print("후보 없음 — 먼저 kq-collect 로 데이터를 수집하세요.")
+        return 1
+
+    print("→ ranking.png")
+    plot_ranking(result, OUT_DIR / "ranking.png", top=15)
+
+    print("→ backtest.png")
+    merged, summary = backtest(df, formation_days=12, min_days=8, max_range_pct=0.15)
+    plot_score_vs_return(merged, summary, OUT_DIR / "backtest.png")
+    print(f"   Spearman={summary['spearman']:.3f}  n={summary['n']}  "
+          f"전체평균={summary['universe_mean']:+.2%}")
+
+    top_code = result.iloc[0]["code"]
+    print(f"→ candidate.png ({top_code} {result.iloc[0]['name']})")
+    build_chart(con, top_code, OUT_DIR / "candidate.png")
+
+    con.close()
+    print(f"\n완료: {OUT_DIR}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/kr_quant/strategies/backtest.py

@@ -0,0 +1,152 @@
+"""Lightweight validation of the accumulation score against forward returns.
+
+Splits the collected window into a *formation* period (used to screen and score
+candidates) and a *holdout* period (used to measure what happened next), then
+checks whether a higher accumulation score lines up with a higher subsequent
+return.
+
+This is an **in-sample illustration**, not a rigorous backtest: with a single
+short window there is no rolling out-of-sample evaluation, no transaction costs,
+and no survivorship control. It is meant to show the screener carries signal,
+with the rank correlation and per-quintile spread quantifying how much.
+
+The core :func:`backtest` takes a DataFrame so it is unit-testable without a DB.
+:func:`main` wires it to SQLite for the CLI (``kq-backtest``).
+"""
+
+from __future__ import annotations
+
+import argparse
+
+import pandas as pd
+
+from ..storage import connect, default_db_path
+from .accumulation import load_frame, screen
+
+
+def spearman(a: pd.Series, b: pd.Series) -> float:
+    """Spearman rank correlation (Pearson on ranks — no scipy dependency)."""
+    if len(a) < 2:
+        return float("nan")
+    return float(a.rank().corr(b.rank()))
+
+
+def forward_returns(df: pd.DataFrame, base_date: str, eval_date: str) -> pd.Series:
+    """Per-code return from ``base_date``'s close to ``eval_date``'s close.
+
+    Kiwoom stores a signed close (the sign marks the day's direction), so we
+    take the absolute value to recover the price level.
+    """
+    piv = df.pivot_table(index="code", columns="date", values="close", aggfunc="first").abs()
+    return (piv[eval_date] / piv[base_date] - 1.0).rename("fwd_ret")
+
+
+def backtest(
+    df: pd.DataFrame,
+    *,
+    formation_days: int = 12,
+    quantiles: int = 5,
+    **screen_kwargs: object,
+) -> tuple[pd.DataFrame, dict]:
+    """Score candidates on the formation window, score forward returns on the rest.
+
+    Args:
+        df: Supply/demand rows joined with the stock master (see
+            :func:`kr_quant.strategies.accumulation.load_frame`).
+        formation_days: Number of leading trading days used to screen/score.
+            The remaining days are the holdout used to measure forward return.
+        quantiles: Number of score buckets for the per-quintile summary.
+        **screen_kwargs: Forwarded to :func:`screen` (e.g. ``max_range_pct``).
+
+    Returns:
+        ``(merged, summary)`` where ``merged`` is the candidate table plus a
+        ``fwd_ret`` column (sorted by score, descending), and ``summary`` holds
+        ``n``, ``spearman``, ``universe_mean`` and a ``buckets`` DataFrame of
+        mean forward return per score quantile.
+    """
+    dates = sorted(df["date"].unique())
+    if len(dates) < formation_days + 2:
+        raise ValueError(
+            f"백테스트에 거래일이 부족합니다: {len(dates)}일 (형성 {formation_days}일 + 보유 ≥2일 필요)"
+        )
+
+    form_dates = dates[:formation_days]
+    base_date, eval_date = form_dates[-1], dates[-1]
+
+    candidates = screen(df[df["date"].isin(form_dates)], **screen_kwargs)  # type: ignore[arg-type]
+    fwd = forward_returns(df, base_date, eval_date)
+    merged = (
+        candidates.merge(fwd, left_on="code", right_index=True, how="inner")
+        .dropna(subset=["fwd_ret"])
+        .sort_values("score", ascending=False)
+        .reset_index(drop=True)
+    )
+
+    buckets = _quantile_summary(merged, quantiles)
+    summary = {
+        "formation_days": formation_days,
+        "base_date": base_date,
+        "eval_date": eval_date,
+        "n": len(merged),
+        "spearman": spearman(merged["score"], merged["fwd_ret"]) if not merged.empty else float("nan"),
+        "universe_mean": float(fwd.dropna().mean()),
+        "buckets": buckets,
+    }
+    return merged, summary
+
+
+def _quantile_summary(merged: pd.DataFrame, quantiles: int) -> pd.DataFrame:
+    """Mean forward return + hit rate per score quantile (Q1 = highest score)."""
+    cols = ["quantile", "n", "mean_fwd", "hit_rate"]
+    if len(merged) < quantiles:
+        return pd.DataFrame(columns=cols)
+    # rank=False so Q1 is the top score bucket; labels 1..quantiles.
+    q = pd.qcut(merged["score"].rank(method="first", ascending=False), quantiles, labels=False) + 1
+    out = (
+        merged.assign(_q=q)
+        .groupby("_q")
+        .agg(n=("fwd_ret", "size"), mean_fwd=("fwd_ret", "mean"), hit_rate=("fwd_ret", lambda s: (s > 0).mean()))
+        .reset_index()
+        .rename(columns={"_q": "quantile"})
+    )
+    return out[cols]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="매집 점수 vs 후속 수익률 검증 (형성구간 스크리닝 → 보유구간 수익률)"
+    )
+    parser.add_argument("--db", default=str(default_db_path()))
+    parser.add_argument("--formation-days", type=int, default=12,
+                        help="형성구간(스크리닝) 거래일 수 — 나머지는 보유구간")
+    parser.add_argument("--max-range", type=float, default=0.15)
+    parser.add_argument("--min-days", type=int, default=8)
+    parser.add_argument("--quantiles", type=int, default=5)
+    args = parser.parse_args()
+
+    con = connect(args.db)
+    df = load_frame(con)
+    con.close()
+
+    merged, summary = backtest(
+        df,
+        formation_days=args.formation_days,
+        quantiles=args.quantiles,
+        min_days=args.min_days,
+        max_range_pct=args.max_range,
+    )
+    print(f"형성구간: {df['date'].min()}..{summary['base_date']} ({args.formation_days}일)  "
+          f"보유구간: {summary['base_date']}..{summary['eval_date']}")
+    print(f"후보 {summary['n']}개 | Spearman(점수~수익률) = {summary['spearman']:.3f} | "
+          f"전체 평균 수익률 {summary['universe_mean']:+.2%}")
+    if not summary["buckets"].empty:
+        b = summary["buckets"].copy()
+        b["mean_fwd"] = b["mean_fwd"].map("{:+.2%}".format)
+        b["hit_rate"] = b["hit_rate"].map("{:.0%}".format)
+        print("\n점수 분위별 (Q1=최고점):")
+        print(b.to_string(index=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/kr_quant/viz/portfolio.py

@@ -0,0 +1,95 @@
+"""Portfolio-cover figures: screener ranking and accumulation-score validation.
+
+These render the headline visuals shown in the README. Functions take plain
+DataFrames (from the screener / backtest) so they stay decoupled from the DB,
+and save PNGs that work headless via the Agg backend. Korean labels render if
+NanumGothic is installed.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import matplotlib
+
+matplotlib.use("Agg")
+import matplotlib.font_manager as fm  # noqa: E402
+import matplotlib.pyplot as plt  # noqa: E402
+import pandas as pd  # noqa: E402
+
+for _fp in (Path.home() / ".local/share/fonts").glob("NanumGothic*.ttf"):
+    fm.fontManager.addfont(str(_fp))
+if any("NanumGothic" in f.name for f in fm.fontManager.ttflist):
+    plt.rcParams["font.family"] = "NanumGothic"
+plt.rcParams["axes.unicode_minus"] = False
+
+_FOREIGN = "#d62728"
+_INST = "#2ca02c"
+
+
+def plot_ranking(result: pd.DataFrame, out_path: str | Path, *, top: int = 15) -> Path:
+    """Horizontal bar chart of the top-N accumulation candidates by score."""
+    top_df = result.head(top).iloc[::-1]  # highest score at the top of the chart
+    labels = [f"{n}\n{c}" for n, c in zip(top_df["name"], top_df["code"])]
+    # market is stored in Korean: '거래소' (KOSPI) / '코스닥' (KOSDAQ).
+    colors = ["#ff7f0e" if "코스닥" in str(m) else "#1f77b4" for m in top_df["market"]]
+
+    fig, ax = plt.subplots(figsize=(11, 7))
+    bars = ax.barh(range(len(top_df)), top_df["score"], color=colors)
+    ax.set_yticks(range(len(top_df)))
+    ax.set_yticklabels(labels, fontsize=8)
+    ax.set_xlabel("매집 점수 (유동성 대비 매집 강도 ÷ 변동범위)")
+    ax.set_title(f"매집 후보 상위 {len(top_df)}종목", fontsize=14, fontweight="bold")
+    ax.grid(True, axis="x", alpha=0.3)
+    for bar, val in zip(bars, top_df["score"]):
+        ax.text(bar.get_width(), bar.get_y() + bar.get_height() / 2,
+                f" {val:.2f}", va="center", fontsize=8)
+    handles = [plt.Rectangle((0, 0), 1, 1, color="#1f77b4"),
+               plt.Rectangle((0, 0), 1, 1, color="#ff7f0e")]
+    ax.legend(handles, ["거래소(KOSPI)", "코스닥(KOSDAQ)"], loc="lower right")
+    fig.tight_layout()
+    return _save(fig, out_path)
+
+
+def plot_score_vs_return(merged: pd.DataFrame, summary: dict, out_path: str | Path) -> Path:
+    """Two-panel validation: score-vs-return scatter + per-quantile mean bars."""
+    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(13, 5.5), gridspec_kw={"width_ratios": [1.4, 1]})
+
+    # Left: scatter of accumulation score vs holdout return.
+    ax1.scatter(merged["score"], merged["fwd_ret"] * 100, s=28, alpha=0.6,
+                color="#1f77b4", edgecolor="white", linewidth=0.4)
+    ax1.axhline(0, color="gray", lw=0.8, ls="--")
+    ax1.axhline(summary["universe_mean"] * 100, color="#d62728", lw=1.0, ls=":",
+                label=f"전체 평균 {summary['universe_mean']:+.1%}")
+    ax1.set_xscale("log")
+    ax1.set_xlabel("매집 점수 (log)")
+    ax1.set_ylabel("보유구간 수익률 (%)")
+    ax1.set_title(f"매집 점수 vs 후속 수익률  (Spearman={summary['spearman']:.2f}, n={summary['n']})",
+                  fontsize=12, fontweight="bold")
+    ax1.grid(True, alpha=0.3)
+    ax1.legend(loc="upper left")
+
+    # Right: mean forward return per score quantile (Q1 = highest score).
+    b = summary["buckets"]
+    qcolors = ["#2ca02c" if v > 0 else "#d62728" for v in b["mean_fwd"]]
+    ax2.bar([f"Q{int(q)}" for q in b["quantile"]], b["mean_fwd"] * 100, color=qcolors)
+    ax2.axhline(summary["universe_mean"] * 100, color="gray", lw=1.0, ls=":")
+    ax2.set_xlabel("점수 분위 (Q1=최고점)")
+    ax2.set_ylabel("평균 수익률 (%)")
+    ax2.set_title("점수 분위별 평균 수익률", fontsize=12, fontweight="bold")
+    ax2.grid(True, axis="y", alpha=0.3)
+    for i, v in enumerate(b["mean_fwd"]):
+        ax2.text(i, v * 100, f"{v:+.1%}", ha="center",
+                 va="bottom" if v >= 0 else "top", fontsize=8)
+
+    fig.suptitle("매집 스크리너 신호 검증 (in-sample 예시)", fontsize=14, fontweight="bold")
+    fig.tight_layout(rect=(0, 0, 1, 0.96))
+    return _save(fig, out_path)
+
+
+def _save(fig: plt.Figure, out_path: str | Path) -> Path:
+    out = Path(out_path)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    fig.savefig(out, dpi=120)
+    plt.close(fig)
+    return out