※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。
「株式投資で資産を増やしたい」という願いは誰もが持っていますが、その実現方法は人によって大きく異なります。ロボアドバイザーに任せるのか、自分で運用するのか、それとも両者を組み合わせるのか。本シリーズを通じて、個人投資家が「自分専用のアルゴリズム」を構築できることを示してきました。
しかし、アルゴリズムを「作る」ことと、それを「長期間安定的に運用する」ことは別問題です。バックテストで素晴らしい成績が出ても、実運用で同じ成果が得られるとは限りません。また、運用開始から数年経つと、市場環境の変化に対応して戦略を調整する必要が生じます。
本記事では、単なる「アルゴリズムの構築」ではなく、「自動化されたシステムで資産を長期運用する」という全体的なフレームワークを提示します。データ分析から売買シグナル生成、リバランス管理、パフォーマンス追跡、そして戦略の動的調整まで、実務的な資産運用システムの完全な設計図を示します。
目次
長期資産運用において重要な5つの原則
Pythonでアルゴリズムを開発する前に、長期資産運用の本質的な原則を理解することが重要です。
原則1:「複利」と「時間」が最強の武器
投資において最も強力なのは、手数料0.1%の効率化ではなく、時間です。以下のシミュレーションをご覧ください。
初期投資:100万円
年間リターン:5%(税引き前)
年数 | 手数料0% | 手数料1% | 差分
-----|---------|---------|-----
10年 | 163万 | 145万 | -18万
20年 | 265万 | 210万 | -55万
30年 | 432万 | 305万 | -127万
40年 | 704万 | 442万 | -262万この表から分かる通り、時間が長くなるほど、わずかな手数料が複利で蓄積され、莫大な機会損失を生み出します。個人でアルゴリズムを構築するメリットは、まさにこの手数料ゼロを30年~40年という長期間維持することにあります。
原則2:「勝つ」のではなく「負けない」ことが重要
多くの投資家が誤解していることが「勝率の高さ」です。実際には、以下のような現実があります。
| 条件 | 期待値 |
|---|---|
| 勝率50%、勝つ時の利益+1%、負ける時の損失-1% | 0%(コストで赤字) |
| 勝率50%、勝つ時の利益+1.5%、負ける時の損失-1% | +0.25%(取引コスト控除後) |
| 勝率45%、勝つ時の利益+1%、負ける時の損失-0.5% | +0.20%(取引コスト控除後) |
重要なのは「どれだけ負けを小さくするか」です。損失を0.5%に抑えれば、勝率が50%でなくても、期待値は正になります。
原則3:「分散」は無料のリスク削減
1つの銘柄に集中投資することのリスクは計り知れません。一方、複数銘柄に分散することで、同じリターン期待値でリスクを削減できます。これは「無料」です。
| ポートフォリオ | 期待リターン | ボラティリティ | シャープレシオ |
|---|---|---|---|
| 単一銘柄 | 5% | 25% | 0.20 |
| 10銘柄均等配分 | 5% | 12% | 0.42 |
| 20銘柄均等配分 | 5% | 9% | 0.56 |
| 業種別+資産別 | 5% | 7% | 0.71 |
原則4:「リバランス」が複利を最大化させる
ポートフォリオを構築した後は、定期的なリバランスが重要です。これは「売買」ではなく、資産配分を目標比率に戻すだけですが、長期的には大きなリターン向上につながります。
原則5:「感情」を排除することが最大の敵
多くの個人投資家が負ける理由は「アルゴリズムが悪い」のではなく、「感情的な判断が介入する」からです。アルゴリズムを自動化することの最大のメリットは、この感情排除にあります。
【コピペOK】長期資産運用システムの完全実装
では、以上の原則に基づいた「長期資産運用システム」を実装します。このシステムは、複数銘柄の監視、売買シグナル生成、自動リバランス、パフォーマンス追跡までを統合しています。
import yfinance as yf
import pandas as pd
import numpy as np
import json
import logging
from datetime import datetime, timedelta
from pathlib import Path
import requests
# ==============================
# ロギング設定
# ==============================
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s] %(levelname)s: %(message)s',
handlers=[
logging.FileHandler('asset_management.log'),
logging.StreamHandler()
]
)
# ==============================
# システム設定
# ==============================
class AssetManagementConfig:
"""資産運用システム設定"""
# ポートフォリオ構成(目標配分)
PORTFOLIO = {
'JP_STOCKS': {
'symbols': ['7203.T', '7201.T', '8058.T', '9984.T', '6758.T'],
'target_weight': 0.40,
'rebalance_threshold': 0.05,
},
'FOREIGN_STOCKS': {
'symbols': ['VTI', 'EFA', 'SCHF'],
'target_weight': 0.40,
'rebalance_threshold': 0.05,
},
'BONDS': {
'symbols': ['BND', 'AGG'],
'target_weight': 0.15,
'rebalance_threshold': 0.05,
},
'CASH': {
'target_weight': 0.05,
},
}
# 運用パラメータ
INITIAL_CAPITAL = 1000000 # 初期資金100万円
MONTHLY_CONTRIBUTION = 50000 # 月次積立額
# テクニカル指標パラメータ
MA_SHORT = 20
MA_LONG = 50
RSI_PERIOD = 14
# リバランス設定
REBALANCE_FREQUENCY = 30 # 日数(30日ごと)
REBALANCE_DAY_OF_MONTH = 1 # 月初にリバランス
# LINE通知設定
LINE_TOKEN = "YOUR_LINE_NOTIFY_TOKEN"
LINE_API_URL = "https://notify-api.line.me/api/notify"
# ファイル設定
STATE_FILE = "portfolio_state.json"
PERFORMANCE_LOG = "performance_log.csv"
# ==============================
# ポートフォリオ状態管理
# ==============================
class PortfolioState:
"""ポートフォリオの状態を保存・管理"""
def __init__(self, config_file=AssetManagementConfig.STATE_FILE):
"""初期化"""
self.config_file = config_file
self.state = self._load_state()
def _load_state(self):
"""状態ファイルから状態を読み込む"""
if Path(self.config_file).exists():
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
logging.error(f"状態読み込みエラー: {e}")
# デフォルト状態
return {
'created_at': datetime.now().isoformat(),
'last_rebalance': None,
'total_invested': AssetManagementConfig.INITIAL_CAPITAL,
'positions': {},
'cash': AssetManagementConfig.INITIAL_CAPITAL,
'performance_history': [],
}
def save(self):
"""状態をファイルに保存"""
try:
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(self.state, f, ensure_ascii=False, indent=2, default=str)
logging.info("状態を保存しました")
except Exception as e:
logging.error(f"状態保存エラー: {e}")
def update_position(self, symbol, price, shares, asset_class):
"""ポジションを更新"""
if symbol not in self.state['positions']:
self.state['positions'][symbol] = {}
self.state['positions'][symbol] = {
'price': price,
'shares': shares,
'value': price * shares,
'asset_class': asset_class,
'last_updated': datetime.now().isoformat(),
}
def get_total_value(self):
"""ポートフォリオ総価値を計算"""
position_value = sum(p['value'] for p in self.state['positions'].values())
return position_value + self.state['cash']
def get_weights(self):
"""各資産クラスの現在のウェイトを計算"""
total_value = self.get_total_value()
weights = {}
for symbol, position in self.state['positions'].items():
asset_class = position['asset_class']
if asset_class not in weights:
weights[asset_class] = 0
weights[asset_class] += position['value'] / total_value
weights['CASH'] = self.state['cash'] / total_value
return weights
# ==============================
# 銘柄分析エンジン
# ==============================
class StockAnalyzer:
"""個別銘柄の分析"""
def __init__(self, symbol):
"""初期化"""
self.symbol = symbol
self.df = None
def fetch_data(self, period='1y'):
"""データを取得"""
try:
logging.info(f"データ取得: {self.symbol}")
ticker = yf.Ticker(self.symbol)
self.df = ticker.history(period=period)
if self.df.empty:
return False
self.df = self.df.fillna(method='ffill')
return True
except Exception as e:
logging.error(f"データ取得エラー({self.symbol}): {e}")
return False
def calculate_indicators(self):
"""テクニカル指標を計算"""
if self.df is None:
return False
try:
# 移動平均線
self.df['MA_SHORT'] = self.df['Close'].rolling(
window=AssetManagementConfig.MA_SHORT).mean()
self.df['MA_LONG'] = self.df['Close'].rolling(
window=AssetManagementConfig.MA_LONG).mean()
# RSI
delta = self.df['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(
window=AssetManagementConfig.RSI_PERIOD).mean()
loss = (-delta.where(delta < 0, 0)).rolling(
window=AssetManagementConfig.RSI_PERIOD).mean()
rs = gain / loss
self.df['RSI'] = 100 - (100 / (1 + rs))
return True
except Exception as e:
logging.error(f"指標計算エラー({self.symbol}): {e}")
return False
def get_signal(self):
"""売買シグナルを取得"""
if self.df is None or len(self.df) < 2:
return 'HOLD'
current = self.df.iloc[-1]
previous = self.df.iloc[-2]
# データ不足チェック
if pd.isna(current['MA_SHORT']) or pd.isna(current['MA_LONG']):
return 'HOLD'
# ゴールデンクロス
if (previous['MA_SHORT'] <= previous['MA_LONG'] and
current['MA_SHORT'] > current['MA_LONG']):
return 'BUY'
# デッドクロス
if (previous['MA_SHORT'] >= previous['MA_LONG'] and
current['MA_SHORT'] < current['MA_LONG']):
return 'SELL'
return 'HOLD'
def get_current_price(self):
"""現在価格を取得"""
if self.df is None or self.df.empty:
return None
return self.df['Close'].iloc[-1]
# ==============================
# リバランス管理
# ==============================
class RebalanceManager:
"""ポートフォリオのリバランス管理"""
def __init__(self, portfolio_state):
"""初期化"""
self.state = portfolio_state
def should_rebalance(self):
"""リバランスが必要かチェック"""
# 前回のリバランス日時を確認
last_rebalance = self.state.state.get('last_rebalance')
if last_rebalance is None:
return True # 初回
last_rebalance_date = datetime.fromisoformat(last_rebalance).date()
today = datetime.now().date()
# 月初でリバランスするか、指定日数が経過したかチェック
if today.day == AssetManagementConfig.REBALANCE_DAY_OF_MONTH:
if (today - last_rebalance_date).days >= 1:
return True
return False
def calculate_rebalance_trades(self, current_prices):
"""
リバランスに必要な取引を計算
Returns:
dict: 各銘柄の売買指示
"""
total_value = self.state.get_total_value()
current_weights = self.state.get_weights()
target_weights = self._get_target_weights()
trades = {}
for asset_class, config in AssetManagementConfig.PORTFOLIO.items():
if asset_class == 'CASH':
continue
current_weight = current_weights.get(asset_class, 0)
target_weight = target_weights[asset_class]
deviation = abs(current_weight - target_weight)
# 閾値を超えた場合のみリバランス
if deviation > config['rebalance_threshold']:
logging.info(f"リバランス必要: {asset_class} "
f"(現在: {current_weight*100:.1f}% → 目標: {target_weight*100:.1f}%)")
trades[asset_class] = {
'current_weight': current_weight,
'target_weight': target_weight,
'target_value': total_value * target_weight,
}
return trades
def _get_target_weights(self):
"""目標ウェイトを計算"""
return {
asset_class: config['target_weight']
for asset_class, config in AssetManagementConfig.PORTFOLIO.items()
if asset_class != 'CASH'
}
# ==============================
# パフォーマンス追跡
# ==============================
class PerformanceTracker:
"""運用パフォーマンスを追跡"""
def __init__(self, log_file=AssetManagementConfig.PERFORMANCE_LOG):
"""初期化"""
self.log_file = log_file
self.load_history()
def load_history(self):
"""履歴を読み込む"""
if Path(self.log_file).exists():
self.history = pd.read_csv(self.log_file)
else:
self.history = pd.DataFrame()
def record_performance(self, portfolio_state):
"""パフォーマンスを記録"""
total_value = portfolio_state.get_total_value()
total_invested = portfolio_state.state['total_invested']
gain = total_value - total_invested
gain_rate = gain / total_invested if total_invested > 0 else 0
record = {
'date': datetime.now().isoformat(),
'total_value': total_value,
'total_invested': total_invested,
'gain': gain,
'gain_rate': gain_rate,
}
# CSVに追記
record_df = pd.DataFrame([record])
if Path(self.log_file).exists():
record_df.to_csv(self.log_file, mode='a', header=False, index=False)
else:
record_df.to_csv(self.log_file, index=False)
logging.info(f"パフォーマンス記録: "
f"資産額¥{total_value:,.0f}, 利益¥{gain:,.0f} ({gain_rate*100:+.2f}%)")
def display_summary(self):
"""パフォーマンスサマリーを表示"""
if self.history.empty:
return
latest = self.history.iloc[-1]
print(f"\n" + "="*70)
print("パフォーマンスサマリー")
print("="*70)
print(f"総資産額: ¥{latest['total_value']:,.0f}")
print(f"累計投資額: ¥{latest['total_invested']:,.0f}")
print(f"含み益: ¥{latest['gain']:,.0f}")
print(f"運用利回り: {latest['gain_rate']*100:+.2f}%")
# 期間リターンを計算
if len(self.history) > 1:
first_value = self.history.iloc[0]['total_value']
last_value = self.history.iloc[-1]['total_value']
period_return = (last_value - first_value) / first_value
print(f"期間リターン: {period_return*100:+.2f}%")
# ==============================
# 通知システム
# ==============================
class NotificationManager:
"""LINE Notifyを使った通知"""
@staticmethod
def send_notification(message, token=AssetManagementConfig.LINE_TOKEN):
"""通知を送信"""
if token == "YOUR_LINE_NOTIFY_TOKEN":
logging.warning("LINE_TOKEN が設定されていません")
return False
try:
headers = {"Authorization": f"Bearer {token}"}
data = {"message": message}
response = requests.post(
AssetManagementConfig.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
# ==============================
# メインシステム
# ==============================
class AssetManagementSystem:
"""統合資産運用システム"""
def __init__(self):
"""初期化"""
self.state = PortfolioState()
self.rebalancer = RebalanceManager(self.state)
self.tracker = PerformanceTracker()
def run_daily_analysis(self):
"""日次分析を実行"""
logging.info("="*70)
logging.info("日次分析開始")
logging.info("="*70)
all_symbols = []
for asset_class, config in AssetManagementConfig.PORTFOLIO.items():
if 'symbols' in config:
all_symbols.extend(config['symbols'])
signals = []
for symbol in all_symbols:
analyzer = StockAnalyzer(symbol)
if not analyzer.fetch_data():
continue
if not analyzer.calculate_indicators():
continue
signal = analyzer.get_signal()
price = analyzer.get_current_price()
if signal != 'HOLD':
signals.append({
'symbol': symbol,
'signal': signal,
'price': price,
})
logging.info(f"シグナル検出: {symbol} - {signal} @ ¥{price:,.0f}")
return signals
def check_rebalance(self):
"""リバランスが必要か確認"""
if self.rebalancer.should_rebalance():
logging.info("リバランス実行")
# リバランス処理はここに実装
self.state.state['last_rebalance'] = datetime.now().isoformat()
self.state.save()
return True
return False
def record_performance(self):
"""パフォーマンスを記録"""
self.tracker.record_performance(self.state)
def send_daily_report(self, signals):
"""日次レポートを送信"""
message = f"\n【資産運用システム 日次レポート】\n"
message += f"実行時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"
# ポートフォリオ情報
total_value = self.state.get_total_value()
total_invested = self.state.state['total_invested']
gain = total_value - total_invested
message += f"【ポートフォリオ】\n"
message += f"総資産額: ¥{total_value:,.0f}\n"
message += f"含み益: ¥{gain:,.0f} ({gain/total_invested*100:+.2f}%)\n\n"
# シグナル情報
if signals:
message += f"【シグナル】\n"
for s in signals:
message += f"{s['symbol']}: {s['signal']} @ ¥{s['price']:,.0f}\n"
else:
message += "【シグナル】特にシグナルはありません\n"
message += "\n※このメッセージは自動で生成されています。"
NotificationManager.send_notification(message)
def run(self):
"""システムを実行"""
try:
# 日次分析を実行
signals = self.run_daily_analysis()
# リバランス確認
self.check_rebalance()
# パフォーマンス記録
self.record_performance()
# レポート送信
self.send_daily_report(signals)
# パフォーマンス表示
self.tracker.display_summary()
except Exception as e:
logging.error(f"実行エラー: {e}")
finally:
self.state.save()
logging.info("実行完了\n")
# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
system = AssetManagementSystem()
system.run()このシステムの特徴:
- 完全な自動化: yfinanceを使ったデータ取得から通知まで全て自動
- 複数銘柄対応: 日本株から外国株まで複数資産クラスを統合管理
- 自動リバランス: 定期的に目標配分に戻す
- パフォーマンス追跡: 運用成績を継続的に記録
- LINE通知: 重要なイベントを即座に通知
【コピペOK】月次積立を含むシステム拡張
長期資産運用では、定期的な積立も重要です。以下は月次積立機能を追加したコードです。
class ContributionManager:
"""定期積立管理"""
def __init__(self, monthly_amount=AssetManagementConfig.MONTHLY_CONTRIBUTION):
"""初期化"""
self.monthly_amount = monthly_amount
self.contribution_log = []
def should_contribute(self, last_contribution_date):
"""月次積立の実施判定"""
today = datetime.now().date()
if last_contribution_date is None:
return True
last_date = datetime.fromisoformat(last_contribution_date).date()
# 月が変わったか確認
if today.month != last_date.month or today.year != last_date.year:
return True
return False
def execute_contribution(self, portfolio_state, allocation):
"""
月次積立を実行
Args:
portfolio_state: ポートフォリオ状態
allocation (dict): 各資産クラスの配分比率
"""
logging.info(f"月次積立実行: ¥{self.monthly_amount:,.0f}")
# 現在のキャッシュに追加
portfolio_state.state['cash'] += self.monthly_amount
portfolio_state.state['total_invested'] += self.monthly_amount
# 配分に従って各資産クラスに割り当て
for asset_class, ratio in allocation.items():
contribution = self.monthly_amount * ratio
logging.info(f" {asset_class}: ¥{contribution:,.0f}")
self.contribution_log.append({
'date': datetime.now().isoformat(),
'amount': self.monthly_amount,
'allocation': allocation,
})
portfolio_state.state['last_contribution'] = datetime.now().isoformat()長期運用における動的な戦略調整
市場環境の変化に対応して、戦略を動的に調整することが重要です。
class StrategyAdapter:
"""市場環境に応じた戦略の動的調整"""
@staticmethod
def adjust_allocation_by_market_regime(market_indicators):
"""
市場環境に応じてアセットアロケーションを調整
Args:
market_indicators (dict): 市場指標(VIX、金利など)
Returns:
dict: 調整後の配分
"""
vix = market_indicators.get('vix', 15) # VIX指数(デフォルト値15)
if vix > 30:
# 高ボラティリティ環境:防御的な配分
return {
'JP_STOCKS': 0.25,
'FOREIGN_STOCKS': 0.25,
'BONDS': 0.40,
'CASH': 0.10,
}
elif vix > 20:
# 中程度のボラティリティ:バランス配分
return {
'JP_STOCKS': 0.35,
'FOREIGN_STOCKS': 0.35,
'BONDS': 0.20,
'CASH': 0.10,
}
else:
# 低ボラティリティ環境:積極的な配分
return {
'JP_STOCKS': 0.40,
'FOREIGN_STOCKS': 0.40,
'BONDS': 0.15,
'CASH': 0.05,
}よくあるエラーと対処法
「データが取得できません」というエラー
原因: yfinanceのサービス一時停止、またはネットワークエラー
対処法:
# リトライロジックを実装
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_with_retry(self, period='1y'):
"""リトライ機能付きデータ取得"""
return yf.Ticker(self.symbol).history(period=period)「LINE通知が送信されません」
原因: LINE_TOKEN が設定されていない、または無効期限切れ
対処法:
環境変数から安全に取得:
import os
from dotenv import load_dotenv
load_dotenv()
LINE_TOKEN = os.getenv('LINE_NOTIFY_TOKEN')リバランス計算で「配分の合計が100%にならない」
原因: 浮動小数点数の丸め誤差
対処法:
# 配分の正規化
def normalize_allocation(allocation):
"""配分の合計を100%に正規化"""
total = sum(allocation.values())
return {k: v/total for k, v in allocation.items()}まとめ
本記事では、Pythonを使った「長期資産運用システム」の完全な実装を提示しました。
要点を整理します。
- 複利と時間が最強: 40年で手数料1%の差が262万円の機会損失を生む
- 「勝つ」のではなく「負けない」: リスク管理が勝率より重要
- 分散は無料のリスク削減: 複数資産クラス、複数銘柄への分散が必須
- 定期的なリバランス: 目標配分を維持することで複利を最大化
- 感情排除が最大の武器: アルゴリズムの自動化が勝つ鍵
- 市場環境への適応: VIXなどの指標で戦略を動的に調整
本シリーズを通じて、個人投資家が「自分専用のアルゴリズム」を構築できることを示してきました。ロボアドバイザーの手数料に悩む必要はありません。Pythonとyfinanceを使えば、十分な機能を持つシステムが、実装可能です。
次のステップ:
- 本記事のコードで小額(月1万円程度)のデモ運用を開始
- 3~6ヶ月間、システムが正常に動作することを確認
- 本格的な資金投入を開始
- 年1回程度、戦略を見直す
30年、40年という長期間、あなたの資産を確実に育てていくシステムを手に入れてください。
