backtest/ui_server.py

"""金仓 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                 # noqa: E402
from overlays import build_overlays   # noqa: E402
from run import one_stats, yearly_table   # noqa: E402

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:                                   # noqa: BLE001
            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()