feat(PORTFOLIO-002): add position storage costs

This commit is contained in:
Bu5hm4nn
2026-03-28 23:48:41 +01:00
parent e148d55cda
commit 0e972e9dd6
5 changed files with 501 additions and 1 deletions

View File

@@ -22,6 +22,9 @@ class Position:
entry_date: Date of position entry (for historical conversion lookups) entry_date: Date of position entry (for historical conversion lookups)
entry_basis_mode: Entry basis mode ("weight" or "value_price") entry_basis_mode: Entry basis mode ("weight" or "value_price")
notes: Optional notes about this position 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 created_at: Timestamp when position was created
""" """
@@ -33,6 +36,9 @@ class Position:
entry_date: date entry_date: date
entry_basis_mode: str = "weight" entry_basis_mode: str = "weight"
notes: str = "" 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)) created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
def __post_init__(self) -> None: def __post_init__(self) -> None:
@@ -67,6 +73,9 @@ class Position:
"entry_date": self.entry_date.isoformat(), "entry_date": self.entry_date.isoformat(),
"entry_basis_mode": self.entry_basis_mode, "entry_basis_mode": self.entry_basis_mode,
"notes": self.notes, "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(), "created_at": self.created_at.isoformat(),
} }
@@ -82,6 +91,11 @@ class Position:
entry_date=date.fromisoformat(data["entry_date"]), entry_date=date.fromisoformat(data["entry_date"]),
entry_basis_mode=data.get("entry_basis_mode", "weight"), entry_basis_mode=data.get("entry_basis_mode", "weight"),
notes=data.get("notes", ""), 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), 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_date: date | None = None,
entry_basis_mode: str = "weight", entry_basis_mode: str = "weight",
notes: str = "", notes: str = "",
storage_cost_basis: Decimal | None = None,
storage_cost_period: str | None = None,
storage_cost_currency: str = "USD",
) -> Position: ) -> Position:
"""Create a new position with sensible defaults. """Create a new position with sensible defaults.
@@ -105,6 +122,9 @@ def create_position(
entry_date: Entry date (default: today) entry_date: Entry date (default: today)
entry_basis_mode: Entry basis mode (default: "weight") entry_basis_mode: Entry basis mode (default: "weight")
notes: Optional notes 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( return Position(
id=uuid4(), id=uuid4(),
@@ -115,4 +135,7 @@ def create_position(
entry_date=entry_date or date.today(), entry_date=entry_date or date.today(),
entry_basis_mode=entry_basis_mode, entry_basis_mode=entry_basis_mode,
notes=notes, notes=notes,
storage_cost_basis=storage_cost_basis,
storage_cost_period=storage_cost_period,
storage_cost_currency=storage_cost_currency,
) )

View File

@@ -2,6 +2,7 @@ from __future__ import annotations
import logging import logging
from datetime import datetime, timezone from datetime import datetime, timezone
from decimal import Decimal
from fastapi import Request from fastapi import Request
from fastapi.responses import RedirectResponse 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.alerts import AlertService, build_portfolio_alert_context
from app.services.ltv_history import LtvHistoryChartModel, LtvHistoryService from app.services.ltv_history import LtvHistoryChartModel, LtvHistoryService
from app.services.runtime import get_data_service 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 from app.services.turnstile import load_turnstile_settings
logger = logging.getLogger(__name__) 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) 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["cash_buffer"] = max(float(portfolio["gold_value"]) - configured_gold_value, 0.0) + _DEFAULT_CASH_BUFFER
portfolio["hedge_budget"] = float(config.monthly_budget) 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) alert_status = AlertService().evaluate(config, portfolio)
ltv_history_service = LtvHistoryService(repository=LtvHistoryRepository(base_path=repo.base_path)) ltv_history_service = LtvHistoryService(repository=LtvHistoryRepository(base_path=repo.base_path))
ltv_history_notice: str | None = None ltv_history_notice: str | None = None
@@ -362,6 +375,11 @@ async def overview_page(workspace_id: str) -> None:
f"${portfolio['hedge_budget']:,.0f}", f"${portfolio['hedge_budget']:,.0f}",
"Monthly budget from saved settings", "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: for title, value, caption in summary_cards:
with ui.card().classes( with ui.card().classes(

View File

@@ -15,6 +15,7 @@ from app.models.workspace import get_workspace_repository
from app.pages.common import dashboard_page, split_page_panes from app.pages.common import dashboard_page, split_page_panes
from app.services.alerts import AlertService, build_portfolio_alert_context from app.services.alerts import AlertService, build_portfolio_alert_context
from app.services.settings_status import save_status_text 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__) logger = logging.getLogger(__name__)
@@ -334,6 +335,19 @@ def settings_page(workspace_id: str) -> None:
label="Underlying", label="Underlying",
).classes("w-full") ).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( pos_quantity = ui.number(
"Quantity", "Quantity",
value=100.0, value=100.0,
@@ -369,6 +383,23 @@ def settings_page(workspace_id: str) -> None:
placeholder="Add notes about this position...", placeholder="Add notes about this position...",
).classes("w-full") ).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"): with ui.row().classes("w-full gap-3 mt-4"):
ui.button("Cancel", on_click=lambda: add_position_dialog.close()).props("outline") 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") 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: def add_position_from_form() -> None:
"""Add a new position from the form.""" """Add a new position from the form."""
try: 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( new_position = Position(
id=uuid4(), id=uuid4(),
underlying=str(pos_underlying.value), underlying=underlying,
quantity=Decimal(str(pos_quantity.value)), quantity=Decimal(str(pos_quantity.value)),
unit=str(pos_unit.value), unit=str(pos_unit.value),
entry_price=Decimal(str(pos_entry_price.value)), entry_price=Decimal(str(pos_entry_price.value)),
entry_date=date.fromisoformat(str(pos_entry_date.value)), entry_date=date.fromisoformat(str(pos_entry_date.value)),
entry_basis_mode="weight", entry_basis_mode="weight",
notes=str(pos_notes.value or ""), 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) workspace_repo.add_position(workspace_id, new_position)
add_position_dialog.close() 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( ui.label(f"Value: ${float(pos.entry_value):,.2f}").classes(
"text-xs font-semibold text-emerald-600 dark:text-emerald-400" "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"): with ui.row().classes("gap-1"):
ui.button( ui.button(
icon="delete", icon="delete",

View File

@@ -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

303
tests/test_storage_costs.py Normal file
View File

@@ -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"