| """金仓 v3 回测对比台（本地 UI，纯标准库）。 |
| |
| 启动：python ui_server.py → http://127.0.0.1:8321 |
| |
| GET / 前端页面（ui/index.html） |
| GET /api/presets 预设方案（v2.1 / F5 / 组合 等，2013–2025 基准） |
| POST /api/backtest 自由参数回测：{target_vol, cash_yield, exit_line, floor, breaker, start, end} |
| |
| 提醒（纪律）：正式研究判读窗口为 2013–2025；2026-09-12 起为前向影子期。 |
| UI 的自由参数仅供探索与核对，正式采纳任何配置仍须走计划书预注册流程。 |
| """ |
| import json |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from pathlib import Path |
|
|
| import engine as E |
| from overlays import build_overlays |
| from run import one_stats, yearly_table |
|
|
| HERE = Path(__file__).resolve().parent |
| PORT = 8321 |
| OV = build_overlays() |
| FLOORS = {"sma": "sma12m_cap", "mom": "mom12_1_cap", "climate": "climate_cap"} |
| PRESET_DEFS = [ |
| ("v2.1（12%，固定102）", "退役对照", 12.0, None), |
| ("F5（12%，动态线）", "默认候选", 12.0, {"cb_line": OV["breakeven"]}), |
| ("组合（8%+动量地板+动态线）", "参考配置（D19 后未达采纳线）", 8.0, |
| {"cap_half": OV["mom12_1_cap"], "cb_line": OV["breakeven"]}), |
| ("F2b 动量地板（12%）", "单项", 12.0, {"cap_half": OV["mom12_1_cap"]}), |
| ("F5+动量地板（12%）", "单项组合", 12.0, |
| {"cap_half": OV["mom12_1_cap"], "cb_line": OV["breakeven"]}), |
| ] |
|
|
|
|
| def run_full(start, end, vol, overlay): |
| summary, st, data = E.run(start, end, target_vol_ann=vol, overlay=overlay, tag=None) |
| curve = data["curve"] |
| return summary, st, curve, data["trades"], one_stats([c["V"] for c in curve]) |
|
|
|
|
| _PRESET_CACHE = None |
|
|
|
|
| def preset_payload(): |
| global _PRESET_CACHE |
| if _PRESET_CACHE is not None: |
| return _PRESET_CACHE |
| out = [] |
| for label, note, vol, ov in PRESET_DEFS: |
| summary, _, _, _, s = run_full("2013-01-01", "2025-12-31", vol, ov) |
| out.append({"label": label, "note": note, "stats": s, |
| "n_trades": summary["n_trades"], "fees": round(summary["fees"], 1)}) |
| _PRESET_CACHE = out |
| return out |
|
|
|
|
| def overlay_from(p): |
| ov = {} |
| if p.get("cash_yield"): |
| ov["cash_yield"] = float(p["cash_yield"]) |
| if p.get("exit_line") == "dynamic": |
| ov["cb_line"] = OV["breakeven"] |
| fl = p.get("floor", "none") |
| if fl in FLOORS: |
| ov["cap_half"] = OV[FLOORS[fl]] |
| if p.get("breaker"): |
| ov["breaker"] = True |
| return ov |
|
|
|
|
| class Handler(BaseHTTPRequestHandler): |
| def _send(self, code, body, ctype="application/json; charset=utf-8"): |
| self.send_response(code) |
| self.send_header("Content-Type", ctype) |
| self.send_header("Content-Length", str(len(body))) |
| self.end_headers() |
| self.wfile.write(body) |
|
|
| def do_GET(self): |
| if self.path in ("/", "/index.html"): |
| self._send(200, (HERE / "ui" / "index.html").read_bytes(), "text/html; charset=utf-8") |
| elif self.path == "/api/presets": |
| self._send(200, json.dumps(preset_payload(), ensure_ascii=False).encode("utf-8")) |
| else: |
| self._send(404, b'{"error": "not found"}') |
|
|
| def do_POST(self): |
| if self.path != "/api/backtest": |
| self._send(404, b'{"error": "not found"}') |
| return |
| try: |
| n = int(self.headers.get("Content-Length", 0)) |
| p = json.loads(self.rfile.read(n)) |
| start = p.get("start", "2013-01-01") |
| end = p.get("end", "2025-12-31") |
| vol = float(p.get("target_vol", 12.0)) |
| ov = overlay_from(p) |
| summary, st, curve, trades, s = run_full(start, end, vol, ov) |
| bh_s = one_stats([c["bh_V"] for c in curve]) |
| ctx = {c["date"]: c for c in curve} |
| t_rows = [{"date": t["date"], |
| "action": "买入" if t["action"] == "BUY" else ("卖出" if t["action"] == "SELL" else "作废"), |
| "price": t["price"], "amount": round(t["amount_usd"], 0), |
| "reason": t["reason"], "V": round(t["V_after"], 0), |
| "sigma": round((ctx.get(t["date"], {}).get("sigma") or 0) * 100, 1), |
| "q": (ctx.get(t["date"], {}).get("q") or 0) and round(ctx[t["date"]]["q"]), |
| "pace": (ctx.get(t["date"], {}).get("pace") or 0) and round(ctx[t["date"]]["pace"])} |
| for t in trades] |
| step = max(1, len(curve) // 1000) |
| idx = list(range(0, len(curve), step)) |
| if idx[-1] != len(curve) - 1: |
| idx.append(len(curve) - 1) |
| payload = {"params": p, |
| "stats": s, "bh": bh_s, |
| "n_trades": summary["n_trades"], "fees": round(summary["fees"], 1), |
| "final_V": round(summary["V"], 0), |
| "yearly": yearly_table(curve), |
| "exit_spans": summary["exit_spans"], |
| "curve": {"dates": [curve[i]["date"] for i in idx], |
| "v": [round(curve[i]["V"], 2) for i in idx], |
| "bh": [round(curve[i]["bh_V"], 2) for i in idx]}, |
| "trades": t_rows} |
| self._send(200, json.dumps(payload, ensure_ascii=False).encode("utf-8")) |
| except Exception as exc: |
| self._send(500, json.dumps({"error": f"{type(exc).__name__}: {exc}"}, |
| ensure_ascii=False).encode("utf-8")) |
|
|
| def log_message(self, *a): |
| pass |
|
|
|
|
| if __name__ == "__main__": |
| print(f"启动金仓对比台 http://127.0.0.1:{PORT} （Ctrl+C 退出）") |
| ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever() |
|
|
