"""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: . 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}