Blame for app.py

"""Portfolio Analyst — FastAPI app (HF Space, sdk: docker, port 7860).

Transport adapted from crucible/app.py: SSE progress stream + serialized runs + watchdog +
per-day run cap. Simplification vs crucible: the pipeline runs in-process and pushes events
onto a per-run queue — no subprocess, no stdout sentinel parsing.
"""
from __future__ import annotations

import json
import os
import queue
import threading
from concurrent.futures import ThreadPoolExecutor
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path

from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from src import ledger
from src.orchestrator import safe_run
from src.portfolio import LookupUnavailable, resolve_symbol, validate_row

ROOT = Path(__file__).resolve().parent
DAILY_CAP = int(os.environ.get("RUNS_PER_DAY", "25"))
RUN_TIMEOUT_S = int(os.environ.get("RUN_TIMEOUT_S", "720"))
MAX_HOLDINGS = int(os.environ.get("MAX_HOLDINGS", "50"))

app = FastAPI(title="Vitals")

try:   # European factor panel: refresh in the background at boot (never blocks, yields to runs)
    from src import europanel as _EP
    _EP.BUSY = lambda: _run_lock._value < MAX_CONCURRENT_RUNS     # noqa — analyses hold the semaphore
    _EP.ensure_async()
except Exception:
    pass

# ---------------- run manager: one analysis at a time, events on a queue ----------------

_runs: dict[str, dict] = {}          # run_id -> {queue, status, result, started}
MAX_CONCURRENT_RUNS = int(os.environ.get("MAX_CONCURRENT_RUNS", "2"))
_run_lock = threading.BoundedSemaphore(MAX_CONCURRENT_RUNS)   # a few analyses at once, not one
_day_count = {"day": "", "n": 0}


def _reserve_run() -> None:
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    if _day_count["day"] != today:
        _day_count.update(day=today, n=0)
    if _day_count["n"] >= DAILY_CAP:
        raise HTTPException(429, f"Daily analysis cap reached ({DAILY_CAP}/day). Try again tomorrow.")
    _day_count["n"] += 1


def _worker(run_id: str, rows: list[dict], context: dict | None) -> None:
    run = _runs[run_id]

    def emit(pct: int, caption: str, chip: str | None = None) -> None:
        run["queue"].put({"type": "progress", "pct": pct, "caption": caption, "chip": chip})

    with _run_lock:
        try:
            result = safe_run(rows, emit, context, budget_s=run.get("timeout", RUN_TIMEOUT_S))
            run["result"] = result
            run["status"] = "done"
            run["queue"].put({"type": "done", "result": result})
        except Exception as e:
            run["status"] = "error"
            run["queue"].put({"type": "error", "message": str(e)})


class ResolveReq(BaseModel):
    query: str
    entry: float | None = None


class AnalyzeReq(BaseModel):
    rows: list[dict]
    context: dict | None = None


_prewarm_pool = ThreadPoolExecutor(max_workers=3)
_prewarmed: dict[str, float] = {}


def _prewarm(symbol: str, data: dict, price) -> None:
    """Background: everything the analysis will need for this holding, fetched while the
    user is still reviewing rows — deep fundamentals, price history, street view, news, events."""
    try:
        from src import fundamentals as FU
        from src.market_data import fetch_history, fetch_news, fetch_events
        if FU.eligible(symbol, data.get("quote_type")):
            FU.enrich_us(data, price)
        fetch_history([symbol], period="8y")
        fetch_news(symbol)
        fetch_events(symbol)
    except Exception:
        pass


def _resolve_one(query: str, entry) -> dict:
    try:
        r = resolve_symbol(query)
    except LookupUnavailable as e:
        return {"ok": False, "error": str(e), "retry": True}
    except ValueError as e:
        return {"ok": False, "error": str(e)}
    except Exception as e:                 # a bug or an outage must read as "retry", never as a 500
        import traceback
        traceback.print_exc()
        return {"ok": False, "error": f"Lookup for {query!r} hit an error ({type(e).__name__}) — click to retry.",
                "retry": True}
    if not r.get("price"):
        return {"ok": False, "error": f"No live price for {query!r} right now — click to retry.", "retry": True}
    out = {"ok": True, "symbol": r["symbol"], "name": r.get("name"), "price": r["price"],
           "currency": r["data"].get("currency") or "USD", "sector": r["data"].get("sector")}
    if entry and entry > 0:
        out["pl_pct"] = r["price"] / entry - 1
        try:   # anomaly nudges (pre-split entry, wrong listing, implausible P/L)
            out["warnings"] = validate_row(r["symbol"], r["data"], r["price"], entry, query)
            for w in out["warnings"]:
                # the wire code differs from the internal one on purpose: a browser still running
                # last week's script auto-applied "presplit"; on "presplit_ask" it falls back to its
                # generic "apply or keep yours" prompt — a stale client can no longer rewrite an entry
                if w.get("code") == "presplit":
                    w["code"] = "presplit_ask"
        except Exception:
            out["warnings"] = []
    now = time.time()
    try:
        if now - _prewarmed.get(r["symbol"], 0) > 3600:      # once an hour per symbol
            _prewarmed[r["symbol"]] = now
            _prewarm_pool.submit(_prewarm, r["symbol"], dict(r["data"]), r["price"])
    except Exception:
        pass                                   # pre-warming is a courtesy, never a failed lookup
    return out


@app.post("/api/resolve")
def api_resolve(req: ResolveReq):
    """Instant per-row feedback while the user types: resolved name + live price + P/L."""
    return _resolve_one(req.query, req.entry)


class ResolveBatchReq(BaseModel):
    rows: list[dict]


@app.post("/api/resolve_batch")
def api_resolve_batch(req: ResolveBatchReq):
    """Import prefill: resolve every row in parallel on the server (one round trip)."""
    rows = [r for r in req.rows if str(r.get("query", "")).strip()][:MAX_HOLDINGS]
    def one(r):
        try:
            return _resolve_one(str(r.get("query", "")).strip(), r.get("entry"))
        except Exception as e:
            return {"ok": False, "error": str(e)[:120]}
    with ThreadPoolExecutor(max_workers=8) as ex:
        results = list(ex.map(one, rows))
    syms = [r["symbol"] for r in results if r.get("ok")]
    if syms:   # batched Sharadar statements + batched news for the whole import, in the background
        def _batch():
            try:
                from src import fundamentals as FU
                from src.market_data import fetch_news_batch
                FU.prefetch_batch(syms)
                fetch_news_batch(syms)
            except Exception:
                pass
        _prewarm_pool.submit(_batch)
    return {"results": results}


@app.post("/api/analyze")
def api_analyze(req: AnalyzeReq):
    rows = [r for r in req.rows if str(r.get("query", "")).strip()][:MAX_HOLDINGS]
    if not rows:
        raise HTTPException(400, "Add at least one holding (ticker/company + entry price).")
    _reserve_run()
    run_id = uuid.uuid4().hex[:12]
    _runs[run_id] = {"queue": queue.Queue(), "status": "running",
                     "result": None, "started": time.time(),
                     "timeout": max(RUN_TIMEOUT_S, 360 + 14 * len(rows))}   # a 47-name book gets ~17 min
    allowed = {"horizon", "risk", "objective", "loss_tolerance", "liquidity_need", "exclusions", "max_positions", "wrapper"}
    context = {k: str(v)[:120] for k, v in (req.context or {}).items() if k in allowed and v}
    _runs[run_id]["rows"] = rows
    _runs[run_id]["context"] = context or None
    threading.Thread(target=_worker, args=(run_id, rows, context or None), daemon=True).start()
    # opportunistic cleanup of runs older than an hour
    cutoff = time.time() - 3600
    for rid in [r for r, v in _runs.items() if v["started"] < cutoff]:
        _runs.pop(rid, None)
    return {"run_id": run_id}


@app.get("/api/events/{run_id}")
def api_events(run_id: str):
    run = _runs.get(run_id)
    if not run:
        raise HTTPException(404, "Unknown run — it may have expired. Start a new analysis.")

    def stream():
        # replay terminal state for reconnects (browser refresh mid-run)
        if run["status"] == "done" and run["result"]:
            yield "data: " + json.dumps({"type": "done", "result": run["result"]}) + "\n\n"
            return
        deadline = run["started"] + run.get("timeout", RUN_TIMEOUT_S)
        while True:
            try:
                ev = run["queue"].get(timeout=15)
                yield "data: " + json.dumps(ev, default=str) + "\n\n"
                if ev["type"] in ("done", "error"):
                    return
            except queue.Empty:
                if time.time() > deadline:  # watchdog: a hung run can't hold the stream forever
                    yield "data: " + json.dumps(
                        {"type": "error", "message": "Analysis timed out — please try again."}) + "\n\n"
                    return
                yield ": keepalive\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


@app.get("/api/result/{run_id}")
def api_result(run_id: str):
    run = _runs.get(run_id)
    if not run or not run.get("result"):
        raise HTTPException(404, "No result for this run.")
    return run["result"]


@app.get("/api/report/{run_id}")
def api_report(run_id: str):
    run = _runs.get(run_id)
    if not run or not run.get("result"):
        raise HTTPException(404, "No report for this run.")
    from src.pdf_report import build_pdf
    try:
        pdf = build_pdf(run["result"])
    except Exception:   # PDF layout must never lose the report — fall back to markdown
        import traceback
        traceback.print_exc()
        md = run["result"].get("report_md", "")
        return PlainTextResponse(md, media_type="text/markdown", headers={
            "Content-Disposition": f'attachment; filename="vitals-report-{run_id}.md"'})
    from fastapi.responses import Response
    return Response(pdf, media_type="application/pdf", headers={
        "Content-Disposition": f'attachment; filename="vitals-report-{run_id}.pdf"'})


@app.get("/api/audit/{run_id}")
def api_audit(run_id: str):
    """Full run artifact — inputs, data snapshot with sources/as-of, every lens output,
    model + prompt versions, decisions. What lets a verdict be reproduced and defended."""
    run = _runs.get(run_id)
    if not run or not run.get("result"):
        raise HTTPException(404, "No audit trail for this run.")
    body = json.dumps(run["result"].get("audit") or {}, indent=1, default=str)
    return PlainTextResponse(body, media_type="application/json", headers={
        "Content-Disposition": f'attachment; filename="vitals-audit-{run_id}.json"'})


class SaveReq(BaseModel):
    run_id: str
    code: str | None = None


@app.post("/api/save")
def api_save(req: SaveReq):
    """Save a finished run into the verdict ledger (new code), or append the run to an
    already-saved portfolio's history when a code is supplied."""
    run = _runs.get(req.run_id)
    if not run or not run.get("result"):
        raise HTTPException(404, "Run the analysis first — nothing to save yet.")
    if req.code and ledger.get_portfolio(req.code):
        ledger.append_run(req.code, run.get("rows") or [], run.get("context"), run["result"])
        return {"code": req.code.strip().lower(), "appended": True}
    try:
        code = ledger.save_portfolio(run.get("rows") or [], run.get("context"), run["result"])
    except ValueError as e:
        raise HTTPException(429, str(e))
    return {"code": code, "appended": False}


@app.get("/api/portfolio/{code}")
def api_portfolio(code: str):
    p = ledger.get_portfolio(code)
    if not p:
        raise HTTPException(404, "No portfolio under that code.")
    runs = p.get("runs") or []
    return {"code": p["code"], "rows": p.get("rows"), "context": p.get("context"),
            "created": p.get("created"),
            "last_run": {"ts": runs[-1]["ts"], "health": runs[-1]["health"],
                         "n_holdings": len(runs[-1]["verdicts"])} if runs else None,
            "n_runs": len(runs),
            "alerts": list(reversed(p.get("alerts") or []))[:12],
            "track": ledger.score_track_record(p)}


_check_count = {"day": "", "n": 0}


@app.post("/api/check/{code}")
def api_check(code: str):
    """User-triggered standing-watch check: deterministic alerts + LLM event scan."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    if _check_count["day"] != today:
        _check_count.update(day=today, n=0)
    if _check_count["n"] >= 100:
        raise HTTPException(429, "Daily check limit reached — try tomorrow.")
    _check_count["n"] += 1
    out = ledger.check_portfolio(code, with_events=True)
    if out is None:
        raise HTTPException(404, "No portfolio under that code.")
    return out


@app.get("/api/stats")
def api_stats():
    return ledger.global_stats()


_import_count = {"day": "", "n": 0}


@app.post("/api/import")
async def api_import(file: UploadFile = File(...)):
    """Extract holdings from an uploaded CSV/Excel/PDF. Parsed in memory, never stored;
    the result only prefills the editable table for the user to review and confirm."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    if _import_count["day"] != today:
        _import_count.update(day=today, n=0)
    if _import_count["n"] >= 50:
        raise HTTPException(429, "Daily import limit reached — enter holdings manually.")
    _import_count["n"] += 1
    from src.importer import extract_rows
    data = await file.read()
    try:
        return extract_rows(file.filename or "", data)
    except ValueError as e:
        raise HTTPException(422, str(e))


@app.get("/api/sample")
def api_sample():
    return json.loads((ROOT / "sample_portfolio.json").read_text())


def _build_version() -> str:
    """Hash of the server code: lets a deploy gate tell the rebuilt container from the old one."""
    import hashlib
    h = hashlib.sha1()
    for f in sorted((ROOT / "src").glob("*.py")) + [ROOT / "app.py"]:
        try:
            h.update(f.read_bytes())
        except Exception:
            pass
    return h.hexdigest()[:10]


BUILD_VERSION = _build_version()


@app.get("/api/health")
def health():
    return {"ok": True, "build": BUILD_VERSION, "assets": ASSET_VERSION}


def _asset_version() -> str:
    """Short hash of the front-end files: a redeploy changes the URLs the page loads, so a
    browser can never keep last week's app.js (a user once saw the pre-split auto-rewrite a
    day after it had been replaced by the confirmation prompt — heuristic caching on a
    'Last-Modified' with no Cache-Control)."""
    import hashlib
    h = hashlib.sha1()
    for name in ("app.js", "style.css"):
        try:
            h.update((ROOT / "static" / name).read_bytes())
        except Exception:
            pass
    return h.hexdigest()[:10]


ASSET_VERSION = _asset_version()


@app.get("/")
def index():
    html = (ROOT / "static" / "index.html").read_text()
    html = html.replace('/static/app.js"', f'/static/app.js?v={ASSET_VERSION}"') \
               .replace('/static/style.css"', f'/static/style.css?v={ASSET_VERSION}"')
    return HTMLResponse(html, headers={"Cache-Control": "no-cache, must-revalidate"})


@app.middleware("http")
async def _no_stale_assets(request, call_next):
    """Static files revalidate on every load (ETag makes that a 304 — cheap) instead of being
    heuristically cached for days."""
    resp = await call_next(request)
    if request.url.path.startswith("/static/") or request.url.path == "/":
        resp.headers["Cache-Control"] = "no-cache, must-revalidate"
    return resp


app.mount("/static", StaticFiles(directory=ROOT / "static"), name="static")

ledger.start_sweep()   # standing watch: quant-only re-checks while the Space is awake