【初心者向け開発ロードマップ】株アルゴリズムをPython×yfinanceでゼロから自作する完全手順

※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。

「Pythonでアルゴリズム取引システムを開発したい」という願いを持ちながらも、どこから始めたら良いか分からないという初心者は多いです。プログラミング経験がない、Pythonの知識がない、金融の専門知識がない…こうした「ない」の積み重ねが、多くの個人投資家をアクション不足に陥らせています。

しかし現実は、初心者でも段階的に進めば、十分にアルゴリズム取引システムを構築できます。本記事は、本シリーズの集大成として、「ゼロから完成させる」までの全ステップを、初心者向けに最適化して提示します。

環境構築から始まり、Python の基礎、yfinanceの使い方、売買ロジックの実装、バックテスト、実運用まで。すべてをコード付きで、段階的に学べるように設計しました。

ステップ0:前提知識と環境確認

まず、あなたが本当に「ゼロ」の状態なのか確認しましょう。

必要な環境

• PC: Windows、Mac、Linux のいずれか
• インターネット接続: 必須
• テキストエディタ or IDE: VS Code(無料)推奨

必要な知識

• プログラミング: 不要。本記事で学びます。
• Python: 不要。本記事で学びます。
• 金融知識: 基本的な株式概念(買う・売る)があれば十分。

ステップ1:Python のインストールと環境構築

すべての基礎となるステップです。焦らず、確実に進めてください。

1-1. Python のインストール

Windows の場合:

1. https://www.python.org/downloads/ にアクセス
2. 「Download Python 3.11」(最新版)をクリック
3. ダウンロードされた .exe ファイルをダブルクリック
4. 重要: 「Add Python 3.11 to PATH」にチェックを入れる
5. 「Install Now」をクリック
6. インストール完了を待つ

インストール確認:

コマンドプロンプト(Win + R → cmd → Enter)を開いて、以下を入力:

python --version

Python 3.11.x と表示されたら成功です。

Mac の場合:

# Homebrewがインストールされていない場合
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

# Pythonをインストール
brew install python3

# バージョン確認
python3 --version

1-2. 作業フォルダの作成

以下の構造でフォルダを作成してください。

C:\Users\YourName\stock_algorithm\
├── data/                    # データ保存フォルダ
├── results/                 # 結果保存フォルダ
├── scripts/                 # Pythonスクリプト
│   ├── step1_fetch_data.py
│   ├── step2_backtest.py
│   └── step3_live_trading.py
└── requirements.txt         # ライブラリリスト

Windows でフォルダを作成:

# コマンドプロンプトで実行
mkdir C:\Users\YourName\stock_algorithm
cd C:\Users\YourName\stock_algorithm
mkdir data results scripts

1-3. 仮想環境の作成(推奨)

仮想環境を使うことで、ライブラリの競合を避けられます。

# 仮想環境を作成
python -m venv venv

# 仮想環境を有効化(Windows)
venv\Scripts\activate

# 仮想環境を有効化(Mac/Linux)
source venv/bin/activate

仮想環境が有効になると、コマンドプロンプトの先頭に (venv) が表示されます。

1-4. 必要なライブラリをインストール

pip install yfinance pandas numpy scikit-learn tensorflow requests schedule

インストール状況を確認:

pip list

以下が表示されたら成功です:

yfinance          X.X.X
pandas            X.X.X
numpy             X.X.X
requests          X.X.X

ステップ2:Python の最小限の知識

アルゴリズム開発に必要な Python の基礎を、実例を通じて学びます。

2-1. 変数と型

# 変数の宣言と代入
symbol = "7203.T"           # 文字列
price = 2500.50             # 小数点数
quantity = 100              # 整数
is_buy = True               # ブール値(True/False)

# データの型を確認
print(type(price))          # <class 'float'>
print(type(quantity))       # <class 'int'>

2-2. リストと辞書

# リスト(複数のデータを順序付きで格納)
symbols = ["7203.T", "7201.T", "8058.T"]
print(symbols[0])           # "7203.T"(最初の要素)

# 辞書(キーと値のペア)
portfolio = {
    "7203.T": 100,          # トヨタを100株
    "7201.T": 50,           # 日産を50株
}
print(portfolio["7203.T"])  # 100

2-3. 条件分岐

price = 2500

# if文
if price > 2000:
    print("高い")
elif price > 1500:
    print("中程度")
else:
    print("安い")

2-4. ループ

# forループ:リストを順に処理
symbols = ["7203.T", "7201.T", "8058.T"]
for symbol in symbols:
    print(f"銘柄: {symbol}")

# whileループ:条件が真の間、繰り返す
count = 0
while count < 3:
    print(f"カウント: {count}")
    count += 1

2-5. 関数の定義

# 関数を定義
def calculate_return(entry_price, exit_price):
    """
    リターン率を計算

    Args:
        entry_price (float): 買値
        exit_price (float): 売値

    Returns:
        float: リターン率
    """
    return (exit_price - entry_price) / entry_price

# 関数を呼び出す
ret = calculate_return(2000, 2100)
print(f"リターン: {ret*100:.2f}%")  # リターン: 5.00%

ステップ3:yfinance でデータを取得する

実際の株価データを取得してみましょう。

【コピペOK】ステップ3:データ取得スクリプト

scripts/step1_fetch_data.py に以下を保存します。

import yfinance as yf
import pandas as pd
from datetime import datetime

# ==============================
# 設定エリア
# ==============================
SYMBOL = "7203.T"              # 銘柄コード(トヨタ)
START_DATE = "2023-01-01"
END_DATE = datetime.now().strftime('%Y-%m-%d')
OUTPUT_FILE = "data/stock_data.csv"

# ==============================
# ステップ1:データ取得
# ==============================
def fetch_stock_data(symbol, start_date, end_date):
    """
    yfinanceから株価データを取得

    Args:
        symbol (str): 銘柄コード
        start_date (str): 開始日付
        end_date (str): 終了日付

    Returns:
        pd.DataFrame: 株価データ
    """
    print(f"[情報] {symbol} のデータを取得中...")

    try:
        ticker = yf.Ticker(symbol)
        df = ticker.history(start=start_date, end=end_date)

        print(f"[成功] {len(df)}営業日分のデータを取得しました")
        return df

    except Exception as e:
        print(f"[エラー] データ取得に失敗しました: {e}")
        return None

# ==============================
# ステップ2:データの確認と保存
# ==============================
def analyze_and_save(df, output_file):
    """
    データを分析して保存

    Args:
        df (pd.DataFrame): 株価データ
        output_file (str): 保存先ファイル
    """
    if df is None or df.empty:
        print("[エラー] データが空です")
        return

    # 基本統計量を表示
    print(f"\n【データサマリー】")
    print(f"期間: {df.index[0].date()} ~ {df.index[-1].date()}")
    print(f"データ数: {len(df)}行")
    print(f"終値(Close)の最小値: ¥{df['Close'].min():,.0f}")
    print(f"終値(Close)の最大値: ¥{df['Close'].max():,.0f}")
    print(f"終値(Close)の平均値: ¥{df['Close'].mean():,.0f}")
    print(f"出来高の平均: {df['Volume'].mean():,.0f}株")

    # CSVファイルに保存
    df.to_csv(output_file)
    print(f"\n[成功] データを保存しました: {output_file}")

    # 先頭5行を表示
    print(f"\n【データの先頭5行】")
    print(df.head())

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    print("="*70)
    print("ステップ1:株価データ取得")
    print("="*70)
    print()

    # データを取得
    df = fetch_stock_data(SYMBOL, START_DATE, END_DATE)

    # データを分析して保存
    analyze_and_save(df, OUTPUT_FILE)

実行方法:

# 仮想環境を有効化(既に有効の場合は不要)
venv\Scripts\activate

# スクリプトを実行
python scripts/step1_fetch_data.py

期待される出力:

======================================================================
ステップ1:株価データ取得
======================================================================

[情報] 7203.T のデータを取得中...
[成功] 252営業日分のデータを取得しました

【データサマリー】
期間: 2023-01-01 ~ 2024-01-01
データ数: 252行
終値(Close)の最小値: ¥2,150
終値(Close)の最大値: ¥2,850
終値(Close)の平均値: ¥2,500
出来高の平均: 50,000,000株

[成功] データを保存しました: data/stock_data.csv

ステップ4:売買ロジックを実装する

データが揃ったら、実際の売買判定ロジックを実装します。

【コピペOK】ステップ4:売買シグナル生成スクリプト

scripts/step2_signals.py に以下を保存します。

import pandas as pd
import numpy as np

# ==============================
# ステップ1:移動平均線の計算
# ==============================
def calculate_moving_averages(df, short_period=20, long_period=50):
    """
    移動平均線を計算

    Args:
        df (pd.DataFrame): 株価データ
        short_period (int): 短期MA の期間
        long_period (int): 長期MA の期間

    Returns:
        pd.DataFrame: MAが追加されたデータ
    """
    df['MA_SHORT'] = df['Close'].rolling(window=short_period).mean()
    df['MA_LONG'] = df['Close'].rolling(window=long_period).mean()

    return df

# ==============================
# ステップ2:売買シグナルの生成
# ==============================
def generate_signals(df):
    """
    移動平均線のクロスからシグナルを生成

    Returns:
        pd.Series: シグナル (1=買い, -1=売り, 0=シグナルなし)
    """
    signals = pd.Series(0, index=df.index)

    for i in range(1, len(df)):
        # 前営業日と当営業日のデータを比較
        prev_short = df['MA_SHORT'].iloc[i-1]
        prev_long = df['MA_LONG'].iloc[i-1]
        curr_short = df['MA_SHORT'].iloc[i]
        curr_long = df['MA_LONG'].iloc[i]

        # データが不完全な場合はスキップ
        if pd.isna(prev_short) or pd.isna(curr_short):
            continue

        # ゴールデンクロス(買いシグナル)
        # 条件:前日は短期MA < 長期MA、今日は短期MA > 長期MA
        if prev_short < prev_long and curr_short > curr_long:
            signals.iloc[i] = 1

        # デッドクロス(売りシグナル)
        # 条件:前日は短期MA > 長期MA、今日は短期MA < 長期MA
        elif prev_short > prev_long and curr_short < curr_long:
            signals.iloc[i] = -1

    return signals

# ==============================
# ステップ3:シグナルの可視化
# ==============================
def display_signals(df, signals):
    """
    シグナルを表示

    Args:
        df (pd.DataFrame): 株価データ
        signals (pd.Series): シグナル
    """
    # シグナルが発生した日付を抽出
    buy_signals = df[signals == 1]
    sell_signals = df[signals == -1]

    print("\n【買いシグナル】")
    if len(buy_signals) > 0:
        for date, row in buy_signals.iterrows():
            print(f"  {date.date()}: ¥{row['Close']:,.0f}")
    else:
        print("  買いシグナルなし")

    print("\n【売りシグナル】")
    if len(sell_signals) > 0:
        for date, row in sell_signals.iterrows():
            print(f"  {date.date()}: ¥{row['Close']:,.0f}")
    else:
        print("  売りシグナルなし")

    print(f"\n【統計】")
    print(f"買いシグナル: {len(buy_signals)}件")
    print(f"売りシグナル: {len(sell_signals)}件")

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    print("="*70)
    print("ステップ2:売買シグナル生成")
    print("="*70)
    print()

    # CSVから前回取得したデータを読み込む
    df = pd.read_csv("data/stock_data.csv", index_col=0, parse_dates=True)

    # 移動平均線を計算
    print("[進行中] 移動平均線を計算中...")
    df = calculate_moving_averages(df, short_period=20, long_period=50)

    # シグナルを生成
    print("[進行中] シグナルを生成中...")
    signals = generate_signals(df)

    # シグナルをDataFrameに追加
    df['Signal'] = signals

    # シグナルを表示
    display_signals(df, signals)

    # 結果を保存
    df.to_csv("data/signals.csv")
    print("\n[成功] シグナルを保存しました: data/signals.csv")

実行方法:

python scripts/step2_signals.py

ステップ5:バックテストを実装する

生成したシグナルで、過去データを使ったシミュレーション(バックテスト)を行います。

【コピペOK】ステップ5:バックテストスクリプト

scripts/step3_backtest.py に以下を保存します。

import pandas as pd
import numpy as np

# ==============================
# 設定エリア
# ==============================
INITIAL_CAPITAL = 1000000      # 初期資金100万円
COMMISSION_RATE = 0.001        # 手数料0.1%

# ==============================
# ステップ1:バックテスト実行
# ==============================
def backtest(df, initial_capital, commission_rate):
    """
    バックテストを実行

    Args:
        df (pd.DataFrame): シグナルが含まれたデータ
        initial_capital (float): 初期資金
        commission_rate (float): 手数料率

    Returns:
        dict: バックテスト結果
    """
    cash = initial_capital           # 現金
    position = 0                     # 保有株数
    portfolio_values = []            # ポートフォリオ価値の推移
    trades = []                      # 取引履歴

    for i in range(len(df)):
        current_price = df['Close'].iloc[i]
        signal = df['Signal'].iloc[i]
        date = df.index[i]

        # 買いシグナル
        if signal == 1 and position == 0:
            # 買える株数を計算
            shares = int(cash / current_price * (1 - commission_rate))

            if shares > 0:
                cost = shares * current_price * (1 + commission_rate)
                cash -= cost
                position = shares

                trades.append({
                    'date': date,
                    'type': 'BUY',
                    'price': current_price,
                    'shares': shares,
                })

        # 売りシグナル
        elif signal == -1 and position > 0:
            revenue = position * current_price * (1 - commission_rate)
            cash += revenue

            profit = revenue - (trades[-1]['price'] * position) if trades else 0

            trades.append({
                'date': date,
                'type': 'SELL',
                'price': current_price,
                'shares': position,
                'profit': profit,
            })

            position = 0

        # ポートフォリオ価値を記録
        portfolio_value = cash + (position * current_price)
        portfolio_values.append(portfolio_value)

    # 最後にポジションが残っていたら売却
    if position > 0:
        final_price = df['Close'].iloc[-1]
        revenue = position * final_price * (1 - commission_rate)
        cash += revenue

    # 結果を計算
    total_return = (cash - initial_capital) / initial_capital
    final_value = cash

    return {
        'final_cash': final_value,
        'total_return': total_return,
        'trades': trades,
        'portfolio_values': portfolio_values,
    }

# ==============================
# ステップ2:結果を表示
# ==============================
def display_backtest_result(result, initial_capital):
    """
    バックテスト結果を表示

    Args:
        result (dict): バックテスト結果
        initial_capital (float): 初期資金
    """
    print("\n" + "="*70)
    print("バックテスト結果")
    print("="*70)

    print(f"\n【成績】")
    print(f"初期資金: ¥{initial_capital:,.0f}")
    print(f"最終資金: ¥{result['final_cash']:,.0f}")
    print(f"総リターン: {result['total_return']*100:+.2f}%")

    print(f"\n【取引】")
    print(f"総取引数: {len(result['trades'])}件")

    # 利益のある取引を集計
    profits = [t['profit'] for t in result['trades'] if 'profit' in t and t['profit'] > 0]
    losses = [t['profit'] for t in result['trades'] if 'profit' in t and t['profit'] < 0]

    if profits:
        print(f"勝ち取引: {len(profits)}件(平均 ¥{np.mean(profits):,.0f})")

    if losses:
        print(f"負け取引: {len(losses)}件(平均 ¥{np.mean(losses):,.0f})")

    # 直近の取引を表示
    print(f"\n【直近5件の取引】")
    for trade in result['trades'][-5:]:
        trade_type = trade['type']
        date = trade['date'].strftime('%Y-%m-%d')
        price = trade['price']
        shares = trade['shares']

        if trade_type == 'BUY':
            print(f"  {date} {trade_type}: ¥{price:,.0f} × {shares}株")
        else:
            profit = trade.get('profit', 0)
            print(f"  {date} {trade_type}: ¥{price:,.0f} × {shares}株 → 利益 ¥{profit:,.0f}")

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    print("="*70)
    print("ステップ3:バックテスト実行")
    print("="*70)
    print()

    # CSVからシグナル入りデータを読み込む
    df = pd.read_csv("data/signals.csv", index_col=0, parse_dates=True)

    # バックテストを実行
    print("[進行中] バックテストを実行中...")
    result = backtest(df, INITIAL_CAPITAL, COMMISSION_RATE)

    # 結果を表示
    display_backtest_result(result, INITIAL_CAPITAL)

    # 結果をCSVに保存
    trades_df = pd.DataFrame(result['trades'])
    trades_df.to_csv("results/backtest_trades.csv", index=False)
    print(f"\n[成功] 結果を保存しました: results/backtest_trades.csv")

実行方法:

python scripts/step3_backtest.py

ステップ6:実運用システムの構築

バックテストで検証したロジックを、実運用可能なシステムに昇華させます。

【コピペOK】ステップ6:実運用スクリプト

scripts/step4_live_trading.py に以下を保存します。

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime
import json
import requests
import logging

# ==============================
# ロギング設定
# ==============================
logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] %(levelname)s: %(message)s'
)

# ==============================
# 設定エリア
# ==============================
SYMBOL = "7203.T"
LINE_TOKEN = "YOUR_LINE_NOTIFY_TOKEN"  # LINE Notifyのトークンに置き換え
STATE_FILE = "trading_state.json"

# ==============================
# ステップ1:状態の保存・読み込み
# ==============================
class TradingState:
    """取引状態を管理"""

    @staticmethod
    def load():
        """状態を読み込む"""
        try:
            with open(STATE_FILE, 'r') as f:
                return json.load(f)
        except:
            return {
                'position': 0,
                'entry_price': 0,
                'last_update': None,
            }

    @staticmethod
    def save(state):
        """状態を保存"""
        with open(STATE_FILE, 'w') as f:
            json.dump(state, f, indent=2, default=str)

# ==============================
# ステップ2:シグナル検出
# ==============================
def detect_signal(symbol):
    """
    リアルタイムでシグナルを検出

    Returns:
        str: 'BUY', 'SELL', または 'HOLD'
    """
    try:
        # データを取得
        df = yf.download(symbol, period='3mo', progress=False)

        # 移動平均線を計算
        df['MA_SHORT'] = df['Close'].rolling(window=20).mean()
        df['MA_LONG'] = df['Close'].rolling(window=50).mean()

        # シグナル判定
        if len(df) < 2:
            return 'HOLD'

        prev_short = df['MA_SHORT'].iloc[-2]
        prev_long = df['MA_LONG'].iloc[-2]
        curr_short = df['MA_SHORT'].iloc[-1]
        curr_long = df['MA_LONG'].iloc[-1]

        if pd.isna(prev_short) or pd.isna(curr_short):
            return 'HOLD'

        # ゴールデンクロス
        if prev_short < prev_long and curr_short > curr_long:
            return 'BUY'

        # デッドクロス
        if prev_short > prev_long and curr_short < curr_long:
            return 'SELL'

        return 'HOLD'

    except Exception as e:
        logging.error(f"シグナル検出エラー: {e}")
        return 'HOLD'

# ==============================
# ステップ3:LINE通知
# ==============================
def send_line_notification(message):
    """
    LINE Notifyで通知を送信

    Args:
        message (str): 通知メッセージ
    """
    if LINE_TOKEN == "YOUR_LINE_NOTIFY_TOKEN":
        logging.warning("LINE_TOKEN が設定されていません")
        return

    try:
        headers = {"Authorization": f"Bearer {LINE_TOKEN}"}
        data = {"message": message}

        requests.post(
            "https://notify-api.line.me/api/notify",
            headers=headers,
            data=data
        )

        logging.info("LINE通知を送信しました")

    except Exception as e:
        logging.error(f"LINE通知エラー: {e}")

# ==============================
# ステップ4:メイン処理
# ==============================
def run_trading_system():
    """取引システムを実行"""
    logging.info("="*70)
    logging.info("リアルタイム取引システム 開始")
    logging.info("="*70)

    # 状態を読み込む
    state = TradingState.load()

    # シグナルを検出
    signal = detect_signal(SYMBOL)

    # 現在価格を取得
    ticker = yf.Ticker(SYMBOL)
    current_price = ticker.info['currentPrice']

    logging.info(f"銘柄: {SYMBOL}")
    logging.info(f"現在価格: ¥{current_price:,.0f}")
    logging.info(f"シグナル: {signal}")

    # シグナルに応じて行動
    if signal == 'BUY' and state['position'] == 0:
        # 買いシグナル
        state['position'] = 1
        state['entry_price'] = current_price
        state['last_update'] = datetime.now().isoformat()
        TradingState.save(state)

        message = f"\n【買いシグナル発生】\n{SYMBOL}\n価格: ¥{current_price:,.0f}\n時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        send_line_notification(message)
        logging.info("買いシグナルを発信しました")

    elif signal == 'SELL' and state['position'] == 1:
        # 売りシグナル
        profit = (current_price - state['entry_price']) / state['entry_price']
        state['position'] = 0
        state['last_update'] = datetime.now().isoformat()
        TradingState.save(state)

        message = f"\n【売りシグナル発生】\n{SYMBOL}\n売値: ¥{current_price:,.0f}\n利益率: {profit*100:+.2f}%\n時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        send_line_notification(message)
        logging.info("売りシグナルを発信しました")

    logging.info("実行完了\n")

# ==============================
# エントリーポイント
# ==============================
if __name__ == "__main__":
    run_trading_system()

実行方法:

python scripts/step4_live_trading.py

ステップ7:Windowsのタスクスケジューラで自動実行

毎営業日の引け後(16時30分)に自動実行する設定:

1. タスクスケジューラを開く: Win + R → taskschd.msc
2. 基本タスクを作成: 右ペインから「基本タスクを作成」
3. 名前と説明: 「Stock Trading Bot」
4. トリガー: 「毎日」→ 時刻「16:30」
5. 操作:
   • プログラム: C:\Users\YourName\stock_algorithm\venv\Scripts\python.exe
   • 引数: C:\Users\YourName\stock_algorithm\scripts\step4_live_trading.py
   • 開始位置: C:\Users\YourName\stock_algorithm
6. 完了

よくあるエラーと対処法

「ModuleNotFoundError: No module named ‘yfinance’」

原因: ライブラリがインストールされていない

対処法:

# 仮想環境が有効になっていることを確認
# (venv) と表示されていることを確認してから実行
pip install yfinance pandas numpy

「PermissionError」でファイル保存に失敗

原因: ファイルが他のプログラムで開かれている

対処法:

• CSVファイルを閉じる(Excelなど)
• 別の名前で保存するように変更

LINE通知が来ない

原因: LINE_TOKEN が間違っている

対処法:

• https://notify-bot.line.me/my/ で新しいトークンを生成
• スクリプトに正確にコピー・ペースト

まとめ

本記事では、「ゼロから」アルゴリズム取引システムを構築する全7ステップを、初心者向けに最適化して提示しました。

要点を整理します。

• ステップ1-2: 環境構築とPython基礎(3時間程度)
• ステップ3: データ取得の実装(1時間程度)
• ステップ4: 売買ロジックの実装(2時間程度)
• ステップ5: バックテストの実装(1時間程度)
• ステップ6: 実運用システムの構築(2時間程度)
• ステップ7: 自動化の設定(30分程度)

合計:約10時間で、基本的なシステムが完成します。

次のステップ:

1. このシリーズの他の記事を読み、より高度な機能を追加
2. 複数銘柄対応、リバランス機能、パフォーマンス追跡など
3. 本記事のコードはあくまで「教育用」。実運用時は十分なテストと小額資金から始めてください

あなたも今日から、自分専用の株式アルゴリズムを持つことができます。焦らず、段階的に進めてください。