diff --git a/app/models/position.py b/app/models/position.py index 10062e0..e5ae77c 100644 --- a/app/models/position.py +++ b/app/models/position.py @@ -21,6 +21,8 @@ class Position: entry_price: Price per unit at purchase (in USD) entry_date: Date of position entry (for historical conversion lookups) entry_basis_mode: Entry basis mode ("weight" or "value_price") + purchase_premium: Dealer markup over spot as percentage (e.g., Decimal("0.04") for 4%) + bid_ask_spread: Expected sale discount below spot as percentage (e.g., Decimal("0.03") for 3%) 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") @@ -35,6 +37,8 @@ class Position: entry_price: Decimal entry_date: date entry_basis_mode: str = "weight" + purchase_premium: Decimal | None = None + bid_ask_spread: Decimal | None = None notes: str = "" storage_cost_basis: Decimal | None = None storage_cost_period: str | None = None @@ -72,6 +76,8 @@ class Position: "entry_price": str(self.entry_price), "entry_date": self.entry_date.isoformat(), "entry_basis_mode": self.entry_basis_mode, + "purchase_premium": str(self.purchase_premium) if self.purchase_premium is not None else None, + "bid_ask_spread": str(self.bid_ask_spread) if self.bid_ask_spread is not None else None, "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, @@ -90,6 +96,8 @@ class Position: entry_price=Decimal(data["entry_price"]), entry_date=date.fromisoformat(data["entry_date"]), entry_basis_mode=data.get("entry_basis_mode", "weight"), + purchase_premium=(Decimal(data["purchase_premium"]) if data.get("purchase_premium") is not None else None), + bid_ask_spread=(Decimal(data["bid_ask_spread"]) if data.get("bid_ask_spread") is not None else None), notes=data.get("notes", ""), storage_cost_basis=( Decimal(data["storage_cost_basis"]) if data.get("storage_cost_basis") is not None else None @@ -107,6 +115,8 @@ def create_position( entry_price: Decimal | None = None, entry_date: date | None = None, entry_basis_mode: str = "weight", + purchase_premium: Decimal | None = None, + bid_ask_spread: Decimal | None = None, notes: str = "", storage_cost_basis: Decimal | None = None, storage_cost_period: str | None = None, @@ -121,6 +131,8 @@ def create_position( entry_price: Entry price per unit (default: Decimal("2150")) entry_date: Entry date (default: today) entry_basis_mode: Entry basis mode (default: "weight") + purchase_premium: Dealer markup over spot as percentage (default: None) + bid_ask_spread: Expected sale discount below spot as percentage (default: None) 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) @@ -134,6 +146,8 @@ def create_position( entry_price=entry_price if entry_price is not None else Decimal("2150"), entry_date=entry_date or date.today(), entry_basis_mode=entry_basis_mode, + purchase_premium=purchase_premium, + bid_ask_spread=bid_ask_spread, notes=notes, storage_cost_basis=storage_cost_basis, storage_cost_period=storage_cost_period, diff --git a/app/pages/settings.py b/app/pages/settings.py index d6d3b46..1724e2e 100644 --- a/app/pages/settings.py +++ b/app/pages/settings.py @@ -400,6 +400,26 @@ def settings_page(workspace_id: str) -> None: label="Cost period", ).classes("w-full") + ui.separator().classes("my-3") + ui.label("Premium & Spread (optional)").classes("text-sm font-semibold text-slate-700 dark:text-slate-300") + ui.label("For physical gold, accounts for dealer markup and bid/ask spread.").classes("text-xs text-slate-500 dark:text-slate-400 mb-2") + + pos_purchase_premium = ui.number( + "Purchase premium over spot (%)", + value=0.0, + min=0.0, + max=100.0, + step=0.1, + ).classes("w-full") + + pos_bid_ask_spread = ui.number( + "Bid/ask spread on exit (%)", + value=0.0, + min=0.0, + max=100.0, + step=0.1, + ).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") @@ -411,6 +431,10 @@ def settings_page(workspace_id: str) -> None: 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 + purchase_premium_val = float(pos_purchase_premium.value) + purchase_premium = Decimal(str(purchase_premium_val / 100)) if purchase_premium_val > 0 else None + bid_ask_spread_val = float(pos_bid_ask_spread.value) + bid_ask_spread = Decimal(str(bid_ask_spread_val / 100)) if bid_ask_spread_val > 0 else None new_position = Position( id=uuid4(), @@ -420,6 +444,8 @@ def settings_page(workspace_id: str) -> None: entry_price=Decimal(str(pos_entry_price.value)), entry_date=date.fromisoformat(str(pos_entry_date.value)), entry_basis_mode="weight", + purchase_premium=purchase_premium, + bid_ask_spread=bid_ask_spread, notes=str(pos_notes.value or ""), storage_cost_basis=storage_cost_basis, storage_cost_period=storage_cost_period, diff --git a/app/services/position_costs.py b/app/services/position_costs.py new file mode 100644 index 0000000..24941e7 --- /dev/null +++ b/app/services/position_costs.py @@ -0,0 +1,118 @@ +"""Position cost calculations for premium, spread, and storage costs.""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any + +from app.models.position import Position + + +def calculate_effective_entry( + entry_price: Decimal, + purchase_premium: Decimal | None = None, +) -> Decimal: + """Calculate effective entry cost including dealer premium. + + Args: + entry_price: Spot price at entry (per unit) + purchase_premium: Dealer markup over spot as percentage (e.g., 0.04 for 4%) + + Returns: + Effective entry cost per unit + """ + if purchase_premium is None or purchase_premium == 0: + return entry_price + return entry_price * (Decimal("1") + purchase_premium) + + +def calculate_effective_exit( + current_spot: Decimal, + bid_ask_spread: Decimal | None = None, +) -> Decimal: + """Calculate effective exit value after bid/ask spread. + + Args: + current_spot: Current spot price (per unit) + bid_ask_spread: Expected sale discount below spot as percentage (e.g., 0.03 for 3%) + + Returns: + Effective exit value per unit + """ + if bid_ask_spread is None or bid_ask_spread == 0: + return current_spot + return current_spot * (Decimal("1") - bid_ask_spread) + + +def calculate_true_pnl( + position: Position, + current_spot: Decimal, +) -> dict[str, Any]: + """Calculate true P&L accounting for premium and spread. + + Args: + position: Position to calculate P&L for + current_spot: Current spot price per unit + + Returns: + Dict with paper_pnl, realized_pnl, effective_entry, effective_exit, entry_value, exit_value + """ + # Effective entry cost (includes premium) + effective_entry = calculate_effective_entry(position.entry_price, position.purchase_premium) + + # Effective exit value (after spread) + effective_exit = calculate_effective_exit(current_spot, position.bid_ask_spread) + + # Paper P&L (without premium/spread) + paper_pnl = (current_spot - position.entry_price) * position.quantity + + # True P&L (with premium/spread) + true_pnl = (effective_exit - effective_entry) * position.quantity + + # Entry and exit values + entry_value = position.entry_price * position.quantity + exit_value = current_spot * position.quantity + + return { + "paper_pnl": float(paper_pnl), + "true_pnl": float(true_pnl), + "effective_entry": float(effective_entry), + "effective_exit": float(effective_exit), + "entry_value": float(entry_value), + "exit_value": float(exit_value), + "premium_impact": float((position.purchase_premium or 0) * entry_value), + "spread_impact": float((position.bid_ask_spread or 0) * exit_value), + } + + +def get_default_premium_for_product(underlying: str, product_type: str = "default") -> Decimal | None: + """Get default premium/spread for common gold products. + + Args: + underlying: Underlying instrument ("GLD", "GC=F", "XAU") + product_type: Product type ("default", "coin_1oz", "bar_1kg", "allocated") + + Returns: + Tuple of (purchase_premium, bid_ask_spread) or None if not applicable + """ + # GLD/GLDM: ETF is liquid, minimal spread + if underlying in ("GLD", "GLDM"): + # ETF spread is minimal, premium is 0 + return Decimal("0"), Decimal("0.001") # 0% premium, 0.1% spread + + # GC=F: Futures roll costs are handled separately (GCF-001) + if underlying == "GC=F": + return None, None + + # XAU: Physical gold + if underlying == "XAU": + defaults = { + "default": (Decimal("0.04"), Decimal("0.03")), # 4% premium, 3% spread + "coin_1oz": (Decimal("0.04"), Decimal("0.03")), # 1oz coins: 4% premium, 3% spread + "bar_1kg": (Decimal("0.015"), Decimal("0.015")), # 1kg bars: 1.5% premium, 1.5% spread + "allocated": (Decimal("0.001"), Decimal("0.003")), # Allocated: 0.1% premium, 0.3% spread + } + return defaults.get(product_type, defaults["default"]) + + # Unknown underlying + return None, None diff --git a/tests/test_position_costs.py b/tests/test_position_costs.py new file mode 100644 index 0000000..3268419 --- /dev/null +++ b/tests/test_position_costs.py @@ -0,0 +1,271 @@ +"""Tests for position cost calculations.""" + +from __future__ import annotations + +from decimal import Decimal + +from app.models.position import Position, create_position +from app.services.position_costs import ( + calculate_effective_entry, + calculate_effective_exit, + calculate_true_pnl, + get_default_premium_for_product, +) + + +class TestEffectiveEntry: + """Tests for calculate_effective_entry.""" + + def test_no_premium_returns_entry_price(self) -> None: + """Entry price unchanged when no premium.""" + result = calculate_effective_entry(Decimal("2000"), None) + assert result == Decimal("2000") + + def test_zero_premium_returns_entry_price(self) -> None: + """Entry price unchanged with zero premium.""" + result = calculate_effective_entry(Decimal("2000"), Decimal("0")) + assert result == Decimal("2000") + + def test_premium_adds_to_entry(self) -> None: + """Premium adds to entry cost.""" + # 4% premium on $2000 = $80, total $2080 + result = calculate_effective_entry(Decimal("2000"), Decimal("0.04")) + assert result == Decimal("2080") + + def test_large_premium(self) -> None: + """Large premium (10%) calculated correctly.""" + result = calculate_effective_entry(Decimal("2000"), Decimal("0.10")) + assert result == Decimal("2200") + + +class TestEffectiveExit: + """Tests for calculate_effective_exit.""" + + def test_no_spread_returns_spot(self) -> None: + """Spot unchanged when no spread.""" + result = calculate_effective_exit(Decimal("2000"), None) + assert result == Decimal("2000") + + def test_zero_spread_returns_spot(self) -> None: + """Spot unchanged with zero spread.""" + result = calculate_effective_exit(Decimal("2000"), Decimal("0")) + assert result == Decimal("2000") + + def test_spread_subtracts_from_exit(self) -> None: + """Spread subtracts from exit value.""" + # 3% spread on $2000 = $60, result $1940 + result = calculate_effective_exit(Decimal("2000"), Decimal("0.03")) + assert result == Decimal("1940") + + def test_large_spread(self) -> None: + """Large spread (5%) calculated correctly.""" + result = calculate_effective_exit(Decimal("2000"), Decimal("0.05")) + assert result == Decimal("1900") + + +class TestTruePnL: + """Tests for calculate_true_pnl.""" + + def test_gld_position_no_premium_spread(self) -> None: + """GLD position with no premium/spread shows paper P&L equals true P&L.""" + position = create_position( + underlying="GLD", + quantity=Decimal("100"), + entry_price=Decimal("400"), + ) + result = calculate_true_pnl(position, Decimal("420")) + + # Paper P&L: (420 - 400) * 100 = 2000 + assert result["paper_pnl"] == 2000.0 + assert result["true_pnl"] == 2000.0 + assert result["effective_entry"] == 400.0 + assert result["effective_exit"] == 420.0 + + def test_xau_position_with_premium(self) -> None: + """Physical gold position with premium shows higher effective entry.""" + position = create_position( + underlying="XAU", + quantity=Decimal("10"), # 10 oz + entry_price=Decimal("2000"), + purchase_premium=Decimal("0.04"), # 4% + ) + result = calculate_true_pnl(position, Decimal("2200")) + + # Effective entry: 2000 * 1.04 = 2080 + # Paper P&L: (2200 - 2000) * 10 = 2000 + # True P&L: (2200 - 2080) * 10 = 1200 + assert result["paper_pnl"] == 2000.0 + assert result["true_pnl"] == 1200.0 + assert result["effective_entry"] == 2080.0 + assert result["premium_impact"] == 800.0 # 4% of 2000 * 10 + + def test_xau_position_with_spread(self) -> None: + """Physical gold position with spread shows lower effective exit.""" + position = create_position( + underlying="XAU", + quantity=Decimal("10"), + entry_price=Decimal("2000"), + bid_ask_spread=Decimal("0.03"), # 3% + ) + result = calculate_true_pnl(position, Decimal("2200")) + + # Effective exit: 2200 * 0.97 = 2134 + # Paper P&L: (2200 - 2000) * 10 = 2000 + # True P&L: (2134 - 2000) * 10 = 1340 + assert result["paper_pnl"] == 2000.0 + assert result["true_pnl"] == 1340.0 + assert result["effective_exit"] == 2134.0 + assert result["spread_impact"] == 660.0 # 3% of 2200 * 10 + + def test_xau_position_with_both_premium_and_spread(self) -> None: + """Physical gold with both premium and spread.""" + position = create_position( + underlying="XAU", + quantity=Decimal("10"), + entry_price=Decimal("2000"), + purchase_premium=Decimal("0.04"), # 4% + bid_ask_spread=Decimal("0.03"), # 3% + ) + result = calculate_true_pnl(position, Decimal("2200")) + + # Effective entry: 2000 * 1.04 = 2080 + # Effective exit: 2200 * 0.97 = 2134 + # Paper P&L: (2200 - 2000) * 10 = 2000 + # True P&L: (2134 - 2080) * 10 = 540 + assert result["paper_pnl"] == 2000.0 + assert result["true_pnl"] == 540.0 + assert result["effective_entry"] == 2080.0 + assert result["effective_exit"] == 2134.0 + assert result["premium_impact"] == 800.0 + assert result["spread_impact"] == 660.0 + + +class TestDefaultPremiumForProduct: + """Tests for get_default_premium_for_product.""" + + def test_gld_defaults(self) -> None: + """GLD has minimal premium/spread.""" + premium, spread = get_default_premium_for_product("GLD") + assert premium == Decimal("0") + assert spread == Decimal("0.001") + + def test_gldm_defaults(self) -> None: + """GLDM has minimal premium/spread.""" + premium, spread = get_default_premium_for_product("GLDM") + assert premium == Decimal("0") + assert spread == Decimal("0.001") + + def test_gcf_returns_none(self) -> None: + """GC=F futures have no premium/spread (roll costs handled separately).""" + result = get_default_premium_for_product("GC=F") + assert result == (None, None) + + def test_xau_default(self) -> None: + """XAU default uses coin defaults.""" + premium, spread = get_default_premium_for_product("XAU") + assert premium == Decimal("0.04") + assert spread == Decimal("0.03") + + def test_xau_coin_1oz(self) -> None: + """XAU 1oz coins have 4%/3%.""" + premium, spread = get_default_premium_for_product("XAU", "coin_1oz") + assert premium == Decimal("0.04") + assert spread == Decimal("0.03") + + def test_xau_bar_1kg(self) -> None: + """XAU 1kg bars have 1.5%/1.5%.""" + premium, spread = get_default_premium_for_product("XAU", "bar_1kg") + assert premium == Decimal("0.015") + assert spread == Decimal("0.015") + + def test_xau_allocated(self) -> None: + """XAU allocated storage has minimal 0.1%/0.3%.""" + premium, spread = get_default_premium_for_product("XAU", "allocated") + assert premium == Decimal("0.001") + assert spread == Decimal("0.003") + + def test_unknown_underlying_returns_none(self) -> None: + """Unknown underlying returns None.""" + result = get_default_premium_for_product("UNKNOWN") + assert result == (None, None) + + +class TestPositionWithPremiumSpread: + """Tests for Position model with premium/spread fields.""" + + def test_create_position_with_premium(self) -> None: + """Create position with purchase premium.""" + position = create_position( + underlying="XAU", + quantity=Decimal("10"), + entry_price=Decimal("2000"), + purchase_premium=Decimal("0.04"), + ) + assert position.purchase_premium == Decimal("0.04") + assert position.bid_ask_spread is None + + def test_create_position_with_spread(self) -> None: + """Create position with bid/ask spread.""" + position = create_position( + underlying="XAU", + quantity=Decimal("10"), + entry_price=Decimal("2000"), + bid_ask_spread=Decimal("0.03"), + ) + assert position.bid_ask_spread == Decimal("0.03") + assert position.purchase_premium is None + + def test_create_position_with_both(self) -> None: + """Create position with both premium and spread.""" + position = create_position( + underlying="XAU", + quantity=Decimal("10"), + entry_price=Decimal("2000"), + purchase_premium=Decimal("0.04"), + bid_ask_spread=Decimal("0.03"), + ) + assert position.purchase_premium == Decimal("0.04") + assert position.bid_ask_spread == Decimal("0.03") + + def test_position_serialization_with_premium_spread(self) -> None: + """Position serializes premium and spread correctly.""" + position = create_position( + underlying="XAU", + quantity=Decimal("10"), + entry_price=Decimal("2000"), + purchase_premium=Decimal("0.04"), + bid_ask_spread=Decimal("0.03"), + ) + data = position.to_dict() + assert data["purchase_premium"] == "0.04" + assert data["bid_ask_spread"] == "0.03" + + def test_position_deserialization_with_premium_spread(self) -> None: + """Position deserializes premium and spread correctly.""" + data = { + "id": "12345678-1234-5678-1234-567812345678", + "underlying": "XAU", + "quantity": "10", + "unit": "oz", + "entry_price": "2000", + "entry_date": "2024-01-01", + "purchase_premium": "0.04", + "bid_ask_spread": "0.03", + } + position = Position.from_dict(data) + assert position.purchase_premium == Decimal("0.04") + assert position.bid_ask_spread == Decimal("0.03") + + def test_position_deserialization_without_premium_spread(self) -> None: + """Position deserializes without premium and spread (backward compat).""" + data = { + "id": "12345678-1234-5678-1234-567812345678", + "underlying": "GLD", + "quantity": "100", + "unit": "shares", + "entry_price": "400", + "entry_date": "2024-01-01", + } + position = Position.from_dict(data) + assert position.purchase_premium is None + assert position.bid_ask_spread is None