blackopsrepl/portfolio-optimization-python

1"""
2REST API for Portfolio Optimization
3
4This module provides HTTP endpoints for the portfolio optimization quickstart:
5
6Endpoints:
7- GET  /demo-data           - List available demo datasets
8- GET  /demo-data/{id}      - Load a specific demo dataset
9- POST /portfolios          - Submit a portfolio for optimization
10- GET  /portfolios/{id}     - Get current solution for a job
11- GET  /portfolios/{id}/status - Get solving status
12- DELETE /portfolios/{id}   - Stop solving
13- PUT  /portfolios/analyze  - Analyze a submitted portfolio's score
14
15The API follows the same patterns as other SolverForge quickstarts.
16"""
17from fastapi import FastAPI, Request
18from fastapi.staticfiles import StaticFiles
19from uuid import uuid4
20from dataclasses import replace
21from typing import Any
22
23from solverforge_legacy.solver import SolverManager, SolverFactory
24
25from .domain import PortfolioOptimizationPlan, PortfolioOptimizationPlanModel
26from .converters import plan_to_model, model_to_plan
27from .demo_data import DemoData, generate_demo_data
28from .solver import solver_manager, solution_manager, create_solver_config
29from .score_analysis import ConstraintAnalysisDTO, MatchAnalysisDTO
30
31
32app = FastAPI(
33    title="Portfolio Optimization Quickstart",
34    description="SolverForge quickstart for stock portfolio optimization",
35    docs_url='/q/swagger-ui'
36)
37
38# In-memory storage for submitted portfolios and their solver managers
39data_sets: dict[str, PortfolioOptimizationPlan] = {}
40solver_managers: dict[str, SolverManager] = {}
41
42
43@app.get("/demo-data")
44async def demo_data_list() -> list[DemoData]:
45    """List available demo datasets."""
46    return [e for e in DemoData]
47
48
49@app.get("/demo-data/{dataset_id}", response_model_exclude_none=True)
50async def get_demo_data(dataset_id: str) -> PortfolioOptimizationPlanModel:
51    """Load a specific demo dataset."""
52    demo_data = getattr(DemoData, dataset_id)
53    domain_plan = generate_demo_data(demo_data)
54    return plan_to_model(domain_plan)
55
56
57@app.get("/portfolios/{problem_id}", response_model_exclude_none=True)
58async def get_portfolio(problem_id: str) -> PortfolioOptimizationPlanModel:
59    """Get current solution for a portfolio optimization job."""
60    plan = data_sets[problem_id]
61    # Use per-job solver manager if available, otherwise use default
62    manager = solver_managers.get(problem_id, solver_manager)
63    updated_plan = replace(plan, solver_status=manager.get_solver_status(problem_id))
64    return plan_to_model(updated_plan)
65
66
67def update_portfolio(problem_id: str, plan: PortfolioOptimizationPlan) -> None:
68    """Callback to update the stored solution as solver improves it."""
69    global data_sets
70    data_sets[problem_id] = plan
71
72
73@app.post("/portfolios")
74async def solve_portfolio(plan_model: PortfolioOptimizationPlanModel) -> str:
75    """
76    Submit a portfolio for optimization.
77
78    Returns a job ID that can be used to retrieve the solution.
79    Supports custom solver configuration via solverConfig field.
80    """
81    job_id = str(uuid4())
82    plan = model_to_plan(plan_model)
83    data_sets[job_id] = plan
84
85    # Get termination time from config or use default
86    termination_seconds = 30
87    if plan_model.solver_config and plan_model.solver_config.termination_seconds:
88        termination_seconds = plan_model.solver_config.termination_seconds
89
90    # Create solver with dynamic config
91    config = create_solver_config(termination_seconds)
92    manager: SolverManager = SolverManager.create(SolverFactory.create(config))
93    solver_managers[job_id] = manager
94
95    manager.solve_and_listen(
96        job_id,
97        plan,
98        lambda solution: update_portfolio(job_id, solution)
99    )
100    return job_id
101
102
103@app.get("/portfolios")
104async def list_portfolios() -> list[str]:
105    """List all job IDs of submitted portfolios."""
106    return list(data_sets.keys())
107
108
109@app.get("/portfolios/{problem_id}/status")
110async def get_status(problem_id: str) -> dict[str, Any]:
111    """Get the portfolio status and score for a given job ID."""
112    if problem_id not in data_sets:
113        raise ValueError(f"No portfolio found with ID {problem_id}")
114
115    plan = data_sets[problem_id]
116    # Use per-job solver manager if available, otherwise use default
117    manager = solver_managers.get(problem_id, solver_manager)
118    solver_status = manager.get_solver_status(problem_id)
119
120    # Calculate additional metrics
121    selected_count = plan.get_selected_count()
122    expected_return = plan.get_expected_return() if selected_count > 0 else 0
123
124    return {
125        "score": {
126            "hardScore": plan.score.hard_score if plan.score else 0,
127            "softScore": plan.score.soft_score if plan.score else 0,
128        },
129        "solverStatus": solver_status.name,
130        "selectedCount": selected_count,
131        "expectedReturn": expected_return,
132        "sectorWeights": plan.get_sector_weights() if selected_count > 0 else {},
133    }
134
135
136@app.delete("/portfolios/{problem_id}")
137async def stop_solving(problem_id: str) -> PortfolioOptimizationPlanModel:
138    """Terminate solving for a given job ID."""
139    if problem_id not in data_sets:
140        raise ValueError(f"No portfolio found with ID {problem_id}")
141
142    # Use per-job solver manager if available, otherwise use default
143    manager = solver_managers.get(problem_id, solver_manager)
144    try:
145        manager.terminate_early(problem_id)
146    except Exception as e:
147        print(f"Warning: terminate_early failed for {problem_id}: {e}")
148
149    return await get_portfolio(problem_id)
150
151
152@app.put("/portfolios/analyze")
153async def analyze_portfolio(request: Request) -> dict[str, Any]:
154    """Submit a portfolio to analyze its score in detail."""
155    json_data = await request.json()
156
157    # Parse the incoming JSON using Pydantic models
158    plan_model = PortfolioOptimizationPlanModel.model_validate(json_data)
159
160    # Convert to domain model for analysis
161    domain_plan = model_to_plan(plan_model)
162
163    analysis = solution_manager.analyze(domain_plan)
164
165    # Convert to DTOs for proper serialization
166    constraints = []
167    for constraint in getattr(analysis, 'constraint_analyses', []) or []:
168        matches = [
169            MatchAnalysisDTO(
170                name=str(getattr(getattr(match, 'constraint_ref', None), 'constraint_name', "")),
171                score=str(getattr(match, 'score', "0hard/0soft")),
172                justification=str(getattr(match, 'justification', "")),
173            )
174            for match in getattr(constraint, 'matches', []) or []
175        ]
176
177        constraint_dto = ConstraintAnalysisDTO(
178            name=str(getattr(constraint, 'constraint_name', "")),
179            weight=str(getattr(constraint, 'weight', "0hard/0soft")),
180            score=str(getattr(constraint, 'score', "0hard/0soft")),
181            matches=matches,
182        )
183        constraints.append(constraint_dto)
184
185    return {"constraints": [constraint.model_dump() for constraint in constraints]}
186
187
188# Mount static files for the web UI
189app.mount("/", StaticFiles(directory="static", html=True), name="static")
190