"""对照任务（计划书 §9，D15 预注册）：底仓 60% + 网格 40% vs v2.1 同窗口对比。 底仓子账户 = v2.1 规则原样（引擎 V0=60%；引擎对规模严格线性，已断言校验）。 网格子账户按 §9.2： 锚点随底仓 ENTRY；档位金额 = 锚点日网格净值/12（本轮固定）； 买入档 = 上一锚点/成交价×(1−0.5×当月σ)（在锚点/成交/恢复时一步算定）； 挂单卖出档 = 成交价×(1+0.5×成交日当月σ)（成交即挂，不重定价）； q_eff≥90 停新买；底仓退出模式停买停挂、并在底仓退出卖出日同步卖 1/3（FIFO）、 底仓残值清仓日全清；12 档耗尽停用，价格收复最近成交档后重开梯子。 执行纪律同主任务：收盘触发、次日 PM 成交、单边 0.05%。 用法：python grid.py 判读（预注册 §9.3）：任一震荡窗口年化增量 < +1% 或任一趋势窗口额外回撤 > 3pp → 不增价值。 """ import csv import json from pathlib import Path import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 import engine as E # noqa: E402 from run import one_stats, stats_full # noqa: E402 matplotlib.rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei", "DejaVu Sans"] matplotlib.rcParams["axes.unicode_minus"] = False HERE = Path(__file__).resolve().parent START, END = "2013-01-01", "2026-09-07" GRID_SHARE, N_SLOTS, STEP = 0.40, 12, 0.5 # D16 存档更正：2026 已按 OOS 政策恢复无接触地位，判读窗口仅保留 2013-2025 段 WINDOWS = [("震荡 2015-2016", "2015-01-01", "2016-12-31", "osc"), ("趋势 2013-2015", "2013-01-01", "2015-12-31", "trend"), ("趋势 2022-2024", "2022-01-01", "2024-12-31", "trend")] def main(): S = E.build_signals() v2_rows = list(csv.DictReader(open(HERE / "results" / "final" / "curve.csv", encoding="utf-8"))) v2_V = {r["date"]: float(r["V"]) for r in v2_rows} bh_V = {r["date"]: float(r["bh_V"]) for r in v2_rows} _, _, base_data = E.run(start=START, end=END, target_vol_ann=12.0, V0=(1 - GRID_SHARE) * 10000.0, tag="grid_base") base_curve, base_trades = base_data["curve"], base_data["trades"] base_V = {c["date"]: c["V"] for c in base_curve} # 规模线性校验：底仓 ≡ 0.6×v2 lin = max(abs(base_V[c["date"]] - (1 - GRID_SHARE) * v2_V[c["date"]]) / v2_V[c["date"]] for c in base_curve) assert lin < 1e-9, f"底仓与 0.6×v2 不一致: {lin}" pm = dict(E.load_pm()) days = [(c["date"], pm[c["date"]]) for c in base_curve] entry_px = {t["date"]: t["price"] for t in base_trades if t["reason"] == "ENTRY"} exit_days = {t["date"] for t in base_trades if t["reason"] == "EXIT pace<102"} final_days = {t["date"] for t in base_trades if t["reason"].endswith("+final")} mode_of = {c["date"]: c["mode"] for c in base_curve} # ---- 网格子账户 ---- cash = GRID_SHARE * 10000.0 lots = [] # {units, fill, sell_level, pending} next_buy = None lot_amt = None active = False exhausted = False ladder_fills = 0 last_fill = None pending = [] # [(exec_j, 'sell'|'buy', lot索引 或 金额)]；调度到执行之间 lots 不会增删 gtrades = [] gcurve = [] for j, (d, px) in enumerate(days): ym = d[:7] s = S["sig"].get(ym) q_eff = E.month_q_eff(S, S["qrows"], ym) exiting = mode_of.get(d, "IN") == "EXITING" # 1) 执行到期挂单 still = [] for pj, kind, ref in pending: if pj != j: still.append((pj, kind, ref)) continue if kind == "sell": lot = lots[ref] lot["pending"] = False if exiting or d in exit_days or d in final_days or lot["units"] <= 1e-12: continue # 风险日不卖档，档位保留 amt = lot["units"] * px cash += amt * (1 - E.COST) lot["units"] = 0.0 gtrades.append({"date": d, "action": "GRID_SELL", "reason": "档位+1格", "amount_usd": round(amt, 2), "price": px}) else: if exiting or not active: continue # 退出/停机日买入作废 amt = min(ref, cash / (1 + E.COST)) if amt <= 1e-9: continue cash -= amt * (1 + E.COST) lots.append({"units": amt / px, "fill": px, "sell_level": px * (1 + STEP * s), "pending": False}) ladder_fills += 1 last_fill = px next_buy = px * (1 - STEP * s) if ladder_fills >= N_SLOTS: exhausted = True gtrades.append({"date": d, "action": "GRID_BUY", "reason": f"第{ladder_fills}/{N_SLOTS}档", "amount_usd": round(amt, 2), "price": px}) pending = still lots = [l for l in lots if l["units"] > 1e-12] # 2) 锚点 / 同步退出 / 残值全清 if d in entry_px: anchor = entry_px[d] lots = [] lot_amt = cash / N_SLOTS active, exhausted, ladder_fills = True, False, 0 last_fill = anchor next_buy = anchor * (1 - STEP * s) if s else None gtrades.append({"date": d, "action": "GRID_ANCHOR", "reason": f"底仓ENTRY 锚{anchor:.0f} 每档{lot_amt:.0f}", "amount_usd": 0, "price": anchor}) elif d in final_days: amt = sum(l["units"] for l in lots) * px if amt > 1e-9: cash += amt * (1 - E.COST) gtrades.append({"date": d, "action": "GRID_SELL", "reason": "底仓残值清仓→网格全清", "amount_usd": round(amt, 2), "price": px}) lots, active, next_buy = [], False, None elif d in exit_days: units = sum(l["units"] for l in lots) target = units / 3.0 rem = target for l in lots: # FIFO 冲减 if rem <= 1e-12: break take = min(l["units"], rem) cash += take * px * (1 - E.COST) l["units"] -= take rem -= take lots = [l for l in lots if l["units"] > 1e-12] gtrades.append({"date": d, "action": "GRID_SELL", "reason": "同步退出1/3", "amount_usd": round(target * px, 2), "price": px}) # 3) 挂单卖出触发（收盘 ≥ 冻结档位）——调度到执行之间 lots 不会增删（步骤顺序保证） if active and not exiting and d not in exit_days: for i, l in enumerate(lots): if not l["pending"] and px >= l["sell_level"]: pending.append((j + 1, "sell", i)) l["pending"] = True # 4) 买入触发（收盘 ≤ 下一买入档） if (active and not exiting and not exhausted and ladder_fills < N_SLOTS and next_buy is not None and px <= next_buy and (q_eff is None or q_eff < 90) and not any(k == "buy" for _, k, _ in pending)): pending.append((j + 1, "buy", lot_amt)) # 5) 耗尽恢复：价格收复最近成交档 if exhausted and last_fill is not None and px > last_fill: exhausted, ladder_fills = False, 0 next_buy = px * (1 - STEP * s) if s else None units = sum(l["units"] for l in lots) gcurve.append({"date": d, "grid_V": cash + units * px}) # ---- 合成与输出 ---- out = HERE / "results" / "grid" out.mkdir(parents=True, exist_ok=True) dates = [c["date"] for c in gcurve] gV = {c["date"]: c["grid_V"] for c in gcurve} variant = {d: gV[d] + base_V[d] for d in dates} control = {d: base_V[d] + GRID_SHARE * 10000.0 for d in dates} with open(out / "variant_curve.csv", "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(["date", "v2", "variant", "control", "bh", "grid_V"]) for d in dates: w.writerow([d, round(v2_V[d], 4), round(variant[d], 4), round(control[d], 4), round(bh_V[d], 4), round(gV[d], 4)]) with open(out / "grid_trades.csv", "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=["date", "action", "reason", "amount_usd", "price"]) w.writeheader() w.writerows(gtrades) def series(m): return [m[d] for d in dates] rows, osc_ok, trend_ok = [], True, True for name, lo, hi, kind in WINDOWS: idx = [i for i, d in enumerate(dates) if lo <= d <= hi] seg = lambda m: [m[dates[i]] for i in idx] # noqa: E731 sv, bv = one_stats(seg(variant)), one_stats(seg(v2_V)) delta = sv["cagr"] - bv["cagr"] extra_dd = abs(sv["maxdd"]) - abs(bv["maxdd"]) ok = (delta >= 0.01) if kind == "osc" else (extra_dd <= 0.03) if kind == "osc": osc_ok = osc_ok and ok else: trend_ok = trend_ok and ok rows.append({"window": name, "kind": kind, "v2_cagr": bv["cagr"], "var_cagr": sv["cagr"], "delta_cagr": delta, "v2_maxdd": bv["maxdd"], "var_maxdd": sv["maxdd"], "extra_dd": extra_dd, "pass": ok}) summary = {"windows": rows, "osc_all_pass": osc_ok, "trend_all_pass": trend_ok, "verdict": "采纳（v3 候选）" if (osc_ok and trend_ok) else "不增价值，不采纳", "grid_sleeve": one_stats(series(gV)), "full": {"v2": one_stats(series(v2_V)), "variant": one_stats(series(variant)), "control": one_stats(series(control)), "bh": one_stats(series(bh_V))}, "n_grid_trades": sum(1 for t in gtrades if t["action"].startswith("GRID_") and t["action"] != "GRID_ANCHOR")} (out / "grid_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1, default=str), encoding="utf-8") # ---- 图 ---- fig, ax = plt.subplots(figsize=(11.5, 6), dpi=150) ax.plot(dates, series(bh_V), lw=1.1, color="#b8b8b8", label="买入持有") ax.plot(dates, series(v2_V), lw=1.5, color="#c8912a", label="v2.1（100%）") ax.plot(dates, series(control), lw=1.3, color="#4569a0", ls="--", label="0.6×v2 + 40%现金（对照）") ax.plot(dates, series(variant), lw=1.5, color="#7a4ea3", label="底仓60% + 网格40%") ax.set_yscale("log") ax.axvline("2026-01-01", color="#4569a0", ls=":", lw=1) ax.annotate("2026 样本外 →", xy=(0.868, 0.045), xycoords="axes fraction", fontsize=9, color="#4569a0") tick_idx = [i for i, d in enumerate(dates) if d[5:7] == "01" and (i == 0 or dates[i - 1][5:7] != "01")] step = max(1, len(tick_idx) // 8) ax.set_xticks([dates[i] for i in tick_idx[::step]]) ax.set_xticklabels([dates[i][:4] for i in tick_idx[::step]], fontsize=9) ax.set_ylabel("净值（V0 = 10,000，对数轴）", fontsize=10) ax.set_title(f"对照任务：底仓+网格 变体 {dates[0]} → {dates[-1]}（USD，成本后）", fontsize=12) ax.grid(alpha=0.25, which="both") ax.legend(loc="upper left", fontsize=9) fig.tight_layout() fig.savefig(out / "grid_equity.png") # ---- 打印 ---- gs = summary["grid_sleeve"] print(f"网格子账户: {summary['n_grid_trades']} 笔 期末 {series(gV)[-1]:.0f}（V0 4,000） " f"年化 {gs['cagr']:+.2%} 回撤 {gs['maxdd']:.1%}") print(f"{'窗口':16}{'v2年化':>9}{'变体年化':>9}{'增量':>8}{'v2回撤':>9}{'变体回撤':>9}{'额外回撤':>10} 判定") for r in rows: print(f"{r['window']:16}{r['v2_cagr']:>+9.1%}{r['var_cagr']:>+9.1%}{r['delta_cagr']:>+8.1%}" f"{r['v2_maxdd']:>9.1%}{r['var_maxdd']:>9.1%}{r['extra_dd']:>+9.1f}pp " f"{'PASS' if r['pass'] else 'FAIL'}") print(f"判读（预注册 §9.3）: {summary['verdict']}") f_ = summary["full"] print(f"全期: v2 {f_['v2']['cagr']:+.1%}/回撤{f_['v2']['maxdd']:.1%} " f"变体 {f_['variant']['cagr']:+.1%}/回撤{f_['variant']['maxdd']:.1%} " f"对照 {f_['control']['cagr']:+.1%}/回撤{f_['control']['maxdd']:.1%} " f"持有 {f_['bh']['cagr']:+.1%}/回撤{f_['bh']['maxdd']:.1%}") if __name__ == "__main__": main()