Blame for app.py

"""Updraft (the Quality-Trend 3 strategy) — automated monthly strategy on an Alpaca paper account, with a public live page.

Public (read-only): GET /, /api/health, /api/status, /api/portfolio, /api/performance, /api/orders,
/api/history (rebalance by rebalance), /api/ranking, /api/plan (dry-run of what a rebalance would do now).
Private: POST /api/rebalance?dry=0 and POST /api/refresh with header X-Rebalance-Token: <REBALANCE_TOKEN>.
The Space schedules itself (qt3/scheduler.py: a 15-minute tick that rebuilds the ranking daily and attempts the
guarded rebalance on weekdays 10:00-15:30 New York); an outside cron calling /api/rebalance is optional. Either
way it trades only on the first trading days of a month (once), while the market is open, on fresh data.
"""
from __future__ import annotations

import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path

from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from starlette.concurrency import run_in_threadpool

from qt3 import config, history, performance, ranking, rebalance, scheduler
from qt3.broker import Alpaca, BrokerError


@asynccontextmanager
async def lifespan(_app):
    if config.NDL_KEY:
        ranking.refresh_async()              # build today's ranking in the background at start-up
    scheduler.start()                        # the Space's own monthly rebalance timer
    yield


app = FastAPI(title="Updraft", lifespan=lifespan)
STATIC = Path(__file__).parent / "static"
app.mount("/static", StaticFiles(directory=STATIC), name="static")
_cache: dict = {}


def cached(key: str, ttl: float, fn):
    hit = _cache.get(key)
    if hit and time.time() - hit[0] < ttl:
        return hit[1]
    val = fn()
    _cache[key] = (time.time(), val)
    return val


def broker() -> Alpaca:
    if not (config.ALPACA_KEY and config.ALPACA_SECRET):
        raise HTTPException(503, "Alpaca keys are not configured on this Space")
    return Alpaca()


def _authorised(token: str) -> None:
    if not config.REBALANCE_TOKEN or token != config.REBALANCE_TOKEN:
        raise HTTPException(403, "rebalance disabled or bad token (set REBALANCE_TOKEN, send X-Rebalance-Token)")


@app.get("/")
def index():
    return FileResponse(STATIC / "index.html")


@app.get("/api/health")
def health():
    return {"ok": True, **config.status(), "ranking_state": ranking.state(), "scheduler": scheduler.STATE}


@app.get("/api/ranking")
def get_ranking():
    if config.NDL_KEY:
        ranking.refresh_async()
    r = ranking.cached() or {}
    age = ranking.age_hours(r or None)                  # inf until the first build finishes (not valid JSON)
    return {**r, "rule": {"names": config.N, "keep_within": config.KEEP, "pool": f"top {round((1 - config.POOL) * 100)}%"},
            "building": ranking.state()["building"], "age_hours": round(age, 1) if age != float("inf") else None}


@app.get("/api/portfolio")
async def portfolio():
    def load():
        a = broker()
        acct, pos = a.account(), a.positions()
        rk = {r["ticker"]: r for r in (ranking.cached() or {}).get("ranked", [])}
        rows = []
        for s, p in sorted(pos.items(), key=lambda x: -x[1]["market_value"]):
            r = rk.get(s, {})
            rows.append({"symbol": s, "name": r.get("name"), "sector": r.get("sector"), "qty": p["qty"],
                         "price": p["price"], "value": p["market_value"],
                         "weight": p["market_value"] / acct["equity"] if acct["equity"] else None,
                         "pl": p["unrealized_pl"], "pl_pct": p["unrealized_plpc"], "today": p["change_today"],
                         "rank_now": r.get("rank")})
        return {"paper": a.paper, "equity": acct["equity"], "cash": acct["cash"],
                "today_change": acct["equity"] - acct["last_equity"] if acct["last_equity"] else None,
                "today_pct": acct["equity"] / acct["last_equity"] - 1 if acct["last_equity"] else None,
                "positions": rows, "as_of": datetime.now(timezone.utc).isoformat(timespec="seconds")}
    try:
        return await run_in_threadpool(cached, "portfolio", 30, load)
    except BrokerError as e:
        raise HTTPException(502, str(e))


@app.get("/api/performance")
async def perf():
    try:
        return await run_in_threadpool(cached, "performance", 60, lambda: performance.series(broker()))
    except BrokerError as e:
        raise HTTPException(502, str(e))


@app.get("/api/orders")
async def orders():
    def load():
        a = broker()
        out = []
        for o in a.orders(status="all"):
            if not str(o.get("client_order_id", "")).startswith(config.TAG + "-"):
                continue
            out.append({"date": (o.get("filled_at") or o.get("submitted_at") or "")[:10], "symbol": o.get("symbol"),
                        "side": o.get("side"), "qty": o.get("filled_qty") or o.get("qty"),
                        "notional": o.get("notional"), "price": o.get("filled_avg_price"), "status": o.get("status")})
        return out[:200]
    try:
        return await run_in_threadpool(cached, "orders", 120, load)
    except BrokerError as e:
        raise HTTPException(502, str(e))


_names: dict = {}


def _short_name(n: str | None) -> str | None:
    """Alpaca's listing name without the share-class tail ("Micron Technology, Inc. Common Stock" → "Micron Technology, Inc.")."""
    if not n:
        return n
    for tail in (" Common Stock", " Class A", " Class B", " Class C", " Ordinary Shares", " American Depositary", " Common Shares"):
        i = n.find(tail)
        if i > 0:
            n = n[:i]
    return n.strip()


@app.get("/api/history")
async def get_history():
    def load():
        a = broker()
        orders, pos = a.all_orders(), a.positions()
        flows = {}
        for f in a.cash_flows():
            flows[f["date"]] = flows.get(f["date"], 0.0) + f["amount"]
        syms = {o["symbol"] for o in orders if str(o.get("client_order_id", "")).startswith(config.TAG + "-")}
        rk = {r["ticker"]: r.get("name") for r in (ranking.cached() or {}).get("ranked", [])}
        missing = sorted(s for s in syms if s not in _names and not rk.get(s))
        if missing:                                  # company names from Alpaca, once per symbol
            try:
                _names.update({s: _short_name(v.get("name")) for s, v in a.assets(missing).items()})
            except BrokerError:
                pass
        return history.build(orders, pos, flows, {s: rk.get(s) or _names.get(s) for s in syms})
    try:
        return await run_in_threadpool(cached, "history", 300, load)
    except BrokerError as e:
        raise HTTPException(502, str(e))


@app.get("/api/status")
async def status():
    def load():
        a = broker()
        chk = rebalance.check(a, ranking.cached())
        clk = a.clock()
        nxt = str(datetime.now(rebalance.NY).date()) if (chk["in_window"] or chk["first_run"]) and not chk["done_this_month"] \
            else chk["next_rebalance"]
        return {"market_open": bool(clk.get("is_open")), "next_open": clk.get("next_open"),
                "next_rebalance": nxt, "done_this_month": chk["done_this_month"], "first_run": chk["first_run"],
                "would_trade_now": chk["ok"], "blocked_by": chk["reasons"], "scheduler": scheduler.STATE,
                **config.status()}
    try:
        return await run_in_threadpool(cached, "status", 60, load)
    except BrokerError as e:
        raise HTTPException(502, str(e))


@app.get("/api/plan")
async def dry_plan():
    r = ranking.cached()
    if not r:
        raise HTTPException(503, "ranking not built yet")
    try:
        return await run_in_threadpool(cached, "plan", 120, lambda: rebalance.build_plan(broker(), r))
    except (BrokerError, ValueError) as e:
        raise HTTPException(502, str(e))


@app.post("/api/refresh")
async def force_refresh(x_rebalance_token: str = Header(default="")):
    _authorised(x_rebalance_token)
    r = await run_in_threadpool(ranking.refresh, True)
    return {"prices_as_of": (r or {}).get("prices_as_of"), "error": ranking.state()["error"]}


@app.post("/api/rebalance")
async def run_rebalance(dry: int = 1, force: int = 0, x_rebalance_token: str = Header(default="")):
    """Dry-run by default. A real run needs dry=0 and the token; it then refreshes the ranking if needed,
    checks the guards (market open, rebalance window, once a month, fresh data) and executes."""
    r = await run_in_threadpool(ranking.refresh)
    if dry:
        return {"dry_run": True, **(await dry_plan())}
    _authorised(x_rebalance_token)
    a = broker()
    try:
        out = await run_in_threadpool(scheduler.run_once, bool(force), a)
    except (BrokerError, ValueError) as e:
        raise HTTPException(502, f"rebalance failed: {e}")
    if out.get("executed"):
        _cache.clear()                       # portfolio, performance and history show the new trades at once
    return {"dry_run": False, **out}