移動平均クロス戦略を 5 銘柄でバックテスト

Date,Code,C
2022-01-04,7203,2110.5
2022-01-05,7203,2128.0
...
2026-04-30,4063,4910.0

1
"""sma_cross_backtest.py
2
短期 SMA と長期 SMA のクロスでロング・キャッシュを切り替えるバックテスト。
3
"""
4
from __future__ import annotations
5

6
import numpy as np
7
import pandas as pd
8
import matplotlib.pyplot as plt
9

10
SHORT_WIN = 25
11
LONG_WIN = 75
12
COST = 0.001            # 片道 0.1%
13
INITIAL_CAPITAL = 1_000_000
14

15

16
def add_signals(df: pd.DataFrame) -> pd.DataFrame:
17
    """Code ごとに SMA とポジションを計算して列を追加する。"""
18
    df = df.sort_values(["Code", "Date"]).copy()
19
    grouped = df.groupby("Code", group_keys=False)
20
    df["sma_short"] = grouped["C"].transform(lambda s: s.rolling(SHORT_WIN).mean())
21
    df["sma_long"]  = grouped["C"].transform(lambda s: s.rolling(LONG_WIN).mean())
22
    # シグナル: 短期 > 長期 を 1、それ以外を 0
23
    df["signal"] = (df["sma_short"] > df["sma_long"]).astype(int)
24
    # 翌営業日からポジションを取る(先読み防止)
25
    df["position"] = grouped["signal"].shift(1)
26
    return df
27

28

29
def backtest_one(df_one: pd.DataFrame) -> pd.DataFrame:
30
    """単一銘柄のバックテスト。資産曲線を返す。"""
31
    df = df_one.dropna(subset=["position"]).copy()
32
    df["ret"] = df["C"].pct_change().fillna(0)
33

34
    # ポジションが切り替わった日にコストを計上
35
    df["pos_change"] = df["position"].diff().abs().fillna(df["position"])
36
    df["cost"] = df["pos_change"] * COST
37

38
    # 戦略のリターン: ポジション * 当日リターン - コスト
39
    df["strategy_ret"] = df["position"] * df["ret"] - df["cost"]
40
    df["equity"] = INITIAL_CAPITAL * (1 + df["strategy_ret"]).cumprod()
41
    return df
42

43

44
def metrics(df: pd.DataFrame) -> dict[str, float]:
45
    """資産曲線から代表的な評価指標を返す。"""
46
    eq = df["equity"]
47
    ret = df["strategy_ret"]
48
    n = len(ret)
49

50
    cum = eq.iloc[-1] / INITIAL_CAPITAL - 1
51
    years = n / 252
52
    cagr = (eq.iloc[-1] / INITIAL_CAPITAL) ** (1 / years) - 1 if years > 0 else np.nan
53
    sharpe = (ret.mean() / ret.std(ddof=0)) * np.sqrt(252) if ret.std(ddof=0) > 0 else np.nan
54
    rolling_max = eq.cummax()
55
    mdd = ((eq - rolling_max) / rolling_max).min()
56

57
    return {
58
        "n_days": n,
59
        "cum_return": cum,
60
        "cagr": cagr,
61
        "sharpe": sharpe,
62
        "max_drawdown": mdd,
63
    }
64

65

66
def main() -> None:
67
    df = pd.read_csv("prices.csv", parse_dates=["Date"])
68
    df = add_signals(df)
69

70
    rows = []
71
    equity_curves = {}
72
    for code, df_one in df.groupby("Code"):
73
        bt = backtest_one(df_one)
74
        rows.append({"Code": code, **metrics(bt)})
75
        equity_curves[code] = bt.set_index("Date")["equity"]
76

77
    summary = pd.DataFrame(rows).round(4)
78
    print(summary)
79
    summary.to_csv("backtest_summary.csv", index=False)
80

81
    # 資産曲線の比較プロット
82
    fig, ax = plt.subplots(figsize=(10, 5))
83
    for code, eq in equity_curves.items():
84
        ax.plot(eq.index, eq, label=code)
85
    ax.set_title(f"SMA({SHORT_WIN})/{LONG_WIN}) cross strategy — equity curves")
86
    ax.set_ylabel("Eq (JPY)")
87
    ax.set_xlabel("Date")
88
    ax.grid(alpha=0.3)
89
    ax.legend()
90
    plt.tight_layout()
91
    plt.savefig("equity_curves.png", dpi=120)
92
    plt.close(fig)
93

94

95
if __name__ == "__main__":
96
    main()

  Code  n_days  cum_return    cagr  sharpe  max_drawdown
0  4063    1004      0.4221  0.0930  0.6125       -0.2310
1    6758    1004      0.5108  0.1112  0.5840       -0.2854
2    7203    1004      0.1823  0.0429  0.3018       -0.1992
3    8306    1004      0.6012  0.1287  0.7233       -0.1644
4    9984    1004     -0.0820 -0.0214 -0.0805       -0.4115

sma_cross_backtest.py の構造はそのままに、シグナルを次のものに変更した
版を作ってください。

新しいシグナル:
- RSI(14) を計算
- RSI が 30 を下から上抜けたとき: ロングエントリー
- RSI が 70 を上から下抜けたとき: ロングを閉じる(キャッシュ)
- それ以外: 直前のポジションを維持

要件:
- ポジションは翌営業日に切り替え(先読み禁止)
- 取引コスト・評価指標・出力フォーマットは既存のまま
- 関数 `compute_rsi(close, window=14)` を別途用意