bu5hm4nn/vault-dash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(CORE-002): add GLD share quote conversion seam

Browse Source
This commit is contained in:
Bu5hm4nn
2026-03-25 14:52:48 +01:00
parent 1a2dfaff01
commit f0d7ab5748
10 changed files with 425 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/domain/__init__.py
View File

@@ -5,9 +5,16 @@ from app.domain.backtesting_math import (
asset_quantity_from_money,
materialize_backtest_portfolio_state,
)
from app.domain.instruments import (
asset_quantity_from_weight,
instrument_metadata,
price_per_weight_from_asset_price,
weight_from_asset_quantity,
)
from app.domain.portfolio_math import (
build_alert_context,
portfolio_snapshot_from_config,
resolve_portfolio_spot_from_quote,
strategy_metrics_from_snapshot,
)
from app.domain.units import (
@@ -35,5 +42,10 @@ __all__ = [
"decimal_from_float",
"portfolio_snapshot_from_config",
"build_alert_context",
"resolve_portfolio_spot_from_quote",
"strategy_metrics_from_snapshot",
"instrument_metadata",
"price_per_weight_from_asset_price",
"weight_from_asset_quantity",
"asset_quantity_from_weight",
]

96
app/domain/instruments.py Normal file
View File

@@ -0,0 +1,96 @@
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from app.domain.backtesting_math import AssetQuantity, PricePerAsset
from app.domain.units import BaseCurrency, PricePerWeight, Weight, WeightUnit
@dataclass(frozen=True, slots=True)
class InstrumentMetadata:
symbol: str
quote_currency: BaseCurrency | str
weight_per_share: Weight
def __post_init__(self) -> None:
normalized_symbol = str(self.symbol).strip().upper()
if not normalized_symbol:
raise ValueError("Instrument symbol is required")
object.__setattr__(self, "symbol", normalized_symbol)
object.__setattr__(self, "quote_currency", BaseCurrency(self.quote_currency))
object.__setattr__(self, "weight_per_share", self.weight_per_share)
def assert_symbol(self, symbol: str) -> InstrumentMetadata:
normalized = str(symbol).strip().upper()
if self.symbol != normalized:
raise ValueError(f"Instrument symbol mismatch: {self.symbol} != {normalized}")
return self
def assert_currency(self, currency: BaseCurrency | str) -> InstrumentMetadata:
normalized = BaseCurrency(currency)
if self.quote_currency is not normalized:
raise ValueError(f"Instrument currency mismatch: {self.quote_currency} != {normalized}")
return self
def price_per_weight_from_asset_price(
self,
price: PricePerAsset,
*,
per_unit: WeightUnit = WeightUnit.OUNCE_TROY,
) -> PricePerWeight:
self.assert_symbol(price.symbol)
self.assert_currency(price.currency)
weight_per_share = self.weight_per_share.to_unit(per_unit)
if weight_per_share.amount <= 0:
raise ValueError("Instrument weight_per_share must be positive")
return PricePerWeight(
amount=price.amount / weight_per_share.amount,
currency=price.currency,
per_unit=per_unit,
)
def weight_from_asset_quantity(self, quantity: AssetQuantity) -> Weight:
self.assert_symbol(quantity.symbol)
return Weight(amount=quantity.amount * self.weight_per_share.amount, unit=self.weight_per_share.unit)
def asset_quantity_from_weight(self, weight: Weight) -> AssetQuantity:
normalized_weight = weight.to_unit(self.weight_per_share.unit)
if self.weight_per_share.amount <= 0:
raise ValueError("Instrument weight_per_share must be positive")
return AssetQuantity(amount=normalized_weight.amount / self.weight_per_share.amount, symbol=self.symbol)
_GLD = InstrumentMetadata(
symbol="GLD",
quote_currency=BaseCurrency.USD,
weight_per_share=Weight(amount=Decimal("0.1"), unit=WeightUnit.OUNCE_TROY),
)
_INSTRUMENTS: dict[str, InstrumentMetadata] = {
_GLD.symbol: _GLD,
}
def instrument_metadata(symbol: str) -> InstrumentMetadata:
normalized = str(symbol).strip().upper()
metadata = _INSTRUMENTS.get(normalized)
if metadata is None:
raise ValueError(f"Unsupported instrument metadata: {normalized or symbol!r}")
return metadata
def price_per_weight_from_asset_price(
price: PricePerAsset,
*,
per_unit: WeightUnit = WeightUnit.OUNCE_TROY,
) -> PricePerWeight:
return instrument_metadata(price.symbol).price_per_weight_from_asset_price(price, per_unit=per_unit)
def weight_from_asset_quantity(quantity: AssetQuantity) -> Weight:
return instrument_metadata(quantity.symbol).weight_from_asset_quantity(quantity)
def asset_quantity_from_weight(symbol: str, weight: Weight) -> AssetQuantity:
return instrument_metadata(symbol).asset_quantity_from_weight(weight)

42
app/domain/portfolio_math.py
View File

@@ -1,8 +1,10 @@
from __future__ import annotations
from decimal import Decimal
from typing import Any
from typing import Any, Mapping
from app.domain.backtesting_math import PricePerAsset
from app.domain.instruments import instrument_metadata
from app.domain.units import BaseCurrency, Money, PricePerWeight, Weight, WeightUnit, decimal_from_float
from app.models.portfolio import PortfolioConfig
@@ -42,6 +44,44 @@ def _gold_weight(gold_ounces: float) -> Weight:
return Weight(amount=decimal_from_float(gold_ounces), unit=WeightUnit.OUNCE_TROY)
def _safe_quote_price(value: object) -> float:
try:
parsed = float(value)
except (TypeError, ValueError):
return 0.0
if parsed <= 0:
return 0.0
return parsed
def resolve_portfolio_spot_from_quote(
config: PortfolioConfig,
quote: Mapping[str, object],
*,
fallback_symbol: str | None = None,
) -> tuple[float, str, str]:
configured_price = float(config.entry_price or 0.0)
quote_price = _safe_quote_price(quote.get("price"))
quote_source = str(quote.get("source", "unknown"))
quote_updated_at = str(quote.get("updated_at", ""))
quote_symbol = str(quote.get("symbol", fallback_symbol or "")).strip().upper()
quote_unit = str(quote.get("quote_unit", "")).strip().lower()
if quote_price <= 0 or not quote_symbol or quote_unit != "share":
return configured_price, "configured_entry_price", ""
try:
metadata = instrument_metadata(quote_symbol)
except ValueError:
return configured_price, "configured_entry_price", ""
converted_spot = metadata.price_per_weight_from_asset_price(
PricePerAsset(amount=decimal_from_float(quote_price), currency=BaseCurrency.USD, symbol=quote_symbol),
per_unit=WeightUnit.OUNCE_TROY,
)
return _decimal_to_float(converted_spot.amount), quote_source, quote_updated_at
def portfolio_snapshot_from_config(config: PortfolioConfig | None = None) -> dict[str, float]:
if config is None:
gold_weight = Weight(amount=Decimal("1000"), unit=WeightUnit.OUNCE_TROY)

36
app/pages/overview.py
View File

@@ -7,6 +7,7 @@ from fastapi.responses import RedirectResponse
from nicegui import ui
from app.components import PortfolioOverview
from app.domain.portfolio_math import resolve_portfolio_spot_from_quote
from app.models.workspace import WORKSPACE_COOKIE, get_workspace_repository
from app.pages.common import dashboard_page, quick_recommendations, recommendation_style, strategy_catalog
from app.services.alerts import AlertService, build_portfolio_alert_context
@@ -16,20 +17,10 @@ from app.services.turnstile import load_turnstile_settings
_DEFAULT_CASH_BUFFER = 18_500.0
def _resolve_overview_spot(config, quote: dict[str, object]) -> tuple[float, str, str]:
configured_price = float(config.entry_price or 0.0)
quote_price = float(quote.get("price", configured_price or 0.0) or 0.0)
quote_source = str(quote.get("source", "unknown"))
quote_updated_at = str(quote.get("updated_at", ""))
if configured_price > 0 and quote_price > 0:
ratio = max(configured_price / quote_price, quote_price / configured_price)
if ratio > 1.5:
return configured_price, "configured_entry_price", ""
if quote_price > 0:
return quote_price, quote_source, quote_updated_at
return configured_price, "configured_entry_price", ""
def _resolve_overview_spot(
config, quote: dict[str, object], *, fallback_symbol: str | None = None
) -> tuple[float, str, str]:
return resolve_portfolio_spot_from_quote(config, quote, fallback_symbol=fallback_symbol)
def _format_timestamp(value: str | None) -> str:
@@ -116,7 +107,9 @@ async def overview_page(workspace_id: str) -> None:
data_service = get_data_service()
symbol = data_service.default_symbol
quote = await data_service.get_quote(symbol)
overview_spot_price, overview_source, overview_updated_at = _resolve_overview_spot(config, quote)
overview_spot_price, overview_source, overview_updated_at = _resolve_overview_spot(
config, quote, fallback_symbol=symbol
)
portfolio = build_portfolio_alert_context(
config,
spot_price=overview_spot_price,
@@ -132,6 +125,7 @@ async def overview_page(workspace_id: str) -> None:
else:
quote_status = (
f"Live quote source: {portfolio['quote_source']} · "
f"GLD share quote converted to ozt-equivalent spot · "
f"Last updated {_format_timestamp(str(portfolio['quote_updated_at']))}"
)
@@ -161,15 +155,21 @@ async def overview_page(workspace_id: str) -> None:
if alert_status.history:
latest = alert_status.history[0]
ui.label(
f"Latest alert logged {_format_timestamp(latest.updated_at)} at spot ${latest.spot_price:,.2f}"
f"Latest alert logged {_format_timestamp(latest.updated_at)} at collateral spot ${latest.spot_price:,.2f}"
).classes("text-xs text-slate-500 dark:text-slate-400")
spot_caption = (
f"{symbol} share quote converted to USD/ozt via {portfolio['quote_source']}"
if portfolio["quote_source"] != "configured_entry_price"
else "Configured entry price fallback in USD/ozt"
)
with ui.grid(columns=4).classes("w-full gap-4 max-lg:grid-cols-2 max-sm:grid-cols-1"):
summary_cards = [
(
"Spot Price",
"Collateral Spot Price",
f"${portfolio['spot_price']:,.2f}",
f"{symbol} live quote via {portfolio['quote_source']}",
spot_caption,
),
(
"Margin Call Price",

21
app/services/data_service.py
View File

@@ -49,12 +49,16 @@ class DataService:
return portfolio
async def get_quote(self, symbol: str) -> dict[str, Any]:
cache_key = f"quote:{symbol}"
normalized_symbol = symbol.upper()
cache_key = f"quote:{normalized_symbol}"
cached = await self.cache.get_json(cache_key)
if cached and isinstance(cached, dict):
return cached
normalized_cached = self._normalize_quote_payload(cached, normalized_symbol)
if normalized_cached != cached:
await self.cache.set_json(cache_key, normalized_cached)
return normalized_cached
quote = await self._fetch_quote(symbol)
quote = self._normalize_quote_payload(await self._fetch_quote(normalized_symbol), normalized_symbol)
await self.cache.set_json(cache_key, quote)
return quote
@@ -250,6 +254,7 @@ class DataService:
return {
"symbol": symbol,
"price": round(last, 4),
"quote_unit": "share",
"change": change,
"change_percent": change_percent,
"updated_at": datetime.now(timezone.utc).isoformat(),
@@ -358,11 +363,21 @@ class DataService:
return round((bid + ask) / 2, 4)
return max(bid, ask, 0.0)
@staticmethod
def _normalize_quote_payload(payload: dict[str, Any], symbol: str) -> dict[str, Any]:
normalized = dict(payload)
normalized_symbol = symbol.upper()
normalized["symbol"] = str(normalized.get("symbol", normalized_symbol)).upper()
if normalized["symbol"] == "GLD" and not normalized.get("quote_unit"):
normalized["quote_unit"] = "share"
return normalized
@staticmethod
def _fallback_quote(symbol: str, source: str) -> dict[str, Any]:
return {
"symbol": symbol,
"price": 215.0,
"quote_unit": "share",
"change": 0.0,
"change_percent": 0.0,
"updated_at": datetime.now(timezone.utc).isoformat(),