移動平均クロス戦略は、短期移動平均が長期移動平均を上抜けたら買い、下抜けたら売るシンプルなトレンドフォロー戦略です。実装は簡単ですが、パラメータ次第で収益性が大きく変わります。ということで、この記事ではバックテストの実装からグリッドサーチによる最適化まで手順をまとめます。
📘 外部参考：Backtesting.py（公式ドキュメント） ／ Backtrader 公式
📘 外部参考：Trend Trading（Investopedia）
📘 外部参考：移動平均（Wikipedia 日本語） ／ Moving Average（Investopedia）
📘 外部参考：Golden Cross（Investopedia） ／ ゴールデンクロス（Wikipedia）
基本的な移動平均クロス戦略
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import itertools
def get_data(ticker, period='5y'):
df = yf.download(ticker, period=period, progress=False)
return df[['Close', 'Volume']].copy()
def moving_average_cross(df, short=20, long=60):
"""移動平均クロス戦略のシグナル生成"""
df = df.copy()
df[f'MA{short}'] = df['Close'].rolling(short).mean()
df[f'MA{long}'] = df['Close'].rolling(long).mean()
# ゴールデンクロス: 1, デッドクロス: -1, 保持: 0
df['position'] = 0
mask_buy = (df[f'MA{short}'] > df[f'MA{long}']) & (df[f'MA{short}'].shift(1) <= df[f'MA{long}'].shift(1))
mask_sell = (df[f'MA{short}'] < df[f'MA{long}']) & (df[f'MA{short}'].shift(1) >= df[f'MA{long}'].shift(1))
df.loc[mask_buy, 'position'] = 1
df.loc[mask_sell, 'position'] = -1
# 保有状態を維持
df['holding'] = df['position'].replace(0, np.nan).ffill().fillna(0).clip(lower=0)
return df.dropna()
df = get_data('7203.T') # トヨタ
result = moving_average_cross(df, short=20, long=60)
print(f"シグナル数 - 買い: {(result['position']==1).sum()}, 売り: {(result['position']==-1).sum()}")バックテストエンジン
def backtest(df, commission=0.001, slippage=0.001):
"""
バックテストエンジン
commission: 手数料率（0.1%）
slippage: スリッページ（0.1%）
"""
df = df.copy()
capital = 1_000_000 # 初期資金100万円
position = 0
entry_price = 0
history = []
for idx, row in df.iterrows():
price = row['Close']
if row['position'] == 1 and position == 0: # 買いシグナル
cost = price * (1 + commission + slippage)
shares = int(capital // cost)
if shares > 0:
capital -= shares * cost
position = shares
entry_price = cost
elif row['position'] == -1 and position > 0: # 売りシグナル
revenue = price * (1 - commission - slippage)
capital += position * revenue
position = 0
total = capital + position * price
history.append(total)
# 最終決済
if position > 0:
capital += position * df['Close'].iloc[-1]
equity = pd.Series(history, index=df.index)
# パフォーマンス指標
total_return = (capital - 1_000_000) / 1_000_000
daily_returns = equity.pct_change().dropna()
sharpe = daily_returns.mean() / daily_returns.std() * np.sqrt(252)
rolling_max = equity.cummax()
drawdown = (equity - rolling_max) / rolling_max
max_dd = drawdown.min()
buy_hold = (df['Close'].iloc[-1] - df['Close'].iloc[0]) / df['Close'].iloc[0]
return {
'total_return': total_return,
'sharpe_ratio': sharpe,
'max_drawdown': max_dd,
'buy_hold_return': buy_hold,
'equity': equity
}
result_df = moving_average_cross(df, 20, 60)
perf = backtest(result_df)
print(f"総リターン: {perf['total_return']:.2%}")
print(f"シャープレシオ: {perf['sharpe_ratio']:.2f}")
print(f"最大ドローダウン: {perf['max_drawdown']:.2%}")
print(f"バイ&ホールド: {perf['buy_hold_return']:.2%}")グリッドサーチによるパラメータ最適化
def grid_search_ma(df, short_range=range(5, 30, 5), long_range=range(20, 120, 10)):
"""移動平均期間のグリッドサーチ"""
results = []
for short, long in itertools.product(short_range, long_range):
if short >= long:
continue
try:
result_df = moving_average_cross(df, short, long)
if len(result_df) < long:
continue
perf = backtest(result_df)
results.append({
'short': short,
'long': long,
'total_return': perf['total_return'],
'sharpe_ratio': perf['sharpe_ratio'],
'max_drawdown': perf['max_drawdown']
})
except Exception:
continue
results_df = pd.DataFrame(results)
# シャープレシオで最適化
best = results_df.nlargest(5, 'sharpe_ratio')
print("シャープレシオ上位5パラメータ:")
print(best[['short', 'long', 'total_return', 'sharpe_ratio', 'max_drawdown']].to_string())
return results_df
results = grid_search_ma(df)
# ヒートマップ描画
pivot = results.pivot(index='short', columns='long', values='sharpe_ratio')
plt.figure(figsize=(12, 6))
plt.imshow(pivot, aspect='auto', cmap='RdYlGn')
plt.colorbar(label='Sharpe Ratio')
plt.title('移動平均クロス パラメータ最適化ヒートマップ')
plt.xlabel('長期MA期間')
plt.ylabel('短期MA期間')
plt.savefig('ma_heatmap.png', dpi=100, bbox_inches='tight')
print("ヒートマップを保存しました")過最適化を避けるウォークフォワードテスト
📘 外部参考：Walk-Forward Optimization（Investopedia）
def walk_forward_test(df, train_years=2, test_years=1):
"""
ウォークフォワードテスト
訓練期間でパラメータを最適化し、テスト期間で評価
"""
df = df.copy()
df.index = pd.to_datetime(df.index)
results = []
start = df.index[0]
end = df.index[-1]
current = start + pd.DateOffset(years=train_years)
while current + pd.DateOffset(years=test_years) <= end:
train = df[df.index < current]
test = df[(df.index >= current) & (df.index < current + pd.DateOffset(years=test_years))]
# 訓練期間で最適パラメータを探索
best_sharpe = -np.inf
best_params = (20, 60)
for short, long in [(5,20),(10,30),(20,60),(5,50),(15,45)]:
try:
r = moving_average_cross(train, short, long)
p = backtest(r)
if p['sharpe_ratio'] > best_sharpe:
best_sharpe = p['sharpe_ratio']
best_params = (short, long)
except Exception:
continue
# テスト期間で評価
test_result = moving_average_cross(test, *best_params)
test_perf = backtest(test_result)
results.append({
'period': f"{current.year}",
'params': best_params,
'test_return': test_perf['total_return'],
'test_sharpe': test_perf['sharpe_ratio']
})
current += pd.DateOffset(years=test_years)
result_df = pd.DataFrame(results)
print(result_df.to_string())
print(f"
平均テストリターン: {result_df['test_return'].mean():.2%}")
return result_df
wf_results = walk_forward_test(df)まとめ
実際に試してみると、グリッドサーチで見つけた最適パラメータでも、別期間に当てると性能が落ちることが多いです。ウォークフォワードテストで「本当に汎化しているか」を確認してから使うのが現実的だと思っています。手数料・スリッページを無視すると実運用でかなりズレるので、そこも含めてバックテストするといいかと思います。
