diff --git a/.gitignore b/.gitignore index 4f34f2c..0cde2f6 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,8 @@ data/ *.csv *.png collect.log +# README 커버 이미지는 추적 (재현: scripts/make_figures.py) +!docs/images/*.png # Python __pycache__/ diff --git a/README.md b/README.md index 4e00813..176910e 100644 --- a/README.md +++ b/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 ``` ## 라이선스 diff --git a/docs/images/backtest.png b/docs/images/backtest.png new file mode 100644 index 0000000..6900e0b Binary files /dev/null and b/docs/images/backtest.png differ diff --git a/docs/images/candidate.png b/docs/images/candidate.png new file mode 100644 index 0000000..8eaeaf7 Binary files /dev/null and b/docs/images/candidate.png differ diff --git a/docs/images/ranking.png b/docs/images/ranking.png new file mode 100644 index 0000000..461c591 Binary files /dev/null and b/docs/images/ranking.png differ diff --git a/pyproject.toml b/pyproject.toml index e3923bd..af0a6e1 100644 --- a/pyproject.toml +++ b/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] diff --git a/scripts/make_figures.py b/scripts/make_figures.py new file mode 100644 index 0000000..0d57c0f --- /dev/null +++ b/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()) diff --git a/src/kr_quant/strategies/backtest.py b/src/kr_quant/strategies/backtest.py new file mode 100644 index 0000000..f203f06 --- /dev/null +++ b/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()) diff --git a/src/kr_quant/viz/portfolio.py b/src/kr_quant/viz/portfolio.py new file mode 100644 index 0000000..30f33d4 --- /dev/null +++ b/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 diff --git a/tests/test_backtest.py b/tests/test_backtest.py new file mode 100644 index 0000000..07aea0a --- /dev/null +++ b/tests/test_backtest.py @@ -0,0 +1,74 @@ +"""Backtest logic: forward returns and score/return validation, no DB needed.""" + +from __future__ import annotations + +import math + +import pandas as pd + +from kr_quant.strategies.backtest import backtest, forward_returns, spearman + + +def _stock_frame(code, closes, foreign, inst, indiv, vol=1_000_000): + n = len(closes) + return pd.DataFrame( + { + "code": [code] * n, + "name": [code] * n, + "market": ["거래소"] * n, + "sector": ["테스트"] * n, + "date": [f"202605{d:02d}" for d in range(1, n + 1)], + "close": closes, + "acc_trde_qty": [vol] * n, + "individual": indiv, + "foreign_": foreign, + "institution": inst, + "penfnd_etc": [0] * n, + "invtrt": [0] * n, + } + ) + + +def test_spearman_perfectly_monotonic(): + a = pd.Series([1, 2, 3, 4]) + b = pd.Series([10, 20, 30, 40]) + assert spearman(a, b) == 1.0 + assert spearman(a, -b) == -1.0 + + +def test_forward_returns_uses_abs_close(): + df = pd.DataFrame( + { + "code": ["A", "A"], + "date": ["20260101", "20260110"], + "close": [-100, 110], # signed close; magnitude is the price level + } + ) + fwd = forward_returns(df, "20260101", "20260110") + assert math.isclose(fwd["A"], 0.10) + + +def test_backtest_splits_formation_and_holdout(): + # 14 days: formation = first 12, holdout = last (sideways accumulation pattern). + closes = [100, 101, 99, 100, 102, 100, 101, 99, 100, 101, 100, 102, 100, 120] + days = len(closes) + df = _stock_frame( + "000001", closes, + foreign=[5000] * days, inst=[3000] * days, indiv=[-8000] * days, + ) + merged, summary = backtest(df, formation_days=12, quantiles=5, min_days=10) + assert summary["base_date"] == "20260512" # 12th day + assert summary["eval_date"] == "20260514" # last day + assert summary["n"] == 1 + # forward return from close 102 (day 12) to 120 (day 14). + assert math.isclose(merged.iloc[0]["fwd_ret"], 120 / 102 - 1) + + +def test_backtest_raises_without_enough_days(): + df = _stock_frame("000001", [100, 101, 102], foreign=[1] * 3, inst=[1] * 3, indiv=[-1] * 3) + try: + backtest(df, formation_days=12) + except ValueError as e: + assert "거래일이 부족" in str(e) + else: + raise AssertionError("expected ValueError for insufficient days")