# -*- coding: utf-8 -*- """ 最近3年（2023-2025）分钟K线回测 策略逻辑与 统计.py 一致： 连续 CONSECUTIVE_POSITIVE 根正K线 且 累计涨幅 > MIN_SUM_THRESHOLD% → 触发信号 下一根K线开盘买入，持有 FOLLOWING_BARS 根K线后卖出 相邻信号间隔 >= MIN_DISTANCE 根 """ import re import os import math import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from matplotlib import font_manager # ==================== 可配置参数（与统计.py保持一致） ==================== CONSECUTIVE_POSITIVE = 3 # 连续正增长的K线数量 MIN_SUM_THRESHOLD = 0.5 # 连续K线增长率总和的最小阈值（%） FOLLOWING_BARS = 90 # 持有K线数量（约1.5小时） MIN_DISTANCE = 120 # 相邻信号的最小间隔（K线数量） COMMISSION = 0.0 # 单边手续费（%），ETF免印花税，可设0.01~0.03 # ================================================================ BASE_DIR = os.path.dirname(os.path.abspath(__file__)) FILES = [ ("2023年", os.path.join(BASE_DIR, "2023K线.txt")), ("2024年", os.path.join(BASE_DIR, "2024K线.txt")), ("2025年", os.path.join(BASE_DIR, "K线.txt")), ] def parse_file(path): """解析 时间戳\t增长率% 格式文件""" rows = [] with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue m = re.match(r"^(\d{8}-\d{2}-\d{2}-\d{2})\s+(-?\d+(?:\.\d+)?)%", line) if m: ts, rate = m.group(1), float(m.group(2)) # ts: YYYYMMDD-HH-MM-SS date = ts[:4] + "-" + ts[4:6] + "-" + ts[6:8] rows.append((date, ts, rate)) return rows def backtest(all_rows, year_filter=None): """对（可选某年的）数据执行信号统计 + 交易模拟""" rates = [r[2] for r in all_rows] n = len(rates) signals = [] i = 0 last_sig = -MIN_DISTANCE while i < n - (CONSECUTIVE_POSITIVE + FOLLOWING_BARS): if rates[i] > 0 and rates[i+1] > 0 and rates[i+2] > 0: csum = rates[i] + rates[i+1] + rates[i+2] if csum > MIN_SUM_THRESHOLD and (i - last_sig) >= MIN_DISTANCE: # 下一根K线开盘买入，持有 FOLLOWING_BARS 根 trade_ret = sum(rates[i+3:i+3+FOLLOWING_BARS]) signals.append({ "idx": i, "date": all_rows[i][0], "ts": all_rows[i][1], "csum": csum, "ret": trade_ret, }) last_sig = i i += 1 return signals def stats_of(signals): if not signals: return None rets = [s["ret"] for s in signals] wins = [r for r in rets if r > 0] losses = [r for r in rets if r < 0] win_rate = len(wins) / len(rets) * 100 avg = sum(rets) / len(rets) avg_win = sum(wins) / len(wins) if wins else 0.0 avg_loss = sum(losses) / len(losses) if losses else 0.0 pl_ratio = abs(avg_win / avg_loss) if avg_loss else float("inf") # 复利累计收益（信号间隔>=120 > 90根持有期，交易不重叠，可复利） equity = 1.0 curve = [] for s in signals: equity *= (1 + (s["ret"] - 2 * COMMISSION) / 100) curve.append((s["date"], s["ts"], equity)) total_ret = (equity - 1) * 100 # 资金曲线最大回撤 peak, max_dd = 1.0, 0.0 for _, _, e in curve: peak = max(peak, e) max_dd = max(max_dd, (peak - e) / peak) return { "n": len(rets), "win_rate": win_rate, "avg": avg, "avg_win": avg_win, "avg_loss": avg_loss, "pl_ratio": pl_ratio, "total_ret": total_ret, "max_dd": max_dd * 100, "curve": curve, "sum_ret": sum(rets), } def fmt_stats(name, st): if st is None: return f"{name}: 无信号" return (f"{name}: 信号{st['n']}个 | 胜率{st['win_rate']:.1f}% | 单次均收益{st['avg']:.3f}% | " f"均盈{st['avg_win']:.3f}% / 均亏{st['avg_loss']:.3f}% | 盈亏比{st['pl_ratio']:.2f} | " f"复利累计{st['total_ret']:.2f}% | 最大回撤{st['max_dd']:.2f}%") def main(): # ---------- 1. 读取三年数据 ---------- all_rows = [] for label, path in FILES: if not os.path.exists(path): print(f"!! 缺少文件: {path}") continue rows = parse_file(path) print(f"{label}: {path} -> {len(rows)} 条 ({rows[0][0]} ~ {rows[-1][0]})") all_rows.extend(rows) all_rows.sort(key=lambda r: r[1]) # 按时间戳排序 print(f"\n合并后总K线: {len(all_rows)} 条 ({all_rows[0][0]} ~ {all_rows[-1][0]})") # ---------- 2. 分年回测 + 总体回测 ---------- results = {} for year in ["2023", "2024", "2025"]: rows_y = [r for r in all_rows if r[0].startswith(year)] sig_y = backtest(rows_y) results[year] = stats_of(sig_y) print(fmt_stats(f"{year}年", results[year])) sig_all = backtest(all_rows) st_all = stats_of(sig_all) print("-" * 100) print(fmt_stats("三年合计", st_all)) # ---------- 3. 买入持有基准 ---------- bh = 1.0 for _, _, r in all_rows: bh *= (1 + r / 100) bh_ret = (bh - 1) * 100 years_span = len(all_rows) / (240 * 250) # 每日240根、每年约250交易日 print(f"买入持有基准（同期）: {bh_ret:.2f}% | 策略超额: {st_all['total_ret'] - bh_ret:+.2f}%") print(f"策略年化(复利, 约{years_span:.1f}年): {((1 + st_all['total_ret']/100) ** (1/years_span) - 1) * 100:.2f}%") # ---------- 4. 绘图 ---------- for f in ["Microsoft YaHei", "SimHei"]: if any(f.lower() in ff.name.lower() for ff in font_manager.fontManager.ttflist): plt.rcParams["font.family"] = f break plt.rcParams["axes.unicode_minus"] = False fig, axes = plt.subplots(2, 1, figsize=(12, 9), gridspec_kw={"height_ratios": [2, 1]}) ax = axes[0] # 策略资金曲线（按交易出场时间） xs = [f"{c[0]} {c[1][9:11]}:{c[1][11:13]}" for c in st_all["curve"]] ys = [c[2] for c in st_all["curve"]] xi = range(len(ys)) ax.plot(xi, ys, color="#d43f3f", lw=1.6, label="策略复利资金曲线") # 买入持有基准（按日） bh_curve, e = [], 1.0 last_date = None for date, _, r in all_rows: e *= (1 + r / 100) if date != last_date: bh_curve.append((date, e)) last_date = date # 将基准对齐到策略曲线的日期轴（近似：直接画独立x轴） ax2 = ax.twiny() ax2.plot(range(len(bh_curve)), [c[1] for c in bh_curve], color="#888888", lw=1.2, ls="--", label="买入持有基准") ax2.set_xticks([]) ax2.spines["top"].set_visible(False) ax.set_xticks(list(xi)[::max(1, len(xi)//8)]) ax.set_xticklabels([xs[k] for k in range(0, len(xs), max(1, len(xs)//8))], rotation=20, fontsize=8) ax.axhline(1.0, color="#bbbbbb", lw=0.8) ax.set_ylabel("净值") ax.set_title(f"最近3年分钟K线回测（2023-2025，数据经核实为159915创业板ETF） 连续{CONSECUTIVE_POSITIVE}根正K线>0.5%，持{FOLLOWING_BARS}根 " f"累计{st_all['total_ret']:.2f}% vs 基准{bh_ret:.2f}%", fontsize=12) lines1, labels1 = ax.get_legend_handles_labels() lines2, labels2 = ax2.get_legend_handles_labels() ax.legend(lines1 + lines2, labels1 + labels2, loc="upper left", fontsize=9) # 分年度柱状图 axb = axes[1] years = ["2023年", "2024年", "2025年", "三年合计"] tots = [results[y]["total_ret"] if results[y] else 0 for y in ["2023", "2024", "2025"]] + [st_all["total_ret"]] wrs = [results[y]["win_rate"] if results[y] else 0 for y in ["2023", "2024", "2025"]] + [st_all["win_rate"]] colors = ["#d43f3f" if t >= 0 else "#2e9e5b" for t in tots] bars = axb.bar(years, tots, color=colors, width=0.55) for b, t, w in zip(bars, tots, wrs): axb.text(b.get_x() + b.get_width()/2, b.get_height() + (0.15 if t >= 0 else -0.6), f"{t:.1f}%\n胜率{w:.0f}%", ha="center", fontsize=9) axb.axhline(0, color="#999999", lw=0.8) axb.set_ylabel("复利累计收益(%)") axb.set_title("分年度收益", fontsize=11) plt.tight_layout() out_png = os.path.join(BASE_DIR, "回测3年_资金曲线.png") plt.savefig(out_png, dpi=130) print(f"\n图表已保存: {out_png}") # ---------- 5. HTML 报告 ---------- def tr_row(y, st): if st is None: return f"{y}无信号" return (f"{y}{st['n']}{st['win_rate']:.1f}%" f"{st['avg']:.3f}%{st['avg_win']:.3f}%{st['avg_loss']:.3f}%" f"=0 else 'neg'}'>{st['total_ret']:.2f}%" f"{st['max_dd']:.2f}%") html = f""" 最近3年分钟K线回测报告（2023-2025）

最近3年分钟K线回测报告（2023-01-03 ~ 2025-11-28，标的经核实为 159915 创业板ETF）

信号总数{st_all['n']}
胜率{st_all['win_rate']:.1f}%
单次均收益{st_all['avg']:.3f}%
盈亏比{st_all['pl_ratio']:.2f}
复利累计{st_all['total_ret']:.2f}%
最大回撤{st_all['max_dd']:.2f}%
{tr_row('2023年', results['2023'])} {tr_row('2024年', results['2024'])} {tr_row('2025年', results['2025'])} {tr_row('三年合计', st_all)}
区间信号数胜率单次均收益均盈均亏复利累计最大回撤

对比基准

同期买入持有：{bh_ret:.2f}% ｜ 策略超额收益：{st_all['total_ret']-bh_ret:+.2f}%

资金曲线

策略规则

连续{CONSECUTIVE_POSITIVE}根1分钟K线为正且累计涨幅 > {MIN_SUM_THRESHOLD}% 时触发信号；下一根K线开盘买入，持有{FOLLOWING_BARS}根（约1.5小时）后卖出；相邻信号间隔 ≥ {MIN_DISTANCE}根（交易不重叠，收益按复利累计）。手续费设为 {COMMISSION}%。2025年数据截至2025-11-28。

""" out_html = os.path.join(BASE_DIR, "回测3年_报告.html") with open(out_html, "w", encoding="utf-8") as f: f.write(html) print(f"报告已保存: {out_html}") return out_html, out_png if __name__ == "__main__": main()