#!/usr/bin/env python3 """ serve.py — RoaringKittyTracker local dashboard server Usage: python serve.py Opens http://localhost:8765 automatically. The dashboard reads data/outputs/scores_compact.csv on every page load, so refreshing the browser always shows the latest scores. Stop with Ctrl+C. """ import csv import http.server import json import math import os import pathlib import threading import urllib.parse import webbrowser PORT = int(os.environ.get("RKT_PORT", "8765")) HOST = os.environ.get("RKT_HOST", "localhost") OPEN_BROWSER = os.environ.get("RKT_OPEN_BROWSER", "1").lower() not in {"0", "false", "no"} ROOT = pathlib.Path(__file__).parent class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(ROOT), **kwargs) def do_GET(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == "/healthz": self._json_response({"status": "ok"}) elif parsed.path == "/api/scores": self._serve_scores() elif parsed.path == "/api/kulamagi": self._serve_csv(ROOT / "data" / "outputs" / "kulamagi_compact.csv") elif parsed.path == "/api/burry": self._serve_csv(ROOT / "data" / "outputs" / "burry_compact.csv") elif parsed.path == "/api/buffett": self._serve_csv(ROOT / "data" / "outputs" / "buffett_compact.csv") elif parsed.path == "/api/combined": self._serve_csv(ROOT / "data" / "share" / "combined_setups.csv") elif parsed.path == "/api/default-watchlist": self._serve_csv(ROOT / "data" / "share" / "default_watchlist.csv") elif parsed.path.startswith("/api/chart/"): ticker = parsed.path.split("/api/chart/")[-1].upper().strip() self._serve_chart(ticker) elif parsed.path.startswith("/api/info/"): ticker = parsed.path.split("/api/info/")[-1].upper().strip() self._serve_info(ticker) else: super().do_GET() def _serve_scores(self): self._serve_csv(ROOT / "data" / "outputs" / "scores_compact.csv") def _serve_csv(self, csv_path: pathlib.Path): if not csv_path.exists(): msg = f"{csv_path.name} not found. Run the tracker first.".encode() self.send_response(404) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", str(len(msg))) self.end_headers() self.wfile.write(msg) return self._csv_mtime = csv_path.stat().st_mtime rows = [] with open(csv_path, newline="", encoding="utf-8") as f: for row in csv.DictReader(f): rec = {} for k, v in row.items(): v = v.strip() if v in ("", "nan", "None", "NaN", "inf", "-inf", "Inf", "-Inf"): rec[k] = None elif v == "True": rec[k] = True elif v == "False": rec[k] = False else: try: f_val = float(v) rec[k] = None if (math.isnan(f_val) or math.isinf(f_val)) else f_val except ValueError: rec[k] = v rows.append(rec) self._json_response(rows) def _serve_chart(self, ticker: str): """Return last 6 months of OHLCV data from the yfinance history cache.""" hist_path = ROOT / "data" / "raw" / f"hist_{ticker}.csv" if not hist_path.exists(): self._json_response({"error": "not_cached"}) return try: rows = [] with open(hist_path, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: try: # Date column may have timezone suffix — strip it date_str = row.get("Date", "").split(" ")[0].split("T")[0] o = float(row.get("Open", 0) or 0) h = float(row.get("High", 0) or 0) l = float(row.get("Low", 0) or 0) c = float(row.get("Close", 0) or 0) v = float(row.get("Volume", 0) or 0) if c > 0: rows.append({"date": date_str, "o": round(o,4), "h": round(h,4), "l": round(l,4), "c": round(c,4), "v": int(v)}) except (ValueError, TypeError): continue # Return all available history (up to 2 years from yfinance cache) self._json_response(rows) except Exception as e: self._json_response({"error": str(e)}) def _serve_info(self, ticker: str): """Return selected fields from the yfinance info cache for one ticker.""" FIELDS = [ "longName", "sector", "industry", "longBusinessSummary", "country", "website", "fullTimeEmployees", "exchange", "currentPrice", "targetMeanPrice", "targetHighPrice", "targetLowPrice", "fiftyTwoWeekHigh", "fiftyTwoWeekLow", "trailingPE", "forwardPE", "dividendYield", "auditRisk", "boardRisk", "compensationRisk", "shareHolderRightsRisk", ] cache_path = ROOT / "data" / "raw" / f"info_{ticker}.json" if not cache_path.exists(): self._json_response({"error": "not_cached"}) return try: raw = json.loads(cache_path.read_text(encoding="utf-8")) out = {} for f in FIELDS: v = raw.get(f) if v is None: out[f] = None elif isinstance(v, float) and (math.isnan(v) or math.isinf(v)): out[f] = None else: out[f] = v self._json_response(out) except Exception as e: self._json_response({"error": str(e)}) def _json_response(self, obj): body = json.dumps(obj, allow_nan=False, default=str).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-cache") if getattr(self, "_csv_mtime", None): self.send_header("X-Data-Mtime", str(self._csv_mtime)) self._csv_mtime = None self.end_headers() self.wfile.write(body) def log_message(self, fmt, *args): # Only log errors, not every request if args and str(args[1]) not in ("200", "304"): super().log_message(fmt, *args) if __name__ == "__main__": server = http.server.ThreadingHTTPServer((HOST, PORT), Handler) url = f"http://{HOST}:{PORT}/dashboard.html" print(f" RoaringKitty Tracker → {url}") print(" Press Ctrl+C to stop.\n") if OPEN_BROWSER: threading.Timer(0.6, lambda: webbrowser.open(url)).start() try: server.serve_forever() except KeyboardInterrupt: print("\nServer stopped.")