Python×yfinanceで株アルゴリズムを構築する現実的ロードマップ決定版

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

「自動売買ロボットを作れば、放置しておくだけで利益が出る」という誘いは、個人投資家が最も陥りやすい幻想です。実際のところ、個人向けの自動売買ロボットを構築することは、技術的には可能ですが、実運用には多くの落とし穴があります。

最大の問題は、SBI証券や楽天証券といった国内主流証券会社が、個人向けに「自由にPythonから発注できるAPI」を提供していないという現実です。非公式なライブラリは存在しますが、セキュリティリスクが高く、仕様変更時の対応負荷も大きいため、実務的ではありません。

一方、yfinanceを使ったデータ取得、売買シグナル生成、LINE自動通知までは、完全に個人で実装可能です。これを「準自動化システム」として運用することで、手動売買の効率化と規則性の向上が実現できます。本記事では、実装可能な範囲での「現実的なロボット構築」をゼロから解説し、将来的に証券会社APIが解放されたときに対応できる設計を提示します。

個人向け自動売買の現状と現実的な選択肢

自動売買ロボットの構築を始める前に、その制約と現実を理解することが重要です。

SBI証券・楽天証券のAPI現状と限界

国内の主流証券会社では、以下の状況が続いています。

証券会社	公開API	個人向け発注API	現状
SBI証券	一部公開	非公開	法人向け、認定パートナー向けのみ
楽天証券	一部公開	非公開	ラッパーAPIは非公式
マネックス証券	一部公開	非公開	データ取得のみ
Liquid by Quoine	一部公開	公開(暗号資産向け)	暗号資産は可能、株式不可
オートレ(自動売買専門)	非公開	専有システムのみ	専用ツール内でのみ利用可

結論としては、日本の主流証券会社からは、個人投資家が自由にPythonで発注できるAPIは事実上利用不可です。

「準自動化システム」という現実的な選択肢

この制約のもとで、個人投資家が実装可能なのは以下のシステムです。

【準自動化システムの流れ】

(自動) yfinanceでデータ取得
    ↓
(自動) テクニカル指標計算・売買シグナル生成
    ↓
(自動) LINE / メール / Slack で通知
    ↓
(手動) 人間が通知を確認
    ↓
(手動) 証券会社のWebサイト or アプリで実際に注文

この仕組みであれば、以下のメリットが得られます。

• セキュリティ: 認証情報をPythonに保存しない(最安全)
• 安定性: 証券会社のサーバーに依存しない(通知だけ)
• 規制準拠: 金融庁の指導に抵触しない
• 検証容易: バックテストと実運用の乖離が小さい

【コピペOK】準自動化ロボットの完全実装

では、実際に動作する「準自動化ロボット」を実装します。このシステムは、毎日指定時刻に自動実行され、取引機会が発生したらLINEで通知します。

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import requests
import json
import logging
from pathlib import Path

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

# ==============================
# 設定エリア
# ==============================
class Config:
    """ロボット設定クラス"""

    # 監視対象銘柄
    SYMBOLS = [
        "7203.T",  # トヨタ
        "7201.T",  # 日産
        "8058.T",  # 三菱UFJフィナンシャル
    ]

    # データ取得期間
    DATA_PERIOD = "1y"

    # テクニカル指標パラメータ
    MA_SHORT = 20
    MA_LONG = 50
    RSI_PERIOD = 14

    # 売買ルール
    GOLDEN_CROSS_THRESHOLD = 0.01  # ゴールデンクロス判定の閾値(1%)
    DEAD_CROSS_THRESHOLD = 0.01    # デッドクロス判定の閾値(1%)
    RSI_OVERSOLD = 30              # RSI売られ過ぎ
    RSI_OVERBOUGHT = 70            # RSI買われ過ぎ

    # LINE Notify設定
    LINE_TOKEN = "YOUR_LINE_NOTIFY_TOKEN"  # ここに自分のトークンを貼り付け
    LINE_API_URL = "https://notify-api.line.me/api/notify"

    # ポジション管理(状態ファイル)
    STATE_FILE = "robot_state.json"

# ==============================
# データ取得・分析エンジン
# ==============================
class StockAnalyzer:
    """株価データの取得と分析"""

    def __init__(self, symbol):
        """
        Args:
            symbol (str): 銘柄コード
        """
        self.symbol = symbol
        self.df = None
        self.current_data = None

    def fetch_data(self):
        """yfinanceからデータを取得"""
        try:
            logging.info(f"データ取得: {self.symbol}")
            ticker = yf.Ticker(self.symbol)
            self.df = ticker.history(period=Config.DATA_PERIOD)

            if self.df.empty:
                logging.warning(f"データ取得失敗: {self.symbol}")
                return False

            # 欠損値を補完
            self.df = self.df.fillna(method='ffill')

            logging.info(f"取得完了: {len(self.df)}営業日分")
            return True

        except Exception as e:
            logging.error(f"エラー: {e}")
            return False

    def calculate_indicators(self):
        """テクニカル指標を計算"""
        if self.df is None:
            return False

        try:
            # 移動平均線
            self.df['MA_SHORT'] = self.df['Close'].rolling(window=Config.MA_SHORT).mean()
            self.df['MA_LONG'] = self.df['Close'].rolling(window=Config.MA_LONG).mean()

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

            return True

        except Exception as e:
            logging.error(f"指標計算エラー: {e}")
            return False

    def detect_signals(self):
        """
        売買シグナルを検出

        Returns:
            dict: シグナル情報
        """
        if self.df is None or len(self.df) < 2:
            return None

        current = self.df.iloc[-1]
        previous = self.df.iloc[-2]

        current_price = current['Close']
        current_rsi = current['RSI']

        signal = {
            'symbol': self.symbol,
            'timestamp': datetime.now().isoformat(),
            'current_price': current_price,
            'current_rsi': current_rsi,
            'signal_type': 'HOLD',
            'reason': 'シグナルなし',
            'confidence': 'LOW',
        }

        # データ不足チェック
        if pd.isna(previous['MA_SHORT']) or pd.isna(current['MA_SHORT']):
            return signal

        prev_short = previous['MA_SHORT']
        prev_long = previous['MA_LONG']
        curr_short = current['MA_SHORT']
        curr_long = current['MA_LONG']

        # ゴールデンクロス検出
        if (prev_short <= prev_long and curr_short > curr_long and 
            (curr_short - curr_long) / curr_long > Config.GOLDEN_CROSS_THRESHOLD):

            signal['signal_type'] = 'BUY'
            signal['reason'] = f"ゴールデンクロス(MA{Config.MA_SHORT} > MA{Config.MA_LONG})"

            # RSIで確度を判定
            if current_rsi < Config.RSI_OVERBOUGHT:
                signal['confidence'] = 'HIGH'
            else:
                signal['confidence'] = 'MEDIUM'

        # デッドクロス検出
        elif (prev_short >= prev_long and curr_short < curr_long and 
              (curr_long - curr_short) / curr_long > Config.DEAD_CROSS_THRESHOLD):

            signal['signal_type'] = 'SELL'
            signal['reason'] = f"デッドクロス(MA{Config.MA_SHORT} < MA{Config.MA_LONG})"

            # RSIで確度を判定
            if current_rsi > Config.RSI_OVERSOLD:
                signal['confidence'] = 'HIGH'
            else:
                signal['confidence'] = 'MEDIUM'

        return signal

# ==============================
# 通知システム
# ==============================
class NotificationManager:
    """LINE Notifyを使った通知管理"""

    @staticmethod
    def send_notification(message, token=Config.LINE_TOKEN):
        """
        LINEで通知を送信

        Args:
            message (str): 送信するメッセージ
            token (str): LINE Notifyトークン

        Returns:
            bool: 送信成功=True, 失敗=False
        """
        if token == "YOUR_LINE_NOTIFY_TOKEN":
            logging.warning("LINE_TOKEN が設定されていません")
            return False

        try:
            headers = {"Authorization": f"Bearer {token}"}
            data = {"message": message}
            response = requests.post(Config.LINE_API_URL, headers=headers, data=data)

            if response.status_code == 200:
                logging.info("LINE通知送信完了")
                return True
            else:
                logging.warning(f"LINE通知送信失敗: {response.status_code}")
                return False

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

    @staticmethod
    def create_message(signals):
        """
        複数のシグナルからメッセージを作成

        Args:
            signals (list): シグナル情報のリスト

        Returns:
            str: LINE通知用メッセージ
        """
        buy_signals = [s for s in signals if s['signal_type'] == 'BUY']
        sell_signals = [s for s in signals if s['signal_type'] == 'SELL']

        message = f"\n【株式自動売買ロボット】\n"
        message += f"実行時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"

        if buy_signals:
            message += "🟢 買いシグナル:\n"
            for signal in buy_signals:
                message += f"{signal['symbol']}\n"
                message += f"  価格: ¥{signal['current_price']:,.0f}\n"
                message += f"  理由: {signal['reason']}\n"
                message += f"  確度: {signal['confidence']}\n"
                message += f"  RSI: {signal['current_rsi']:.1f}\n\n"

        if sell_signals:
            message += "🔴 売りシグナル:\n"
            for signal in sell_signals:
                message += f"{signal['symbol']}\n"
                message += f"  価格: ¥{signal['current_price']:,.0f}\n"
                message += f"  理由: {signal['reason']}\n"
                message += f"  確度: {signal['confidence']}\n"
                message += f"  RSI: {signal['current_rsi']:.1f}\n\n"

        if not buy_signals and not sell_signals:
            message += "特にシグナルはありません。\n"

        message += "※このメッセージは自動売買ロボットからの通知です。\n"
        message += "※実際の取引は、確認の上で自己責任で行ってください。"

        return message

# ==============================
# ポジション・状態管理
# ==============================
class StateManager:
    """ロボットの状態をファイルで管理"""

    @staticmethod
    def load_state():
        """状態ファイルから状態を読み込む"""
        state_file = Path(Config.STATE_FILE)

        if state_file.exists():
            try:
                with open(state_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                logging.error(f"状態読み込みエラー: {e}")
                return {}

        return {}

    @staticmethod
    def save_state(state):
        """状態ファイルに状態を保存"""
        try:
            with open(Config.STATE_FILE, 'w', encoding='utf-8') as f:
                json.dump(state, f, ensure_ascii=False, indent=2, default=str)
            logging.info("状態を保存しました")

        except Exception as e:
            logging.error(f"状態保存エラー: {e}")

    @staticmethod
    def update_position(state, symbol, signal_type):
        """ポジション情報を更新"""
        if 'positions' not in state:
            state['positions'] = {}

        if signal_type == 'BUY':
            state['positions'][symbol] = {
                'status': 'HOLDING',
                'entry_time': datetime.now().isoformat(),
            }
        elif signal_type == 'SELL':
            if symbol in state['positions']:
                del state['positions'][symbol]

        StateManager.save_state(state)

# ==============================
# メインロボットクラス
# ==============================
class TradingRobot:
    """自動売買ロボットのメインエンジン"""

    def __init__(self):
        """初期化"""
        self.analyzers = {}
        self.signals = []
        self.state = StateManager.load_state()

    def initialize(self):
        """ロボットを初期化"""
        logging.info("="*70)
        logging.info("株式自動売買ロボット 実行開始")
        logging.info("="*70)

        # アナライザーを初期化
        for symbol in Config.SYMBOLS:
            self.analyzers[symbol] = StockAnalyzer(symbol)

    def run_analysis(self):
        """すべての銘柄を分析"""
        logging.info(f"\n分析実行: {len(Config.SYMBOLS)}銘柄")

        self.signals = []

        for symbol in Config.SYMBOLS:
            analyzer = self.analyzers[symbol]

            # データ取得
            if not analyzer.fetch_data():
                continue

            # 指標計算
            if not analyzer.calculate_indicators():
                continue

            # シグナル検出
            signal = analyzer.detect_signals()

            if signal:
                self.signals.append(signal)

                if signal['signal_type'] != 'HOLD':
                    logging.info(f"シグナル検出: {signal['symbol']} - {signal['signal_type']}")

        logging.info(f"分析完了: {len(self.signals)}銘柄処理")

    def send_notifications(self):
        """シグナルをLINEで通知"""
        if not self.signals:
            logging.info("通知対象なし")
            return

        # メッセージを作成
        message = NotificationManager.create_message(self.signals)

        # LINE通知を送信
        NotificationManager.send_notification(message)

        # ポジションを更新
        for signal in self.signals:
            if signal['signal_type'] in ['BUY', 'SELL']:
                StateManager.update_position(self.state, signal['symbol'], signal['signal_type'])

    def display_report(self):
        """実行結果をレポート表示"""
        print(f"\n" + "="*70)
        print("実行結果レポート")
        print("="*70)
        print(f"\n実行時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"処理銘柄数: {len(self.signals)}")

        buy_count = sum(1 for s in self.signals if s['signal_type'] == 'BUY')
        sell_count = sum(1 for s in self.signals if s['signal_type'] == 'SELL')
        hold_count = sum(1 for s in self.signals if s['signal_type'] == 'HOLD')

        print(f"買いシグナル: {buy_count}件")
        print(f"売りシグナル: {sell_count}件")
        print(f"シグナルなし: {hold_count}件")

        if buy_count > 0 or sell_count > 0:
            print(f"\n【シグナル詳細】")
            for signal in self.signals:
                if signal['signal_type'] != 'HOLD':
                    print(f"\n{signal['symbol']} - {signal['signal_type']}")
                    print(f"  価格: ¥{signal['current_price']:,.0f}")
                    print(f"  理由: {signal['reason']}")
                    print(f"  確度: {signal['confidence']}")
                    print(f"  RSI: {signal['current_rsi']:.1f}")

    def run(self):
        """ロボット実行(メイン処理)"""
        try:
            self.initialize()
            self.run_analysis()
            self.send_notifications()
            self.display_report()

        except Exception as e:
            logging.error(f"実行エラー: {e}")
            error_message = f"ロボット実行エラー: {e}"
            NotificationManager.send_notification(error_message)

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

# ==============================
# スケジューラー設定
# ==============================
def schedule_robot_execution():
    """
    ロボットを定期実行するためのスケジューラー

    Windowsのタスクスケジューラで、このスクリプトを
    毎営業日16時30分(市場引け後)に実行するよう設定します。
    """
    robot = TradingRobot()
    robot.run()

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    # 直接実行時
    robot = TradingRobot()
    robot.run()

    # または定期実行する場合:
    # import schedule
    # import time
    # schedule.every().weekday(0).at("16:30").do(schedule_robot_execution)  # 月曜日
    # schedule.every().weekday(1).at("16:30").do(schedule_robot_execution)  # 火曜日
    # # ... 金曜日まで
    # while True:
    #     schedule.run_pending()
    #     time.sleep(60)

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

• StockAnalyzer で複数銘柄のデータを取得し、テクニカル指標を計算
• 移動平均線のクロスとRSIでシグナルを検出
• NotificationManager でシグナル情報をLINEで通知
• StateManager で過去のポジション情報を保存・管理
• TradingRobot がメイン実行エンジン

Windowsのタスクスケジューラで自動実行する設定

Pythonスクリプトを毎日自動実行するには、Windowsのタスクスケジューラを使用します。

タスク登録手順

1. スクリプト保存: 上記のコードを trading_robot.py として保存
2. タスクスケジューラを開く:
   • Win + R → taskschd.msc → Enter
3. 基本タスクを作成:
   • 右ペインから「基本タスクを作成」
   • タスク名: Stock Trading Robot
   • 説明: 自動売買ロボット
4. トリガー設定:
   • 「毎日」を選択
   • 時刻: 16:30(株式市場の引け後)
5. 操作を設定:
   • プログラム/スクリプト: C:\Python311\python.exe (Pythonのパス)
   • 引数: C:\Users\YourName\Desktop\trading_robot.py
   • 開始位置: C:\Users\YourName\Desktop
6. 完了して保存

注意: Pythonパスは環境によって異なります。コマンドプロンプルで where python を実行して確認してください。

より発展的な実装:複数ポジション管理と損切り機能

実際の運用では、複数銘柄のポジションを同時に保有し、損切りルールを設定する必要があります。

【コピペOK】ポジション・リスク管理機能の追加

import json
from datetime import datetime, timedelta

# ==============================
# ポジション・リスク管理クラス
# ==============================
class PortfolioManager:
    """複数ポジションと損切りルールを管理"""

    def __init__(self, initial_capital=1000000, max_position_size=100000):
        """
        Args:
            initial_capital (float): 初期資金
            max_position_size (float): 1銘柄あたりの最大投資額
        """
        self.capital = initial_capital
        self.max_position_size = max_position_size
        self.positions = {}  # {symbol: {entry_price, shares, entry_date, stop_loss}}
        self.trade_history = []

    def can_open_position(self, symbol):
        """
        新規ポジションを開くことができるかチェック

        Returns:
            bool: 開可能=True, 開不可=False
        """
        # 既にポジションがあれば不可
        if symbol in self.positions:
            return False

        # 資金が不足していれば不可
        if self.capital < self.max_position_size:
            return False

        return True

    def open_position(self, symbol, entry_price, stop_loss_ratio=0.05):
        """
        新規ポジションを開く

        Args:
            symbol (str): 銘柄コード
            entry_price (float): 買値
            stop_loss_ratio (float): 損切り幅(デフォルト:5%)
        """
        if not self.can_open_position(symbol):
            logging.warning(f"ポジション開設不可: {symbol}")
            return False

        shares = int(self.max_position_size / entry_price)

        self.positions[symbol] = {
            'entry_price': entry_price,
            'shares': shares,
            'entry_date': datetime.now().isoformat(),
            'stop_loss': entry_price * (1 - stop_loss_ratio),
        }

        self.capital -= (shares * entry_price)

        logging.info(f"ポジション開設: {symbol} × {shares}株 @ ¥{entry_price:,.0f}")

        return True

    def close_position(self, symbol, exit_price):
        """
        ポジションをクローズ

        Args:
            symbol (str): 銘柄コード
            exit_price (float): 売値
        """
        if symbol not in self.positions:
            logging.warning(f"ポジションなし: {symbol}")
            return False

        position = self.positions[symbol]

        revenue = position['shares'] * exit_price
        cost = position['shares'] * position['entry_price']
        profit = revenue - cost
        profit_rate = (exit_price - position['entry_price']) / position['entry_price']

        # 資本を回復
        self.capital += revenue

        # 取引履歴を記録
        self.trade_history.append({
            'symbol': symbol,
            'entry_price': position['entry_price'],
            'exit_price': exit_price,
            'shares': position['shares'],
            'profit': profit,
            'profit_rate': profit_rate,
            'duration_days': (datetime.now() - datetime.fromisoformat(position['entry_date'])).days,
        })

        # ポジションを削除
        del self.positions[symbol]

        logging.info(f"ポジションクローズ: {symbol} @ ¥{exit_price:,.0f} / "
                    f"利益 ¥{profit:,.0f} ({profit_rate*100:+.2f}%)")

        return True

    def check_stop_loss(self, symbol, current_price):
        """
        損切り判定

        Args:
            symbol (str): 銘柄コード
            current_price (float): 現在価格

        Returns:
            bool: 損切り対象=True, 対象外=False
        """
        if symbol not in self.positions:
            return False

        position = self.positions[symbol]

        if current_price <= position['stop_loss']:
            logging.warning(f"損切り判定: {symbol} @ ¥{current_price:,.0f}")
            return True

        return False

    def display_portfolio(self):
        """ポートフォリオ状況を表示"""
        print(f"\n" + "="*70)
        print("ポートフォリオ")
        print("="*70)

        print(f"\n【資金状況】")
        print(f"初期資金: ¥{self.capital + sum(p['shares'] * p['entry_price'] for p in self.positions.values()):,.0f}")
        print(f"現金: ¥{self.capital:,.0f}")
        print(f"保有資産: ¥{sum(p['shares'] * p['entry_price'] for p in self.positions.values()):,.0f}")

        if self.positions:
            print(f"\n【保有ポジション】")
            for symbol, position in self.positions.items():
                print(f"{symbol}")
                print(f"  買値: ¥{position['entry_price']:,.0f}")
                print(f"  株数: {position['shares']}")
                print(f"  損切り: ¥{position['stop_loss']:,.0f}")
                print(f"  保有期間: {(datetime.now() - datetime.fromisoformat(position['entry_date'])).days}日")

        if self.trade_history:
            print(f"\n【取引履歴(直近5件)】")
            for trade in self.trade_history[-5:]:
                print(f"{trade['symbol']}: {trade['profit_rate']*100:+.2f}% / ¥{trade['profit']:+,.0f}")

# ==============================
# 拡張型ロボット(ポジション管理付き)
# ==============================
class AdvancedTradingRobot(TradingRobot):
    """ポートフォリオ管理機能を備えた発展版ロボット"""

    def __init__(self, initial_capital=1000000):
        """初期化"""
        super().__init__()
        self.portfolio = PortfolioManager(initial_capital)

    def process_signals(self):
        """シグナルに基づいてポジション管理"""
        for signal in self.signals:
            symbol = signal['symbol']
            current_price = signal['current_price']
            signal_type = signal['signal_type']

            # 損切りチェック
            if self.portfolio.check_stop_loss(symbol, current_price):
                self.portfolio.close_position(symbol, current_price)
                continue

            # 買いシグナル
            if signal_type == 'BUY':
                if self.portfolio.can_open_position(symbol):
                    self.portfolio.open_position(symbol, current_price)

            # 売りシグナル
            elif signal_type == 'SELL':
                if symbol in self.portfolio.positions:
                    self.portfolio.close_position(symbol, current_price)

    def run(self):
        """ロボット実行"""
        try:
            self.initialize()
            self.run_analysis()
            self.process_signals()
            self.send_notifications()
            self.portfolio.display_portfolio()

        except Exception as e:
            logging.error(f"実行エラー: {e}")

よくあるエラーと対処法

タスクスケジューラが指定時刻に実行されません

原因:

• Pythonパスが誤っている
• スクリプトの権限設定に問題がある
• PCがスリープ状態になっている

対処法:

• where python でPythonパスを確認
• タスクスケジューラの詳細設定で「最後に正常に実行された時刻」を確認
• PCの電源設定を「スリープなし」に変更

LINE通知が送信されません

原因:

• LINE_TOKENが間違っている
• トークンの有効期限が切れている
• インターネット接続が切れている

対処法:

• https://notify-bot.line.me/my/ で新しいトークンを生成
• ping google.com でネットワーク接続を確認

ロボットが銘柄を見つけられません

原因:

• 銘柄コードが誤っている
• yfinanceのサービスが一時的に停止している

対処法:

• 銘柄コードの確認:yf.Ticker("7203.T").info を試す
• 別の銘柄で試す:問題が広範囲か確認

まとめ

本記事では、個人投資家が実装可能な「準自動化ロボット」の完全なシステムを解説しました。

要点を整理します。

• SBI証券などの国内証券会社は、個人向けの自由な発注APIを提供していない
• この制約のもとで、yfinanceを使ったデータ取得と自動通知は完全に個人で実装可能
• LINE Notifyを使うことで、取引機会を即座に通知できる
• Windowsのタスクスケジューラで、毎営業日の指定時刻に自動実行が実現できる
• ポジション・リスク管理機能を追加することで、本格的な運用が可能
• 重要な決定(実際の注文実行)は「人間の手」で行うため、セキュリティと規制準拠が確保される

次のステップとしては、本記事のコードを自分の環境で実行し、数週間のデモ運用を行うことをお勧めします。その後、実運用資金を投入してください。

ロボットは、あなたの「規則性」と「判断力」を拡張するツールです。その使い方次第で、投資の質は大きく向上します。