Coverage for src / monte_neo / core / portfolio / manager.py: 72%
116 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-28 16:27 +0200
1"""Portfolio management module.
3Handles multiple indicators, risk allocation, and correlation analysis.
4"""
6from __future__ import annotations
8from dataclasses import dataclass
9from typing import Any
11import numpy as np
12import pandas as pd
14from monte_neo.utils.logger import get_logger
16logger = get_logger(__name__)
18@dataclass
19class PortfolioAsset:
20 """Represents an indicator/strategy in the portfolio."""
21 id: str
22 indicator_path: str
23 symbol: str
24 weight: float = 1.0
25 active: bool = True
26 equity_curve: np.ndarray | None = None
28class PortfolioManager:
29 """Manages a collection of indicators as a single trading portfolio."""
31 def __init__(self, initial_capital: float = 10000.0):
32 self.assets: list[PortfolioAsset] = []
33 self.initial_capital = initial_capital
34 self.correlation_matrix: pd.DataFrame | None = None
36 def add_asset(self, asset: PortfolioAsset):
37 """Add an asset to the portfolio."""
38 self.assets.append(asset)
39 logger.info(f"Added asset {asset.id} for {asset.symbol} to portfolio")
41 def calculate_correlations(self, returns_dict: dict[str, pd.Series]) -> pd.DataFrame:
42 """Calculate correlation matrix between assets based on their returns."""
43 df = pd.DataFrame(returns_dict)
44 self.correlation_matrix = df.corr()
45 return self.correlation_matrix
47 def cluster_assets(self, returns_dict: dict[str, pd.Series]) -> dict[int, list[str]]:
48 """Cluster assets based on correlation to find redundant strategies."""
49 try:
50 from scipy.cluster.hierarchy import fcluster, linkage
51 from scipy.spatial.distance import squareform
52 except ImportError:
53 logger.warning("scipy not installed, skipping clustering")
54 return {0: list(returns_dict.keys())}
56 if len(returns_dict) < 2:
57 return {0: list(returns_dict.keys())}
59 corr = self.calculate_correlations(returns_dict)
60 # Convert correlation to distance (1 - corr)
61 dist = 1 - corr.fillna(0)
63 # Ensure symmetry and 0 diagonal for squareform
64 dist = (dist + dist.T) / 2
65 np.fill_diagonal(dist.values, 0)
67 # Hierarchical clustering
68 try:
69 from scipy.cluster.hierarchy import fcluster, linkage
70 from scipy.spatial.distance import squareform
72 link = linkage(squareform(dist), method='ward')
73 clusters = fcluster(link, t=0.5, criterion='distance')
75 cluster_map = {}
76 for asset_id, cluster_id in zip(corr.index, clusters):
77 if cluster_id not in cluster_map:
78 cluster_map[int(cluster_id)] = []
79 cluster_map[int(cluster_id)].append(asset_id)
81 return cluster_map
82 except Exception as e:
83 logger.error(f"Clustering failed: {e}")
84 return {0: list(returns_dict.keys())}
86 def optimize_weights(self, method: str = "risk_parity", volatilities: list[float] | None = None) -> dict[str, float]:
87 """Optimize asset weights based on selected method."""
88 if not self.assets:
89 return {}
91 if method == "equal":
92 weight = 1.0 / len(self.assets)
93 for asset in self.assets:
94 asset.weight = weight
96 elif method == "risk_parity" and volatilities:
97 from monte_neo.core.portfolio.risk import calculate_risk_parity_weights
98 weights = calculate_risk_parity_weights(volatilities)
99 for asset, weight in zip(self.assets, weights):
100 asset.weight = weight
102 elif method == "kelly":
103 # Advanced Kelly Criterion for multiple assets
104 # f* = (p/a - q/b) where p is win probability, q is loss probability, a is fractional loss, b is fractional gain
105 for asset in self.assets:
106 # Mock values for now, should be calculated from equity_curve
107 win_rate = 0.55
108 win_loss_ratio = 1.2
109 kelly_f = win_rate - (1 - win_rate) / win_loss_ratio
110 asset.weight = max(0, kelly_f * 0.5) # Half-Kelly for safety
112 return {a.id: a.weight for a in self.assets}
114 def run_portfolio_monte_carlo(self, iterations: int = 1000) -> dict[str, Any]:
115 """Runs Monte Carlo simulation on the combined portfolio equity."""
116 combined_equity = self.get_combined_equity()
117 if len(combined_equity) == 0:
118 return {}
120 from monte_neo.monte_carlo.engine import MonteCarloEngine
121 mc_engine = MonteCarloEngine()
123 # We simulate variations of the combined returns
124 returns = np.diff(combined_equity) / combined_equity[:-1]
126 results = []
127 for _ in range(iterations):
128 # Shuffle returns to simulate different sequences
129 shuffled_returns = np.random.permutation(returns)
130 sim_equity = np.cumprod(1 + shuffled_returns)
132 # Calculate drawdown
133 peak = np.maximum.accumulate(sim_equity)
134 drawdown = (peak - sim_equity) / peak
135 results.append({
136 "final_return": sim_equity[-1] - 1,
137 "max_drawdown": np.max(drawdown)
138 })
140 return {
141 "avg_return": np.mean([r["final_return"] for r in results]),
142 "max_drawdown_95th": np.percentile([r["max_drawdown"] for r in results], 95),
143 "var_95": np.percentile([r["final_return"] for r in results], 5)
144 }
146 def get_combined_equity(self) -> np.ndarray:
147 """Calculate the combined equity curve of the portfolio."""
148 if not self.assets:
149 return np.array([])
151 active_assets = [a for a in self.assets if a.active and a.equity_curve is not None]
152 if not active_assets:
153 return np.array([])
155 # Sum weighted equity curves
156 # Assuming all curves are same length for simplicity
157 min_len = min(len(a.equity_curve) for a in active_assets)
158 combined = np.zeros(min_len)
160 for asset in active_assets:
161 # Type narrowing for Mypy
162 equity = asset.equity_curve
163 if equity is not None:
164 combined += equity[:min_len] * asset.weight
166 return combined
168 def get_portfolio_summary(self) -> dict[str, Any]:
169 """Get high-level portfolio statistics."""
170 summary = {
171 "total_assets": len(self.assets),
172 "active_assets": sum(1 for a in self.assets if a.active),
173 "initial_capital": self.initial_capital,
174 "weights": {a.id: a.weight for a in self.assets},
175 "clusters": {}
176 }
178 # Add clustering info if we have enough assets
179 if len(self.assets) >= 2:
180 returns = {a.id: pd.Series(a.equity_curve).pct_change().dropna()
181 for a in self.assets if a.equity_curve is not None}
182 if returns:
183 summary["clusters"] = self.cluster_assets(returns)
185 return summary
187 def auto_rebalance(self, method: str = "risk_parity") -> dict[str, float]:
188 """Automatically rebalance the portfolio based on latest metrics."""
189 volatilities = []
190 for asset in self.assets:
191 if asset.equity_curve is not None:
192 returns = pd.Series(asset.equity_curve).pct_change().dropna()
193 volatilities.append(returns.std())
194 else:
195 volatilities.append(1.0) # Default
197 return self.optimize_weights(method=method, volatilities=volatilities)