86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
from __future__ import annotations
|
|
import pandas as pd
|
|
import yfinance as yf
|
|
from core.market_data import get_quote, get_history
|
|
|
|
|
|
def calculate_pnl(positions: list[dict], quotes: dict[str, dict]) -> list[dict]:
|
|
result = []
|
|
for pos in positions:
|
|
sym = pos["symbol"]
|
|
q = quotes.get(sym, {})
|
|
price = q.get("price", 0.0)
|
|
shares = pos.get("shares", 0.0)
|
|
avg_cost = pos.get("avg_cost", 0.0)
|
|
market_value = price * shares
|
|
cost_basis = avg_cost * shares
|
|
pnl = market_value - cost_basis
|
|
pnl_pct = (pnl / cost_basis * 100) if cost_basis else 0.0
|
|
result.append({
|
|
**pos,
|
|
"price": price,
|
|
"market_value": market_value,
|
|
"cost_basis": cost_basis,
|
|
"pnl": pnl,
|
|
"pnl_pct": pnl_pct,
|
|
"change_pct": q.get("change_pct", 0.0),
|
|
})
|
|
return result
|
|
|
|
|
|
def get_benchmark_performance(period: str = "1y") -> pd.DataFrame:
|
|
return get_history("SPY", period=period)
|
|
|
|
|
|
def calculate_sector_allocation(positions: list[dict]) -> dict[str, float]:
|
|
sector_values: dict[str, float] = {}
|
|
for pos in positions:
|
|
sector = pos.get("sector", "Unknown") or "Unknown"
|
|
value = pos.get("market_value", 0.0)
|
|
sector_values[sector] = sector_values.get(sector, 0.0) + value
|
|
total = sum(sector_values.values())
|
|
if total == 0:
|
|
return {}
|
|
return {s: (v / total * 100) for s, v in sorted(sector_values.items(), key=lambda x: -x[1])}
|
|
|
|
|
|
def get_portfolio_performance(positions: list[dict], period: str = "1y") -> pd.DataFrame:
|
|
if not positions:
|
|
return pd.DataFrame()
|
|
|
|
symbols = [p["symbol"] for p in positions]
|
|
weights = {}
|
|
total_value = sum(p.get("market_value", 0) for p in positions)
|
|
|
|
if total_value == 0:
|
|
return pd.DataFrame()
|
|
|
|
for p in positions:
|
|
weights[p["symbol"]] = p.get("market_value", 0) / total_value
|
|
|
|
frames = []
|
|
for sym in symbols:
|
|
hist = get_history(sym, period=period)
|
|
if not hist.empty:
|
|
pct = hist["Close"].pct_change().fillna(0)
|
|
pct.name = sym
|
|
frames.append(pct * weights.get(sym, 0))
|
|
|
|
if not frames:
|
|
return pd.DataFrame()
|
|
|
|
combined = pd.concat(frames, axis=1).fillna(0)
|
|
portfolio_returns = combined.sum(axis=1)
|
|
portfolio_cumulative = (1 + portfolio_returns).cumprod() - 1
|
|
|
|
spy = get_history("SPY", period=period)
|
|
if not spy.empty:
|
|
spy_returns = spy["Close"].pct_change().fillna(0)
|
|
spy_cumulative = (1 + spy_returns).cumprod() - 1
|
|
return pd.DataFrame({
|
|
"portfolio": portfolio_cumulative,
|
|
"spy": spy_cumulative,
|
|
}).dropna()
|
|
|
|
return pd.DataFrame({"portfolio": portfolio_cumulative}).dropna()
|