#!/usr/bin/env python3 """极简 HTTP server：把 bot 的实时账本 JSON 暴露成可读的网页。 服务在 0.0.0.0:18080（端口通过 PORT 环境变量改）。 只依赖 Python 3 stdlib。 路由： GET / -> 重定向到 /bot.html GET /bot.html -> 中文实时仪表盘 GET /api/feed -> live/copybot_live.json（paper bot feed） GET /api/live -> live/copybot_live_real.json（live bot feed） GET /api/portfolio -> live/portfolio.json GET /api/health -> 系统健康（服务状态 + 代理 + RTDS + CASH≠CHAIN） GET /api/daily -> daily.sh 最近运行状态 GET /api/wallets -> 钱包建议（当前跟的 vs watch_sharps 排名） """ import http.server import json import os import socketserver import subprocess import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) # /opt/winning-wallet-finder class Handler(http.server.SimpleHTTPRequestHandler): """默认行为是当文件服务器，但 /api/* 走自定义 JSON 接口。""" def do_GET(self): if self.path in ("/", "/bot.html"): self.path = "/bot_dashboard.html" elif self.path == "/api/feed": self._serve_json("copybot_live.json", "paper feed") return elif self.path == "/api/live": self._serve_json("copybot_live_real.json", "live feed") return elif self.path == "/api/portfolio": self._serve_json("portfolio.json", "portfolio") return elif self.path == "/api/health": self._serve_health() return elif self.path == "/api/daily": self._serve_daily() return elif self.path == "/api/wallets": self._serve_wallets() return # 让父类继续处理（serve 文件） return super().do_GET() def _serve_json(self, filename, kind): path = os.path.join(HERE, filename) if not os.path.exists(path): self.send_error(404, f"{kind} not yet written ({filename} missing)") return try: with open(path, "rb") as f: data = f.read() self._send_json_bytes(data) except OSError as e: self.send_error(500, f"read {filename} failed: {e}") def _send_json_bytes(self, data): self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _send_json_obj(self, obj): self._send_json_bytes(json.dumps(obj, ensure_ascii=False).encode()) def _serve_health(self): """系统健康状态：4 个服务 + 代理 + RTDS + CASH≠CHAIN + 心跳。""" health = {"services": {}, "proxy": None, "rtds": None, "cash_drift": None, "last_heartbeat": None, "ok": True} # 1. 4 个 systemd 服务状态 for svc in ("copybot-paper", "copybot-live", "copybot-dashboard", "mihomo"): try: r = subprocess.run(["systemctl", "is-active", svc], capture_output=True, text=True, timeout=3) state = r.stdout.strip() health["services"][svc] = state if state != "active": health["ok"] = False except Exception: health["services"][svc] = "unknown" health["ok"] = False # 2. 代理状态（跑 healthcheck 脚本，3 秒超时） try: r = subprocess.run(["/opt/mihomo-healthcheck.sh"], capture_output=True, timeout=8) health["proxy"] = "ok" if r.returncode == 0 else "fail" if r.returncode != 0: health["ok"] = False except Exception: health["proxy"] = "unknown" # 3. 从 live feed 读 RTDS / CASH≠CHAIN / 心跳 feed_path = os.path.join(HERE, "copybot_live_real.json") if os.path.exists(feed_path): try: feed = json.load(open(feed_path)) updated = feed.get("updated", 0) health["last_heartbeat"] = updated age = time.time() - updated if updated else 999 health["heartbeat_age_s"] = int(age) if age > 300: health["ok"] = False drift = feed.get("ledger_drift", 0) health["cash_drift"] = drift if abs(drift) > 0.5: health["ok"] = False except Exception: pass # 也读 paper feed 的心跳 paper_path = os.path.join(HERE, "copybot_live.json") if os.path.exists(paper_path): try: feed = json.load(open(paper_path)) health["paper_heartbeat"] = feed.get("updated", 0) except Exception: pass self._send_json_obj(health) def _serve_daily(self): """daily.sh 最近运行状态。""" daily = {"last_run": None, "status": "unknown", "skilled_count": 0, "sharps_count": 0} # 读 daily.log 最后一行带时间戳的 log_path = os.path.join(HERE, "daily.log") if os.path.exists(log_path): try: lines = open(log_path, encoding="utf-8", errors="replace").readlines() for line in reversed(lines): if "done ->" in line: daily["status"] = "ok" # 提取时间 "2026-07-19 02:36:56" if "] " in line: ts_part = line.split("] ")[1].split(" ")[0:2] daily["last_run"] = " ".join(ts_part) break elif "[daily]" in line and "error" in line.lower(): daily["status"] = "error" break if daily["status"] == "unknown" and lines: daily["last_run"] = "有日志但未完成" except Exception: pass # 钱包池数量 for name, key in [("watch_skilled.json", "skilled_count"), ("watch_sharps.json", "sharps_count")]: p = os.path.join(HERE, name) if os.path.exists(p): try: data = json.load(open(p)) daily[key] = len(data) if isinstance(data, list) else 0 except Exception: pass self._send_json_obj(daily) def _serve_wallets(self): """钱包建议：当前跟的 vs watch_sharps 排名 + 前瞻盈亏。""" result = {"followed": [], "suggestions": [], "warnings": []} # 1. 读当前跟的钱包（config.live.json） live_cfg_path = os.path.join(ROOT, "config.live.json") followed = [] if os.path.exists(live_cfg_path): try: cfg = json.load(open(live_cfg_path)) followed = cfg.get("wallets", []) except Exception: pass # 2. 读 watch_sharps.json 排名（copy_pnl = in-sample） sharps_path = os.path.join(HERE, "watch_sharps.json") sharps = [] if os.path.exists(sharps_path): try: sharps = json.load(open(sharps_path)) except Exception: pass # 按 copy_pnl 排序 sharps_sorted = sorted(sharps, key=lambda w: w.get("copy_pnl", 0), reverse=True) sharps_addrs = {w.get("wallet", "").lower(): i + 1 for i, w in enumerate(sharps_sorted)} # 3. 读 portfolio.json 的前瞻 realized（out-of-sample，真实表现） fwd_pnl = {} # wallet_addr_lower -> {realized, bets, wins} portfolio_path = os.path.join(HERE, "portfolio.json") if os.path.exists(portfolio_path): try: pf = json.load(open(portfolio_path)) for w in pf.get("wallets", []): addr = w.get("wallet", "").lower() if addr: fwd_pnl[addr] = { "realized": w.get("realized", 0), "bets": w.get("bets", 0), "won": w.get("won", 0), "lost": w.get("lost", 0), "sold": w.get("sold", 0), } except Exception: pass # 4. 标记当前跟的钱包（加前瞻数据） for w in followed: addr = w.get("wallet", "").lower() rank = sharps_addrs.get(addr) name = w.get("name") or addr[:10] fwd = fwd_pnl.get(addr, {}) fwd_realized = fwd.get("realized") entry = { "name": name, "rank": rank, "copy_pnl": None, "fwd_realized": fwd_realized, "fwd_bets": fwd.get("bets", 0), "conv_win": None, "status": "ok" } if rank: sharp_data = next((s for s in sharps_sorted if s.get("wallet", "").lower() == addr), {}) entry["copy_pnl"] = sharp_data.get("copy_pnl", 0) entry["conv_win"] = sharp_data.get("conv_win") else: entry["status"] = "eliminated" result["followed"].append(entry) # 5. 建议加的钱包（前瞻盈利 + copy_pnl 排名前 8 未跟） # 永久排除的钱包（作者明确拒绝，不论前瞻表现） never_follow = {"0xd96750bf8d941a8186e592b0ae6e096da66aa266"} # imwalkinghere: 5min BTC，结算比跟单快 followed_addrs = {w.get("wallet", "").lower() for w in followed} for i, s in enumerate(sharps_sorted): addr = s.get("wallet", "").lower() if addr in followed_addrs or addr in never_follow: continue if i >= 8: break name = s.get("name") or addr[:10] fwd = fwd_pnl.get(addr, {}) fwd_realized = fwd.get("realized") entry = { "name": name, "rank": i + 1, "copy_pnl": s.get("copy_pnl", 0), "fwd_realized": fwd_realized, "fwd_bets": fwd.get("bets", 0), "conv_win": s.get("conv_win"), "wallet": addr } # 前瞻盈利 -> 建议；前瞻亏损 -> 警告；无前瞻数据 -> 未知 if fwd_realized is not None and fwd_realized > 0: result["suggestions"].append(entry) elif fwd_realized is not None and fwd_realized < 0: result["warnings"].append(entry) # 无前瞻数据的不推荐也不警告 self._send_json_obj(result) def end_headers(self): self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") super().end_headers() def log_message(self, fmt, *args): sys.stderr.write(f"[dashboard] {self.address_string()} - {fmt % args}\n") class ReusableTCPServer(socketserver.TCPServer): """默认 TCPServer 关闭前不释放端口（TIME_WAIT），加 SO_REUSEADDR 让重启无缝。""" allow_reuse_address = True def main(): port = int(os.environ.get("PORT", "18080")) os.chdir(HERE) httpd = ReusableTCPServer(("0.0.0.0", port), Handler) sys.stderr.write(f"[dashboard] serving {HERE} on 0.0.0.0:{port}\n") try: httpd.serve_forever() except KeyboardInterrupt: pass finally: httpd.server_close() if __name__ == "__main__": main()