megaproxy
cdc1e67321
- Add four portfolio charts: allocation donut by holding, allocation donut by asset type, cost basis vs current value bar, return % bar - Fix asset search to use yf.Search() full-text instead of ticker-only lookup — name searches like "vanguard ftse all world" now work - Fix holding creation double-quantity bug: holdings now created with quantity=0 so buy transaction is sole source of quantity/cost basis - Add per-share / total price toggle in Add Holding modal with live calculated equivalent shown as you type - Add ErrorBoundary in AppShell so render errors show a message instead of a blank page - Fix donut charts using || instead of ?? when falling back from current_value to cost_basis_total (0 was not falling through ??) - Allow HoldingCreate.quantity >= 0 (was gt=0) to support zero-init - Fix error display for Pydantic v2 array-of-objects validation errors Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
"""
|
|
Live price fetching: yfinance for stocks/ETFs, CoinGecko for crypto.
|
|
Falls back gracefully — never raises, always returns None on failure.
|
|
"""
|
|
import asyncio
|
|
from datetime import date, datetime, timezone, timedelta
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
import structlog
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
async def _run_sync(fn, *args):
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(None, fn, *args)
|
|
|
|
|
|
def _fetch_yahoo(symbol: str) -> dict | None:
|
|
try:
|
|
import yfinance as yf
|
|
ticker = yf.Ticker(symbol)
|
|
info = ticker.fast_info
|
|
price = getattr(info, "last_price", None) or getattr(info, "regularMarketPrice", None)
|
|
prev = getattr(info, "previous_close", None)
|
|
if price is None:
|
|
return None
|
|
change_24h = None
|
|
if prev and prev > 0:
|
|
change_24h = round((price - prev) / prev * 100, 4)
|
|
return {
|
|
"price": Decimal(str(round(price, 8))),
|
|
"change_24h": Decimal(str(change_24h)) if change_24h is not None else None,
|
|
"currency": (getattr(info, "currency", None) or "USD").upper(),
|
|
"name": getattr(info, "long_name", None) or symbol,
|
|
"exchange": getattr(info, "exchange", None),
|
|
}
|
|
except Exception as exc:
|
|
logger.warning("yahoo_fetch_failed", symbol=symbol, error=str(exc))
|
|
return None
|
|
|
|
|
|
def _fetch_coingecko(coin_id: str) -> dict | None:
|
|
try:
|
|
import requests
|
|
r = requests.get(
|
|
f"https://api.coingecko.com/api/v3/simple/price",
|
|
params={"ids": coin_id, "vs_currencies": "usd,gbp", "include_24hr_change": "true"},
|
|
timeout=10,
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json().get(coin_id, {})
|
|
if not data:
|
|
return None
|
|
return {
|
|
"price": Decimal(str(data.get("gbp", data.get("usd", 0)))),
|
|
"change_24h": Decimal(str(round(data.get("gbp_24h_change", 0), 4))),
|
|
"currency": "GBP",
|
|
"name": coin_id,
|
|
}
|
|
except Exception as exc:
|
|
logger.warning("coingecko_fetch_failed", coin_id=coin_id, error=str(exc))
|
|
return None
|
|
|
|
|
|
def _fetch_yahoo_history(symbol: str, days: int = 365) -> list[dict]:
|
|
try:
|
|
import yfinance as yf
|
|
ticker = yf.Ticker(symbol)
|
|
hist = ticker.history(period=f"{days}d", interval="1d")
|
|
rows = []
|
|
for ts, row in hist.iterrows():
|
|
rows.append({
|
|
"date": ts.date(),
|
|
"open": Decimal(str(round(float(row["Open"]), 8))),
|
|
"high": Decimal(str(round(float(row["High"]), 8))),
|
|
"low": Decimal(str(round(float(row["Low"]), 8))),
|
|
"close": Decimal(str(round(float(row["Close"]), 8))),
|
|
"volume": Decimal(str(int(row.get("Volume", 0) or 0))),
|
|
})
|
|
return rows
|
|
except Exception as exc:
|
|
logger.warning("yahoo_history_failed", symbol=symbol, error=str(exc))
|
|
return []
|
|
|
|
|
|
async def fetch_price(symbol: str, data_source: str, data_source_id: str | None) -> dict | None:
|
|
if data_source == "coingecko":
|
|
return await _run_sync(_fetch_coingecko, data_source_id or symbol.lower())
|
|
return await _run_sync(_fetch_yahoo, symbol)
|
|
|
|
|
|
async def fetch_history(symbol: str, days: int = 365) -> list[dict]:
|
|
return await _run_sync(_fetch_yahoo_history, symbol, days)
|
|
|
|
|
|
def search_yahoo(query: str) -> list[dict]:
|
|
try:
|
|
import yfinance as yf
|
|
results = yf.Search(query, max_results=10).quotes
|
|
out = []
|
|
for r in results:
|
|
symbol = r.get("symbol")
|
|
if not symbol:
|
|
continue
|
|
quote_type = (r.get("quoteType") or r.get("typeDisp") or "stock").lower()
|
|
type_map = {"etf": "etf", "equity": "stock", "cryptocurrency": "crypto",
|
|
"mutualfund": "fund", "bond": "bond"}
|
|
asset_type = type_map.get(quote_type, "stock")
|
|
out.append({
|
|
"symbol": symbol,
|
|
"name": r.get("longname") or r.get("shortname") or symbol,
|
|
"type": asset_type,
|
|
"currency": "USD", # updated on first price sync
|
|
"exchange": r.get("exchDisp") or r.get("exchange"),
|
|
"data_source": "yahoo_finance",
|
|
"data_source_id": None,
|
|
})
|
|
return out
|
|
except Exception:
|
|
pass
|
|
return []
|