giorgio/read.markets
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

phase G: data minimisation + passwordless auth + DeepSeek-first LLM

Browse source 
Server no longer holds portfolios. Holdings live in the browser
(localStorage); the server publishes an anonymous ticker_universe and a
gzipped /api/universe payload identical for every authenticated user, so
access patterns can't betray which tickers a user holds. AI commentary
is generated ephemerally from the browser-supplied pie and the cost
ledger row records no positions. Migrations 0009-0011 added the
universe table and dropped positions / portfolio_snapshots /
portfolios.

Authentication is now e-mail OTP only. Migration 0010 dropped
password_hash and email_verified (every active session is by
construction proof of email control). The /signup endpoint is gone;
signup and login share a single email-entry page. Email rendering is
HTML+plain-text multipart with a shared brand palette (app/branding.py)
asserted in sync with the CSS by a drift-detection test.

LLM provider defaults to DeepSeek-direct (cheaper, api.deepseek.com)
with OpenRouter as automatic fallback if DeepSeek fails. ai_log_job and
indicator_summary_job now iterate the two tones (NOVICE, INTERMEDIATE)
per cycle so the dashboard's tone toggle is instant; PROMPT_VERSION
bumped to 6 with an educational anti-TA / anti-gambling stance baked
into _CORE. NOVICE mode renders a curated glossary inline (CBOE VIX,
yield curve, HY OAS, etc.) with JS-positioned tooltips that survive
viewport edges and sticky bars. Model name and tokens hidden from the
user UI; still recorded in StrategicLog.model and AICall for admin.

Layout adds a sticky top nav, a sticky bottom markets bar (one chip per
exchange with status LED + headline index + 1d change), and
Phase H feedback reporting is queued in tasks/todo.md.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Giorgio Gilestro 2026-05-18 14:16:57 +01:00
parent 480fd311c5
commit 6e7f57c6b2
54 changed files with 5005 additions and 916 deletions
Download patch file Download diff file Expand all files Collapse all files

273
app/routers/api.py
View file

@ -34,9 +34,6 @@ from app.models import (
Headline,
IndicatorSummary,
JobRun,
Portfolio,
PortfolioSnapshot,
Position,
Quote,
StrategicLog,
)
@ -44,7 +41,6 @@ from app.schemas import (
HealthOut,
HeadlineOut,
JobStatus,
PortfolioSummary,
QuoteOut,
StrategicLogOut,
)
@ -52,7 +48,8 @@ from app.schemas import (
router = APIRouter(dependencies=[Depends(require_token)])
JOB_NAMES = ("market_job", "news_job", "portfolio_job", "ai_log_job", "rollup_job")
JOB_NAMES = ("market_job", "news_job", "ai_log_job", "rollup_job",
"indicator_summary_job", "universe_flush_job")
JOB_STALE_HOURS = 2.0 # job is "warn" if its last success was >2h ago
# Per-group expected freshness — bonds and intraday tape want daily data,
@ -133,6 +130,7 @@ async def indicators(
group: str,
request: Request,
as_: str | None = Query(default=None, alias="as"),
tone: str | None = Query(default=None),
session: AsyncSession = Depends(get_session),
):
sub = (
@ -170,12 +168,22 @@ async def indicators(
rows = [r for r in rows if r.symbol in configured]
has_anchor = any((r.changes or {}).get("anchor") is not None for r in rows)
wanted_tone = _resolve_tone_param(tone)
summary = (await session.execute(
select(IndicatorSummary)
.where(IndicatorSummary.group_name == group)
.where(IndicatorSummary.tone == wanted_tone)
.order_by(desc(IndicatorSummary.generated_at))
.limit(1)
)).scalar_one_or_none()
if summary is None:
# Fallback during rollout: any tone for this group.
summary = (await session.execute(
select(IndicatorSummary)
.where(IndicatorSummary.group_name == group)
.order_by(desc(IndicatorSummary.generated_at))
.limit(1)
)).scalar_one_or_none()
# Mark rows whose `as_of` is older than the group-specific threshold.
# Daily-tape groups (bonds, rates, equity, ...) flag stale earlier
@ -195,7 +203,8 @@ async def indicators(
request, "partials/indicators.html",
{"quotes": rows, "has_anchor": has_anchor,
"summary": summary, "notes": notes,
"stale_symbols": stale_symbols},
"stale_symbols": stale_symbols,
"tone": wanted_tone},
)
return [QuoteOut.model_validate(r, from_attributes=True) for r in rows]
@ -257,19 +266,42 @@ def _log_partial_payload(row: StrategicLog | None) -> dict | None:
}
def _resolve_tone_param(tone: str | None) -> str:
"""Normalise a query-param tone to one of the two valid values.
PRO is silently mapped to INTERMEDIATE (see openrouter.PROMPT_VERSION 6)."""
if not tone:
return get_settings().CASSANDRA_TONE.upper()
upper = tone.upper().strip()
if upper in ("NOVICE", "INTERMEDIATE"):
return upper
return "INTERMEDIATE"
@router.get("/log/latest")
async def log_latest(
request: Request,
session: AsyncSession = Depends(get_session),
as_: str | None = Query(default=None, alias="as"),
tone: str | None = Query(default=None),
):
wanted_tone = _resolve_tone_param(tone)
row = (await session.execute(
select(StrategicLog).order_by(desc(StrategicLog.generated_at)).limit(1)
select(StrategicLog)
.where(StrategicLog.tone == wanted_tone)
.order_by(desc(StrategicLog.generated_at))
.limit(1)
)).scalar_one_or_none()
# Fallback during rollout: if the requested tone isn't produced yet,
# serve whatever is latest rather than 404 the panel.
if row is None:
row = (await session.execute(
select(StrategicLog).order_by(desc(StrategicLog.generated_at)).limit(1)
)).scalar_one_or_none()
if as_ == "html":
return templates.TemplateResponse(
request, "partials/log.html", {"log": _log_partial_payload(row)},
request, "partials/log.html",
{"log": _log_partial_payload(row), "tone": wanted_tone},
)
if row is None:
@ -283,22 +315,35 @@ async def log_by_date(
day: str,
session: AsyncSession = Depends(get_session),
as_: str | None = Query(default=None, alias="as"),
tone: str | None = Query(default=None),
):
"""Canonical log for a given day = MAX(generated_at) within that day."""
"""Canonical log for a given day = MAX(generated_at) within that day,
filtered by tone (NOVICE | INTERMEDIATE; default from settings)."""
try:
target = datetime.strptime(day, "%Y-%m-%d").date()
except ValueError:
raise HTTPException(status_code=400, detail="day must be YYYY-MM-DD")
wanted_tone = _resolve_tone_param(tone)
row = (await session.execute(
select(StrategicLog)
.where(func.date(StrategicLog.generated_at) == target)
.where(StrategicLog.tone == wanted_tone)
.order_by(desc(StrategicLog.generated_at))
.limit(1)
)).scalar_one_or_none()
if row is None:
# Fallback: any tone for that day.
row = (await session.execute(
select(StrategicLog)
.where(func.date(StrategicLog.generated_at) == target)
.order_by(desc(StrategicLog.generated_at))
.limit(1)
)).scalar_one_or_none()
if as_ == "html":
return templates.TemplateResponse(
request, "partials/log.html", {"log": _log_partial_payload(row)},
request, "partials/log.html",
{"log": _log_partial_payload(row), "tone": wanted_tone},
)
if row is None:
raise HTTPException(status_code=404, detail="No log on this date")
@ -380,119 +425,9 @@ async def log_days(
return templates.TemplateResponse(request, "partials/calendar.html", payload)
# --- Portfolios --------------------------------------------------------------
# 2 MiB max for CSV uploads — T212 pies don't exceed a few KB in practice.
# Keeps the abuse vector small without rejecting legitimate exports.
_MAX_CSV_BYTES = 2 * 1024 * 1024
@router.post("/portfolios/upload")
async def upload_portfolio_csv(
file: UploadFile = File(...),
portfolio_name: str | None = Form(default=None),
currency: str = Form(default="GBP"),
session: AsyncSession = Depends(get_session),
):
"""Import a Trading 212 pie-export CSV. Parses, resolves each Slice to a
T212 ticker + Yahoo symbol via InstrumentMap, and persists a new
PortfolioSnapshot + Position rows.
No user-id scoping yet — that lands in phase C. Until then, all uploads
land in the single shared portfolio identified by name."""
from app.services.csv_import import CSVImportError, parse_t212_csv, persist_pie
if not file.filename:
raise HTTPException(status_code=400, detail="No file uploaded")
if not file.filename.lower().endswith(".csv"):
raise HTTPException(status_code=400, detail="File must have .csv extension")
raw = await file.read(_MAX_CSV_BYTES + 1)
if len(raw) > _MAX_CSV_BYTES:
raise HTTPException(status_code=413, detail=f"File exceeds {_MAX_CSV_BYTES} bytes")
if not raw:
raise HTTPException(status_code=400, detail="File is empty")
try:
pie = parse_t212_csv(raw)
except CSVImportError as e:
raise HTTPException(status_code=400, detail=str(e))
try:
result = await persist_pie(
session, pie,
portfolio_name=portfolio_name,
currency=currency,
)
except Exception as e:
# Roll back; surface a clean error
await session.rollback()
raise HTTPException(status_code=500, detail=f"Persist failed: {e}")
return {
"portfolio_id": result.portfolio_id,
"snapshot_id": result.snapshot_id,
"portfolio_name": result.portfolio_name,
"is_new_portfolio": result.is_new_portfolio,
"positions": result.positions_written,
"unmapped": result.unmapped_slices,
"invested": pie.invested,
"value": pie.value,
"result": pie.result,
}
@router.get("/portfolios")
async def portfolios(
request: Request,
session: AsyncSession = Depends(get_session),
as_: str | None = Query(default=None, alias="as"),
):
rows: list[PortfolioSummary] = []
for p in (await session.execute(select(Portfolio))).scalars().all():
snap = (await session.execute(
select(PortfolioSnapshot)
.where(PortfolioSnapshot.portfolio_id == p.id)
.order_by(desc(PortfolioSnapshot.snapshot_at))
.limit(1)
)).scalar_one_or_none()
positions: list = []
if snap is not None:
pos = (await session.execute(
select(Position).where(Position.snapshot_id == snap.id)
.order_by(desc(
(Position.quantity * Position.current_price).label("v")
))
)).scalars().all()
positions = [
{"ticker": x.ticker, "name": x.name, "quantity": x.quantity,
"average_price": x.average_price, "current_price": x.current_price,
"ppl": x.ppl,
"ppl_pct": (
(x.current_price - x.average_price) / x.average_price * 100
if x.average_price and x.current_price else None
)}
for x in pos
]
raw = (snap.raw_json or {}) if snap else {}
inv = raw.get("investments") or {}
rows.append(PortfolioSummary(
name=p.name, currency=p.currency,
snapshot_at=snap.snapshot_at if snap else None,
total_value=snap.total_value if snap else None,
cash=snap.cash if snap else None,
invested=snap.invested if snap else None,
total_cost=inv.get("totalCost"),
unrealized_ppl=inv.get("unrealizedProfitLoss"),
realized_ppl=inv.get("realizedProfitLoss"),
positions=positions,
))
if as_ == "html":
return templates.TemplateResponse(
request, "partials/portfolio.html", {"portfolios": rows},
)
return rows
# Portfolio endpoints moved to app/routers/universe.py (Phase G). The
# server no longer persists per-user portfolio data; holdings live in
# the browser's localStorage and prices come from /api/universe.
# --- Health / ops footer -----------------------------------------------------
@ -509,13 +444,23 @@ async def aggregate_summary(
request: Request,
session: AsyncSession = Depends(get_session),
as_: str | None = Query(default=None, alias="as"),
tone: str | None = Query(default=None),
):
wanted_tone = _resolve_tone_param(tone)
row = (await session.execute(
select(IndicatorSummary)
.where(IndicatorSummary.group_name == AGGREGATE_GROUP_NAME)
.where(IndicatorSummary.tone == wanted_tone)
.order_by(desc(IndicatorSummary.generated_at))
.limit(1)
)).scalar_one_or_none()
if row is None:
row = (await session.execute(
select(IndicatorSummary)
.where(IndicatorSummary.group_name == AGGREGATE_GROUP_NAME)
.order_by(desc(IndicatorSummary.generated_at))
.limit(1)
)).scalar_one_or_none()
from app.services.markets import all_statuses
statuses = all_statuses()
@ -523,7 +468,7 @@ async def aggregate_summary(
if as_ == "html":
return templates.TemplateResponse(
request, "partials/dashboard_header.html",
{"summary": row, "markets": statuses},
{"summary": row, "markets": statuses, "tone": wanted_tone},
)
return {
"summary": (
@ -538,6 +483,86 @@ async def aggregate_summary(
}
# Market → headline index mapping for the sticky bottom bar. Symbols must
# be present in config/default.toml so market_job populates `quotes`.
_MARKET_INDEX = {
"NYSE": ("^GSPC", "S&P 500"),
"LSE": ("^FTSE", "FTSE 100"),
# XETRA → Euro Stoxx 50 rather than ^GDAXI: Yahoo's DAX ticker is
# patchy via the chart endpoint, and ^STOXX50E is already tracked in
# config/default.toml's equity group.
"XETRA": ("^STOXX50E", "STOXX 50"),
"JPX": ("^N225", "Nikkei 225"),
"HKEX": ("^HSI", "Hang Seng"),
"SSE": ("000300.SS", "CSI 300"),
}
def _fmt_price(p: float | None) -> str:
if p is None:
return "—"
if abs(p) >= 1000:
return f"{p:,.0f}"
if abs(p) >= 100:
return f"{p:,.1f}"
return f"{p:,.2f}"
@router.get("/markets-bar", response_class=HTMLResponse, include_in_schema=False)
async def markets_bar(
request: Request,
session: AsyncSession = Depends(get_session),
as_: str | None = Query(default=None, alias="as"),
):
"""The sticky bottom-bar payload: per-market open/close status with the
market's headline index price + 1d change. Refreshed by HTMX every 60s.
"""
from app.services.markets import all_statuses
statuses = all_statuses()
# Latest quote per headline-index symbol in one query.
wanted_syms = [sym for sym, _ in _MARKET_INDEX.values()]
sub = (
select(Quote.symbol, func.max(Quote.fetched_at).label("mx"))
.where(Quote.symbol.in_(wanted_syms))
.group_by(Quote.symbol)
.subquery()
)
rows = (await session.execute(
select(Quote).join(
sub,
(Quote.symbol == sub.c.symbol) & (Quote.fetched_at == sub.c.mx),
)
)).scalars().all()
by_sym = {q.symbol: q for q in rows}
markets: list[dict] = []
for st in statuses:
sym, label = _MARKET_INDEX.get(st["code"], (None, None))
q = by_sym.get(sym) if sym else None
idx = None
if q is not None and q.price is not None:
idx = {
"symbol": q.symbol,
"label": label,
"price_fmt": _fmt_price(q.price),
"change_1d_pct": (q.changes or {}).get("1d"),
}
markets.append({
"code": st["code"],
"label": st["label"],
"open": st["open"],
"until_iso": st["until"].isoformat(),
"until_hhmm": st["until"].strftime("%H:%M"),
"index": idx,
})
return templates.TemplateResponse(
request, "partials/markets_bar.html",
{"markets": markets},
)
@router.get("/health", response_class=HTMLResponse, include_in_schema=False)
async def health_html(
request: Request,

190
app/routers/auth.py
View file

@ -1,8 +1,19 @@
"""Authentication routes: /login, /signup, /logout.
"""Authentication routes: /login, /verify, /verify/resend, /logout.
These do NOT depend on require_auth (they're how you become authenticated).
The router is included separately in app/main.py without a router-level
auth dependency.
Cassandra is passwordless. Single auth flow:
GET /login → enter email
POST /login → get_or_create_user → issue OTP → send → 303 /verify
GET /verify → enter 6-digit code (email shown from pending cookie)
POST /verify → validate → set session → 303 /
POST /verify/resend → reissue OTP (rate-limited)
Signup and login are intentionally the same path — typing your email is
sign-in if you've been here before, sign-up otherwise. No UI signal
distinguishes the two, which also masks user-enumeration.
The /signup endpoints from the previous auth scheme are gone. Anything
that linked to /signup should now link to /login.
"""
from __future__ import annotations
@ -12,13 +23,26 @@ from fastapi import APIRouter, Depends, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import SESSION_COOKIE_NAME, SESSION_TTL_SECONDS, sign_session
from app.auth import (
PENDING_COOKIE_NAME,
PENDING_TTL_SECONDS,
SESSION_COOKIE_NAME,
SESSION_TTL_SECONDS,
sign_pending,
sign_session,
verify_pending,
)
from app.config import get_settings
from app.db import get_session
from app.services.auth_service import AuthError, authenticate, create_user
from app.db import get_session, utcnow
from app.logging import get_logger
from app.services.auth_service import AuthError, get_or_create_user, get_user
from app.services import otp_service
from app.services.email_service import EmailSendError, send_otp
from app.templates_env import templates
log = get_logger("auth_router")
router = APIRouter(tags=["auth"])
@ -26,7 +50,6 @@ def _safe_next(next_value: str | None) -> str:
"""Only allow same-origin relative paths to prevent open-redirect."""
if not next_value or not next_value.startswith("/") or next_value.startswith("//"):
return "/"
# Block any embedded scheme or host.
if urlparse(next_value).netloc:
return "/"
return next_value
@ -39,19 +62,49 @@ def _set_session_cookie(response: RedirectResponse, user_id: int) -> None:
max_age=SESSION_TTL_SECONDS,
httponly=True,
samesite="lax",
# `secure=True` requires HTTPS; the operator should enable this in
# production via a reverse proxy. Off for local-dev convenience.
secure=False,
path="/",
)
def _set_pending_cookie(response: RedirectResponse, email: str, user_id: int) -> None:
response.set_cookie(
key=PENDING_COOKIE_NAME,
value=sign_pending(email, user_id),
max_age=PENDING_TTL_SECONDS,
httponly=True,
samesite="lax",
secure=False,
path="/",
)
def _clear_pending_cookie(response) -> None:
response.delete_cookie(PENDING_COOKIE_NAME, path="/")
async def _issue_and_send_otp(session: AsyncSession, email: str) -> bool:
"""Generate a code, persist its hash, send the email. Returns True on
success. Returns False (and logs) if SMTP submission fails — the OTP
row is still created so the user can hit /verify/resend."""
code = await otp_service.issue(session, email, purpose="auth")
try:
await send_otp(email, code, otp_service.OTP_TTL_MINUTES)
return True
except EmailSendError:
return False
# ---------------------------------------------------------------------------
# Login (email entry)
# ---------------------------------------------------------------------------
@router.get("/login", response_class=HTMLResponse)
async def login_page(request: Request, next: str | None = None, error: str | None = None):
return templates.TemplateResponse(
request, "login.html",
{"next_path": _safe_next(next), "error": error,
"signup_enabled": get_settings().CASSANDRA_SIGNUP_ENABLED},
{"next_path": _safe_next(next), "error": error},
)
@ -59,73 +112,124 @@ async def login_page(request: Request, next: str | None = None, error: str | Non
async def login_submit(
request: Request,
email: str = Form(...),
password: str = Form(...),
next: str | None = Form(default=None),
session: AsyncSession = Depends(get_session),
):
s = get_settings()
try:
user = await authenticate(session, email, password)
user = await get_or_create_user(
session, email, create_if_missing=s.CASSANDRA_SIGNUP_ENABLED,
)
except AuthError as e:
return templates.TemplateResponse(
request, "login.html",
{"next_path": _safe_next(next), "error": str(e),
"email": email,
"signup_enabled": get_settings().CASSANDRA_SIGNUP_ENABLED},
{"next_path": _safe_next(next), "error": str(e), "email": email},
status_code=400,
)
target = _safe_next(next)
resp = RedirectResponse(url=target, status_code=303)
_set_session_cookie(resp, user.id)
# Issue OTP only if cooldown allows; if a fresh one was sent in the
# last 60s we just reuse the existing one (silently) to avoid
# spamming the user's inbox on a refreshed form submit.
allowed, _ = await otp_service.can_request_new(session, user.email)
if allowed:
await _issue_and_send_otp(session, user.email)
resp = RedirectResponse(url="/verify", status_code=303)
_set_pending_cookie(resp, user.email, user.id)
return resp
@router.get("/signup", response_class=HTMLResponse)
async def signup_page(request: Request, error: str | None = None):
s = get_settings()
if not s.CASSANDRA_SIGNUP_ENABLED:
return templates.TemplateResponse(
request, "login.html",
{"next_path": "/", "error": "Sign-ups are currently disabled. Ask the operator.",
"signup_enabled": False},
status_code=403,
)
# ---------------------------------------------------------------------------
# Verify (code entry)
# ---------------------------------------------------------------------------
@router.get("/verify", response_class=HTMLResponse)
async def verify_page(request: Request, error: str | None = None, sent: str | None = None):
cookie = request.cookies.get(PENDING_COOKIE_NAME)
pending = verify_pending(cookie) if cookie else None
if pending is None:
return RedirectResponse(url="/login", status_code=303)
return templates.TemplateResponse(
request, "signup.html",
{"error": error},
request, "verify.html",
{"email": pending["email"], "error": error, "sent": sent,
"ttl_minutes": otp_service.OTP_TTL_MINUTES,
"resend_cooldown": otp_service.RESEND_COOLDOWN_SECONDS},
)
@router.post("/signup")
async def signup_submit(
@router.post("/verify")
async def verify_submit(
request: Request,
email: str = Form(...),
password: str = Form(...),
code: str = Form(...),
session: AsyncSession = Depends(get_session),
):
s = get_settings()
if not s.CASSANDRA_SIGNUP_ENABLED:
cookie = request.cookies.get(PENDING_COOKIE_NAME)
pending = verify_pending(cookie) if cookie else None
if pending is None:
return RedirectResponse(url="/login", status_code=303)
email = pending["email"]
try:
user = await create_user(session, email, password)
except AuthError as e:
await otp_service.verify(session, email, code)
except otp_service.OTPError as e:
return templates.TemplateResponse(
request, "signup.html",
{"error": str(e), "email": email},
request, "verify.html",
{"email": email, "error": str(e),
"ttl_minutes": otp_service.OTP_TTL_MINUTES,
"resend_cooldown": otp_service.RESEND_COOLDOWN_SECONDS},
status_code=400,
)
user = await get_user(session, pending["uid"])
if user is None:
# User row vanished between cookie issue and verify. Restart flow.
return RedirectResponse(url="/login", status_code=303)
user.last_login_at = utcnow()
await session.commit()
log.info("user.login", user_id=user.id, email=email)
resp = RedirectResponse(url="/", status_code=303)
_set_session_cookie(resp, user.id)
_clear_pending_cookie(resp)
return resp
@router.post("/verify/resend")
async def verify_resend(
request: Request,
session: AsyncSession = Depends(get_session),
):
cookie = request.cookies.get(PENDING_COOKIE_NAME)
pending = verify_pending(cookie) if cookie else None
if pending is None:
return RedirectResponse(url="/login", status_code=303)
email = pending["email"]
allowed, wait = await otp_service.can_request_new(session, email)
if not allowed:
return RedirectResponse(
url=f"/verify?error=Please+wait+{wait}s+before+requesting+another+code",
status_code=303,
)
ok = await _issue_and_send_otp(session, email)
msg = "A new code has been sent" if ok else "Could not send email — try again shortly"
return RedirectResponse(url=f"/verify?sent={msg}", status_code=303)
# ---------------------------------------------------------------------------
# Logout
# ---------------------------------------------------------------------------
@router.post("/logout")
async def logout(request: Request):
resp = RedirectResponse(url="/login", status_code=303)
resp.delete_cookie(SESSION_COOKIE_NAME, path="/")
_clear_pending_cookie(resp)
return resp
@router.get("/logout")
async def logout_get(request: Request):
# Convenience for users who click a logout link rather than POSTing.
return await logout(request)

351
app/routers/universe.py Normal file
View file

@ -0,0 +1,351 @@
"""Phase G endpoints — the data-minimised path that replaces per-user
portfolio persistence.
Four routes:
- GET /api/universe Full ticker universe + prices.
Identical payload for every
authenticated user — request
bodies don't leak which
tickers belong to which user.
- GET /api/universe/sparkline/{ticker} Lazy per-ticker sparkline,
fetched on hover from the
browser.
- POST /api/portfolio/parse CSV → parsed pie back to
browser localStorage. Seeds
ticker_universe (no user FK).
No DB writes for positions.
- POST /api/analyze Ephemeral AI commentary.
Pie passed in via JSON body,
held in memory for one LLM
call, discarded on response.
All routes require authentication (session cookie OR bearer token). The
old endpoints in `app/routers/api.py` (`/api/portfolios/upload`,
`/api/portfolio/{name}/summary`) remain live until step 10 of the Phase G
plan, when they're removed alongside the table drops.
"""
from __future__ import annotations
import asyncio
from datetime import datetime, timedelta, timezone
import httpx
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
from fastapi.responses import JSONResponse
from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import require_auth
from app.config import get_settings
from app.db import get_session, utcnow
from app.logging import get_logger
from app.models import Quote, QuoteDaily
from app.services import fx, portfolio_analysis, ticker_universe
from app.services.csv_import import CSVImportError, parse_t212_csv
from app.services.instrument_map import resolve_slice
from app.services.market import fetch as market_fetch
log = get_logger("universe_router")
router = APIRouter(dependencies=[Depends(require_auth)])
# Hard caps on inbound payload sizes. Anything bigger is rejected with 4xx
# rather than tying up an LLM call or a CSV parser.
MAX_CSV_BYTES = 1_048_576 # 1 MB
MAX_ANALYZE_JSON_BYTES = 256 * 1024 # 256 KB
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
# ---------------------------------------------------------------------------
# GET /api/universe — full ticker universe with prices
# ---------------------------------------------------------------------------
@router.get("/universe")
async def get_universe(session: AsyncSession = Depends(get_session)) -> JSONResponse:
"""Return every ticker tracked by Cassandra, with its latest quote.
The response is intentionally the *whole* universe — never filtered
per user — so the access pattern (request body, return body) carries
no information about which tickers belong to which user. Browser
filters down to its own holdings client-side.
Cache-Control: 60s — the browser refreshes once a minute, matching
market_job's hourly write cadence with slack."""
tickers = await ticker_universe.get_all_tickers(session)
out: dict[str, dict] = {}
if tickers:
# Latest quote per ticker within the last 24h. Older = considered
# broken feed; we drop it rather than serve stale data.
cutoff = _utcnow() - timedelta(hours=24)
subq = (
select(Quote.symbol, func.max(Quote.fetched_at).label("max_fetched"))
.where(Quote.symbol.in_(tickers))
.where(Quote.fetched_at >= cutoff)
.group_by(Quote.symbol)
.subquery()
)
stmt = (
select(Quote)
.join(
subq,
and_(
Quote.symbol == subq.c.symbol,
Quote.fetched_at == subq.c.max_fetched,
),
)
)
rows = (await session.execute(stmt)).scalars().all()
for q in rows:
if q.price is None:
continue
price = q.price
currency = q.currency
# LSE tickers come back from Yahoo in pence (GBp / GBX) but
# T212 CSV invested-value is reported in pounds. Normalise to
# pounds here so the browser never has to know about the
# pence quirk. Daily change percentages are unit-independent.
if currency in ("GBp", "GBX"):
price = price / 100.0
currency = "GBP"
out[q.symbol] = {
"p": price,
"c": currency,
"d": q.changes or {},
}
# FX rates for every currency present, against a USD pivot. Browser
# uses these to convert each position into the pie's base currency
# before computing P/L. Same payload for every user.
needed_ccy = {q.get("c") for q in out.values() if q.get("c")}
# Always include the common bases so the browser has them even if
# no current position is denominated in them (e.g. avg cost in GBP
# but no LSE holding right now).
needed_ccy.update({"USD", "EUR", "GBP"})
try:
fx_rates = await fx.get_rates(needed_ccy)
except Exception as e:
log.warning("universe.fx_failed", error=str(e)[:200])
fx_rates = {"USD": 1.0}
body = {
"as_of": _utcnow().isoformat(),
"tickers": out,
"fx": fx_rates,
}
return JSONResponse(
body,
headers={
"Cache-Control": "max-age=60",
"Vary": "Accept-Encoding",
},
)
# ---------------------------------------------------------------------------
# GET /api/universe/sparkline/{ticker} — lazy per-ticker history
# ---------------------------------------------------------------------------
@router.get("/universe/sparkline/{ticker}")
async def get_sparkline(
ticker: str,
session: AsyncSession = Depends(get_session),
) -> JSONResponse:
"""Daily closes for the last ~60 days. Browser fetches on hover, so
we cache aggressively. 404 if the symbol has no daily rollup yet."""
ticker = ticker.strip().upper()[:32]
if not ticker:
raise HTTPException(status_code=400, detail="ticker required")
rows = (await session.execute(
select(QuoteDaily.date, QuoteDaily.close)
.where(QuoteDaily.symbol == ticker)
.where(QuoteDaily.close.is_not(None))
.order_by(QuoteDaily.date.desc())
.limit(60)
)).all()
if not rows:
raise HTTPException(status_code=404, detail=f"no sparkline data for {ticker}")
series = [{"d": r.date.isoformat(), "c": r.close} for r in reversed(rows)]
return JSONResponse(
{"ticker": ticker, "series": series},
headers={"Cache-Control": "max-age=300"},
)
# ---------------------------------------------------------------------------
# POST /api/portfolio/parse — CSV → parsed pie for browser localStorage
# ---------------------------------------------------------------------------
@router.post("/portfolio/parse")
async def parse_portfolio(
file: UploadFile = File(...),
session: AsyncSession = Depends(get_session),
) -> dict:
"""Parse a T212 pie-export CSV. Returns the structured pie to the
browser to be stashed in localStorage. **Does NOT persist holdings.**
Side effects on the server:
- Resolved Yahoo tickers are buffered into ticker_universe (no user
FK, timing-leak mitigation via 5-min batch flush in scheduler).
- last_referenced_at is bumped on any ticker already in the universe.
"""
raw = await file.read()
if len(raw) > MAX_CSV_BYTES:
raise HTTPException(status_code=413, detail="CSV too large (1 MB max)")
if not raw:
raise HTTPException(status_code=400, detail="empty CSV")
try:
pie = parse_t212_csv(raw)
except CSVImportError as e:
raise HTTPException(status_code=400, detail=str(e))
positions_out: list[dict] = []
yahoo_tickers: list[str] = []
unmapped: list[str] = []
for p in pie.positions:
resolved = await resolve_slice(session, p.slice)
if resolved is None or not resolved.yahoo_ticker:
unmapped.append(p.slice or p.name or "?")
continue
positions_out.append({
"yahoo_ticker": resolved.yahoo_ticker,
"t212_slice": p.slice,
"name": resolved.name or p.name,
"qty": p.quantity,
"avg_cost": p.average_price, # @property — no call parens
"currency": resolved.currency,
})
yahoo_tickers.append(resolved.yahoo_ticker)
# Synchronous upsert: bypass the Redis buffer so the dashboard has
# live prices immediately. The buffer + flush machinery remains for
# multi-user timing-mitigation when we hit >=10 concurrent users.
upserted = await ticker_universe.upsert_tickers(session, yahoo_tickers)
# Also drop into the Redis buffer so flush_buffer's existing tests +
# ledger remain coherent if/when we re-enable buffered-only mode.
buffered = await ticker_universe.buffer_tickers(yahoo_tickers)
# Inline price fetch for the resolved tickers, so /api/universe has
# something to return on the very first dashboard load after upload.
# Bounded concurrency to keep Yahoo happy.
fetched_ok = 0
if yahoo_tickers:
anchor = get_settings().CASSANDRA_ANCHOR_DATE or None
now = utcnow()
sem = asyncio.Semaphore(16)
async def _fetch_one(client, sym):
async with sem:
return await market_fetch(client, sym, sym, "", anchor)
try:
async with httpx.AsyncClient(follow_redirects=True, timeout=20) as client:
quotes = await asyncio.gather(
*(_fetch_one(client, t) for t in yahoo_tickers),
return_exceptions=True,
)
for sym, q in zip(yahoo_tickers, quotes):
if isinstance(q, Exception):
log.warning("portfolio.parse.fetch_failed", symbol=sym, error=str(q)[:120])
continue
session.add(Quote(
symbol=q.symbol, source=q.source, label=q.label,
group_name="universe", price=q.price, currency=q.currency,
as_of=q.as_of, changes=q.changes or None,
error=(q.error[:250] if q.error else None),
fetched_at=now,
))
if q.price is not None:
fetched_ok += 1
await session.commit()
except Exception as e:
log.error("portfolio.parse.fetch_block_failed", error=str(e)[:200])
log.info(
"portfolio.parse",
positions=len(positions_out),
unmapped=len(unmapped),
upserted=upserted,
buffered=buffered,
priced=fetched_ok,
)
warnings = []
if unmapped:
warnings.append(
f"{len(unmapped)} position(s) could not be resolved to Yahoo tickers: "
+ ", ".join(unmapped[:10])
+ (" ..." if len(unmapped) > 10 else "")
)
return {
"pie_name": pie.name,
"base_currency": "GBP",
"positions": positions_out,
"totals": {
"invested": pie.invested,
"value": pie.value,
"result": pie.result,
},
"warnings": warnings,
}
# ---------------------------------------------------------------------------
# POST /api/analyze — ephemeral AI commentary
# ---------------------------------------------------------------------------
@router.post("/analyze")
async def analyze_portfolio(
request: Request,
session: AsyncSession = Depends(get_session),
) -> dict:
"""Generate AI commentary for the supplied pie. The pie is held in
memory only for the duration of the LLM call; nothing about holdings
is persisted. The ai_calls ledger row records tokens + cost, never
holdings."""
# Read JSON body manually so we can enforce a hard size cap. FastAPI's
# default body limit is generous; we want tighter control here.
body = await request.body()
if len(body) > MAX_ANALYZE_JSON_BYTES:
raise HTTPException(status_code=413, detail="payload too large")
try:
payload = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="malformed JSON body")
try:
req = portfolio_analysis.parse_request(payload)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
try:
result = await portfolio_analysis.analyse(session, req)
except RuntimeError as e:
log.error("analyze.llm_failed", error=str(e)[:200])
raise HTTPException(status_code=502, detail="analysis failed — try again")
return {
"content": result.content,
"model": result.model,
"generated_at": result.generated_at.isoformat(),
"prompt_tokens": result.prompt_tokens,
"completion_tokens": result.completion_tokens,
"cost_usd": result.cost_usd,
}