#!/usr/bin/env python3 """ Lightweight API server for the Swing Trader Dashboard. Serves static files + /api/live endpoint for real-time price updates. """ import http.server import json import urllib.request import os import threading import time from datetime import datetime PORTFOLIO_PATH = "/Users/vpragent/.openclaw/workspace-joe/PORTFOLIO.md" DATA_PATH = "/Users/vpragent/.openclaw/swing-trader/dashboard_data.json" STATIC_DIR = "/Users/vpragent/.openclaw/swing-trader" # Cache to avoid hammering Yahoo _cache = {"data": None, "ts": 0} CACHE_TTL = 45 # seconds def parse_portfolio(): positions = [] try: with open(PORTFOLIO_PATH, "r") as f: content = f.read() for line in content.split("\n"): line = line.strip() if line.startswith("|") and not line.startswith("| #") and not line.startswith("|--"): cols = [c.strip() for c in line.split("|")[1:-1]] if len(cols) >= 7 and cols[0].isdigit(): positions.append({ "ticker": cols[1].strip(), "direction": cols[2].strip(), "entry": float(cols[3].replace("$", "").replace(",", "")), "size": float(cols[4].replace("$", "").replace(",", "")), "sl": float(cols[5].replace("$", "").replace(",", "")), "tp1": float(cols[6].replace("$", "").replace(",", "")), "date_opened": cols[7].strip() if len(cols) > 7 else "" }) except Exception as e: print(f"Error parsing portfolio: {e}") return positions def fetch_price(symbol): try: url = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?interval=1d&range=10d" req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=10) as r: data = json.loads(r.read()) result = data["chart"]["result"][0] meta = result["meta"] price = meta["regularMarketPrice"] # Use actual last two closes for accurate day change closes = [c for c in result["indicators"]["quote"][0]["close"] if c is not None] prev_close = closes[-2] if len(closes) >= 2 else meta.get("chartPreviousClose", 0) day_chg = ((price - prev_close) / prev_close * 100) if prev_close else 0 return {"price": round(price, 2), "prev_close": round(prev_close, 2), "day_change_pct": round(day_chg, 2)} except Exception as e: return {"price": 0, "prev_close": 0, "day_change_pct": 0, "error": str(e)} def get_live_data(): now = time.time() if _cache["data"] and (now - _cache["ts"]) < CACHE_TTL: return _cache["data"] positions = parse_portfolio() pos_data = [] total_pnl = 0 total_invested = 0 for pos in positions: quote = fetch_price(pos["ticker"]) price = quote["price"] entry = pos["entry"] size = pos["size"] shares = size / entry if entry else 0 pnl = (price - entry) * shares if pos["direction"] == "LONG" else (entry - price) * shares pnl_pct = ((price - entry) / entry * 100) if pos["direction"] == "LONG" else ((entry - price) / entry * 100) dist_sl = abs(price - pos["sl"]) / price * 100 if price else 0 dist_tp = abs(pos["tp1"] - price) / price * 100 if price else 0 total_pnl += pnl total_invested += size pos_data.append({ "ticker": pos["ticker"], "direction": pos["direction"], "entry": entry, "size": size, "sl": pos["sl"], "tp1": pos["tp1"], "date_opened": pos["date_opened"], "current_price": quote["price"], "prev_close": quote["prev_close"], "day_change_pct": quote["day_change_pct"], "pnl": round(pnl, 2), "pnl_pct": round(pnl_pct, 2), "dist_sl_pct": round(dist_sl, 2), "dist_tp_pct": round(dist_tp, 2) }) # Market data market_syms = { "S&P 500": "^GSPC", "Nasdaq": "^IXIC", "VIX": "^VIX", "DXY": "DX-Y.NYB", "Oil": "CL=F", "Gold": "GC=F", "10Y": "^TNX", "BTC": "BTC-USD" } market = {} for name, sym in market_syms.items(): q = fetch_price(sym) market[name] = {"price": q["price"], "change": q["day_change_pct"]} # Load historical snapshots snapshots = [] try: with open(DATA_PATH, "r") as f: snapshots = json.load(f).get("snapshots", []) except: pass result = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "positions": pos_data, "total_pnl": round(total_pnl, 2), "total_invested": total_invested, "total_pnl_pct": round(total_pnl / total_invested * 100, 2) if total_invested else 0, "market": market, "snapshots": snapshots[-100:] # last 100 for equity curve } _cache["data"] = result _cache["ts"] = now return result class DashboardHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=STATIC_DIR, **kwargs) def do_GET(self): if self.path.startswith("/api/live"): data = get_live_data() payload = json.dumps(data).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "no-cache, no-store") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) elif self.path == "/dashboard" or self.path == "/dashboard/": # Redirect /dashboard to /dashboard.html self.send_response(301) self.send_header("Location", "/dashboard.html") self.end_headers() else: super().do_GET() def do_HEAD(self): if self.path.startswith("/api/live"): self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() else: super().do_HEAD() def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS") self.send_header("Access-Control-Allow-Headers", "*") self.end_headers() def log_message(self, format, *args): pass # Silence request logs def background_refresh(): """Pre-warm cache every 45 seconds so API responds instantly.""" while True: try: get_live_data() except: pass time.sleep(45) if __name__ == "__main__": port = 8888 # Start background cache warmer t = threading.Thread(target=background_refresh, daemon=True) t.start() # Pre-warm cache immediately print("Pre-warming cache...") get_live_data() print("Cache ready!") server = http.server.HTTPServer(("0.0.0.0", port), DashboardHandler) print(f"Dashboard server running on http://0.0.0.0:{port}") server.serve_forever()