diff --git a/deploy/profile-scripts/hk_backtest.py b/deploy/profile-scripts/hk_backtest.py index ed661cd4..d1a247a7 100644 --- a/deploy/profile-scripts/hk_backtest.py +++ b/deploy/profile-scripts/hk_backtest.py @@ -96,8 +96,11 @@ def gen_trades(panel, strat): break if exit_p is None: exit_p, reason, hold = fut.iloc[-1]["close"], "time", maxh + # 退出日期(用该股票日历往后推 hold 个交易日) + exit_date = fut.iloc[min(hold - 1, len(fut) - 1)]["date"] if hold > 0 else s["date"] trades.append({ "code": s["code"], "name": s["code"], "entry_date": s["date"], + "exit_date": exit_date, "entry_price": round(ep, 2), "exit_price": round(exit_p, 2), "profit_pct": round((exit_p - ep) / ep * 100, 2), "exit_reason": reason, "hold_days": hold, @@ -108,28 +111,94 @@ def gen_trades(panel, strat): return trades +def portfolio_nav(trades, capital=1000000, slots=8): + """8槽资金管理净值曲线:每日结算到期→入场(仓位满跳过)→持仓按成本估值。 + 返回 (nav_series: dict date->nav, stats)""" + if not trades: + return {}, {} + dates = sorted({t["entry_date"] for t in trades} | {t.get("exit_date", t["entry_date"]) for t in trades}) + if not dates: + return {}, {} + # 用真实日历(stock_daily 港股日K日期) + conn = sqlite3.connect(DB) + cal = [r[0] for r in conn.execute( + "SELECT DISTINCT date FROM stock_daily WHERE date>=? AND date<=? AND length(code)=5 ORDER BY date", + (dates[0], dates[-1])).fetchall()] + conn.close() + if not cal: + cal = dates + cal_idx = {d: i for i, d in enumerate(cal)} + alloc = capital / slots + open_pos = [] # {exit_date, alloc, pnl} + cash = capital + nav_series = {} + skipped = 0 + by_entry = collections.defaultdict(list) + for t in trades: + by_entry[t["entry_date"]].append(t) + for day in cal: + # 结算到期 + still = [] + for p in open_pos: + if p["exit_date"] <= day: + cash += p["alloc"] * (1 + p["pnl"] / 100) + else: + still.append(p) + open_pos = still + # 入场 + for t in by_entry.get(day, []): + if len(open_pos) >= slots or cash < alloc: + skipped += 1 + continue + open_pos.append({"exit_date": t.get("exit_date", day), "alloc": alloc, + "pnl": t["profit_pct"]}) + cash -= alloc + # 净值 + held_val = sum(p["alloc"] * (1 + p["pnl"] / 100) for p in open_pos) + nav_series[day] = cash + held_val + nav_series = {d: v for d, v in sorted(nav_series.items())} + return nav_series, {"skipped": skipped} + + +def window_returns(nav_series): + """从净值曲线算窗口收益(近1年/6月/3月,对照最后日期)""" + if not nav_series: + return {"year1": None, "month6": None, "month3": None} + items = sorted(nav_series.items()) + last_d, last_v = items[-1] + last_dt = datetime.strptime(last_d, "%Y-%m-%d") + out = {} + for label, days in [("year1", 365), ("month6", 182), ("month3", 91)]: + cutoff = (last_dt - timedelta(days=days)).strftime("%Y-%m-%d") + # 取 cutoff 后最近的净值点 + base = None + for d, v in items: + if d >= cutoff: + base = v + break + out[label] = (last_v / base - 1) * 100 if base else None + return out + + def portfolio_metrics(trades, capital=1000000, slots=8): - """资金模拟 + 时间窗收益(Ralph Loop验收标准)""" - import strategy_lab as lab - pf = lab.portfolio_sim(trades, capital, max_positions=slots) - years_span = 7.5 - cagr = pf.get("cagr_pct") - # 时间窗收益(按 entry_date 过滤 trades,8槽等权复利净值) - def window_return(months): - cutoff = (datetime(2026, 7, 24) - timedelta(days=int(months * 30.4))).strftime("%Y-%m-%d") - wt = [t for t in trades if t["entry_date"] >= cutoff] - if not wt: - return None - nav = 1.0 - for t in sorted(wt, key=lambda x: x["entry_date"]): - nav *= (1 + t["profit_pct"] / 100 / slots) - return (nav - 1) * 100 + """资金模拟 + 窗口收益(净值曲线)""" + nav, stats = portfolio_nav(trades, capital, slots) + win = window_returns(nav) + # 年化(用首末净值) + cagr = None + if nav: + items = sorted(nav.items()) + d0, v0 = items[0] + d1, v1 = items[-1] + yrs = max((datetime.strptime(d1, "%Y-%m-%d") - datetime.strptime(d0, "%Y-%m-%d")).days / 365.0, 0.5) + cagr = (((v1 / v0) ** (1 / yrs)) - 1) * 100 if v0 > 0 else None return { "cagr": cagr, - "year1": window_return(12), - "month6": window_return(6), - "month3": window_return(3), + "year1": win.get("year1"), + "month6": win.get("month6"), + "month3": win.get("month3"), "trades": len(trades), + "skipped": stats.get("skipped", 0), "win_rate": sum(1 for t in trades if t["profit_pct"] > 0) / len(trades) * 100 if trades else 0, }