diff --git a/app/models/position.py b/app/models/position.py index 3ae20c8..10062e0 100644 --- a/app/models/position.py +++ b/app/models/position.py @@ -22,6 +22,9 @@ class Position: entry_date: Date of position entry (for historical conversion lookups) entry_basis_mode: Entry basis mode ("weight" or "value_price") notes: Optional notes about this position + storage_cost_basis: Annual storage cost as percentage (e.g., Decimal("0.12") for 0.12%) or fixed amount + storage_cost_period: Period for storage cost ("annual" or "monthly") + storage_cost_currency: Currency for fixed amount costs (default "USD") created_at: Timestamp when position was created """ @@ -33,6 +36,9 @@ class Position: entry_date: date entry_basis_mode: str = "weight" notes: str = "" + storage_cost_basis: Decimal | None = None + storage_cost_period: str | None = None + storage_cost_currency: str = "USD" created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) def __post_init__(self) -> None: @@ -67,6 +73,9 @@ class Position: "entry_date": self.entry_date.isoformat(), "entry_basis_mode": self.entry_basis_mode, "notes": self.notes, + "storage_cost_basis": str(self.storage_cost_basis) if self.storage_cost_basis is not None else None, + "storage_cost_period": self.storage_cost_period, + "storage_cost_currency": self.storage_cost_currency, "created_at": self.created_at.isoformat(), } @@ -82,6 +91,11 @@ class Position: entry_date=date.fromisoformat(data["entry_date"]), entry_basis_mode=data.get("entry_basis_mode", "weight"), notes=data.get("notes", ""), + storage_cost_basis=( + Decimal(data["storage_cost_basis"]) if data.get("storage_cost_basis") is not None else None + ), + storage_cost_period=data.get("storage_cost_period"), + storage_cost_currency=data.get("storage_cost_currency", "USD"), created_at=datetime.fromisoformat(data["created_at"]) if "created_at" in data else datetime.now(UTC), ) @@ -94,6 +108,9 @@ def create_position( entry_date: date | None = None, entry_basis_mode: str = "weight", notes: str = "", + storage_cost_basis: Decimal | None = None, + storage_cost_period: str | None = None, + storage_cost_currency: str = "USD", ) -> Position: """Create a new position with sensible defaults. @@ -105,6 +122,9 @@ def create_position( entry_date: Entry date (default: today) entry_basis_mode: Entry basis mode (default: "weight") notes: Optional notes + storage_cost_basis: Annual storage cost as percentage or fixed amount (default: None) + storage_cost_period: Period for storage cost ("annual" or "monthly", default: None) + storage_cost_currency: Currency for fixed amount costs (default: "USD") """ return Position( id=uuid4(), @@ -115,4 +135,7 @@ def create_position( entry_date=entry_date or date.today(), entry_basis_mode=entry_basis_mode, notes=notes, + storage_cost_basis=storage_cost_basis, + storage_cost_period=storage_cost_period, + storage_cost_currency=storage_cost_currency, ) diff --git a/app/pages/overview.py b/app/pages/overview.py index 683fd20..926bda0 100644 --- a/app/pages/overview.py +++ b/app/pages/overview.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging from datetime import datetime, timezone +from decimal import Decimal from fastapi import Request from fastapi.responses import RedirectResponse @@ -21,6 +22,7 @@ from app.pages.common import ( from app.services.alerts import AlertService, build_portfolio_alert_context from app.services.ltv_history import LtvHistoryChartModel, LtvHistoryService from app.services.runtime import get_data_service +from app.services.storage_costs import calculate_total_storage_cost from app.services.turnstile import load_turnstile_settings logger = logging.getLogger(__name__) @@ -163,6 +165,17 @@ async def overview_page(workspace_id: str) -> None: configured_gold_value = float(config.gold_value or 0.0) portfolio["cash_buffer"] = max(float(portfolio["gold_value"]) - configured_gold_value, 0.0) + _DEFAULT_CASH_BUFFER portfolio["hedge_budget"] = float(config.monthly_budget) + + # Calculate storage costs for positions + positions = config.positions + current_values: dict[str, Decimal] = {} + for pos in positions: + # Use entry value as proxy for current value (would need live prices for accurate calc) + current_values[str(pos.id)] = pos.entry_value + total_annual_storage_cost = calculate_total_storage_cost(positions, current_values) + portfolio["annual_storage_cost"] = float(total_annual_storage_cost) + portfolio["storage_cost_pct"] = (float(total_annual_storage_cost) / float(portfolio["gold_value"]) * 100) if portfolio["gold_value"] > 0 else 0.0 + alert_status = AlertService().evaluate(config, portfolio) ltv_history_service = LtvHistoryService(repository=LtvHistoryRepository(base_path=repo.base_path)) ltv_history_notice: str | None = None @@ -362,6 +375,11 @@ async def overview_page(workspace_id: str) -> None: f"${portfolio['hedge_budget']:,.0f}", "Monthly budget from saved settings", ), + ( + "Storage Costs", + f"${portfolio['annual_storage_cost']:,.2f}/yr ({portfolio['storage_cost_pct']:.2f}%)", + "Annual vault storage for physical positions (GLD expense ratio baked into share price)", + ), ] for title, value, caption in summary_cards: with ui.card().classes( diff --git a/app/pages/settings.py b/app/pages/settings.py index 61984e4..d6d3b46 100644 --- a/app/pages/settings.py +++ b/app/pages/settings.py @@ -15,6 +15,7 @@ from app.models.workspace import get_workspace_repository from app.pages.common import dashboard_page, split_page_panes from app.services.alerts import AlertService, build_portfolio_alert_context from app.services.settings_status import save_status_text +from app.services.storage_costs import get_default_storage_cost_for_underlying logger = logging.getLogger(__name__) @@ -334,6 +335,19 @@ def settings_page(workspace_id: str) -> None: label="Underlying", ).classes("w-full") + def update_storage_cost_default() -> None: + """Update storage cost defaults based on underlying selection.""" + underlying = str(pos_underlying.value) + default_basis, default_period = get_default_storage_cost_for_underlying(underlying) + if default_basis is not None: + pos_storage_cost_basis.value = float(default_basis) + pos_storage_cost_period.value = default_period or "annual" + else: + pos_storage_cost_basis.value = 0.0 + pos_storage_cost_period.value = "annual" + + pos_underlying.on_value_change(lambda _: update_storage_cost_default()) + pos_quantity = ui.number( "Quantity", value=100.0, @@ -369,6 +383,23 @@ def settings_page(workspace_id: str) -> None: placeholder="Add notes about this position...", ).classes("w-full") + ui.separator().classes("my-3") + ui.label("Storage Costs (optional)").classes("text-sm font-semibold text-slate-700 dark:text-slate-300") + ui.label("For physical gold (XAU), defaults to 0.12% annual vault storage.").classes("text-xs text-slate-500 dark:text-slate-400 mb-2") + + pos_storage_cost_basis = ui.number( + "Storage cost (% per year or fixed $)", + value=0.0, + min=0.0, + step=0.01, + ).classes("w-full") + + pos_storage_cost_period = ui.select( + {"annual": "Annual", "monthly": "Monthly"}, + value="annual", + label="Cost period", + ).classes("w-full") + with ui.row().classes("w-full gap-3 mt-4"): ui.button("Cancel", on_click=lambda: add_position_dialog.close()).props("outline") ui.button("Add Position", on_click=lambda: add_position_from_form()).props("color=primary") @@ -376,15 +407,22 @@ def settings_page(workspace_id: str) -> None: def add_position_from_form() -> None: """Add a new position from the form.""" try: + underlying = str(pos_underlying.value) + storage_cost_basis_val = float(pos_storage_cost_basis.value) + storage_cost_basis = Decimal(str(storage_cost_basis_val)) if storage_cost_basis_val > 0 else None + storage_cost_period = str(pos_storage_cost_period.value) if storage_cost_basis else None + new_position = Position( id=uuid4(), - underlying=str(pos_underlying.value), + underlying=underlying, quantity=Decimal(str(pos_quantity.value)), unit=str(pos_unit.value), entry_price=Decimal(str(pos_entry_price.value)), entry_date=date.fromisoformat(str(pos_entry_date.value)), entry_basis_mode="weight", notes=str(pos_notes.value or ""), + storage_cost_basis=storage_cost_basis, + storage_cost_period=storage_cost_period, ) workspace_repo.add_position(workspace_id, new_position) add_position_dialog.close() @@ -423,6 +461,19 @@ def settings_page(workspace_id: str) -> None: ui.label(f"Value: ${float(pos.entry_value):,.2f}").classes( "text-xs font-semibold text-emerald-600 dark:text-emerald-400" ) + # Show storage cost if configured + if pos.storage_cost_basis is not None: + basis_val = float(pos.storage_cost_basis) + period = pos.storage_cost_period or "annual" + if basis_val < 1: + # Percentage + storage_label = f"{basis_val:.2f}% {period} storage" + else: + # Fixed amount + storage_label = f"${basis_val:,.2f} {period} storage" + ui.label(f"Storage: {storage_label}").classes( + "text-xs text-slate-500 dark:text-slate-400" + ) with ui.row().classes("gap-1"): ui.button( icon="delete", diff --git a/app/services/storage_costs.py b/app/services/storage_costs.py new file mode 100644 index 0000000..2ec8106 --- /dev/null +++ b/app/services/storage_costs.py @@ -0,0 +1,105 @@ +"""Storage cost calculation service for positions with physical storage requirements.""" + +from __future__ import annotations + +from decimal import Decimal + +from app.models.position import Position + +_DECIMAL_ZERO = Decimal("0") +_DECIMAL_ONE = Decimal("1") +_DECIMAL_HUNDRED = Decimal("100") +_DECIMAL_TWELVE = Decimal("12") + + +def calculate_annual_storage_cost(position: Position, current_value: Decimal) -> Decimal: + """Calculate annual storage cost for a single position. + + Args: + position: Position with optional storage_cost_basis and storage_cost_period + current_value: Current market value of the position (quantity × current_price) + + Returns: + Annual storage cost in position's storage_cost_currency (default USD) + + Notes: + - If storage_cost_basis is None, returns 0 (no storage cost) + - If storage_cost_period is "monthly", annualizes the cost (×12) + - If storage_cost_basis is a percentage, applies it to current_value + - If storage_cost_basis is a fixed amount, uses it directly + """ + if position.storage_cost_basis is None: + return _DECIMAL_ZERO + + basis = position.storage_cost_basis + period = position.storage_cost_period or "annual" + + # Determine if basis is a percentage (e.g., 0.12 for 0.12%) or fixed amount + # Heuristic: if basis < 1, treat as percentage; otherwise as fixed amount + if basis < _DECIMAL_ONE: + # Percentage-based cost + if period == "monthly": + # Monthly percentage, annualize it + annual_rate = basis * _DECIMAL_TWELVE + else: + # Already annual + annual_rate = basis + + # Apply percentage to current value + return (current_value * annual_rate) / _DECIMAL_HUNDRED + else: + # Fixed amount + if period == "monthly": + # Monthly fixed cost, annualize it + return basis * _DECIMAL_TWELVE + else: + # Already annual fixed cost + return basis + + +def calculate_total_storage_cost( + positions: list[Position], + current_values: dict[str, Decimal], +) -> Decimal: + """Calculate total annual storage cost across all positions. + + Args: + positions: List of positions with optional storage costs + current_values: Mapping of position ID (str) to current market value + + Returns: + Total annual storage cost in USD (assumes all positions use USD) + """ + total = _DECIMAL_ZERO + for position in positions: + current_value = current_values.get(str(position.id), _DECIMAL_ZERO) + cost = calculate_annual_storage_cost(position, current_value) + total += cost + return total + + +def get_default_storage_cost_for_underlying(underlying: str) -> tuple[Decimal | None, str | None]: + """Get default storage cost settings for a given underlying instrument. + + Args: + underlying: Instrument symbol (e.g., "XAU", "GLD", "GC=F") + + Returns: + Tuple of (storage_cost_basis, storage_cost_period) or (None, None) if no default + + Notes: + - XAU (physical gold): 0.12% annual for allocated vault storage + - GLD: None (expense ratio baked into share price) + - GC=F: None (roll costs are the storage analog, handled separately) + """ + if underlying == "XAU": + # Physical gold: 0.12% annual storage cost for allocated vault storage + return Decimal("0.12"), "annual" + elif underlying == "GLD": + # GLD: expense ratio is implicit in share price, no separate storage cost + return None, None + elif underlying == "GC=F": + # Futures: roll costs are the storage analog (deferred to GCF-001) + return None, None + else: + return None, None diff --git a/tests/test_storage_costs.py b/tests/test_storage_costs.py new file mode 100644 index 0000000..c3d15e1 --- /dev/null +++ b/tests/test_storage_costs.py @@ -0,0 +1,303 @@ +"""Tests for storage cost calculations.""" + +from __future__ import annotations + +from decimal import Decimal +from uuid import uuid4 + +from app.models.position import Position, create_position +from app.services.storage_costs import ( + calculate_annual_storage_cost, + calculate_total_storage_cost, + get_default_storage_cost_for_underlying, +) + + +class TestCalculateAnnualStorageCost: + """Test calculate_annual_storage_cost function.""" + + def test_no_storage_cost_returns_zero(self) -> None: + """Test that position without storage cost returns zero.""" + pos = create_position( + underlying="GLD", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=None, + ) + current_value = Decimal("215000") + + result = calculate_annual_storage_cost(pos, current_value) + + assert result == Decimal("0") + + def test_percentage_annual_cost(self) -> None: + """Test percentage-based annual storage cost.""" + pos = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.12"), # 0.12% + storage_cost_period="annual", + ) + current_value = Decimal("215000") + + result = calculate_annual_storage_cost(pos, current_value) + + # 0.12% of 215000 = 258 + expected = Decimal("258") + assert result == expected + + def test_percentage_monthly_cost_annualized(self) -> None: + """Test percentage-based monthly storage cost is annualized.""" + pos = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.01"), # 0.01% per month + storage_cost_period="monthly", + ) + current_value = Decimal("215000") + + result = calculate_annual_storage_cost(pos, current_value) + + # 0.01% monthly × 12 = 0.12% annual + # 0.12% of 215000 = 258 + expected = Decimal("258") + assert result == expected + + def test_fixed_annual_cost(self) -> None: + """Test fixed annual storage cost.""" + pos = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("300"), # $300 per year + storage_cost_period="annual", + ) + current_value = Decimal("215000") + + result = calculate_annual_storage_cost(pos, current_value) + + assert result == Decimal("300") + + def test_fixed_monthly_cost_annualized(self) -> None: + """Test fixed monthly storage cost is annualized.""" + pos = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("25"), # $25 per month + storage_cost_period="monthly", + ) + current_value = Decimal("215000") + + result = calculate_annual_storage_cost(pos, current_value) + + # $25 × 12 = $300 per year + expected = Decimal("300") + assert result == expected + + def test_percentage_boundary_at_one(self) -> None: + """Test that basis < 1 is treated as percentage, >= 1 as fixed.""" + # 0.99 should be treated as percentage + pos_pct = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.99"), + storage_cost_period="annual", + ) + current_value = Decimal("100000") + result_pct = calculate_annual_storage_cost(pos_pct, current_value) + # 0.99% of 100000 = 990 + assert result_pct == Decimal("990") + + # 1.00 should be treated as fixed + pos_fixed = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("1"), + storage_cost_period="annual", + ) + result_fixed = calculate_annual_storage_cost(pos_fixed, current_value) + assert result_fixed == Decimal("1") + + +class TestCalculateTotalStorageCost: + """Test calculate_total_storage_cost function.""" + + def test_empty_positions_returns_zero(self) -> None: + """Test that empty position list returns zero.""" + result = calculate_total_storage_cost([], {}) + assert result == Decimal("0") + + def test_multiple_positions_summed(self) -> None: + """Test that multiple positions have costs summed.""" + pos1 = create_position( + underlying="XAU", + quantity=Decimal("50"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.12"), + storage_cost_period="annual", + ) + pos2 = create_position( + underlying="XAU", + quantity=Decimal("50"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.12"), + storage_cost_period="annual", + ) + positions = [pos1, pos2] + current_values = { + str(pos1.id): Decimal("107500"), # 50 × 2150 + str(pos2.id): Decimal("107500"), + } + + result = calculate_total_storage_cost(positions, current_values) + + # Each: 0.12% of 107500 = 129 + # Total: 129 + 129 = 258 + expected = Decimal("258") + assert result == expected + + def test_mixed_positions_with_and_without_costs(self) -> None: + """Test positions with and without storage costs.""" + pos_xau = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.12"), + storage_cost_period="annual", + ) + pos_gld = create_position( + underlying="GLD", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=None, # GLD has no storage cost + ) + positions = [pos_xau, pos_gld] + current_values = { + str(pos_xau.id): Decimal("215000"), + str(pos_gld.id): Decimal("215000"), + } + + result = calculate_total_storage_cost(positions, current_values) + + # Only XAU position has cost: 0.12% of 215000 = 258 + expected = Decimal("258") + assert result == expected + + +class TestGetDefaultStorageCostForUnderlying: + """Test get_default_storage_cost_for_underlying function.""" + + def test_xau_default(self) -> None: + """Test XAU gets 0.12% annual default.""" + basis, period = get_default_storage_cost_for_underlying("XAU") + + assert basis == Decimal("0.12") + assert period == "annual" + + def test_gld_default(self) -> None: + """Test GLD gets no storage cost (expense ratio in price).""" + basis, period = get_default_storage_cost_for_underlying("GLD") + + assert basis is None + assert period is None + + def test_gc_f_default(self) -> None: + """Test GC=F gets no storage cost (roll costs deferred).""" + basis, period = get_default_storage_cost_for_underlying("GC=F") + + assert basis is None + assert period is None + + def test_unknown_underlying_default(self) -> None: + """Test unknown underlying gets no storage cost.""" + basis, period = get_default_storage_cost_for_underlying("UNKNOWN") + + assert basis is None + assert period is None + + +class TestPositionStorageCostFields: + """Test Position model storage cost fields.""" + + def test_position_with_storage_cost_fields(self) -> None: + """Test Position can be created with storage cost fields.""" + pos = Position( + id=uuid4(), + underlying="XAU", + quantity=Decimal("100"), + unit="oz", + entry_price=Decimal("2150"), + entry_date="2025-01-01", + storage_cost_basis=Decimal("0.12"), + storage_cost_period="annual", + storage_cost_currency="USD", + ) + + assert pos.storage_cost_basis == Decimal("0.12") + assert pos.storage_cost_period == "annual" + assert pos.storage_cost_currency == "USD" + + def test_position_default_storage_cost_fields(self) -> None: + """Test Position defaults for storage cost fields.""" + pos = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + ) + + assert pos.storage_cost_basis is None + assert pos.storage_cost_period is None + assert pos.storage_cost_currency == "USD" + + def test_position_serialization_with_storage_costs(self) -> None: + """Test Position serialization includes storage cost fields.""" + pos = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.12"), + storage_cost_period="annual", + ) + + data = pos.to_dict() + + assert data["storage_cost_basis"] == "0.12" + assert data["storage_cost_period"] == "annual" + assert data["storage_cost_currency"] == "USD" + + def test_position_deserialization_with_storage_costs(self) -> None: + """Test Position deserialization restores storage cost fields.""" + pos = create_position( + underlying="XAU", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=Decimal("0.12"), + storage_cost_period="annual", + ) + + data = pos.to_dict() + restored = Position.from_dict(data) + + assert restored.storage_cost_basis == Decimal("0.12") + assert restored.storage_cost_period == "annual" + assert restored.storage_cost_currency == "USD" + + def test_position_serialization_with_null_storage_costs(self) -> None: + """Test Position serialization handles null storage cost fields.""" + pos = create_position( + underlying="GLD", + quantity=Decimal("100"), + entry_price=Decimal("2150"), + storage_cost_basis=None, + ) + + data = pos.to_dict() + + assert data["storage_cost_basis"] is None + assert data["storage_cost_period"] is None + assert data["storage_cost_currency"] == "USD"