Spaces:
Running
Running
File size: 3,437 Bytes
cafdd88 c3aab0c 44f08fc cafdd88 2750cce cafdd88 c3aab0c aca4b95 cafdd88 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | from typing import List, Dict, Optional
from pydantic import BaseModel, Field, validator
import pandas as pd
from datetime import date
class TickerData(BaseModel):
"""
Represents a single stock's metadata and price history.
"""
symbol: str
sector: str
price_history: Dict[str, float] = Field(default_factory=dict, description="Date (ISO) -> Adj Close Price")
@property
def latest_price(self) -> float:
if not self.price_history:
return 0.0
# Sort by date key and get last value
return self.price_history[sorted(self.price_history.keys())[-1]]
class OptimizationRequest(BaseModel):
"""
User request for portfolio optimization.
"""
client_id: str
initial_investment: float = 100000.0
excluded_sectors: List[str] = Field(default_factory=list, description="List of sectors to exclude (e.g., ['Energy'])")
excluded_tickers: List[str] = Field(default_factory=list, description="List of specific tickers to exclude (e.g., ['AMZN'])")
max_weight: Optional[float] = Field(None, description="Maximum weight for any single asset (e.g., 0.05)")
strategy: Optional[str] = Field(None, description="Global Filter Strategy: 'smallest_market_cap' or 'largest_market_cap'")
top_n: Optional[int] = Field(None, description="Number of assets to select for strategy (e.g. 50)")
benchmark: str = "^GSPC"
user_prompt: Optional[str] = Field(None, description="Raw user input for LLM intent parsing")
class Config:
json_schema_extra = {
"example": {
"client_id": "Demo_User_1",
"initial_investment": 100000.0,
"excluded_sectors": ["Energy"],
"excluded_tickers": ["AMZN"],
"benchmark": "^GSPC"
}
}
class OptimizationResult(BaseModel):
"""
Output of the optimization engine.
"""
weights: Dict[str, float] = Field(..., description="Ticker -> Optimal Weight")
tracking_error: float
status: str
@validator('weights')
def validate_weights(cls, v):
# Filter out near-zero weights for cleanliness
return {k: val for k, val in v.items() if val > 0.0001}
class TaxLot(BaseModel):
"""
A specific purchase lot of a stock.
"""
symbol: str
purchase_date: date
quantity: int
cost_basis_per_share: float
current_price: float
@property
def unrealized_pl(self) -> float:
return (self.current_price - self.cost_basis_per_share) * self.quantity
@property
def is_loss(self) -> bool:
return self.unrealized_pl < 0
@property
def loss_percentage(self) -> float:
if self.cost_basis_per_share == 0: return 0.0
return (self.current_price - self.cost_basis_per_share) / self.cost_basis_per_share
class HarvestOpportunity(BaseModel):
"""
A suggestion to harvest a tax loss.
"""
sell_ticker: str
buy_proxy_ticker: str
quantity: int
estimated_loss_harvested: float
reason: str
class AttributionReport(BaseModel):
"""
Brinson Attribution Data.
"""
allocation_effect: float
selection_effect: float
total_active_return: float
top_contributors: List[Dict]
top_detractors: List[Dict]
sector_exposure: Optional[List[Dict]] = Field(default_factory=list, description="Sector Level Attribution (Truth Table)")
narrative: str
|