mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
OpenClaw Sub-agent
5252222674
- Add django.contrib.humanize; apply intcomma to every currency amount across invest templates (dashboard, portfolio detail, transactions)
- get_risk_summary now aggregates top_positions by ticker (MRVL across MOMO+IBKR merges into one row with Accounts column), keeping Top 1/3/5 cards and table on the same basis
- Mobile: metric cards all 2-col (2x2 + 2+1 with Last Snapshot spanning full width), no more 1-col break
- Transactions show source badges: purple 🤖 AI (source ai/ocr), grey 手动, amber ⚠ 低置信 when confidence < 0.9
- Delete unused invest/base.html (dead nav; page extends GoLinks base.html)
- tailwind.config.js: add invest/jbot/routermon template dirs so their classes are scanned (text-[10px] etc. were silently missing)
783 lines
30 KiB
Python
783 lines
30 KiB
Python
"""
|
|
Service layer for the invest app.
|
|
|
|
Design goals:
|
|
- Keep ticker/quantity sync simple for AI/OCR workflows.
|
|
- Treat transaction price/currency/fee as optional.
|
|
- Separate account-value growth from cash-flow-adjusted investment return.
|
|
"""
|
|
import json
|
|
import logging
|
|
from datetime import date as date_cls
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from typing import Iterable, Optional
|
|
|
|
from django.db.models import Sum
|
|
from django.utils import timezone
|
|
|
|
from .models import BenchmarkPrice, CashFlow, Portfolio, PortfolioSnapshot, Stock
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Price cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_price_cache: dict[str, tuple[float, datetime]] = {}
|
|
_historical_price_cache: dict[str, tuple[Optional[float], datetime]] = {}
|
|
_chart_cache: dict = {}
|
|
_PRICE_CACHE_TTL_SECONDS = 300
|
|
_HISTORICAL_CACHE_TTL_SECONDS = 3600
|
|
_CHART_CACHE_TTL = 900
|
|
|
|
|
|
SEMI_TICKERS = {'NVDA', 'AMD', 'AVGO', 'TSM', 'ASML', 'MU', 'MRVL', 'INTC', 'SOXX', 'DRAM'}
|
|
AI_CLOUD_TICKERS = {'NVDA', 'AMD', 'AVGO', 'TSM', 'ASML', 'MU', 'MRVL', 'INTC', 'SOXX', 'DRAM', 'NET', 'DDOG', 'GOOG', 'GOOGL', 'MSFT', 'AMZN'}
|
|
|
|
|
|
def _to_float(value) -> Optional[float]:
|
|
if value is None:
|
|
return None
|
|
return float(value)
|
|
|
|
|
|
def _as_date(value) -> Optional[date_cls]:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return timezone.localtime(value).date() if timezone.is_aware(value) else value.date()
|
|
if hasattr(value, 'date') and not isinstance(value, date_cls):
|
|
return value.date()
|
|
if isinstance(value, date_cls):
|
|
return value
|
|
if isinstance(value, str):
|
|
return date_cls.fromisoformat(value)
|
|
return value
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Market data
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _get_yfinance_price(stock_code: str) -> Optional[float]:
|
|
try:
|
|
import yfinance as yf
|
|
|
|
ticker = yf.Ticker(stock_code)
|
|
hist = ticker.history(period="1d")
|
|
if hist.empty:
|
|
return None
|
|
return float(hist["Close"].iloc[-1])
|
|
except Exception as exc:
|
|
logger.warning("yfinance failed for %s: %s", stock_code, exc)
|
|
return None
|
|
|
|
|
|
def get_current_price(stock_code: str) -> Optional[float]:
|
|
now = datetime.now()
|
|
stock_code = stock_code.upper()
|
|
cached = _price_cache.get(stock_code)
|
|
if cached:
|
|
price, cached_at = cached
|
|
if (now - cached_at).total_seconds() < _PRICE_CACHE_TTL_SECONDS:
|
|
return price
|
|
price = _get_yfinance_price(stock_code)
|
|
if price is not None:
|
|
_price_cache[stock_code] = (price, now)
|
|
return price
|
|
if cached:
|
|
logger.info("Using stale cached price for %s", stock_code)
|
|
return cached[0]
|
|
return None
|
|
|
|
|
|
def _get_historical_price(stock_code: str, ref_date) -> Optional[float]:
|
|
"""Return the close on or before ref_date, using BenchmarkPrice then yfinance fallback."""
|
|
ref_date = _as_date(ref_date)
|
|
if not ref_date:
|
|
return None
|
|
stock_code = stock_code.upper()
|
|
cache_key = f"{stock_code}:{ref_date.isoformat()}"
|
|
now = datetime.now()
|
|
|
|
# Prefer explicit DB fixtures/cache rows over in-process cache. Tests and manual backfills
|
|
# may create BenchmarkPrice rows after a previous best-effort yfinance lookup.
|
|
db_price = (
|
|
BenchmarkPrice.objects.filter(ticker=stock_code, date__lte=ref_date)
|
|
.order_by('-date')
|
|
.values_list('close', flat=True)
|
|
.first()
|
|
)
|
|
if db_price is not None:
|
|
price = float(db_price)
|
|
_historical_price_cache[cache_key] = (price, now)
|
|
return price
|
|
|
|
cached = _historical_price_cache.get(cache_key)
|
|
if cached and cached[0] is not None and (now - cached[1]).total_seconds() < _HISTORICAL_CACHE_TTL_SECONDS:
|
|
return cached[0]
|
|
|
|
price = None
|
|
try:
|
|
import yfinance as yf
|
|
|
|
start = ref_date - timedelta(days=7)
|
|
end = ref_date + timedelta(days=1)
|
|
hist = yf.Ticker(stock_code).history(start=start.isoformat(), end=end.isoformat())
|
|
if not hist.empty:
|
|
price = float(hist["Close"].iloc[-1])
|
|
BenchmarkPrice.objects.update_or_create(
|
|
ticker=stock_code,
|
|
date=hist.index[-1].date() if hasattr(hist.index[-1], 'date') else ref_date,
|
|
defaults={'close': Decimal(str(round(price, 6)))},
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("historical price failed for %s @ %s: %s", stock_code, ref_date, exc)
|
|
|
|
_historical_price_cache[cache_key] = (price, now)
|
|
return price
|
|
|
|
|
|
def refresh_benchmark_prices(tickers: Iterable[str] = ('SPY', 'QQQ'), days: int = 540) -> int:
|
|
"""Best-effort benchmark cache refresh. Returns number of rows upserted."""
|
|
try:
|
|
import yfinance as yf
|
|
except Exception as exc:
|
|
logger.warning("yfinance unavailable for benchmark refresh: %s", exc)
|
|
return 0
|
|
|
|
end = timezone.now().date() + timedelta(days=1)
|
|
start = end - timedelta(days=days)
|
|
count = 0
|
|
for ticker in tickers:
|
|
try:
|
|
hist = yf.Ticker(ticker).history(start=start.isoformat(), end=end.isoformat())
|
|
rows = []
|
|
for d, v in hist['Close'].items():
|
|
row_date = d.date() if hasattr(d, 'date') else d
|
|
rows.append(BenchmarkPrice(ticker=ticker.upper(), date=row_date, close=Decimal(str(round(float(v), 6)))))
|
|
if rows:
|
|
BenchmarkPrice.objects.bulk_create(
|
|
rows,
|
|
update_conflicts=True,
|
|
unique_fields=['ticker', 'date'],
|
|
update_fields=['close'],
|
|
)
|
|
count += len(rows)
|
|
except Exception as exc:
|
|
logger.warning("benchmark refresh failed for %s: %s", ticker, exc)
|
|
return count
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Portfolio values
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_portfolio_value(portfolio: Portfolio, reference_date=None) -> dict:
|
|
"""Return live holdings with current prices and optional change vs reference_date."""
|
|
holdings = []
|
|
total_value = Decimal('0')
|
|
|
|
for stock in portfolio.stocks.filter(quantity__gt=0):
|
|
ticker = stock.stock_code.upper()
|
|
price = get_current_price(ticker) or 0.0
|
|
value = Decimal(str(price)) * stock.quantity
|
|
ref_price = _get_historical_price(ticker, reference_date) if reference_date else None
|
|
|
|
price_change = None
|
|
price_change_pct = None
|
|
value_change = None
|
|
if price and ref_price and ref_price > 0:
|
|
price_change = round(price - ref_price, 4)
|
|
price_change_pct = round((price_change / ref_price) * 100, 2)
|
|
value_change = round(price_change * float(stock.quantity), 2)
|
|
|
|
holdings.append({
|
|
'stock_code': ticker,
|
|
'quantity': float(stock.quantity),
|
|
'current_price': price,
|
|
'current_value': float(value),
|
|
'ref_price': ref_price,
|
|
'price_change': price_change,
|
|
'price_change_pct': price_change_pct,
|
|
'value_change': value_change,
|
|
})
|
|
total_value += value
|
|
|
|
return {
|
|
'portfolio_id': portfolio.id,
|
|
'portfolio_name': portfolio.name,
|
|
'holdings': holdings,
|
|
'total_value': float(total_value),
|
|
}
|
|
|
|
|
|
def get_all_holdings(reference_date=None) -> list[dict]:
|
|
palette = [
|
|
{'badge': 'bg-indigo-100 text-indigo-800', 'row': 'bg-indigo-50', 'border': 'border-indigo-200'},
|
|
{'badge': 'bg-emerald-100 text-emerald-800', 'row': 'bg-emerald-50', 'border': 'border-emerald-200'},
|
|
{'badge': 'bg-amber-100 text-amber-800', 'row': 'bg-amber-50', 'border': 'border-amber-200'},
|
|
{'badge': 'bg-rose-100 text-rose-800', 'row': 'bg-rose-50', 'border': 'border-rose-200'},
|
|
{'badge': 'bg-sky-100 text-sky-800', 'row': 'bg-sky-50', 'border': 'border-sky-200'},
|
|
]
|
|
|
|
result = []
|
|
for idx, portfolio in enumerate(Portfolio.objects.all()):
|
|
data = get_portfolio_value(portfolio, reference_date=reference_date)
|
|
result.append({
|
|
'portfolio': portfolio,
|
|
'colors': palette[idx % len(palette)],
|
|
'holdings': data['holdings'],
|
|
'total_value': data['total_value'],
|
|
})
|
|
return result
|
|
|
|
|
|
def _snapshot_asof(portfolio: Portfolio, target_date) -> Optional[float]:
|
|
target_date = _as_date(target_date)
|
|
if not target_date:
|
|
return None
|
|
snap = (
|
|
PortfolioSnapshot.objects.filter(portfolio=portfolio, captured_at__date__lte=target_date)
|
|
.order_by('-captured_at')
|
|
.first()
|
|
)
|
|
return float(snap.total_value) if snap else None
|
|
|
|
|
|
def get_total_value_asof(target_date=None, live_if_today: bool = True) -> Optional[float]:
|
|
target_date = _as_date(target_date)
|
|
today = timezone.now().date()
|
|
portfolios = list(Portfolio.objects.prefetch_related('stocks').all())
|
|
if target_date is None or (live_if_today and target_date == today):
|
|
total = sum(get_portfolio_value(p)['total_value'] for p in portfolios)
|
|
return float(total)
|
|
|
|
values = [_snapshot_asof(p, target_date) for p in portfolios]
|
|
values = [v for v in values if v is not None]
|
|
if not values:
|
|
return None
|
|
return float(sum(values))
|
|
|
|
|
|
def _distinct_snapshot_dates() -> list[date_cls]:
|
|
days = []
|
|
for dt in PortfolioSnapshot.objects.values_list('captured_at', flat=True).order_by('captured_at'):
|
|
day = _as_date(dt)
|
|
if day and day not in days:
|
|
days.append(day)
|
|
return days
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Weekly overview
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_weekly_overview() -> dict:
|
|
"""
|
|
Compute overview from the latest snapshot and the prior snapshot at least 5 days earlier.
|
|
Uses as-of per-portfolio lookups to avoid duplicate/mixed-market snapshot dates double counting.
|
|
"""
|
|
latest_ts = PortfolioSnapshot.objects.order_by('-captured_at').values_list('captured_at', flat=True).first()
|
|
today = timezone.now().date()
|
|
|
|
if latest_ts:
|
|
this_week_date = _as_date(latest_ts)
|
|
snapshots_are_stale = this_week_date < today
|
|
else:
|
|
this_week_date = today
|
|
snapshots_are_stale = True
|
|
|
|
cutoff = this_week_date - timedelta(days=5)
|
|
prev_ts = (
|
|
PortfolioSnapshot.objects.filter(captured_at__date__lte=cutoff)
|
|
.order_by('-captured_at')
|
|
.values_list('captured_at', flat=True)
|
|
.first()
|
|
)
|
|
last_week_date = _as_date(prev_ts) if prev_ts else None
|
|
|
|
this_week_total = get_total_value_asof(this_week_date if not snapshots_are_stale else today)
|
|
last_week_total = get_total_value_asof(last_week_date, live_if_today=False) if last_week_date else None
|
|
|
|
week_gain = None
|
|
week_change_pct = None
|
|
if this_week_total is not None and last_week_total and last_week_total > 0:
|
|
week_gain = this_week_total - last_week_total
|
|
week_change_pct = round((week_gain / last_week_total) * 100, 2)
|
|
|
|
palette = [
|
|
{'badge': 'bg-indigo-100 text-indigo-800', 'row': 'bg-indigo-50', 'border': 'border-indigo-200'},
|
|
{'badge': 'bg-emerald-100 text-emerald-800', 'row': 'bg-emerald-50', 'border': 'border-emerald-200'},
|
|
{'badge': 'bg-amber-100 text-amber-800', 'row': 'bg-amber-50', 'border': 'border-amber-200'},
|
|
{'badge': 'bg-rose-100 text-rose-800', 'row': 'bg-rose-50', 'border': 'border-rose-200'},
|
|
{'badge': 'bg-sky-100 text-sky-800', 'row': 'bg-sky-50', 'border': 'border-sky-200'},
|
|
]
|
|
|
|
portfolio_rows = []
|
|
for idx, portfolio in enumerate(Portfolio.objects.all()):
|
|
this_val = get_portfolio_value(portfolio)['total_value'] if snapshots_are_stale else _snapshot_asof(portfolio, this_week_date)
|
|
last_val = _snapshot_asof(portfolio, last_week_date) if last_week_date else None
|
|
change = change_pct = None
|
|
if this_val is not None and last_val and last_val > 0:
|
|
change = this_val - last_val
|
|
change_pct = round((change / last_val) * 100, 2)
|
|
portfolio_rows.append({
|
|
'portfolio': portfolio,
|
|
'this_week_value': this_val,
|
|
'last_week_value': last_val,
|
|
'change': change,
|
|
'change_pct': change_pct,
|
|
'position_count': portfolio.stocks.filter(quantity__gt=0).count(),
|
|
'colors': palette[idx % len(palette)],
|
|
})
|
|
|
|
return {
|
|
'this_week_total': this_week_total,
|
|
'last_week_total': last_week_total,
|
|
'this_week_date': this_week_date,
|
|
'last_week_date': last_week_date,
|
|
'week_gain': week_gain,
|
|
'week_change_pct': week_change_pct,
|
|
'portfolio_rows': portfolio_rows,
|
|
'portfolio_count': Portfolio.objects.count(),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cash-flow adjusted performance
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _external_cashflows(start=None, end=None, include_start: bool = False):
|
|
qs = CashFlow.objects.all()
|
|
if start:
|
|
start_date = _as_date(start)
|
|
qs = qs.filter(date__gte=start_date) if include_start else qs.filter(date__gt=start_date)
|
|
if end:
|
|
qs = qs.filter(date__lte=_as_date(end))
|
|
return qs.order_by('date', 'created_at')
|
|
|
|
|
|
def _sum_external_cashflows(start=None, end=None, include_start: bool = False) -> Decimal:
|
|
total = Decimal('0')
|
|
for flow in _external_cashflows(start=start, end=end, include_start=include_start):
|
|
total += flow.external_signed_amount
|
|
return total
|
|
|
|
|
|
def get_net_external_cash_flow(start=None, end=None) -> float:
|
|
"""All external deposits/transfers in minus withdrawals/transfers out."""
|
|
return round(float(_sum_external_cashflows(start=start, end=end, include_start=True)), 2)
|
|
|
|
|
|
def _first_performance_date() -> Optional[date_cls]:
|
|
snapshot_date = PortfolioSnapshot.objects.order_by('captured_at').values_list('captured_at', flat=True).first()
|
|
flow_date = CashFlow.objects.order_by('date').values_list('date', flat=True).first()
|
|
candidates = [_as_date(v) for v in (snapshot_date, flow_date) if v]
|
|
return min(candidates) if candidates else None
|
|
|
|
|
|
def _xirr(cashflows: list[tuple[date_cls, Decimal]]) -> Optional[float]:
|
|
if not cashflows:
|
|
return None
|
|
if not any(amount < 0 for _, amount in cashflows) or not any(amount > 0 for _, amount in cashflows):
|
|
return None
|
|
start = cashflows[0][0]
|
|
|
|
def npv(rate: float) -> float:
|
|
total = 0.0
|
|
for flow_date, amount in cashflows:
|
|
years = (flow_date - start).days / 365.0
|
|
total += float(amount) / ((1 + rate) ** years)
|
|
return total
|
|
|
|
low, high = -0.9999, 10.0
|
|
try:
|
|
for _ in range(100):
|
|
mid = (low + high) / 2
|
|
val = npv(mid)
|
|
if abs(val) < 1e-7:
|
|
return round(mid, 6)
|
|
if val > 0:
|
|
low = mid
|
|
else:
|
|
high = mid
|
|
return round((low + high) / 2, 6)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _benchmark_same_cashflow(ticker: str, start: date_cls, end: date_cls, start_value: float, flows) -> Optional[dict]:
|
|
ticker = ticker.upper()
|
|
start_price = _get_historical_price(ticker, start)
|
|
end_price = _get_historical_price(ticker, end)
|
|
if not start_price or not end_price:
|
|
return None
|
|
|
|
units = Decimal(str(start_value)) / Decimal(str(start_price)) if start_value else Decimal('0')
|
|
net_external = Decimal('0')
|
|
for flow in flows:
|
|
price = _get_historical_price(ticker, flow.date)
|
|
if not price:
|
|
continue
|
|
amount = flow.external_signed_amount
|
|
net_external += amount
|
|
units += amount / Decimal(str(price))
|
|
|
|
end_value = units * Decimal(str(end_price))
|
|
cash_adjusted_gain = end_value - Decimal(str(start_value)) - net_external
|
|
capital_base = Decimal(str(start_value)) + sum(
|
|
f.external_signed_amount for f in flows if f.external_signed_amount > 0
|
|
)
|
|
return {
|
|
'ticker': ticker,
|
|
'start_price': round(start_price, 4),
|
|
'end_price': round(end_price, 4),
|
|
'end_value': round(float(end_value), 2),
|
|
'cash_adjusted_gain': round(float(cash_adjusted_gain), 2),
|
|
'simple_return': round(float(cash_adjusted_gain / capital_base), 6) if capital_base > 0 else None,
|
|
}
|
|
|
|
|
|
def get_cashflow_adjusted_performance(start=None, end=None, benchmark_tickers: Iterable[str] = ('QQQ', 'SPY')) -> dict:
|
|
end_date = _as_date(end) or timezone.now().date()
|
|
explicit_start = start is not None
|
|
start_date = _as_date(start) or _first_performance_date() or end_date
|
|
|
|
start_value = get_total_value_asof(start_date, live_if_today=False)
|
|
if start_value is None:
|
|
start_value = 0.0
|
|
end_value = get_total_value_asof(end_date)
|
|
if end_value is None:
|
|
end_value = 0.0
|
|
|
|
include_start_flows = not explicit_start and start_value == 0
|
|
flows = list(_external_cashflows(start=start_date, end=end_date, include_start=include_start_flows))
|
|
net_external = sum((flow.external_signed_amount for flow in flows), Decimal('0'))
|
|
positive_external = sum((flow.external_signed_amount for flow in flows if flow.external_signed_amount > 0), Decimal('0'))
|
|
cash_adjusted_gain = Decimal(str(end_value)) - Decimal(str(start_value)) - net_external
|
|
capital_base = Decimal(str(start_value)) + positive_external
|
|
simple_return = cash_adjusted_gain / capital_base if capital_base > 0 else None
|
|
|
|
xirr_flows = [(start_date, -Decimal(str(start_value)))] if start_value else []
|
|
for flow in flows:
|
|
xirr_flows.append((flow.date, -flow.external_signed_amount))
|
|
xirr_flows.append((end_date, Decimal(str(end_value))))
|
|
|
|
benchmarks = {}
|
|
for ticker in benchmark_tickers:
|
|
bench = _benchmark_same_cashflow(ticker, start_date, end_date, start_value, flows)
|
|
if bench:
|
|
benchmarks[ticker.upper()] = bench
|
|
|
|
return {
|
|
'start_date': start_date.isoformat(),
|
|
'end_date': end_date.isoformat(),
|
|
'start_value': round(start_value, 2),
|
|
'end_value': round(end_value, 2),
|
|
'net_external_cash_flow': round(float(net_external), 2),
|
|
'positive_external_cash_flow': round(float(positive_external), 2),
|
|
'cash_adjusted_gain': round(float(cash_adjusted_gain), 2),
|
|
'simple_return': round(float(simple_return), 6) if simple_return is not None else None,
|
|
'money_weighted_return': _xirr(xirr_flows),
|
|
'cashflows': [
|
|
{
|
|
'id': flow.id,
|
|
'portfolio_id': flow.portfolio_id,
|
|
'flow_type': flow.flow_type,
|
|
'date': flow.date.isoformat(),
|
|
'amount': float(flow.amount),
|
|
'external_signed_amount': float(flow.external_signed_amount),
|
|
'currency': flow.currency,
|
|
}
|
|
for flow in flows
|
|
],
|
|
'benchmarks': benchmarks,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Risk and agent summary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_risk_summary() -> dict:
|
|
holdings = []
|
|
for group in get_all_holdings():
|
|
for holding in group['holdings']:
|
|
holdings.append({
|
|
'portfolio_id': group['portfolio'].id,
|
|
'portfolio_name': group['portfolio'].name,
|
|
**holding,
|
|
})
|
|
|
|
total_value = sum(h['current_value'] for h in holdings)
|
|
|
|
# 集中度口径统一：按 ticker 聚合（同一股票跨账户合并），
|
|
# 与 Top 1/3/5 权重卡片一致，避免 MRVL 等跨账户持仓在表格里重复出现。
|
|
by_ticker: dict[str, dict] = {}
|
|
for h in holdings:
|
|
agg = by_ticker.setdefault(h['stock_code'], {
|
|
'stock_code': h['stock_code'],
|
|
'current_value': 0.0,
|
|
'portfolio_names': [],
|
|
})
|
|
agg['current_value'] += h['current_value']
|
|
if h['portfolio_name'] not in agg['portfolio_names']:
|
|
agg['portfolio_names'].append(h['portfolio_name'])
|
|
|
|
top_positions = sorted(by_ticker.values(), key=lambda r: r['current_value'], reverse=True)
|
|
for row in top_positions:
|
|
row['weight'] = round(row['current_value'] / total_value, 6) if total_value else 0
|
|
|
|
top_1 = top_positions[0]['weight'] if top_positions else 0
|
|
top_3 = sum(r['weight'] for r in top_positions[:3])
|
|
top_5 = sum(r['weight'] for r in top_positions[:5])
|
|
|
|
semi_weight = sum(r['weight'] for r in top_positions if r['stock_code'] in SEMI_TICKERS)
|
|
ai_cloud_weight = sum(r['weight'] for r in top_positions if r['stock_code'] in AI_CLOUD_TICKERS)
|
|
|
|
concentration_level = 'LOW'
|
|
if top_1 >= 0.25 or top_5 >= 0.70:
|
|
concentration_level = 'HIGH'
|
|
elif top_3 >= 0.50 or top_5 >= 0.55:
|
|
concentration_level = 'MEDIUM'
|
|
|
|
return {
|
|
'total_value': round(total_value, 2),
|
|
'position_count': len(top_positions),
|
|
'top_1_weight': round(top_1, 6),
|
|
'top_3_weight': round(top_3, 6),
|
|
'top_5_weight': round(top_5, 6),
|
|
'concentration_level': concentration_level,
|
|
'max_position': top_positions[0] if top_positions else None,
|
|
'top_positions': top_positions[:10],
|
|
'theme_exposure': {
|
|
'semiconductors': round(semi_weight, 6),
|
|
'ai_cloud': round(ai_cloud_weight, 6),
|
|
},
|
|
}
|
|
|
|
|
|
def get_agent_summary() -> dict:
|
|
total_value = get_total_value_asof()
|
|
net_external_all_time = _sum_external_cashflows()
|
|
performance = get_cashflow_adjusted_performance()
|
|
risk = get_risk_summary()
|
|
return {
|
|
'as_of': timezone.now().isoformat(),
|
|
'portfolio_count': Portfolio.objects.count(),
|
|
'total_value': round(total_value or 0, 2),
|
|
'net_external_cash_flow': round(float(net_external_all_time), 2),
|
|
'performance': performance,
|
|
'risk': risk,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Performance chart data (snapshot value % vs benchmarks)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_performance_chart_data() -> Optional[str]:
|
|
latest_snapshot = PortfolioSnapshot.objects.order_by('-id').values_list('id', flat=True).first()
|
|
snapshot_count = PortfolioSnapshot.objects.count()
|
|
cache_key = f'performance:{latest_snapshot}:{snapshot_count}'
|
|
now = datetime.now()
|
|
cached = _chart_cache.get(cache_key)
|
|
if cached:
|
|
data, cached_at = cached
|
|
if (now - cached_at).total_seconds() < _CHART_CACHE_TTL:
|
|
return data
|
|
|
|
result = _build_performance_chart_data()
|
|
_chart_cache.clear()
|
|
_chart_cache[cache_key] = (result, now)
|
|
return result
|
|
|
|
|
|
def _line_dataset(label: str, data: list, color: str, dashed: bool = False, width: float = 2) -> dict:
|
|
return {
|
|
'label': label,
|
|
'data': data,
|
|
'borderColor': color,
|
|
'backgroundColor': color,
|
|
'borderWidth': width,
|
|
'pointRadius': 4 if not dashed else 3,
|
|
'pointHoverRadius': 7 if not dashed else 5,
|
|
'tension': 0.3,
|
|
'borderDash': [5, 5] if dashed else [],
|
|
'fill': False,
|
|
}
|
|
|
|
|
|
def _unique_dates(values: Iterable[date_cls]) -> list[date_cls]:
|
|
result = []
|
|
seen = set()
|
|
for value in values:
|
|
if value and value not in seen:
|
|
result.append(value)
|
|
seen.add(value)
|
|
return result
|
|
|
|
|
|
def _build_performance_chart_data() -> Optional[str]:
|
|
all_snaps = list(PortfolioSnapshot.objects.select_related('portfolio').order_by('captured_at'))
|
|
if not all_snaps:
|
|
return None
|
|
|
|
portfolio_weekly: dict[int, dict[tuple[int, int], tuple[date_cls, float]]] = {}
|
|
week_label_date = {}
|
|
for snap in all_snaps:
|
|
day = _as_date(snap.captured_at)
|
|
key = day.isocalendar()[:2]
|
|
portfolio_weekly.setdefault(snap.portfolio_id, {})
|
|
existing = portfolio_weekly[snap.portfolio_id].get(key)
|
|
if existing is None or day > existing[0]:
|
|
portfolio_weekly[snap.portfolio_id][key] = (day, float(snap.total_value))
|
|
if key not in week_label_date or day > week_label_date[key]:
|
|
week_label_date[key] = day
|
|
|
|
if not week_label_date:
|
|
return None
|
|
|
|
earliest_snapshot_date = min(week_label_date.values())
|
|
latest_date = max(week_label_date.values())
|
|
requested_baseline = date_cls(latest_date.year, 1, 1)
|
|
baseline_total = get_total_value_asof(requested_baseline, live_if_today=False)
|
|
if baseline_total is None or baseline_total <= 0:
|
|
requested_baseline = earliest_snapshot_date
|
|
baseline_total = get_total_value_asof(requested_baseline, live_if_today=False)
|
|
if baseline_total is None or baseline_total <= 0:
|
|
return None
|
|
|
|
chart_dates = _unique_dates(
|
|
[requested_baseline]
|
|
+ [day for _, day in sorted(week_label_date.items()) if day > requested_baseline]
|
|
)
|
|
if not chart_dates:
|
|
return None
|
|
|
|
refresh_needed = not BenchmarkPrice.objects.filter(ticker='QQQ', date__gte=requested_baseline).exists()
|
|
if refresh_needed:
|
|
refresh_benchmark_prices()
|
|
|
|
colors = ['#2563EB', '#7C3AED', '#0D9488', '#DB2777', '#EA580C']
|
|
percentage_datasets = []
|
|
|
|
total_values = []
|
|
total_percent = []
|
|
for day in chart_dates:
|
|
value = get_total_value_asof(day, live_if_today=(day == timezone.now().date()))
|
|
rounded_value = round(value, 2) if value is not None else None
|
|
total_values.append(rounded_value)
|
|
total_percent.append(round((value - baseline_total) / baseline_total * 100, 2) if value is not None else None)
|
|
|
|
percentage_datasets.append(_line_dataset('All Portfolios', total_percent, '#111827', width=3))
|
|
|
|
for idx, portfolio in enumerate(Portfolio.objects.all()):
|
|
base_val = _snapshot_asof(portfolio, requested_baseline)
|
|
if not base_val or base_val <= 0:
|
|
weekly = portfolio_weekly.get(portfolio.id, {})
|
|
if not weekly:
|
|
continue
|
|
first_day, base_val = min(weekly.values(), key=lambda item: item[0])
|
|
if not base_val:
|
|
continue
|
|
data = []
|
|
for day in chart_dates:
|
|
value = _snapshot_asof(portfolio, day)
|
|
data.append(round((value - base_val) / base_val * 100, 2) if value is not None else None)
|
|
percentage_datasets.append(_line_dataset(portfolio.name, data, colors[idx % len(colors)], width=1.75))
|
|
|
|
value_datasets = [_line_dataset('All Portfolios', total_values, '#111827', width=3)]
|
|
|
|
def benchmark_datasets(ticker: str, percent_label: str, value_label: str, color: str) -> tuple[Optional[dict], Optional[dict]]:
|
|
base_price = _get_historical_price(ticker, requested_baseline)
|
|
if not base_price:
|
|
return None, None
|
|
percent_data = []
|
|
value_data = []
|
|
for day in chart_dates:
|
|
price = _get_historical_price(ticker, day)
|
|
if not price:
|
|
percent_data.append(None)
|
|
value_data.append(None)
|
|
continue
|
|
growth_ratio = Decimal(str(price)) / Decimal(str(base_price))
|
|
percent_data.append(round((price - base_price) / base_price * 100, 2))
|
|
value_data.append(round(float(Decimal(str(baseline_total)) * growth_ratio), 2))
|
|
return (
|
|
_line_dataset(percent_label, percent_data, color, dashed=True, width=1.5),
|
|
_line_dataset(value_label, value_data, color, dashed=True, width=1.5),
|
|
)
|
|
|
|
for percent_ds, value_ds in (
|
|
benchmark_datasets('SPY', 'S&P 500', 'S&P 500 benchmark', '#D97706'),
|
|
benchmark_datasets('QQQ', 'QQQ', 'QQQ benchmark', '#16A34A'),
|
|
):
|
|
if percent_ds:
|
|
percentage_datasets.append(percent_ds)
|
|
if value_ds:
|
|
value_datasets.append(value_ds)
|
|
|
|
labels = [day.strftime('%b %-d') for day in chart_dates]
|
|
return json.dumps({
|
|
'labels': labels,
|
|
'baseline_date': requested_baseline.isoformat(),
|
|
'modes': {
|
|
'percentage': {
|
|
'unit': 'percent',
|
|
'description': 'Growth/decline since baseline',
|
|
'datasets': percentage_datasets,
|
|
},
|
|
'value': {
|
|
'unit': 'currency',
|
|
'description': 'Portfolio value and same-baseline benchmark value',
|
|
'datasets': value_datasets,
|
|
},
|
|
},
|
|
})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Holdings sync (AI / manual)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def ai_update_holdings(portfolio: Portfolio, holdings: list[dict], reset: bool = False) -> dict:
|
|
"""Update Stock records. No cost/price tracking required."""
|
|
from django.db import transaction as db_transaction
|
|
|
|
results = []
|
|
with db_transaction.atomic():
|
|
if reset:
|
|
portfolio.stocks.all().delete()
|
|
|
|
for item in holdings:
|
|
stock_code = item['stock_code'].upper()
|
|
quantity = Decimal(str(item['quantity']))
|
|
|
|
stock, created = Stock.objects.update_or_create(
|
|
portfolio=portfolio,
|
|
stock_code=stock_code,
|
|
defaults={'quantity': quantity},
|
|
)
|
|
results.append({
|
|
'stock_code': stock.stock_code,
|
|
'quantity': float(quantity),
|
|
'created': created,
|
|
})
|
|
|
|
return {
|
|
'portfolio_id': portfolio.id,
|
|
'portfolio_name': portfolio.name,
|
|
'reset': reset,
|
|
'results': results,
|
|
}
|