blackopsrepl/portfolio-optimization-python
1
1"""
2Portfolio Optimization Domain Model
3
4This module defines the core domain entities for stock portfolio optimization:
5- StockSelection: A stock that can be selected for the portfolio (planning entity)
6- PortfolioOptimizationPlan: The complete portfolio optimization problem (planning solution)
7
8The model uses a Boolean selection approach:
9- Each stock has a `selected` field (True/False)
10- Selected stocks get equal weight (100% / number_selected)
11- This simplifies the optimization while still demonstrating constraint solving
12"""
13from solverforge_legacy.solver import SolverStatus
14from solverforge_legacy.solver.domain import (
15 planning_entity,
16 planning_solution,
17 PlanningId,
18 PlanningVariable,
19 PlanningEntityCollectionProperty,
20 ProblemFactCollectionProperty,
21 ProblemFactProperty,
22 ValueRangeProvider,
23 PlanningScore,
24)
25from solverforge_legacy.solver.score import HardSoftScore
26from typing import Annotated, List, Optional
27from dataclasses import dataclass, field
28from .json_serialization import JsonDomainBase
29from pydantic import Field
30
31
32@dataclass
33class SelectionValue:
34 """
35 Represents a possible selection state for a stock.
36
37 We use this wrapper class instead of raw bool because SolverForge
38 needs a reference type for the value range provider.
39 """
40 value: bool
41
42 def __hash__(self):
43 return hash(self.value)
44
45 def __eq__(self, other):
46 if isinstance(other, SelectionValue):
47 return self.value == other.value
48 return False
49
50
51# Pre-created selection values for the value range
52SELECTED = SelectionValue(True)
53NOT_SELECTED = SelectionValue(False)
54
55
56@dataclass
57class PortfolioConfig:
58 """
59 Configuration parameters for portfolio constraints.
60
61 This is a problem fact that constraints can join against to access
62 configurable threshold values.
63
64 Attributes:
65 target_count: Number of stocks to select (default 20)
66 max_per_sector: Maximum stocks per sector (default 5, which is 25% of 20)
67 unselected_penalty: Soft penalty per unselected stock (default 10000)
68 """
69 target_count: int = 20
70 max_per_sector: int = 5
71 unselected_penalty: int = 10000
72
73 def __hash__(self) -> int:
74 return hash((self.target_count, self.max_per_sector, self.unselected_penalty))
75
76 def __eq__(self, other: object) -> bool:
77 if isinstance(other, PortfolioConfig):
78 return (
79 self.target_count == other.target_count
80 and self.max_per_sector == other.max_per_sector
81 and self.unselected_penalty == other.unselected_penalty
82 )
83 return False
84
85
86@planning_entity
87@dataclass
88class StockSelection:
89 """
90 Represents a stock that can be included in the portfolio.
91
92 This is a planning entity - SolverForge decides whether to include
93 each stock by setting the `selection` field.
94
95 Attributes:
96 stock_id: Unique identifier (ticker symbol, e.g., "AAPL")
97 stock_name: Human-readable name (e.g., "Apple Inc.")
98 sector: Industry sector (e.g., "Technology", "Healthcare")
99 predicted_return: ML-predicted return as decimal (0.12 = 12%)
100 selection: Planning variable - SELECTED or NOT_SELECTED
101 """
102 stock_id: Annotated[str, PlanningId]
103 stock_name: str
104 sector: str
105 predicted_return: float # e.g., 0.12 means 12% expected return
106
107 # THE DECISION: Should we include this stock in the portfolio?
108 # SolverForge will set this to SELECTED or NOT_SELECTED
109 # Note: value_range_provider_refs links to the 'selection_range' field
110 selection: Annotated[
111 SelectionValue | None,
112 PlanningVariable(value_range_provider_refs=["selection_range"])
113 ] = None
114
115 @property
116 def selected(self) -> bool | None:
117 """Convenience property to check if stock is selected."""
118 if self.selection is None:
119 return None
120 return self.selection.value
121
122
123@planning_solution
124@dataclass
125class PortfolioOptimizationPlan:
126 """
127 The complete portfolio optimization problem.
128
129 This is the planning solution that contains:
130 - All candidate stocks (planning entities)
131 - Configuration parameters
132 - The optimization score
133
134 The solver will decide which stocks to select (set selected=True)
135 while respecting constraints and maximizing expected return.
136 """
137 # All stocks we're choosing from (planning entities)
138 stocks: Annotated[
139 list[StockSelection],
140 PlanningEntityCollectionProperty,
141 ValueRangeProvider
142 ]
143
144 # Configuration
145 target_position_count: int = 20 # How many stocks to select
146 max_sector_percentage: float = 0.25 # Max 25% in any sector
147
148 # Constraint configuration (problem fact for constraints to access)
149 # This derives from target_position_count and max_sector_percentage
150 portfolio_config: Annotated[
151 PortfolioConfig,
152 ProblemFactProperty
153 ] = field(default_factory=PortfolioConfig)
154
155 # Value range for the selection
156 # The solver can set `selection` to SELECTED or NOT_SELECTED
157 # Note: id="selection_range" must match the value_range_provider_refs in StockSelection
158 selection_range: Annotated[
159 list[SelectionValue],
160 ValueRangeProvider(id="selection_range"),
161 ProblemFactCollectionProperty
162 ] = field(default_factory=lambda: [SELECTED, NOT_SELECTED])
163
164 # Solution quality score (set by solver)
165 score: Annotated[HardSoftScore | None, PlanningScore] = None
166
167 # Current solver status
168 solver_status: SolverStatus = SolverStatus.NOT_SOLVING
169
170 def get_selected_stocks(self) -> list[StockSelection]:
171 """Return only stocks that are selected for the portfolio."""
172 return [s for s in self.stocks if s.selected is True]
173
174 def get_selected_count(self) -> int:
175 """Return count of selected stocks."""
176 return len(self.get_selected_stocks())
177
178 def get_weight_per_stock(self) -> float:
179 """Calculate equal weight per selected stock (e.g., 20 stocks = 5% each)."""
180 count = self.get_selected_count()
181 return 1.0 / count if count > 0 else 0.0
182
183 def get_sector_weights(self) -> dict[str, float]:
184 """Calculate total weight per sector."""
185 weight = self.get_weight_per_stock()
186 sector_weights: dict[str, float] = {}
187 for stock in self.get_selected_stocks():
188 sector_weights[stock.sector] = sector_weights.get(stock.sector, 0.0) + weight
189 return sector_weights
190
191 def get_expected_return(self) -> float:
192 """Calculate total expected portfolio return."""
193 weight = self.get_weight_per_stock()
194 return sum(s.predicted_return * weight for s in self.get_selected_stocks())
195
196 def get_herfindahl_index(self) -> float:
197 """
198 Calculate the Herfindahl-Hirschman Index (HHI) for sector concentration.
199
200 HHI = sum of (sector_weight)^2
201 - Range: 1/n (perfectly diversified) to 1.0 (all in one sector)
202 - Lower HHI = more diversified
203 - Common thresholds: <0.15 (diversified), 0.15-0.25 (moderate), >0.25 (concentrated)
204 """
205 sector_weights = self.get_sector_weights()
206 if not sector_weights:
207 return 0.0
208 return sum(w * w for w in sector_weights.values())
209
210 def get_diversification_score(self) -> float:
211 """
212 Calculate diversification score as 1 - HHI.
213
214 Range: 0.0 (all in one sector) to 1-1/n (perfectly diversified)
215 Higher = more diversified
216 """
217 return 1.0 - self.get_herfindahl_index()
218
219 def get_max_sector_exposure(self) -> float:
220 """
221 Get the highest single sector weight.
222
223 Returns the weight of the most concentrated sector.
224 Lower is better for diversification.
225 """
226 sector_weights = self.get_sector_weights()
227 if not sector_weights:
228 return 0.0
229 return max(sector_weights.values())
230
231 def get_sector_count(self) -> int:
232 """Return count of unique sectors in selected stocks."""
233 selected = self.get_selected_stocks()
234 return len(set(s.sector for s in selected))
235
236 def get_return_volatility(self) -> float:
237 """
238 Calculate standard deviation of predicted returns (proxy for risk).
239
240 Higher volatility = higher risk portfolio.
241 """
242 selected = self.get_selected_stocks()
243 if len(selected) < 2:
244 return 0.0
245
246 returns = [s.predicted_return for s in selected]
247 mean_return = sum(returns) / len(returns)
248 variance = sum((r - mean_return) ** 2 for r in returns) / len(returns)
249 return variance ** 0.5
250
251 def get_sharpe_proxy(self) -> float:
252 """
253 Calculate a proxy for Sharpe ratio: return / volatility.
254
255 Higher = better risk-adjusted return.
256 Note: This is a simplified proxy, not true Sharpe (no risk-free rate).
257 """
258 volatility = self.get_return_volatility()
259 if volatility == 0:
260 return 0.0
261 return self.get_expected_return() / volatility
262
263
264# ============================================================
265# Pydantic REST Models (for API serialization)
266# ============================================================
267
268class StockSelectionModel(JsonDomainBase):
269 """REST API model for StockSelection."""
270 stock_id: str = Field(..., alias="stockId")
271 stock_name: str = Field(..., alias="stockName")
272 sector: str
273 predicted_return: float = Field(..., alias="predictedReturn")
274 selected: Optional[bool] = None
275
276
277class SolverConfigModel(JsonDomainBase):
278 """REST API model for solver configuration options."""
279 termination_seconds: int = Field(default=30, alias="terminationSeconds", ge=10, le=300)
280
281
282class PortfolioMetricsModel(JsonDomainBase):
283 """
284 REST API model for portfolio business metrics (KPIs).
285
286 These metrics provide business insight beyond the solver score:
287 - Diversification measures (HHI, max sector exposure)
288 - Risk/return measures (expected return, volatility, Sharpe proxy)
289 """
290 expected_return: float = Field(..., alias="expectedReturn")
291 sector_count: int = Field(..., alias="sectorCount")
292 max_sector_exposure: float = Field(..., alias="maxSectorExposure")
293 herfindahl_index: float = Field(..., alias="herfindahlIndex")
294 diversification_score: float = Field(..., alias="diversificationScore")
295 return_volatility: float = Field(..., alias="returnVolatility")
296 sharpe_proxy: float = Field(..., alias="sharpeProxy")
297
298
299class PortfolioOptimizationPlanModel(JsonDomainBase):
300 """REST API model for PortfolioOptimizationPlan."""
301 stocks: List[StockSelectionModel]
302 target_position_count: int = Field(default=20, alias="targetPositionCount")
303 max_sector_percentage: float = Field(default=0.25, alias="maxSectorPercentage")
304 score: Optional[str] = None
305 solver_status: Optional[str] = Field(None, alias="solverStatus")
306 solver_config: Optional[SolverConfigModel] = Field(None, alias="solverConfig")
307 metrics: Optional[PortfolioMetricsModel] = None
308