Python×yfinanceで株アルゴリズムに使うデータ収集の完全手順

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

Pythonでアルゴリズム取引を開発しようとするとき、最初の大きな課題が「データをどこから取得するか」という問題です。金融データベースサービスは月額数万円~数十万円の料金がかかり、個人投資家には現実的ではありません。一方、無料のデータソースは玉石混交で、データの品質や利用可能性が不確定な場合が多いです。

実際のところ、個人投資家が必要とするレベルのデータは、完全に無料で揃えられます。yfinance、pandas_datareader、Krakenなどのオープンソースライブラリを組み合わせれば、日本株の株価、出来高、為替、さらには簡易的な財務データまで取得できるからです。

しかし、取得したデータをそのまま使うと、不完全な値(欠損値)や異常値が混在しており、シグナル生成時に誤った判断を招きます。データの前処理と正規化は、アルゴリズムの精度を大きく左右する重要なステップです。

本記事では、無料で株価データを取得し、正規化・前処理を行い、最終的に売買シグナル生成まで実行する「データエコシステム」の全手順をコード付きで解説します。個別銘柄だけでなく、複数銘柄の一括取得、スクリーニング、異常値検出なども含めて、実務的なワークフローを網羅しています。

無料でアクセス可能な金融データソースの比較

個人投資家が利用できるデータソースは多く存在しますが、それぞれに特徴と制限があります。

主要なデータソースと特徴

データソース	対象市場	更新頻度	データ品質	用途	APIキー
yfinance	日本株、米国株、仮想通貨	リアルタイム(遅延あり)	高い	株価、出来高、配当	不要
pandas_datareader	日本株、米国株、経済指標	日次	高い	株価、マクロデータ	一部必要
Alpha Vantage	米国株、仮想通貨	リアルタイム(遅延あり)	高い	テクニカル指標、業績	必要(無料枠あり)
IEX Cloud	米国株のみ	リアルタイム	非常に高い	詳細な企業情報、ニュース	必要(有料)
FRED	経済指標のみ	月次以上	非常に高い	マクロ経済データ	必要(無料)
Polygon.io	米国株、仮想通貨	リアルタイム	高い	株価、集計データ	必要(無料枠あり)

個人投資家の場合、yfinanceとpandas_datareaderの組み合わせで、大部分の需要を満たせます。これ以上の詳細データが必要な場合は、Alpha VantageやFREDの無料枠を活用することで、追加投資なしに実装可能です。

yfinanceが「世界標準」である理由

yfinanceは、Google Financeの後継として、Yahoo! Financeのデータを自動取得するPythonライブラリです。以下の理由から、個人投資家に最適です。

• 登録不要: APIキーや認証なしで即座に使用開始可能
• 広範な資産カバー: 日本株、米国株、仮想通貨、先物、インデックスなど
• 定期的な更新: 2025年現在も活発にメンテナンスされている
• 実機関でも使用: 金融機関や大学の研究機関でも採用されている実績

ただし、遅延データ(数分~20分程度の遅延)という制限があるため、デイトレード向きではありません。日足以上の取引には十分です。

【コピペOK】単一銘柄の株価データ取得と正規化

まずは、基本となる「1つの銘柄から株価データを取得し、清潔で分析可能な形に整える」という処理を実装します。

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

# ==============================
# 設定エリア
# ==============================
SYMBOL = "7203.T"              # トヨタ自動車
START_DATE = "2020-01-01"
END_DATE = datetime.now().strftime('%Y-%m-%d')
INTERVAL = "1d"                # 日足

# ==============================
# データ取得関数
# ==============================
def fetch_stock_data(symbol, start_date, end_date, interval='1d'):
    """
    yfinanceから株価データを取得

    Args:
        symbol (str): 銘柄コード(例:7203.T)
        start_date (str): 開始日付(YYYY-MM-DD)
        end_date (str): 終了日付(YYYY-MM-DD)
        interval (str): データ間隔(1m, 5m, 15m, 1h, 1d, 1wk, 1mo)

    Returns:
        pd.DataFrame: OHLCV(始値、高値、安値、終値、出来高)データ
    """
    print(f"[{datetime.now()}] データ取得開始: {symbol}")
    print(f"  期間: {start_date} ~ {end_date}")

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

        if df.empty:
            print(f"警告: {symbol} のデータが取得できませんでした")
            return None

        print(f"[{datetime.now()}] 取得完了: {len(df)}行のデータ")
        return df

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

# ==============================
# データクリーニング(欠損値処理)
# ==============================
def handle_missing_values(df, method='forward_fill'):
    """
    欠損値を処理

    Args:
        df (pd.DataFrame): 価格データ
        method (str): 補完方法
                     'forward_fill': 前営業日の値を使う
                     'interpolate': 線形補間
                     'drop': 欠損行を削除

    Returns:
        pd.DataFrame: 欠損値が処理されたデータ
    """
    print(f"\n【欠損値処理】")
    print(f"処理前の欠損値数: {df.isnull().sum().sum()}")

    if method == 'forward_fill':
        df = df.fillna(method='ffill')
    elif method == 'interpolate':
        df = df.interpolate(method='linear')
    elif method == 'drop':
        df = df.dropna()

    print(f"処理後の欠損値数: {df.isnull().sum().sum()}")
    return df

# ==============================
# 異常値検出と処理
# ==============================
def detect_outliers(df, column='Close', threshold=0.1):
    """
    異常値(ギャップアップ・ギャップダウン)を検出

    Args:
        df (pd.DataFrame): 価格データ
        column (str): チェック対象列
        threshold (float): 異常と判定する変動率(デフォルト:10%)

    Returns:
        pd.DataFrame: 異常値フラグが追加されたデータ
    """
    print(f"\n【異常値検出】")

    # 日次変動率を計算
    df['DailyChange'] = df[column].pct_change()

    # 異常値フラグ(変動率が±threshold以上)
    df['IsOutlier'] = (abs(df['DailyChange']) > threshold)

    outlier_count = df['IsOutlier'].sum()
    print(f"検出された異常値: {outlier_count}件(全体の{outlier_count/len(df)*100:.2f}%)")

    if outlier_count > 0:
        print(f"\n異常値の詳細:")
        outlier_rows = df[df['IsOutlier'] == True]
        for idx, row in outlier_rows.iterrows():
            print(f"  {idx.date()}: {row['DailyChange']*100:+.2f}% "
                  f"(¥{row['Close']:,.0f})")

    return df

# ==============================
# データの正規化
# ==============================
def normalize_data(df):
    """
    データを正規化(0~1のスケールに変換)

    Args:
        df (pd.DataFrame): 価格データ

    Returns:
        pd.DataFrame: 正規化されたデータ
    """
    print(f"\n【データの正規化】")

    df_normalized = df.copy()

    # 価格の正規化(最小値を0、最大値を1に変換)
    for col in ['Open', 'High', 'Low', 'Close']:
        col_min = df_normalized[col].min()
        col_max = df_normalized[col].max()
        df_normalized[f'{col}_Norm'] = (df_normalized[col] - col_min) / (col_max - col_min)

    # 出来高の正規化
    vol_min = df_normalized['Volume'].min()
    vol_max = df_normalized['Volume'].max()
    df_normalized['Volume_Norm'] = (df_normalized['Volume'] - vol_min) / (vol_max - vol_min)

    print(f"正規化完了: Close_Norm の範囲: {df_normalized['Close_Norm'].min():.4f} ~ {df_normalized['Close_Norm'].max():.4f}")

    return df_normalized

# ==============================
# 基本統計情報の表示
# ==============================
def display_data_summary(df, symbol):
    """データの要約統計情報を表示"""
    print(f"\n" + "="*70)
    print(f"データサマリー({symbol})")
    print("="*70)

    print(f"\n【期間】")
    print(f"開始日: {df.index[0].date()}")
    print(f"終了日: {df.index[-1].date()}")
    print(f"営業日数: {len(df)}")

    print(f"\n【価格統計】")
    print(f"始値(Open): 平均 ¥{df['Open'].mean():,.0f}, "
          f"最小 ¥{df['Open'].min():,.0f}, 最大 ¥{df['Open'].max():,.0f}")
    print(f"終値(Close): 平均 ¥{df['Close'].mean():,.0f}, "
          f"最小 ¥{df['Close'].min():,.0f}, 最大 ¥{df['Close'].max():,.0f}")
    print(f"変動幅(High - Low): 平均 ¥{(df['High'] - df['Low']).mean():,.0f}")

    print(f"\n【出来高統計】")
    print(f"平均出来高: {df['Volume'].mean():,.0f}株")
    print(f"最小出来高: {df['Volume'].min():,.0f}株")
    print(f"最大出来高: {df['Volume'].max():,.0f}株")

    daily_change = df['Close'].pct_change()
    print(f"\n【変動率統計】")
    print(f"平均日次変動率: {daily_change.mean()*100:+.3f}%")
    print(f"日次変動の標準偏差: {daily_change.std()*100:.3f}%")
    print(f"最大上昇: {daily_change.max()*100:+.2f}%")
    print(f"最大下落: {daily_change.min()*100:+.2f}%")

# ==============================
# データ保存
# ==============================
def save_data_to_csv(df, symbol, folder='stock_data'):
    """DataFrameをCSVファイルとして保存"""
    import os

    if not os.path.exists(folder):
        os.makedirs(folder)

    filename = f"{folder}/{symbol}_{datetime.now().strftime('%Y%m%d')}.csv"
    df.to_csv(filename)
    print(f"\n[OK] データを保存しました: {filename}")
    return filename

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    # データ取得
    df = fetch_stock_data(SYMBOL, START_DATE, END_DATE, INTERVAL)

    if df is not None:
        # 欠損値を処理
        df = handle_missing_values(df, method='forward_fill')

        # 異常値を検出
        df = detect_outliers(df, column='Close', threshold=0.10)

        # データを正規化
        df = normalize_data(df)

        # 要約統計を表示
        display_data_summary(df, SYMBOL)

        # CSVファイルに保存
        save_data_to_csv(df, SYMBOL)

        # データの先頭5行を表示
        print(f"\n【データサンプル(先頭5行)】")
        print(df[['Open', 'High', 'Low', 'Close', 'Volume']].head())

このコードの処理フロー:

• fetch_stock_data() でyfinanceから株価データを取得
• handle_missing_values() で欠損値(祝日など)を前営業日の値で補完
• detect_outliers() で突然の大きな値動き(ギャップ)を検出
• normalize_data() で価格と出来高を0~1のスケールに変換
• 統計情報を表示し、CSVファイルに保存

複数銘柄の一括取得と効率的な管理

実際の投資では、単一銘柄ではなく複数銘柄を同時に監視することがほとんどです。複数銘柄を効率よく管理するシステムを構築します。

【コピペOK】複数銘柄の一括取得・統合管理システム

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime
import os
import json

# ==============================
# 設定エリア
# ==============================
SYMBOLS = [
    "7203.T",  # トヨタ
    "7201.T",  # 日産
    "8058.T",  # 三菱UFJフィナンシャル
    "9984.T",  # ソフトバンクグループ
    "6758.T",  # ソニー
]

START_DATE = "2022-01-01"
END_DATE = datetime.now().strftime('%Y-%m-%d')
DATA_FOLDER = "multi_stock_data"

# ==============================
# 複数銘柄データ取得エンジン
# ==============================
class MultiStockDataFetcher:
    """複数銘柄のデータを効率よく取得・管理するクラス"""

    def __init__(self, symbols, start_date, end_date, data_folder='stock_data'):
        """
        Args:
            symbols (list): 銘柄コードのリスト
            start_date (str): 開始日付
            end_date (str): 終了日付
            data_folder (str): データ保存フォルダ
        """
        self.symbols = symbols
        self.start_date = start_date
        self.end_date = end_date
        self.data_folder = data_folder
        self.data_dict = {}  # 銘柄ごとのDataFrameを格納
        self.metadata = {}   # メタデータ(銘柄名、セクターなど)

        if not os.path.exists(data_folder):
            os.makedirs(data_folder)

    def fetch_all_symbols(self):
        """
        すべての銘柄のデータを一括取得

        Returns:
            dict: 銘柄ごとのDataFrameの辞書
        """
        print(f"[{datetime.now()}] 複数銘柄のデータ取得を開始")
        print(f"対象銘柄数: {len(self.symbols)}")

        for i, symbol in enumerate(self.symbols, 1):
            print(f"\n[{i}/{len(self.symbols)}] {symbol} を取得中...", end=" ")

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

                if not df.empty:
                    # 欠損値を前営業日の値で補完
                    df = df.fillna(method='ffill')

                    self.data_dict[symbol] = df

                    # メタデータを保存
                    self.metadata[symbol] = {
                        'name': ticker.info.get('longName', 'N/A'),
                        'sector': ticker.info.get('sector', 'N/A'),
                        'data_points': len(df),
                    }

                    print(f"✅ {len(df)}行のデータを取得")
                else:
                    print(f"⚠️ データなし")

            except Exception as e:
                print(f"❌ エラー: {e}")

        print(f"\n[{datetime.now()}] 取得完了")
        return self.data_dict

    def create_unified_dataframe(self, column='Close'):
        """
        複数銘柄のデータを統合して1つのDataFrameにする

        Args:
            column (str): 統合する列(Close, Volume, etc)

        Returns:
            pd.DataFrame: 銘柄ごとの列を持つ統合DataFrame
        """
        print(f"\n[{datetime.now()}] 統合DataFrameを作成中...")

        unified_df = pd.DataFrame()

        for symbol in self.data_dict:
            unified_df[symbol] = self.data_dict[symbol][column]

        # 欠損値を前営業日の値で補完
        unified_df = unified_df.fillna(method='ffill')

        print(f"統合完了: {len(unified_df)}行 × {len(unified_df.columns)}列")
        return unified_df

    def calculate_correlations(self):
        """銘柄間の相関係数を計算"""
        print(f"\n【銘柄間の相関係数】")

        unified_df = self.create_unified_dataframe('Close')

        # 日次リターンを計算
        returns = unified_df.pct_change().dropna()

        # 相関係数を計算
        correlation_matrix = returns.corr()

        print(correlation_matrix.round(3))

        return correlation_matrix

    def detect_low_volume_days(self, threshold=0.5):
        """出来高が異常に少ない営業日を検出"""
        print(f"\n【出来高異常検出(閾値: {threshold*100:.0f}%)】")

        low_volume_count = 0

        for symbol, df in self.data_dict.items():
            avg_volume = df['Volume'].mean()
            low_days = df[df['Volume'] < avg_volume * threshold]

            if len(low_days) > 0:
                print(f"\n{symbol}: {len(low_days)}日検出")
                for idx, row in low_days.tail(3).iterrows():
                    print(f"  {idx.date()}: {row['Volume']:,.0f}株 (平均比 {row['Volume']/avg_volume*100:.1f}%)")
                low_volume_count += len(low_days)

        return low_volume_count

    def generate_daily_report(self):
        """日次レポートを生成"""
        print(f"\n" + "="*70)
        print(f"複数銘柄 日次データレポート")
        print("="*70)

        print(f"\n【取得銘柄一覧】")
        for symbol in self.data_dict:
            info = self.metadata.get(symbol, {})
            df = self.data_dict[symbol]
            latest_price = df['Close'].iloc[-1]
            latest_date = df.index[-1].date()

            print(f"{symbol:8} {info.get('name', 'N/A'):30} "
                  f"最新価格: ¥{latest_price:>8,.0f} ({latest_date})")

    def save_all_data(self):
        """すべてのデータをCSVファイルとして保存"""
        print(f"\n【データを保存中...】")

        for symbol, df in self.data_dict.items():
            filename = f"{self.data_folder}/{symbol}_{datetime.now().strftime('%Y%m%d')}.csv"
            df.to_csv(filename)
            print(f"  ✅ {filename}")

        # 統合DataFrameも保存
        unified_df = self.create_unified_dataframe('Close')
        unified_filename = f"{self.data_folder}/unified_Close_{datetime.now().strftime('%Y%m%d')}.csv"
        unified_df.to_csv(unified_filename)
        print(f"  ✅ {unified_filename}")

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    # 複数銘柄データ取得エンジンを初期化
    fetcher = MultiStockDataFetcher(SYMBOLS, START_DATE, END_DATE, DATA_FOLDER)

    # すべての銘柄のデータを取得
    data_dict = fetcher.fetch_all_symbols()

    # 銘柄間の相関係数を計算
    correlation_matrix = fetcher.calculate_correlations()

    # 出来高の異常を検出
    fetcher.detect_low_volume_days(threshold=0.5)

    # 日次レポートを生成
    fetcher.generate_daily_report()

    # データを保存
    fetcher.save_all_data()

このコードの処理フロー:

• MultiStockDataFetcher クラスで複数銘柄を効率的に管理
• fetch_all_symbols() で全銘柄のデータを順次取得
• create_unified_dataframe() で複数銘柄を1つのDataFrameに統合
• calculate_correlations() で銘柄間の相関係数を計算
• detect_low_volume_days() で出来高の異常を検出

テクニカル指標の自動計算と売買シグナル生成

データを取得した後の最終ステップは、そのデータからテクニカル指標を計算し、売買シグナルを生成することです。

【コピペOK】テクニカル指標の自動計算と信号生成エンジン

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

# ==============================
# 設定エリア
# ==============================
SYMBOL = "7203.T"
START_DATE = "2023-01-01"
END_DATE = datetime.now().strftime('%Y-%m-%d')

# テクニカル指標のパラメータ
INDICATORS = {
    'MA_SHORT': 20,
    'MA_LONG': 50,
    'RSI_PERIOD': 14,
    'MACD_FAST': 12,
    'MACD_SLOW': 26,
    'MACD_SIGNAL': 9,
    'BOLLINGER_PERIOD': 20,
    'BOLLINGER_STD': 2,
}

# ==============================
# テクニカル指標計算関数
# ==============================
class TechnicalIndicatorCalculator:
    """テクニカル指標を計算するクラス"""

    def __init__(self, df):
        """
        Args:
            df (pd.DataFrame): 価格データ(Close, Volumeを含む)
        """
        self.df = df.copy()
        self.df.index = pd.to_datetime(self.df.index)

    def calculate_moving_averages(self, short_period, long_period):
        """移動平均線を計算"""
        print(f"計算中: 移動平均線({short_period}日, {long_period}日)")
        self.df['MA_SHORT'] = self.df['Close'].rolling(window=short_period).mean()
        self.df['MA_LONG'] = self.df['Close'].rolling(window=long_period).mean()
        return self.df

    def calculate_rsi(self, period=14):
        """RSI(相対力指数)を計算"""
        print(f"計算中: RSI({period}日)")

        delta = self.df['Close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
        rs = gain / loss
        self.df['RSI'] = 100 - (100 / (1 + rs))

        return self.df

    def calculate_macd(self, fast=12, slow=26, signal=9):
        """MACD(移動平均収束発散)を計算"""
        print(f"計算中: MACD({fast}, {slow}, {signal})")

        ema_fast = self.df['Close'].ewm(span=fast).mean()
        ema_slow = self.df['Close'].ewm(span=slow).mean()

        self.df['MACD'] = ema_fast - ema_slow
        self.df['Signal'] = self.df['MACD'].ewm(span=signal).mean()
        self.df['MACD_Histogram'] = self.df['MACD'] - self.df['Signal']

        return self.df

    def calculate_bollinger_bands(self, period=20, std_dev=2):
        """ボリンジャーバンドを計算"""
        print(f"計算中: ボリンジャーバンド({period}日, ±{std_dev}σ)")

        middle_band = self.df['Close'].rolling(window=period).mean()
        std = self.df['Close'].rolling(window=period).std()

        self.df['BB_Middle'] = middle_band
        self.df['BB_Upper'] = middle_band + (std_dev * std)
        self.df['BB_Lower'] = middle_band - (std_dev * std)
        self.df['BB_Width'] = self.df['BB_Upper'] - self.df['BB_Lower']

        return self.df

    def calculate_all_indicators(self, config):
        """
        すべてのテクニカル指標を一括計算

        Args:
            config (dict): パラメータ設定
        """
        print(f"\n【テクニカル指標の計算開始】")
        print(f"対象銘柄: {self.df.index[0].date()} ~ {self.df.index[-1].date()}")

        self.calculate_moving_averages(config['MA_SHORT'], config['MA_LONG'])
        self.calculate_rsi(config['RSI_PERIOD'])
        self.calculate_macd(config['MACD_FAST'], config['MACD_SLOW'], config['MACD_SIGNAL'])
        self.calculate_bollinger_bands(config['BOLLINGER_PERIOD'], config['BOLLINGER_STD'])

        print(f"[OK] 計算完了\n")
        return self.df

# ==============================
# 売買シグナル生成エンジン
# ==============================
class SignalGenerator:
    """テクニカル指標からシグナルを生成"""

    def __init__(self, df):
        """Args: df (pd.DataFrame): テクニカル指標が計算されたデータ"""
        self.df = df.copy()
        self.signals = pd.DataFrame(index=self.df.index)

    def generate_ma_crossover_signal(self):
        """移動平均線クロスのシグナルを生成"""
        print("生成中: 移動平均線クロスシグナル")

        self.signals['MA_SIGNAL'] = 0

        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:
                self.signals['MA_SIGNAL'].iloc[i] = 1
            # デッドクロス
            elif prev_short > prev_long and curr_short < curr_long:
                self.signals['MA_SIGNAL'].iloc[i] = -1

        return self.signals

    def generate_rsi_signal(self, oversold=30, overbought=70):
        """RSI反発シグナルを生成"""
        print(f"生成中: RSIシグナル(売られ過ぎ:{oversold}, 買われ過ぎ:{overbought})")

        self.signals['RSI_SIGNAL'] = 0

        for i in range(1, len(self.df)):
            curr_rsi = self.df['RSI'].iloc[i]
            prev_rsi = self.df['RSI'].iloc[i-1]

            if pd.isna(curr_rsi):
                continue

            # 売られ過ぎから回復
            if prev_rsi <= oversold and curr_rsi > oversold:
                self.signals['RSI_SIGNAL'].iloc[i] = 1
            # 買われ過ぎから反転
            elif prev_rsi >= overbought and curr_rsi < overbought:
                self.signals['RSI_SIGNAL'].iloc[i] = -1

        return self.signals

    def generate_macd_signal(self):
        """MACDクロスシグナルを生成"""
        print("生成中: MACDクロスシグナル")

        self.signals['MACD_SIGNAL'] = 0

        for i in range(1, len(self.df)):
            prev_macd = self.df['MACD'].iloc[i-1]
            prev_signal = self.df['Signal'].iloc[i-1]
            curr_macd = self.df['MACD'].iloc[i]
            curr_signal = self.df['Signal'].iloc[i]

            if pd.isna(prev_macd) or pd.isna(curr_macd):
                continue

            # MACD > Signal Line
            if prev_macd <= prev_signal and curr_macd > curr_signal:
                self.signals['MACD_SIGNAL'].iloc[i] = 1
            # MACD < Signal Line
            elif prev_macd >= prev_signal and curr_macd < curr_signal:
                self.signals['MACD_SIGNAL'].iloc[i] = -1

        return self.signals

    def generate_all_signals(self):
        """すべてのシグナルを生成"""
        print(f"\n【シグナル生成開始】")

        self.generate_ma_crossover_signal()
        self.generate_rsi_signal()
        self.generate_macd_signal()

        # 統合シグナル(複数シグナルが一致したときのみ)
        self.signals['CONSENSUS_BUY'] = (
            (self.signals['MA_SIGNAL'] == 1) | 
            (self.signals['RSI_SIGNAL'] == 1) | 
            (self.signals['MACD_SIGNAL'] == 1)
        ).astype(int)

        self.signals['CONSENSUS_SELL'] = (
            (self.signals['MA_SIGNAL'] == -1) | 
            (self.signals['RSI_SIGNAL'] == -1) | 
            (self.signals['MACD_SIGNAL'] == -1)
        ).astype(int)

        print(f"[OK] シグナル生成完了\n")
        return self.signals

    def display_recent_signals(self, rows=10):
        """直近のシグナルを表示"""
        print(f"【直近{rows}日のシグナル】")
        print(self.signals.tail(rows))

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    # データ取得
    print(f"[{datetime.now()}] {SYMBOL} のデータを取得中...")
    ticker = yf.Ticker(SYMBOL)
    df = ticker.history(start=START_DATE, end=END_DATE)
    df = df.fillna(method='ffill')

    # テクニカル指標を計算
    calc = TechnicalIndicatorCalculator(df)
    df_with_indicators = calc.calculate_all_indicators(INDICATORS)

    # シグナルを生成
    signal_gen = SignalGenerator(df_with_indicators)
    signals = signal_gen.generate_all_signals()

    # 結果を統合
    result_df = pd.concat([df_with_indicators, signals], axis=1)

    # 直近のシグナルを表示
    signal_gen.display_recent_signals(10)

    # データを保存
    filename = f"{SYMBOL}_with_signals_{datetime.now().strftime('%Y%m%d')}.csv"
    result_df.to_csv(filename)
    print(f"\n[OK] データを保存しました: {filename}")

このコードの処理フロー:

• TechnicalIndicatorCalculator で複数のテクニカル指標を自動計算
• 移動平均線、RSI、MACD、ボリンジャーバンドなどを計算
• SignalGenerator で複数のシグナルを生成
• 複数シグナルが一致したときのみ取引(フィルター効果)

📘 外部参考：RSI（Wikipedia 日本語） ／ RSI（Investopedia）

📘 外部参考：Bollinger Bands 公式 ／ Wikipedia

📘 外部参考：Moving Average（Investopedia）

📘 外部参考：移動平均（Wikipedia 日本語）

よくあるエラーと対処法

yfinanceでデータが取得できません(特に古いデータ)

原因:

• 指定した期間にYahoo! Financeにデータが存在しない
• 銘柄コードが正確でない
• ネットワークが一時的に切れている

対処法:

• 銘柄コードを確認: yf.Ticker("7203.T").info で仕様確認
• より短い期間から試す: period='1y' で1年分取得してみる
• リトライロジックを実装: 失敗時に数秒待機してから再試行

データに欠損値が多いです

原因:

• 銘柄が上場廃止や合併の対象になった
• データ取得期間に取引がなかった(新興銘柄など)

対処法:

• fillna(method='ffill') で前営業日の値で補完
• 別の補完方法を試す: fillna(method='bfill') (翌営業日の値を使う)
• dropna() で欠損行を削除(ただしデータが失われる)

テクニカル指標がNaN(計算不可)になります

原因:

• 計算に必要な期間分のデータがない
• パラメータ期間が長すぎて、データが足りない

対処法:

• パラメータを短くする: MA20→MA10、RSI14→RSI7など
• より長期のデータを取得する: period='5y' で5年分取得

複数銘柄取得で一部の銘柄だけ失敗します

原因:

• 銘柄が上場廃止になった
• 取引停止中の銘柄が含まれている
• 一時的なネットワークエラー

対処法:

• try-exceptで個別エラーハンドリング
• 成功した銘柄だけを処理継続
• ログに失敗銘柄を記録して後で確認

異常値検出で多くのシグナルが出ます

原因:

• しきい値が低すぎる(例:5%)
• 株式分割や配当落ちの影響で大きな変動が発生

対処法:

• しきい値を高くする: 10%や15%に設定
• 配当落ち日などの企業アクション情報を別途取得して除外

まとめ

本記事では、個人投資家が無料で株式データを取得し、整備し、分析するための完全なワークフローを解説しました。

要点を整理します。

• yfinanceとpandas_datareaderを使うことで、登録不要で高品質なデータを無料で取得可能
• 欠損値処理と異常値検出により、データの品質を確保することが重要
• 複数銘柄を効率よく管理するには、クラスベースの設計が有効
• 移動平均線、RSI、MACD、ボリンジャーバンドなどの主要指標は自動計算可能
• 複数シグナルを組み合わせることで、誤シグナルの削減につながる
• すべてのデータを定期的に保存し、履歴を保全することが実運用では重要

次のステップとしては、本記事のコードで複数銘柄のデータを定期的に取得する自動化パイプラインを構築し、市場のライフサイクルを通じてデータの品質を監視することをお勧めします。その後、バックテストと実運用検証を進めてください。

📘 外部参考：Backtesting.py（公式ドキュメント） ／ Backtrader 公式

データは、アルゴリズム取引の最も重要な資産です。正確で豊富なデータがあれば、あらゆる戦略開発が可能になります。