mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
548 lines
21 KiB
Python
548 lines
21 KiB
Python
"""
|
||
Service layer for the invest app.
|
||
Prices fetched from Yahoo Finance on demand via yfinance.
|
||
No cost basis or P&L tracking.
|
||
"""
|
||
import json
|
||
import logging
|
||
from decimal import Decimal
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional
|
||
|
||
from django.db.models import Sum
|
||
|
||
from .models import Portfolio, Stock, PortfolioSnapshot
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# In-process price cache (5 min TTL) + last-week price cache (1 hour TTL)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_price_cache: dict[str, tuple[float, datetime]] = {}
|
||
_PRICE_CACHE_TTL_SECONDS = 300
|
||
|
||
# Cache for historical prices keyed by (stock_code, date_iso) with 1-hour TTL
|
||
_historical_price_cache: dict[str, tuple[Optional[float], datetime]] = {}
|
||
_HISTORICAL_CACHE_TTL_SECONDS = 3600
|
||
|
||
|
||
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_historical_price(stock_code: str, ref_date) -> Optional[float]:
|
||
"""
|
||
Return the closing price on or just before ref_date (handles weekends/holidays).
|
||
ref_date can be a date or datetime object.
|
||
"""
|
||
import datetime as dt
|
||
if hasattr(ref_date, 'date'):
|
||
ref_date = ref_date.date()
|
||
cache_key = f"{stock_code}:{ref_date.isoformat()}"
|
||
now = datetime.now()
|
||
cached = _historical_price_cache.get(cache_key)
|
||
if cached:
|
||
price, cached_at = cached
|
||
if (now - cached_at).total_seconds() < _HISTORICAL_CACHE_TTL_SECONDS:
|
||
return price
|
||
|
||
try:
|
||
import yfinance as yf
|
||
# Look back up to 7 days to find the nearest prior trading day
|
||
start = ref_date - dt.timedelta(days=7)
|
||
end = ref_date + dt.timedelta(days=1) # end is exclusive in yfinance
|
||
hist = yf.Ticker(stock_code).history(start=start.isoformat(), end=end.isoformat())
|
||
if hist.empty:
|
||
price = None
|
||
else:
|
||
price = float(hist["Close"].iloc[-1])
|
||
except Exception as exc:
|
||
logger.warning("yfinance historical price failed for %s @ %s: %s", stock_code, ref_date, exc)
|
||
price = None
|
||
|
||
_historical_price_cache[cache_key] = (price, now)
|
||
return price
|
||
|
||
|
||
def get_current_price(stock_code: str) -> Optional[float]:
|
||
now = datetime.now()
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Portfolio value (live prices, no cost tracking)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def get_portfolio_value(portfolio: Portfolio, reference_date=None) -> dict:
|
||
"""
|
||
Return live holdings with current prices, total value, and weekly price change per stock.
|
||
When reference_date is provided, per-stock change is relative to the closing price on that date
|
||
(the same baseline used by the portfolio-level change in the dashboard header).
|
||
"""
|
||
holdings = []
|
||
total_value = Decimal('0')
|
||
|
||
for stock in portfolio.stocks.filter(quantity__gt=0):
|
||
price = get_current_price(stock.stock_code) or 0.0
|
||
value = Decimal(str(price)) * stock.quantity
|
||
|
||
ref_price = _get_historical_price(stock.stock_code, 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': stock.stock_code,
|
||
'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),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Weekly snapshot overview
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _get_snapshot_total(date) -> Optional[float]:
|
||
result = PortfolioSnapshot.objects.filter(
|
||
captured_at__date=date
|
||
).aggregate(total=Sum('total_value'))['total']
|
||
return float(result) if result is not None else None
|
||
|
||
|
||
def get_weekly_overview() -> dict:
|
||
"""
|
||
Compute overview from the two most recent weekly snapshots.
|
||
'This week' = most recent snapshot date.
|
||
'Last week' = most recent snapshot date at least 5 days earlier (ensuring different week).
|
||
Per-portfolio values use as-of lookups (latest snapshot on or before the target date).
|
||
"""
|
||
latest_ts = (
|
||
PortfolioSnapshot.objects.order_by('-captured_at')
|
||
.values_list('captured_at', flat=True)
|
||
.first()
|
||
)
|
||
if not latest_ts:
|
||
return {
|
||
'this_week_total': None, 'last_week_total': None,
|
||
'this_week_date': None, 'last_week_date': None,
|
||
'week_gain': None, 'week_change_pct': None,
|
||
'portfolio_rows': [], 'portfolio_count': Portfolio.objects.count(),
|
||
}
|
||
|
||
this_week_date = latest_ts.date() if hasattr(latest_ts, 'date') else latest_ts
|
||
last_week_cutoff = this_week_date - timedelta(days=5)
|
||
|
||
prev_ts = (
|
||
PortfolioSnapshot.objects
|
||
.filter(captured_at__date__lte=last_week_cutoff)
|
||
.order_by('-captured_at')
|
||
.values_list('captured_at', flat=True)
|
||
.first()
|
||
)
|
||
last_week_date = (prev_ts.date() if hasattr(prev_ts, 'date') else prev_ts) if prev_ts else None
|
||
|
||
def _snap_asof(portfolio, date):
|
||
"""Most recent snapshot for portfolio on or before date."""
|
||
if not date:
|
||
return None
|
||
s = (
|
||
PortfolioSnapshot.objects
|
||
.filter(portfolio=portfolio, captured_at__date__lte=date)
|
||
.order_by('-captured_at')
|
||
.first()
|
||
)
|
||
return float(s.total_value) if s else None
|
||
|
||
portfolios = list(Portfolio.objects.all())
|
||
this_week_total = sum(v for p in portfolios if (v := _snap_asof(p, this_week_date)) is not None) or None
|
||
last_week_total = sum(v for p in portfolios if (v := _snap_asof(p, last_week_date)) is not None) if last_week_date else None
|
||
if last_week_total == 0:
|
||
last_week_total = None
|
||
|
||
week_gain = None
|
||
week_change_pct = None
|
||
if this_week_total is not None and last_week_total is not None and last_week_total > 0:
|
||
week_gain = this_week_total - last_week_total
|
||
week_change_pct = round((week_gain / last_week_total) * 100, 2)
|
||
|
||
# Per-portfolio breakdown
|
||
_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 = _snap_asof(portfolio, this_week_date)
|
||
last_val = _snap_asof(portfolio, last_week_date)
|
||
|
||
change = change_pct = None
|
||
if this_val is not None and last_val is not None 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(),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Holdings sync (AI / manual)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def get_all_holdings(reference_date=None) -> list[dict]:
|
||
"""
|
||
Return live holdings for every portfolio, grouped for dashboard display.
|
||
reference_date: if provided, per-stock week change is relative to closing prices on that date.
|
||
Each entry: portfolio, portfolio_color_class, holdings (list), total_value
|
||
"""
|
||
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()):
|
||
colors = palette[idx % len(palette)]
|
||
data = get_portfolio_value(portfolio, reference_date=reference_date)
|
||
result.append({
|
||
'portfolio': portfolio,
|
||
'colors': colors,
|
||
'holdings': data['holdings'],
|
||
'total_value': data['total_value'],
|
||
})
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Performance chart data (cumulative % from first snapshot + benchmarks)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Simple in-process cache — benchmarks don't need to refresh every page load
|
||
_chart_cache: dict = {}
|
||
_CHART_CACHE_TTL = 900 # 15 minutes
|
||
|
||
|
||
def get_performance_chart_data() -> Optional[str]:
|
||
"""
|
||
Build Chart.js-ready JSON with cumulative % return from the earliest snapshot.
|
||
Base week = 0%. Each portfolio gets a series; S&P 500 (SPY) and QQQ added as benchmarks.
|
||
Returns a JSON string (safe to pass directly to the template) or None if no snapshots.
|
||
"""
|
||
now = datetime.now()
|
||
cached = _chart_cache.get('performance')
|
||
if cached:
|
||
data, cached_at = cached
|
||
if (now - cached_at).total_seconds() < _CHART_CACHE_TTL:
|
||
return data
|
||
|
||
result = _build_performance_chart_data()
|
||
_chart_cache['performance'] = (result, now)
|
||
return result
|
||
|
||
|
||
def _build_performance_chart_data() -> Optional[str]:
|
||
# Collect all snapshots, deduplicate by ISO week (keep latest date per portfolio per week)
|
||
# This merges HK-market Friday dates with US-market Monday dates for the same week.
|
||
from collections import defaultdict
|
||
|
||
all_snaps = list(
|
||
PortfolioSnapshot.objects.select_related('portfolio').order_by('captured_at')
|
||
)
|
||
if not all_snaps:
|
||
return None
|
||
|
||
# Group each portfolio's snapshots by ISO year-week, keep last per week
|
||
portfolio_weekly: dict = {} # portfolio_id -> {iso_week_key -> (date, value)}
|
||
for snap in all_snaps:
|
||
d = snap.captured_at.date() if hasattr(snap.captured_at, 'date') else snap.captured_at
|
||
key = d.isocalendar()[:2] # (year, week)
|
||
pid = snap.portfolio_id
|
||
if pid not in portfolio_weekly:
|
||
portfolio_weekly[pid] = {}
|
||
existing = portfolio_weekly[pid].get(key)
|
||
# keep the later date within the same week
|
||
if existing is None or d > existing[0]:
|
||
portfolio_weekly[pid][key] = (d, float(snap.total_value))
|
||
|
||
# Build the union of all week keys, sorted chronologically
|
||
all_week_keys = sorted(
|
||
{wk for pw in portfolio_weekly.values() for wk in pw}
|
||
)
|
||
if not all_week_keys:
|
||
return None
|
||
|
||
# Representative label date: latest date seen in that week across all portfolios
|
||
week_label_date: dict = {}
|
||
for pw in portfolio_weekly.values():
|
||
for wk, (d, _) in pw.items():
|
||
if wk not in week_label_date or d > week_label_date[wk]:
|
||
week_label_date[wk] = d
|
||
|
||
# Date range for price lookups (portfolio series + benchmarks share these)
|
||
earliest_date = week_label_date[all_week_keys[0]]
|
||
latest_date = week_label_date[all_week_keys[-1]]
|
||
start_str = (earliest_date - timedelta(days=7)).isoformat()
|
||
end_str = (latest_date + timedelta(days=5)).isoformat()
|
||
|
||
# Pre-fetch the full closing-price history for every currently-held stock in one
|
||
# yfinance call per ticker (not per week). This avoids O(stocks × weeks) fetches
|
||
# and is the same price series used by the dashboard cards.
|
||
def _fetch_closes(stock_code: str) -> list:
|
||
"""Return [(date, close), …] sorted ascending for stock_code over the chart period."""
|
||
try:
|
||
import yfinance as yf
|
||
hist = yf.Ticker(stock_code).history(start=start_str, end=end_str, auto_adjust=True)
|
||
if hist.empty:
|
||
return []
|
||
closes = []
|
||
for d, price in hist['Close'].items():
|
||
date_val = d.date() if hasattr(d, 'date') else d
|
||
closes.append((date_val, float(price)))
|
||
return sorted(closes)
|
||
except Exception as exc:
|
||
logger.warning("invest: price history failed for %s: %s", stock_code, exc)
|
||
return []
|
||
|
||
def _closest_close(closes: list, target_date) -> Optional[float]:
|
||
"""Most recent closing price on or before target_date."""
|
||
result = None
|
||
for d, price in closes:
|
||
if d <= target_date:
|
||
result = price
|
||
else:
|
||
break
|
||
return result
|
||
|
||
stock_histories: dict = {}
|
||
for _p in Portfolio.objects.prefetch_related('stocks').all():
|
||
for _s in _p.stocks.filter(quantity__gt=0):
|
||
if _s.stock_code not in stock_histories:
|
||
stock_histories[_s.stock_code] = _fetch_closes(_s.stock_code)
|
||
|
||
# Per-portfolio cumulative % series — current holdings × historical prices.
|
||
#
|
||
# Using current share counts valued at each historical week's price removes the
|
||
# distortion caused by transactions (capital injections / withdrawals): a BUY that
|
||
# injects cash no longer inflates subsequent snapshot totals, and a SELL no longer
|
||
# deflates them. The resulting series reflects pure market price performance of the
|
||
# positions actually held today — identical in methodology to the week-change figures
|
||
# shown on each portfolio card on the dashboard.
|
||
portfolio_colors = ['#2563EB', '#7C3AED', '#0D9488', '#DB2777', '#EA580C']
|
||
datasets = []
|
||
|
||
for idx, portfolio in enumerate(Portfolio.objects.prefetch_related('stocks').all()):
|
||
current_stocks = list(portfolio.stocks.filter(quantity__gt=0))
|
||
if not current_stocks:
|
||
continue
|
||
|
||
week_totals = []
|
||
for wk in all_week_keys:
|
||
label_date = week_label_date[wk]
|
||
total = 0.0
|
||
all_priced = True
|
||
for stock in current_stocks:
|
||
closes = stock_histories.get(stock.stock_code, [])
|
||
price = _closest_close(closes, label_date)
|
||
if price is None:
|
||
all_priced = False
|
||
break
|
||
total += price * float(stock.quantity)
|
||
week_totals.append(total if all_priced else None)
|
||
|
||
# Base = first week where every position has a price
|
||
base_val = next((v for v in week_totals if v), None)
|
||
if not base_val:
|
||
continue
|
||
|
||
data_pts = [
|
||
round((v - base_val) / base_val * 100, 2) if v is not None else None
|
||
for v in week_totals
|
||
]
|
||
color = portfolio_colors[idx % len(portfolio_colors)]
|
||
datasets.append({
|
||
'label': portfolio.name,
|
||
'data': data_pts,
|
||
'borderColor': color,
|
||
'backgroundColor': color,
|
||
'borderWidth': 2,
|
||
'pointRadius': 5,
|
||
'pointHoverRadius': 7,
|
||
'tension': 0.3,
|
||
'borderDash': [],
|
||
'fill': False,
|
||
})
|
||
|
||
# Benchmark series — start_str / end_str already computed above
|
||
|
||
def _benchmark(ticker: str, label: str, color: str) -> Optional[dict]:
|
||
from .models import BenchmarkPrice
|
||
import datetime as dt
|
||
today = dt.date.today()
|
||
|
||
# Check DB coverage — refresh if we have no rows or latest price > 7 days stale
|
||
qs = BenchmarkPrice.objects.filter(ticker=ticker, date__gte=earliest_date - timedelta(days=7))
|
||
latest_db_date = qs.order_by('-date').values_list('date', flat=True).first()
|
||
need_refresh = latest_db_date is None or (today - latest_db_date).days > 7
|
||
|
||
if need_refresh:
|
||
try:
|
||
import yfinance as yf
|
||
hist = yf.Ticker(ticker).history(start=start_str, end=end_str)
|
||
if not hist.empty:
|
||
rows = []
|
||
for d, v in hist['Close'].items():
|
||
date_val = d.date() if hasattr(d, 'date') else d
|
||
rows.append(BenchmarkPrice(ticker=ticker, date=date_val, close=round(float(v), 4)))
|
||
BenchmarkPrice.objects.bulk_create(rows, update_conflicts=True,
|
||
unique_fields=['ticker', 'date'],
|
||
update_fields=['close'])
|
||
logger.info("invest: cached %d prices for %s", len(rows), ticker)
|
||
except Exception as exc:
|
||
logger.warning("benchmark %s yfinance fetch failed: %s", ticker, exc)
|
||
|
||
try:
|
||
closes = {
|
||
row.date: float(row.close)
|
||
for row in BenchmarkPrice.objects.filter(
|
||
ticker=ticker,
|
||
date__gte=earliest_date - timedelta(days=7),
|
||
date__lte=latest_date + timedelta(days=5),
|
||
).order_by('date')
|
||
}
|
||
if not closes:
|
||
return None
|
||
sorted_trading_days = sorted(closes.keys())
|
||
|
||
def closest_close(target):
|
||
candidates = [td for td in sorted_trading_days if td <= target]
|
||
return closes[candidates[-1]] if candidates else None
|
||
|
||
base_price = closest_close(earliest_date)
|
||
if not base_price:
|
||
return None
|
||
data_pts = [
|
||
round((closest_close(week_label_date[wk]) - base_price) / base_price * 100, 2)
|
||
if closest_close(week_label_date[wk]) is not None else None
|
||
for wk in all_week_keys
|
||
]
|
||
return {
|
||
'label': label,
|
||
'data': data_pts,
|
||
'borderColor': color,
|
||
'backgroundColor': color,
|
||
'borderWidth': 1.5,
|
||
'pointRadius': 3,
|
||
'pointHoverRadius': 5,
|
||
'tension': 0.3,
|
||
'borderDash': [5, 5],
|
||
'fill': False,
|
||
}
|
||
except Exception as exc:
|
||
logger.warning("benchmark %s failed: %s", ticker, exc)
|
||
return None
|
||
|
||
spy = _benchmark('SPY', 'S&P 500', '#D97706')
|
||
qqq = _benchmark('QQQ', 'QQQ', '#16A34A')
|
||
if spy:
|
||
datasets.append(spy)
|
||
if qqq:
|
||
datasets.append(qqq)
|
||
|
||
labels = [week_label_date[wk].strftime('%b %-d') for wk in all_week_keys]
|
||
return json.dumps({'labels': labels, 'datasets': datasets})
|
||
|
||
|
||
def ai_update_holdings(portfolio: Portfolio, holdings: list[dict], reset: bool = False) -> dict:
|
||
"""Update Stock records. No cost/price tracking."""
|
||
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']
|
||
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_code,
|
||
'quantity': float(quantity),
|
||
'created': created,
|
||
})
|
||
|
||
return {
|
||
'portfolio_id': portfolio.id,
|
||
'portfolio_name': portfolio.name,
|
||
'reset': reset,
|
||
'results': results,
|
||
}
|