megaproxy
61a7884ee5
Full-stack self-hosted finance app with FastAPI backend and React frontend. Features: - Accounts, transactions, budgets, investments with GBP base currency - CSV import with auto-detection for 10 UK bank formats - ML predictions: spending forecast, net worth projection, Monte Carlo - 7 selectable themes (Obsidian, Arctic, Midnight, Vault, Terminal, Synthwave, Ledger) - Receipt/document attachments on transactions (JPEG, PNG, WebP, PDF) - AES-256-GCM field encryption, RS256 JWT, TOTP 2FA, RLS, audit log - Encrypted nightly backups + key rotation script - Mobile-responsive layout with bottom nav Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
116 lines
4 KiB
Python
116 lines
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
|
|
ticker = yf.Ticker(query)
|
|
info = ticker.fast_info
|
|
price = getattr(info, "last_price", None)
|
|
if price:
|
|
return [{
|
|
"symbol": query.upper(),
|
|
"name": getattr(info, "long_name", None) or query.upper(),
|
|
"type": "stock",
|
|
"currency": (getattr(info, "currency", None) or "USD").upper(),
|
|
"exchange": getattr(info, "exchange", None),
|
|
"data_source": "yahoo_finance",
|
|
"data_source_id": None,
|
|
}]
|
|
except Exception:
|
|
pass
|
|
return []
|