# 投資ポートフォリオをシステムとして設計
class InvestmentPortfolio:
def __init__(self):
self.assets = {}
self.target_allocation = {}
self.risk_tolerance = 0.0
self.rebalance_threshold = 0.05  # 5%の乖離で再バランス
def design_portfolio(self, risk_profile, investment_horizon):
        """リスクプロファイルと投資期間に基づくポートフォリオ設計"""
if risk_profile == 'conservative':
self.target_allocation = {
'domestic_stocks': 0.30,
'international_stocks': 0.20,
'bonds': 0.40,
'cash': 0.10
}
elif risk_profile == 'moderate':
self.target_allocation = {
'domestic_stocks': 0.40,
'international_stocks': 0.30,
'bonds': 0.25,
'cash': 0.05
}
elif risk_profile == 'aggressive':
self.target_allocation = {
'domestic_stocks': 0.50,
'international_stocks': 0.35,
'growth_stocks': 0.10,
'bonds': 0.05
}
return self.target_allocation
def calculate_rebalance_needs(self, current_values):
        """リバランス必要性の計算"""
total_value = sum(current_values.values())
current_allocation = {
asset: value / total_value 
for asset, value in current_values.items()
}
rebalance_actions = {}
for asset, target_ratio in self.target_allocation.items():
current_ratio = current_allocation.get(asset, 0)
deviation = abs(current_ratio - target_ratio)
if deviation > self.rebalance_threshold:
rebalance_actions[asset] = {
'current': current_ratio,
'target': target_ratio,
'action': 'buy' if current_ratio < target_ratio else 'sell',
'amount': abs(current_ratio - target_ratio) * total_value
}
return rebalance_actions

import pandas as pd
import numpy as np
import yfinance as yf
class StockAnalyzer:
def __init__(self):
self.fundamental_weights = {
'per': 0.25,
'pbr': 0.20,
'roe': 0.25,
'dividend_yield': 0.15,
'debt_ratio': 0.15
}
def calculate_investment_score(self, ticker):
        """投資スコア計算（エンジニア向け指標）"""
stock = yf.Ticker(ticker)
info = stock.info
# 基本指標取得
per = info.get('trailingPE', 0)
pbr = info.get('priceToBook', 0)
roe = info.get('returnOnEquity', 0)
dividend_yield = info.get('dividendYield', 0) or 0
debt_ratio = info.get('debtToEquity', 0)
# スコア正規化（0-100点）
scores = {
'per': self._normalize_per(per),
'pbr': self._normalize_pbr(pbr),
'roe': self._normalize_roe(roe),
'dividend_yield': self._normalize_dividend(dividend_yield),
'debt_ratio': self._normalize_debt(debt_ratio)
}
# 重み付き総合スコア
total_score = sum(
scores[metric] * weight 
for metric, weight in self.fundamental_weights.items()
)
return {
'total_score': total_score,
'individual_scores': scores,
'raw_data': {
'per': per,
'pbr': pbr,
'roe': roe,
'dividend_yield': dividend_yield,
'debt_ratio': debt_ratio
}
}
def _normalize_per(self, per):
        """PER正規化（低いほど良い）"""
if per <= 0:
return 0
elif per <= 15:
return 100
elif per <= 25:
return 80
elif per <= 35:
return 60
else:
return 40
def _normalize_roe(self, roe):
        """ROE正規化（高いほど良い）"""
if roe >= 0.20:
return 100
elif roe >= 0.15:
return 80
elif roe >= 0.10:
return 60
elif roe >= 0.05:
return 40
else:
return 20

class TechStockStrategy:
def __init__(self):
self.tech_sectors = {
'cloud_computing': ['AMZN', 'MSFT', 'GOOGL'],
'semiconductors': ['NVDA', 'AMD', 'TSM'],
'software': ['CRM', 'ADBE', 'NOW'],
'cybersecurity': ['CRWD', 'ZS', 'OKTA'],
'ai_ml': ['NVDA', 'GOOGL', 'MSFT']
}
self.evaluation_criteria = {
'technical_moat': 0.30,      # 技術的優位性
'market_position': 0.25,     # 市場ポジション
'growth_potential': 0.25,    # 成長性
'financial_health': 0.20     # 財務健全性
}
def evaluate_tech_company(self, ticker, sector):
        """技術企業の評価"""
# 技術的優位性の評価
technical_moat = self._assess_technical_moat(ticker, sector)
# 市場ポジションの評価
market_position = self._assess_market_position(ticker)
# 成長性の評価
growth_potential = self._assess_growth_potential(ticker)
# 財務健全性の評価
financial_health = self._assess_financial_health(ticker)
# 総合スコア計算
total_score = (
technical_moat * self.evaluation_criteria['technical_moat'] +
market_position * self.evaluation_criteria['market_position'] +
growth_potential * self.evaluation_criteria['growth_potential'] +
financial_health * self.evaluation_criteria['financial_health']
)
return {
'ticker': ticker,
'sector': sector,
'total_score': total_score,
'breakdown': {
'technical_moat': technical_moat,
'market_position': market_position,
'growth_potential': growth_potential,
'financial_health': financial_health
},
'recommendation': self._generate_recommendation(total_score)
}
def _assess_technical_moat(self, ticker, sector):
        """技術的優位性評価"""
# 特許数、R&D投資比率、技術革新性などを評価
# 実装では外部APIやデータソースを使用
moat_scores = {
'NVDA': 95,  # GPU技術での圧倒的優位性
'GOOGL': 90, # 検索・AI技術
'MSFT': 85,  # クラウド・OS技術
'AMZN': 80,  # クラウドインフラ
}
return moat_scores.get(ticker, 70)  # デフォルト値

import schedule
import time
from datetime import datetime, timedelta
class AutomatedInvestmentSystem:
def __init__(self, broker_api, portfolio_manager):
self.broker_api = broker_api
self.portfolio_manager = portfolio_manager
self.investment_rules = {}
self.safety_checks = True
def setup_dollar_cost_averaging(self, amount, frequency, assets):
        """ドルコスト平均法の自動化設定"""
def execute_dca():
try:
# 安全性チェック
if self.safety_checks:
if not self._perform_safety_checks():
self.log_event("Safety check failed, skipping DCA")
return
# 各資産への投資実行
for asset, allocation in assets.items():
investment_amount = amount * allocation
# 市場時間チェック
if self._is_market_open():
order_result = self.broker_api.place_order(
symbol=asset,
quantity=investment_amount,
order_type='market'
)
self.log_event(f"DCA executed: {asset}, Amount: ${investment_amount}")
else:
self.log_event("Market closed, DCA postponed")
except Exception as e:
self.log_event(f"DCA execution failed: {str(e)}")
# スケジュール設定
if frequency == 'weekly':
schedule.every().monday.at("09:30").do(execute_dca)
elif frequency == 'monthly':
schedule.every().month.do(execute_dca)
return execute_dca
def setup_rebalancing(self, threshold=0.05):
        """自動リバランシング設定"""
def execute_rebalancing():
try:
current_positions = self.broker_api.get_positions()
rebalance_actions = self.portfolio_manager.calculate_rebalance_needs(
current_positions
)
if rebalance_actions:
for asset, action in rebalance_actions.items():
if action['action'] == 'buy':
self.broker_api.place_order(
symbol=asset,
quantity=action['amount'],
order_type='market'
)
elif action['action'] == 'sell':
self.broker_api.place_order(
symbol=asset,
quantity=-action['amount'],
order_type='market'
)
self.log_event("Portfolio rebalanced successfully")
else:
self.log_event("No rebalancing needed")
except Exception as e:
self.log_event(f"Rebalancing failed: {str(e)}")
# 月次リバランシング
schedule.every().month.do(execute_rebalancing)
return execute_rebalancing
def _perform_safety_checks(self):
        """安全性チェック"""
checks = [
self._check_market_volatility(),
self._check_account_balance(),
self._check_system_health()
]
return all(checks)
def run_system(self):
        """自動投資システム実行"""
self.log_event("Automated investment system started")
while True:
schedule.run_pending()
time.sleep(60)  # 1分間隔でチェック

import numpy as np
import matplotlib.pyplot as plt
class PortfolioSimulator:
def __init__(self):
self.simulation_years = 30
self.simulations = 10000
def monte_carlo_simulation(self, portfolio_returns, portfolio_volatility, 
initial_investment, annual_contribution):
        """モンテカルロシミュレーション実行"""
results = []
for _ in range(self.simulations):
portfolio_value = initial_investment
annual_values = [portfolio_value]
for year in range(self.simulation_years):
# 年間リターンをランダム生成（正規分布）
annual_return = np.random.normal(
portfolio_returns, 
portfolio_volatility
)
# ポートフォリオ価値更新
portfolio_value = (portfolio_value + annual_contribution) * (1 + annual_return)
annual_values.append(portfolio_value)
results.append(annual_values)
return np.array(results)
def analyze_simulation_results(self, simulation_results):
        """シミュレーション結果分析"""
final_values = simulation_results[:, -1]
analysis = {
'median_final_value': np.median(final_values),
'mean_final_value': np.mean(final_values),
'percentile_10': np.percentile(final_values, 10),
'percentile_25': np.percentile(final_values, 25),
'percentile_75': np.percentile(final_values, 75),
'percentile_90': np.percentile(final_values, 90),
'probability_of_loss': np.mean(final_values < 0) * 100,
'probability_of_doubling': np.mean(final_values > initial_investment * 2) * 100
}
return analysis
def visualize_simulation(self, simulation_results):
        """シミュレーション結果可視化"""
plt.figure(figsize=(12, 8))
# パーセンタイル範囲の表示
years = range(self.simulation_years + 1)
percentiles = [10, 25, 50, 75, 90]
for p in percentiles:
values = np.percentile(simulation_results, p, axis=0)
plt.plot(years, values, label=f'{p}th percentile')
plt.xlabel('Years')
plt.ylabel('Portfolio Value ($)')
plt.title('Portfolio Growth Simulation (Monte Carlo)')
plt.legend()
plt.grid(True)
plt.show()
return plt

class TaxOptimizationStrategy:
def __init__(self):
self.tax_brackets = {
2025: [
(0, 195000, 0.05),
(195000, 330000, 0.10),
(330000, 695000, 0.20),
(695000, 900000, 0.23),
(900000, 1800000, 0.33),
(1800000, 4000000, 0.40),
(4000000, float('inf'), 0.45)
]
}
def calculate_tax_efficient_allocation(self, income, investment_amount):
        """税効率的な投資配分計算"""
strategies = {
'nisa_general': {
'limit': 1200000,  # 年間120万円
'tax_benefit': 'tax_free_growth'
},
'nisa_tsumitate': {
'limit': 400000,   # 年間40万円
'tax_benefit': 'tax_free_growth'
},
'ideco': {
'limit': self._calculate_ideco_limit(income),
'tax_benefit': 'tax_deduction'
},
'corporate_pension': {
'limit': 276000,   # 年間27.6万円
'tax_benefit': 'tax_deduction'
}
}
allocation_plan = {}
remaining_amount = investment_amount
# 優先順位に基づく配分
priority_order = ['ideco', 'nisa_tsumitate', 'nisa_general', 'corporate_pension']
for strategy in priority_order:
if remaining_amount <= 0:
break
strategy_info = strategies[strategy]
allocatable_amount = min(remaining_amount, strategy_info['limit'])
if allocatable_amount > 0:
allocation_plan[strategy] = {
'amount': allocatable_amount,
'tax_benefit': strategy_info['tax_benefit'],
'tax_savings': self._calculate_tax_savings(
allocatable_amount, 
strategy_info['tax_benefit'], 
income
)
}
remaining_amount -= allocatable_amount
# 残額は特定口座へ
if remaining_amount > 0:
allocation_plan['taxable_account'] = {
'amount': remaining_amount,
'tax_benefit': 'none',
'tax_savings': 0
}
return allocation_plan
def _calculate_ideco_limit(self, income):
        """iDeCo拠出限度額計算"""
# 会社員の場合の基本限度額
base_limit = 276000  # 年間27.6万円
# 企業年金の有無により調整
# 実際の実装では詳細な条件分岐が必要
return base_limit
def optimize_capital_gains_tax(self, positions):
        """譲渡益税最適化"""
optimization_strategies = []
for position in positions:
if position['unrealized_gain'] > 0:
# 利益確定のタイミング最適化
if position['holding_period'] > 365:
# 長期保有による税率優遇（実際の日本では適用なし）
pass
# 損益通算の機会検討
loss_positions = [p for p in positions if p['unrealized_gain'] < 0]
if loss_positions:
optimization_strategies.append({
'action': 'tax_loss_harvesting',
'profit_position': position,
'loss_positions': loss_positions,
'potential_tax_savings': self._calculate_tax_loss_benefit(
position, loss_positions
)
})
return optimization_strategies

phase1_allocation = {
'emergency_fund': 0.30,      # 生活費6ヶ月分
'nisa_tsumitate': 0.40,      # インデックス投資
'ideco': 0.30                # 老後資金
}

phase2_allocation = {
'nisa_general': 0.35,        # 個別株・ETF
'ideco': 0.25,               # 老後資金
'tech_stocks': 0.25,         # 技術株投資
'international': 0.15        # 海外投資
}

phase3_allocation = {
'growth_stocks': 0.30,       # 成長株
'dividend_stocks': 0.25,     # 配当株
'international': 0.20,       # 海外分散
'alternative': 0.15,         # オルタナティブ投資
'cash': 0.10                 # 機会資金
}

class AnnualInvestmentSchedule:
def __init__(self):
self.monthly_tasks = {
1: ['年間投資計画策定', 'NISA枠リセット対応'],
3: ['決算発表チェック', 'ポートフォリオ見直し'],
6: ['中間決算チェック', 'リバランシング'],
9: ['第3四半期決算チェック'],
12: ['年末調整', '損益通算検討', '来年度計画']
}
def generate_investment_calendar(self, year):
        """年間投資カレンダー生成"""
calendar = {}
for month, tasks in self.monthly_tasks.items():
calendar[f'{year}-{month:02d}'] = {
'tasks': tasks,
'dca_execution': True,  # 毎月のドルコスト平均法
'performance_review': month % 3 == 0  # 四半期レビュー
}
return calendar