※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。
「自動売買ロボットを作れば、放置しておくだけで利益が出る」という誘いは、個人投資家が最も陥りやすい幻想です。実際のところ、個人向けの自動売買ロボットを構築することは、技術的には可能ですが、実運用には多くの落とし穴があります。
最大の問題は、SBI証券や楽天証券といった国内主流証券会社が、個人向けに「自由にPythonから発注できるAPI」を提供していないという現実です。非公式なライブラリは存在しますが、セキュリティリスクが高く、仕様変更時の対応負荷も大きいため、実務的ではありません。
一方、yfinanceを使ったデータ取得、売買シグナル生成、LINE自動通知までは、完全に個人で実装可能です。これを「準自動化システム」として運用することで、手動売買の効率化と規則性の向上が実現できます。本記事では、実装可能な範囲での「現実的なロボット構築」をゼロから解説し、将来的に証券会社APIが解放されたときに対応できる設計を提示します。
📘 外部参考:yfinance 公式GitHubリポジトリ / PyPIページ
個人向け自動売買の現状と現実的な選択肢
自動売買ロボットの構築を始める前に、その制約と現実を理解することが重要です。
SBI証券・楽天証券のAPI現状と限界
国内の主流証券会社では、以下の状況が続いています。
| 証券会社 | 公開API | 個人向け発注API | 現状 |
|---|---|---|---|
| SBI証券 | 一部公開 | 非公開 | 法人向け、認定パートナー向けのみ |
| 楽天証券 | 一部公開 | 非公開 | ラッパーAPIは非公式 |
| マネックス証券 | 一部公開 | 非公開 | データ取得のみ |
| Liquid by Quoine | 一部公開 | 公開(暗号資産向け) | 暗号資産は可能、株式不可 |
| オートレ(自動売買専門) | 非公開 | 専有システムのみ | 専用ツール内でのみ利用可 |
結論としては、日本の主流証券会社からは、個人投資家が自由にPythonで発注できるAPIは事実上利用不可です。
「準自動化システム」という現実的な選択肢
この制約のもとで、個人投資家が実装可能なのは以下のシステムです。
【準自動化システムの流れ】
(自動) yfinanceでデータ取得
↓
(自動) テクニカル指標計算・売買シグナル生成
↓
(自動) LINE / メール / Slack で通知
↓
(手動) 人間が通知を確認
↓
(手動) 証券会社のWebサイト or アプリで実際に注文
この仕組みであれば、以下のメリットが得られます。
- セキュリティ: 認証情報をPythonに保存しない(最安全)
- 安定性: 証券会社のサーバーに依存しない(通知だけ)
- 規制準拠: 金融庁の指導に抵触しない
- 検証容易: バックテストと実運用の乖離が小さい
📘 外部参考:Backtesting.py(公式ドキュメント) / Backtrader 公式
【コピペ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がメイン実行エンジン
📘 外部参考:RSI(Wikipedia 日本語) / Relative Strength Index(Investopedia)
📘 外部参考:移動平均(Wikipedia 日本語) / Moving Average(Investopedia)
Windowsのタスクスケジューラで自動実行する設定
Pythonスクリプトを毎日自動実行するには、Windowsのタスクスケジューラを使用します。
タスク登録手順
- スクリプト保存: 上記のコードを
trading_robot.pyとして保存 - タスクスケジューラを開く:
Win + R→taskschd.msc→ Enter
- 基本タスクを作成:
- 右ペインから「基本タスクを作成」
- タスク名:
Stock Trading Robot - 説明:
自動売買ロボット
- トリガー設定:
- 「毎日」を選択
- 時刻:
16:30(株式市場の引け後)
- 操作を設定:
- プログラム/スクリプト:
C:\Python311\python.exe(Pythonのパス) - 引数:
C:\Users\YourName\Desktop\trading_robot.py - 開始位置:
C:\Users\YourName\Desktop
- プログラム/スクリプト:
- 完了して保存
注意: 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のタスクスケジューラで、毎営業日の指定時刻に自動実行が実現できる
- ポジション・リスク管理機能を追加することで、本格的な運用が可能
- 重要な決定(実際の注文実行)は「人間の手」で行うため、セキュリティと規制準拠が確保される
次のステップとしては、本記事のコードを自分の環境で実行し、数週間のデモ運用を行うことをお勧めします。その後、実運用資金を投入してください。
ロボットは、あなたの「規則性」と「判断力」を拡張するツールです。その使い方次第で、投資の質は大きく向上します。

