※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。
「Pythonで株のアルゴリズム取引システムを自作したい」という願いを持つ個人投資家は多いですが、実現方法が分からず、立ち往生しているケースがほとんどです。本シリーズを通じて提供してきた知識を、いよいよ「完全な実装版」として統合します。
本記事は、本シリーズの「総集編」であり「完全版」です。環境構築から実運用まで、すべての手順をコード付きで解説し、あなたが「今日から始められる」状態に到達させます。
目次
自作アルゴリズム取引システムの全体像
まず、あなたが構築するシステムの全体像を示します。
【構造図】
┌─────────────────────────────────────────────────────────┐
│ Windows PC / Mac / Linux │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Python スクリプト(自動実行) │ │
│ ├─────────────────────────────────────────────────┤ │
│ │ │ │
│ │ ┌────────────────────────────────────────────┐ │ │
│ │ │ ステップ1:yfinanceでデータ取得(自動) │ │ │
│ │ └────────────────────────────────────────────┘ │ │
│ │ ↓ │ │
│ │ ┌────────────────────────────────────────────┐ │ │
│ │ │ ステップ2:テクニカル指標計算(自動) │ │ │
│ │ └────────────────────────────────────────────┘ │ │
│ │ ↓ │ │
│ │ ┌────────────────────────────────────────────┐ │ │
│ │ │ ステップ3:シグナル生成(自動) │ │ │
│ │ └────────────────────────────────────────────┘ │ │
│ │ ↓ │ │
│ │ ┌────────────────────────────────────────────┐ │ │
│ │ │ ステップ4:Discord通知(自動) │ │ │
│ │ └────────────────────────────────────────────┘ │ │
│ │ ↓ │ │
│ │ ┌────────────────────────────────────────────┐ │ │
│ │ │ ステップ5:パフォーマンス追跡(自動) │ │ │
│ │ └────────────────────────────────────────────┘ │ │
│ │ ↓ │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Windows タスクスケジューラ(毎日自動実行) │ │
│ └─────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
↓ ↓ ↓
Discord通知 SBI証券 CSV ログ
(通知) (手動) (記録)フェーズ0:事前準備(1日)
必要な物の確認
□ Windows PC / Mac / Linux のいずれか
□ インターネット接続
□ テキストエディタ(VS Code 推奨、無料)
□ SBI証券の口座(既に持っていること前提)
□ LINE アカウント
□ 30分~1時間の時間ファイル構成
stock-algorithm/
├── venv/ # 仮想環境(後で作成)
├── data/ # 取得したデータ
│ └── backtest_results.csv
├── logs/ # ログファイル
│ └── trading.log
├── scripts/
│ ├── step1_data_fetch.py
│ ├── step2_backtest.py
│ ├── step3_signal_gen.py
│ ├── step4_live_trading.py
│ └── main.py # 統合スクリプト
├── requirements.txt # 依存ライブラリ
└── config.py # 設定ファイルフェーズ1:環境構築(1~2時間)
ステップ1-1. Python のインストール
Windows の場合:
# 1. https://www.python.org/downloads/ から最新版をダウンロード
# 2. インストーラーを実行
# 3. 「Add Python to PATH」にチェック
# 4. 「Install Now」をクリック
# 5. インストール完了後、コマンドプロンプトで確認
python --version
# Python 3.11.x と表示されたら成功ステップ1-2. 仮想環境の作成
# 作業フォルダに移動
cd stock-algorithm
# 仮想環境を作成
python -m venv venv
# 仮想環境を有効化(Windows)
venv\Scripts\activate
# 仮想環境を有効化(Mac/Linux)
source venv/bin/activate
# 有効になると、コマンドプロンプトに (venv) が表示されるステップ1-3. ライブラリをインストール
# requirements.txt を作成
cat > requirements.txt << 'EOF'
yfinance==0.2.34
pandas==2.1.3
numpy==1.26.2
scikit-learn==1.3.2
tensorflow==2.15.0
requests==2.31.0
schedule==1.2.0
EOF
# インストール実行
pip install -r requirements.txt
# インストール確認
pip listフェーズ2:基本的なデータ取得スクリプト(1時間)
【コピペOK】step1_data_fetch.py
# ==============================
# ステップ1:データ取得スクリプト
# ==============================
# 目的:yfinanceから株価データを取得し、CSV保存
import yfinance as yf
import pandas as pd
from datetime import datetime
import logging
# ロギング設定
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s] %(levelname)s: %(message)s'
)
def fetch_data(symbol, period='1y', output_file='data/stock_data.csv'):
"""
株価データを取得してCSVに保存
Args:
symbol (str): 銘柄コード(例:7203.T)
period (str): データ期間(例:1y = 1年)
output_file (str): 出力ファイルパス
"""
print(f"[情報] {symbol} のデータ取得を開始します")
try:
# yfinanceでデータ取得
ticker = yf.Ticker(symbol)
df = ticker.history(period=period)
if df.empty:
print(f"[警告] {symbol} のデータが取得できませんでした")
return None
# 欠損値を埋める
df = df.fillna(method='ffill')
# CSVで保存
df.to_csv(output_file)
print(f"[成功] {len(df)}営業日分のデータを取得しました")
print(f"[成功] {output_file} に保存しました")
# 統計情報を表示
print(f"\n【統計情報】")
print(f"開始日: {df.index[0].date()}")
print(f"終了日: {df.index[-1].date()}")
print(f"終値の最小値: ¥{df['Close'].min():,.0f}")
print(f"終値の最大値: ¥{df['Close'].max():,.0f}")
print(f"終値の平均値: ¥{df['Close'].mean():,.0f}")
return df
except Exception as e:
print(f"[エラー] データ取得に失敗しました: {e}")
return None
if __name__ == "__main__":
# トヨタのデータを取得
fetch_data("7203.T")実行方法:
python scripts/step1_data_fetch.pyフェーズ3:バックテストスクリプト(1.5時間)
📘 外部参考：Backtesting.py（公式ドキュメント） ／ Backtrader 公式
【コピペOK】step2_backtest.py
# ==============================
# ステップ2:バックテストスクリプト
# ==============================
# 目的:過去データでアルゴリズムを検証
import pandas as pd
import numpy as np
from datetime import datetime
import logging
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s] %(levelname)s: %(message)s'
)
class BacktestEngine:
"""バックテストエンジン"""
def __init__(self, df, initial_capital=1000000, commission_rate=0.001):
"""
Args:
df (pd.DataFrame): 株価データ
initial_capital (float): 初期資金
commission_rate (float): 手数料率
"""
self.df = df.copy()
self.initial_capital = initial_capital
self.commission_rate = commission_rate
self.trades = []
self.portfolio_values = []
def calculate_indicators(self):
"""テクニカル指標を計算"""
print("[進行中] テクニカル指標を計算中...")
# 移動平均線
self.df['MA_SHORT'] = self.df['Close'].rolling(window=20).mean()
self.df['MA_LONG'] = self.df['Close'].rolling(window=50).mean()
# RSI
delta = self.df['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
self.df['RSI'] = 100 - (100 / (1 + rs))
print("[成功] 計算完了")
def generate_signals(self):
"""シグナルを生成"""
print("[進行中] シグナルを生成中...")
signals = pd.Series(0, index=self.df.index)
for i in range(1, len(self.df)):
prev_short = self.df['MA_SHORT'].iloc[i-1]
prev_long = self.df['MA_LONG'].iloc[i-1]
curr_short = self.df['MA_SHORT'].iloc[i]
curr_long = self.df['MA_LONG'].iloc[i]
if pd.isna(prev_short) or pd.isna(curr_short):
continue
# ゴールデンクロス
if prev_short <= prev_long and curr_short > curr_long:
signals.iloc[i] = 1
# デッドクロス
elif prev_short >= prev_long and curr_short < curr_long:
signals.iloc[i] = -1
self.df['Signal'] = signals
print(f"[成功] {len([s for s in signals if s != 0])}個のシグナルを生成")
def run_backtest(self):
"""バックテストを実行"""
print("[進行中] バックテストを実行中...")
cash = self.initial_capital
position = 0
for i in range(len(self.df)):
price = self.df['Close'].iloc[i]
signal = self.df['Signal'].iloc[i]
# 買いシグナル
if signal == 1 and position == 0:
shares = int(cash / price * (1 - self.commission_rate))
if shares > 0:
cost = shares * price * (1 + self.commission_rate)
cash -= cost
position = shares
self.trades.append({
'date': self.df.index[i],
'type': 'BUY',
'price': price,
'shares': shares,
})
# 売りシグナル
elif signal == -1 and position > 0:
revenue = position * price * (1 - self.commission_rate)
profit = revenue - (self.trades[-1]['price'] * position if self.trades else 0)
cash += revenue
self.trades.append({
'date': self.df.index[i],
'type': 'SELL',
'price': price,
'profit': profit,
})
position = 0
portfolio_value = cash + (position * price)
self.portfolio_values.append(portfolio_value)
print(f"[成功] バックテスト完了")
def display_results(self):
"""結果を表示"""
print("\n" + "="*70)
print("バックテスト結果")
print("="*70)
final_cash = self.portfolio_values[-1] if self.portfolio_values else self.initial_capital
total_return = (final_cash - self.initial_capital) / self.initial_capital
print(f"\n初期資金: ¥{self.initial_capital:,.0f}")
print(f"最終資金: ¥{final_cash:,.0f}")
print(f"総利益: ¥{final_cash - self.initial_capital:,.0f}")
print(f"利益率: {total_return*100:+.2f}%")
print(f"\n取引数: {len([t for t in self.trades if t['type'] == 'BUY'])}回")
# 勝ち・負け統計
winning_trades = [t for t in self.trades if t.get('profit', 0) > 0]
losing_trades = [t for t in self.trades if t.get('profit', 0) < 0]
if winning_trades:
print(f"勝ち取引: {len(winning_trades)}回 (平均 ¥{np.mean([t['profit'] for t in winning_trades]):,.0f})")
if losing_trades:
print(f"負け取引: {len(losing_trades)}回 (平均 ¥{np.mean([t['profit'] for t in losing_trades]):,.0f})")
def run_backtest_from_csv(csv_file):
"""CSVファイルからバックテストを実行"""
print("="*70)
print("バックテスト開始")
print("="*70)
print()
# CSVを読み込む
df = pd.read_csv(csv_file, index_col=0, parse_dates=True)
# バックテストエンジンを初期化
engine = BacktestEngine(df)
# 指標を計算
engine.calculate_indicators()
# シグナルを生成
engine.generate_signals()
# バックテストを実行
engine.run_backtest()
# 結果を表示
engine.display_results()
if __name__ == "__main__":
run_backtest_from_csv("data/stock_data.csv")実行方法:
python scripts/step2_backtest.pyフェーズ4:リアルタイム信号生成(1.5時間)
【コピペOK】step3_signal_gen.py
# ==============================
# ステップ3:リアルタイム信号生成スクリプト
# ==============================
# 目的:毎日のシグナル生成とDiscord通知
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime
import requests
import logging
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s] %(levelname)s: %(message)s'
)
class SignalGenerator:
"""リアルタイム信号生成"""
def __init__(self, symbols, discord_webhook_url):
"""
Args:
symbols (list): 銘柄コードのリスト
discord_webhook_url (str): Discord Webhook トークン
"""
self.symbols = symbols
self.discord_webhook_url = discord_webhook_url
self.signals = []
def fetch_and_analyze(self):
"""データ取得と分析"""
print(f"[進行中] {len(self.symbols)}銘柄を分析中...")
for symbol in self.symbols:
try:
# データ取得
df = yf.download(symbol, period='3mo', progress=False)
if df.empty:
continue
# 指標計算
df['MA_SHORT'] = df['Close'].rolling(window=20).mean()
df['MA_LONG'] = df['Close'].rolling(window=50).mean()
# シグナル判定
if len(df) >= 2:
current = df.iloc[-1]
previous = df.iloc[-2]
if (previous['MA_SHORT'] <= previous['MA_LONG'] and
current['MA_SHORT'] > current['MA_LONG']):
self.signals.append({
'symbol': symbol,
'signal': 'BUY',
'price': current['Close'],
})
elif (previous['MA_SHORT'] >= previous['MA_LONG'] and
current['MA_SHORT'] < current['MA_LONG']):
self.signals.append({
'symbol': symbol,
'signal': 'SELL',
'price': current['Close'],
})
except Exception as e:
logging.error(f"分析エラー({symbol}): {e}")
def send_discord_notification(self):
"""Discord通知を送信"""
if not self.signals:
print("[情報] シグナルなし")
return
# メッセージを作成
message = f"\n【取引シグナル】\n"
message += f"時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"
for signal in self.signals:
message += f"{signal['symbol']}: {signal['signal']} @ ¥{signal['price']:,.0f}\n"
# Discord通知を送信
try:
payload = {"content": message}
response = requests.post(self.discord_webhook_url, json=payload)
if response.status_code == 200:
print("[成功] Discord通知を送信しました")
else:
print(f"[警告] Discord通知送信失敗: {response.status_code}")
except Exception as e:
print(f"[エラー] 通知エラー: {e}")
def main():
"""メイン処理"""
print("="*70)
print("リアルタイム信号生成システム")
print("="*70)
print()
# 設定
symbols = ['7203.T', '7201.T', '8058.T', '9984.T']
discord_webhook_url = "YOUR_DISCORD_WEBHOOK_URL" # ここに自分のトークンを設定
if discord_webhook_url == "YOUR_DISCORD_WEBHOOK_URL":
print("[警告] DISCORD_WEBHOOK_URL が設定されていません")
print(" https://discord.com/developers/docs/resources/webhook でトークンを取得してください")
return
# 信号生成
generator = SignalGenerator(symbols, discord_webhook_url)
generator.fetch_and_analyze()
generator.send_discord_notification()
if __name__ == "__main__":
main()フェーズ5:統合スクリプトと自動実行設定(1時間)
【コピペOK】main.py(統合スクリプト)
# ==============================
# メインスクリプト:全処理を統合実行
# ==============================
import sys
from pathlib import Path
# ステップ1: データ取得
from scripts.step1_data_fetch import fetch_data
# ステップ2: バックテスト
from scripts.step2_backtest import run_backtest_from_csv
# ステップ3: シグナル生成
from scripts.step3_signal_gen import SignalGenerator
def main():
"""全処理を実行"""
print("="*70)
print("株式アルゴリズム取引システム 統合実行")
print("="*70)
print()
# ステップ1: データ取得
print("【ステップ1】データ取得")
print("-"*70)
fetch_data("7203.T", period='1y', output_file='data/stock_data.csv')
print()
# ステップ2: バックテスト
print("【ステップ2】バックテスト")
print("-"*70)
run_backtest_from_csv("data/stock_data.csv")
print()
# ステップ3: リアルタイム信号生成
print("【ステップ3】リアルタイム信号生成")
print("-"*70)
symbols = ['7203.T', '7201.T', '8058.T', '9984.T']
discord_webhook_url = "YOUR_DISCORD_WEBHOOK_URL"
if discord_webhook_url != "YOUR_DISCORD_WEBHOOK_URL":
generator = SignalGenerator(symbols, discord_webhook_url)
generator.fetch_and_analyze()
generator.send_discord_notification()
else:
print("[情報] DISCORD_WEBHOOK_URL が設定されていません")
print(" https://discord.com/developers/docs/resources/webhook でトークンを取得し、")
print(" discord_webhook_url に設定してください")
print("\n" + "="*70)
print("実行完了")
print("="*70)
if __name__ == "__main__":
main()Windows タスクスケジューラでの自動実行設定
【手順】
1. タスクスケジューラを開く
→ Win + R → taskschd.msc → Enter
2. 「基本タスクを作成」をクリック
→ 名前: 「Stock Algorithm」
→ 説明: 「株式自動取引分析」
3. トリガーを設定
→ 「毎日」を選択
→ 時刻: 16:30(株式市場の引け後)
4. 操作を設定
→ プログラム: C:\Users\YourName\stock-algorithm\venv\Scripts\python.exe
→ 引数: main.py
→ 開始位置: C:\Users\YourName\stock-algorithm
5. 完了フェーズ6:実運用(1日目以降)
日々の流れ
【毎営業日の流れ】
16:30 → タスクスケジューラが自動実行
↓
1. yfinanceでデータ取得
↓
2. テクニカル指標計算
↓
3. シグナル生成
↓
4. Discord通知送信
↓
スマホに通知が届く
↓
16:31 投資家が通知を確認
↓
16:32 SBI証券サイトで手動発注
↓
完了よくあるエラーと対処法
エラー1:「ModuleNotFoundError: No module named ‘yfinance’」
📘 外部参考：yfinance 公式GitHubリポジトリ ／ PyPIページ
# 仮想環境が有効になっているか確認
# (venv) と表示されているか確認
# なければ再度有効化
venv\Scripts\activate
# インストール実行
pip install yfinanceエラー2:「タスクスケジューラで実行されない」
原因1:Pythonパスが間違っている
→ where python で確認し、完全パスをコピー
原因2:スクリプトパスが間違っている
→ フルパス(C:\Users\...)を使用
原因3:作業ディレクトリが設定されていない
→ 「開始位置」にスクリプトの親ディレクトリを指定エラー3:「Discord通知が来ない」
原因1:DISCORD_WEBHOOK_URL が間違っている
→ https://discord.com/developers/docs/resources/webhook で新しいトークンを生成
原因2:トークンの有効期限が切れている
→ 新しいトークンを生成
原因3:ネットワーク接続がない
→ インターネット接続を確認実運用チェックリスト
本運用を開始する前に、すべてに☑︎をつけてください。
□ Python が正常にインストールされている
□ 仮想環境が作成され、ライブラリがインストールされている
□ データ取得スクリプトが正常に動作する
□ バックテストスクリプトが正常に動作する
□ Discord Webhook トークンを取得し、設定した
□ シグナル生成スクリプトが正常に動作する
□ Discord通知が正常に届く
□ タスクスケジューラで自動実行できる
□ SBI証券の口座にログインできる
□ 小額資金(月1~5万円)で実運用の準備ができているすべてチェックがついたら、本運用開始です。
最後に:心構え
このシステムで実現できること:
✅ 毎日の自動分析
✅ シグナルの自動通知
✅ 感情的判断の排除
✅ 規則的な売買
実現できないこと:
❌ 「完全な自動売買」(SBI証券のAPI制限により)
❌ 年利50%以上(非現実的)
❌ リスクゼロの投資
あなたが手に入れたのは、「機関投資家に勝つシステム」ではなく、「自分の投資判断を機械化するシステム」です。
それで十分です。むしろ、その「シンプルさ」が、長期的な成功の鍵になります。
本シリーズを通じて、個人投資家がアルゴリズム取引システムを構築するための完全なロードマップを提供してきました。後は、あなたの行動だけです。
今日から始めてください。
