아무 종목이나 돌렸더니 498번 매매, 거래대금·ATR로 거르는 법

일 거래대금 = 종가 × 거래량
20일 중앙 거래대금 = 직전 20거래일 일 거래대금의 중앙값

python -m pip install yfinance pandas numpy matplotlib

from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import yfinance as yf
from matplotlib import font_manager


TICKERS = ["005930.KS", "000660.KS", "005380.KS", "000270.KS", "035420.KS", "035720.KS", "068270.KS", "207940.KS", "105560.KS", "055550.KS", "066570.KS", "012330.KS", "028260.KS", "017670.KS", "009150.KS", "003670.KS", "042700.KS", "000810.KS", "196170.KQ", "247540.KQ", "086520.KQ", "058470.KQ", "035900.KQ", "041510.KQ", "067160.KQ", "263750.KQ", "293490.KQ", "112040.KQ", "036930.KQ", "240810.KQ"]
START_DATE = "2018-01-01"
END_DATE = "2026-01-01"
TURNOVER_FLOOR = 10_000_000_000
ATR_MIN = 1.5
ATR_MAX = 6.0
ORDER_VALUE = 10_000_000
VOLUME_MULTIPLIER = 1.2
COST_RATE = 0.0005
CHART_FILE = Path("universe_filter_chart.png")
RESULT_FILE = Path("universe_filter_results.csv")


def prepare_price(raw: pd.DataFrame, ticker: str) -> pd.DataFrame:
    data = raw[ticker].rename(columns=str.lower).dropna().copy()
    data["sma_20"] = data["close"].rolling(20).mean()
    data["sma_60"] = data["close"].rolling(60).mean()
    data["sma_120"] = data["close"].rolling(120).mean()
    data["volume_average_20"] = data["volume"].shift(1).rolling(20).mean()
    data["entry_signal"] = (data["sma_60"] > data["sma_120"]) & (data["close"] > data["sma_120"]) & (data["close"] > data["sma_20"]) & (data["close"].shift(1) <= data["sma_20"].shift(1)) & (data["volume"] >= data["volume_average_20"] * VOLUME_MULTIPLIER)
    data["exit_signal"] = (data["close"] < data["sma_20"]) & (data["close"].shift(1) >= data["sma_20"].shift(1))
    previous_close = data["close"].shift(1)
    true_range = pd.concat([data["high"] - data["low"], (data["high"] - previous_close).abs(), (data["low"] - previous_close).abs()], axis=1).max(axis=1)
    data["turnover_20"] = (data["close"] * data["volume"]).shift(1).rolling(20).median()
    data["atr_percent"] = true_range.shift(1).rolling(14).mean() / data["close"].shift(1) * 100.0
    data["eligible"] = (data["turnover_20"] >= TURNOVER_FLOOR) & data["atr_percent"].between(ATR_MIN, ATR_MAX)
    data["next_open_gap"] = (data["open"].shift(-1) / data["close"] - 1.0).abs()
    return data.dropna(subset=["sma_120", "turnover_20", "atr_percent"])


def run_backtest(data: pd.DataFrame, use_filter: bool) -> tuple[pd.DataFrame, pd.DataFrame]:
    cash = 1.0
    shares = 0.0
    pending_order = None
    current_trade = None
    trades = []
    equity_records = []

    for date, row in data.iterrows():
        if pending_order is not None:
            if pending_order["side"] == "buy" and shares == 0.0:
                capital_before = cash
                entry_price = float(row["open"])
                shares = cash * (1.0 - COST_RATE) / entry_price
                cash = 0.0
                current_trade = {"entry_signal_date": pending_order["signal_date"], "entry_date": date, "entry_price": entry_price, "capital_before": capital_before}
            elif pending_order["side"] == "sell" and shares > 0.0:
                exit_price = float(row["open"])
                cash = shares * exit_price * (1.0 - COST_RATE)
                shares = 0.0
                trades.append({**current_trade, "exit_signal_date": pending_order["signal_date"], "exit_date": date, "exit_price": exit_price, "trade_return": cash / current_trade["capital_before"] - 1.0, "forced_exit": False})
                current_trade = None
            pending_order = None

        if shares > 0.0 and bool(row["exit_signal"]):
            pending_order = {"side": "sell", "signal_date": date}
        elif shares == 0.0 and bool(row["entry_signal"]) and (bool(row["eligible"]) or not use_filter):
            pending_order = {"side": "buy", "signal_date": date}

        equity_records.append({"date": date, "equity": cash + shares * float(row["close"])})

    if shares > 0.0:
        last_date = data.index[-1]
        last_close = float(data.iloc[-1]["close"])
        cash = shares * last_close * (1.0 - COST_RATE)
        trades.append({**current_trade, "exit_signal_date": last_date, "exit_date": last_date, "exit_price": last_close, "trade_return": cash / current_trade["capital_before"] - 1.0, "forced_exit": True})
        equity_records[-1]["equity"] = cash

    return pd.DataFrame(trades), pd.DataFrame(equity_records).set_index("date")


def build_portfolio(price_by_ticker: dict[str, pd.DataFrame], calendar: pd.DatetimeIndex, use_filter: bool) -> tuple[pd.DataFrame, pd.DataFrame]:
    equity_series = []
    trade_frames = []
    for ticker, data in price_by_ticker.items():
        trades, equity = run_backtest(data, use_filter)
        equity_series.append(equity["equity"].reindex(calendar).ffill().fillna(1.0).rename(ticker))
        if not trades.empty:
            trades.insert(0, "ticker", ticker)
            trade_frames.append(trades)
    portfolio = pd.concat(equity_series, axis=1).mean(axis=1).to_frame("equity")
    combined_trades = pd.concat(trade_frames, ignore_index=True)
    return combined_trades, portfolio


def summarize(trades: pd.DataFrame, equity: pd.DataFrame) -> dict[str, float]:
    elapsed_years = (equity.index[-1] - equity.index[0]).days / 365.25
    total_return = float(equity["equity"].iloc[-1] - 1.0)
    cagr = float(equity["equity"].iloc[-1] ** (1.0 / elapsed_years) - 1.0)
    drawdown = equity["equity"] / equity["equity"].cummax() - 1.0
    return {"누적수익률": total_return, "CAGR": cagr, "MDD": float(drawdown.min()), "승률": float((trades["trade_return"] > 0.0).mean()), "평균거래수익률": float(trades["trade_return"].mean()), "거래수": float(len(trades))}


available_fonts = {item.name for item in font_manager.fontManager.ttflist}
for font_name in ("Malgun Gothic", "AppleGothic", "UnDotum", "DejaVu Sans"):
    if font_name in available_fonts:
        plt.rcParams["font.family"] = font_name
        break
plt.rcParams["axes.unicode_minus"] = False

raw = yf.download(TICKERS, start=START_DATE, end=END_DATE, auto_adjust=True, progress=False, group_by="ticker", threads=False)
if raw.empty:
    raise RuntimeError("가격 데이터를 받지 못했습니다.")

price_by_ticker = {ticker: prepare_price(raw, ticker) for ticker in TICKERS}
candidate_frames = []
for ticker, data in price_by_ticker.items():
    candidate = data.loc[data["entry_signal"], ["turnover_20", "atr_percent", "next_open_gap"]].copy()
    candidate["ticker"] = ticker
    candidate_frames.append(candidate)
candidate_signals = pd.concat(candidate_frames)
candidate_signals["group"] = np.select([candidate_signals["turnover_20"] < TURNOVER_FLOOR, candidate_signals["atr_percent"] < ATR_MIN, candidate_signals["atr_percent"] > ATR_MAX], ["거래대금 미달", "변동성 부족", "변동성 과다"], default="필터 통과")

unfiltered_trades, unfiltered_equity = build_portfolio(price_by_ticker, raw.index, False)
filtered_trades, filtered_equity = build_portfolio(price_by_ticker, raw.index, True)
unfiltered_summary = summarize(unfiltered_trades, unfiltered_equity)
filtered_summary = summarize(filtered_trades, filtered_equity)
summary = pd.DataFrame([unfiltered_summary, filtered_summary], index=["필터 없음", "거래대금·ATR 필터"])
summary["거래수"] = summary["거래수"].astype(int)
summary.to_csv(RESULT_FILE, encoding="utf-8-sig")

group_order = ["필터 통과", "거래대금 미달", "변동성 과다", "변동성 부족"]
group_summary = candidate_signals.groupby("group").agg(신호수=("next_open_gap", "size"), 중앙거래대금=("turnover_20", "median"), 중앙ATR=("atr_percent", "median"), 중앙시가괴리=("next_open_gap", "median"), 상위10퍼센트시가괴리=("next_open_gap", lambda values: values.quantile(0.9))).reindex(group_order)
pass_participation = ORDER_VALUE / candidate_signals.loc[candidate_signals["group"] == "필터 통과", "turnover_20"].median()
low_turnover_participation = ORDER_VALUE / candidate_signals.loc[candidate_signals["group"] == "거래대금 미달", "turnover_20"].median()

print(summary.to_string(formatters={"누적수익률": "{:.2%}".format, "CAGR": "{:.2%}".format, "MDD": "{:.2%}".format, "승률": "{:.2%}".format, "평균거래수익률": "{:.2%}".format}))
print(group_summary.to_string(formatters={"중앙거래대금": lambda value: f"{value / 100_000_000:.1f}억", "중앙ATR": "{:.2f}%".format, "중앙시가괴리": "{:.2%}".format, "상위10퍼센트시가괴리": "{:.2%}".format}))
print(f"1,000만 원 주문 참여율 중앙값: 통과 {pass_participation:.3%}, 거래대금 미달 {low_turnover_participation:.3%}")

fig = plt.figure(figsize=(13, 9), facecolor="white")
grid = fig.add_gridspec(2, 2, height_ratios=[2.3, 1.4], hspace=0.35, wspace=0.22)
equity_axis = fig.add_subplot(grid[0, :])
count_axis = fig.add_subplot(grid[1, 0])
participation_axis = fig.add_subplot(grid[1, 1])

equity_axis.plot(unfiltered_equity.index, unfiltered_equity["equity"], color="#F59E0B", linewidth=2.0, label=f"필터 없음 {unfiltered_summary['누적수익률']:.1%}")
equity_axis.plot(filtered_equity.index, filtered_equity["equity"], color="#0F766E", linewidth=2.2, label=f"거래대금·ATR 필터 {filtered_summary['누적수익률']:.1%}")
equity_axis.set_title("같은 전략, 종목 필터 전후 비교", fontsize=18, fontweight="bold", pad=16)
equity_axis.set_ylabel("30종목 동일 비중 누적 자산")
equity_axis.legend(loc="upper left", frameon=False)

bar_colors = ["#0F766E", "#F59E0B", "#DC2626", "#64748B"]
count_bars = count_axis.bar(group_order, group_summary["신호수"], color=bar_colors)
count_axis.set_title("진입 후보가 제외된 이유", fontsize=14, fontweight="bold")
count_axis.set_ylabel("신호 수")
count_axis.tick_params(axis="x", rotation=12)
count_axis.bar_label(count_bars, padding=3)

participation_values = [pass_participation * 100.0, low_turnover_participation * 100.0]
participation_bars = participation_axis.bar(["필터 통과", "거래대금 미달"], participation_values, color=["#0F766E", "#F59E0B"])
participation_axis.set_title("1,000만 원 주문의 거래대금 참여율", fontsize=14, fontweight="bold")
participation_axis.set_ylabel("20일 중앙 거래대금 대비(%)")
participation_axis.bar_label(participation_bars, labels=[f"{value:.3f}%" for value in participation_values], padding=3)

for axis in (equity_axis, count_axis, participation_axis):
    axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
    axis.spines["top"].set_visible(False)
    axis.spines["left"].set_visible(False)
    axis.spines["right"].set_color("#D1D5DB")
    axis.spines["bottom"].set_color("#D1D5DB")
    axis.yaxis.tick_right()
    axis.yaxis.set_label_position("right")

fig.savefig(CHART_FILE, dpi=160, bbox_inches="tight", facecolor="white")
plt.show()

주문 참여율 = 주문 금액 ÷ 직전 20일 중앙 거래대금