#!/usr/bin/env python3
"""
Dashboard updater โ reads portfolio, fetches live prices, updates data file, generates HTML.
Run after each scan or on demand.
"""
import urllib.request
import json
import os
import re
from datetime import datetime
PORTFOLIO_PATH = "/Users/vpragent/.openclaw/workspace-joe/PORTFOLIO.md"
DATA_PATH = "/Users/vpragent/.openclaw/swing-trader/dashboard_data.json"
HTML_PATH = "/Users/vpragent/.openclaw/swing-trader/dashboard.html"
FORECASTS_PATH = "/Users/vpragent/.openclaw/workspace-karina/FORECASTS.md"
def parse_portfolio():
"""Parse PORTFOLIO.md for open positions."""
positions = []
try:
with open(PORTFOLIO_PATH, "r") as f:
content = f.read()
# Find the table rows
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():
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 ""
positions.append({
"ticker": ticker,
"direction": direction,
"entry": entry,
"size": size,
"sl": sl,
"tp1": tp1,
"date_opened": date_opened
})
except Exception as e:
print(f"Error parsing portfolio: {e}")
return positions
def fetch_price(symbol):
"""Fetch current price from Yahoo Finance."""
try:
url = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?interval=1d&range=5d"
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())
meta = data["chart"]["result"][0]["meta"]
price = meta["regularMarketPrice"]
prev = meta.get("chartPreviousClose", meta.get("previousClose", 0))
day_chg = ((price - prev) / prev * 100) if prev else 0
return {"price": price, "prev_close": prev, "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 fetch_market_data():
"""Fetch key market indices."""
symbols = {
"S&P 500": "^GSPC", "Nasdaq": "^IXIC", "VIX": "^VIX",
"DXY": "DX-Y.NYB", "Oil": "CL=F", "Gold": "GC=F",
"10Y": "^TNX", "BTC": "BTC-USD"
}
results = {}
for name, sym in symbols.items():
results[name] = fetch_price(sym)
return results
def update_data():
"""Update dashboard data with latest snapshot."""
# Load existing data
try:
with open(DATA_PATH, "r") as f:
data = json.load(f)
except:
data = {"snapshots": [], "closed_trades": [], "scan_history": []}
# Parse positions and fetch prices
positions = parse_portfolio()
now = datetime.now().strftime("%Y-%m-%d %H:%M")
total_pnl = 0
total_invested = 0
pos_data = []
for pos in positions:
quote = fetch_price(pos["ticker"])
price = quote["price"]
entry = pos["entry"]
size = pos["size"]
shares = size / entry
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
dist_tp = abs(pos["tp1"] - price) / price * 100
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": round(price, 2),
"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)
})
# Fetch market data
market = fetch_market_data()
snapshot = {
"timestamp": now,
"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": {k: {"price": round(v["price"], 2), "change": v["day_change_pct"]} for k, v in market.items()}
}
data["snapshots"].append(snapshot)
# Keep last 500 snapshots (about 2 months of 10x/day)
if len(data["snapshots"]) > 500:
data["snapshots"] = data["snapshots"][-500:]
with open(DATA_PATH, "w") as f:
json.dump(data, f, indent=2)
return data
def generate_html(data):
"""Generate self-contained HTML dashboard."""
html = f"""
๐ฐ Swing Trader Joe โ Dashboard
๐ Open Positions
| Ticker | Dir | Entry | Current | Today | P&L | P&L % | SL Dist | TP1 Dist | Status |
|---|
"""
with open(HTML_PATH, "w") as f:
f.write(html)
print(f"Dashboard generated: {HTML_PATH}")
if __name__ == "__main__":
print("Updating dashboard data...")
data = update_data()
print(f"Snapshot recorded. Total snapshots: {len(data['snapshots'])}")
generate_html(data)
print("Done!")