Chan/scripts/wyckoff_param_grid.py

#!/usr/bin/env python3
"""在最优周期 1h/4h/8h 上扫 ATR 与关键参数。"""
from __future__ import annotations

import itertools
import json
import logging
import re
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(ROOT))

from user_data.Chan.scripts.wyckoff_tf_grid import (  # noqa: E402
	STRAT_PATH,
	install_offline_markets,
	patch_strategy,
	run_one,
)

# 参数名 -> (正则匹配赋值行前缀, 候选值列表)
PARAM_GRID = {
	"atr_sl_mult": (
		r'^(\tatr_sl_mult = DecimalParameter\([^\n]*default=)([0-9.]+)',
		[1.5, 2.0, 2.5, 3.0],
	),
	"vol_spike_mult": (
		r'^(\tvol_spike_mult = DecimalParameter\([^\n]*default=)([0-9.]+)',
		[1.2, 1.4, 1.8],
	),
	"spring_pierce_pct": (
		r'^(\tspring_pierce_pct = DecimalParameter\([^\n]*default=)([0-9.]+)',
		[0.002, 0.004, 0.008],
	),
	"range_lookback": (
		r'^(\trange_lookback = IntParameter\([^\n]*default=)([0-9]+)',
		[18, 24, 36],
	),
}


def set_defaults(text: str, values: dict[str, Any]) -> str:
	for key, (pat, _) in PARAM_GRID.items():
		val = values[key]
		text = re.sub(pat, rf"\g<1>{val}", text, count=1, flags=re.M)
	return text


def main() -> None:
	logging.getLogger("freqtrade").setLevel(logging.ERROR)
	timerange = sys.argv[1] if len(sys.argv) > 1 else "20240101-"
	install_offline_markets()
	orig = STRAT_PATH.read_text()

	keys = list(PARAM_GRID.keys())
	combos = list(itertools.product(*[PARAM_GRID[k][1] for k in keys]))
	# 全组合太多：改为坐标下降式 — 先基线，再逐参扫描
	base = {k: PARAM_GRID[k][1][len(PARAM_GRID[k][1]) // 2] for k in keys}
	# 确保与当前文件接近的中心点
	base.update(
		{
			"atr_sl_mult": 2.0,
			"vol_spike_mult": 1.4,
			"spring_pierce_pct": 0.004,
			"range_lookback": 24,
		}
	)

	trials = [dict(base)]
	for k in keys:
		for v in PARAM_GRID[k][1]:
			if v == base[k]:
				continue
			t = dict(base)
			t[k] = v
			trials.append(t)

	rows = []
	try:
		patch_strategy("1h", "4h", "8h")
		for i, vals in enumerate(trials):
			text = set_defaults(STRAT_PATH.read_text(), vals)
			STRAT_PATH.write_text(text)
			label = ",".join(f"{k}={vals[k]}" for k in keys)
			print(f"[{i+1}/{len(trials)}] {label}", flush=True)
			try:
				res = run_one("1h", timerange)
				res.update(vals)
				res["label"] = label
				res["ok"] = True
			except Exception as e:
				res = {"ok": False, "error": str(e), "label": label, **vals}
			rows.append(res)
			if res.get("ok"):
				print(
					f"  -> profit={res['profit_pct']:.2f}% trades={res['trades']} "
					f"dd={res['dd_pct']:.2f}% pf={res['pf']:.2f}",
					flush=True,
				)
			else:
				print(f"  FAILED {res.get('error')}", flush=True)
	finally:
		STRAT_PATH.write_text(orig)

	ok = [r for r in rows if r.get("ok")]
	ok.sort(key=lambda r: (r["profit_pct"], r["pf"]), reverse=True)
	print("\n========== PARAM RANKING ==========")
	for r in ok[:10]:
		print(
			f"{r['profit_pct']:>7.2f}%  pf={r['pf']:.2f}  dd={r['dd_pct']:.1f}%  "
			f"n={r['trades']:<3}  {r['label']}"
		)
	out = ROOT / "user_data/Chan/scripts/wyckoff_param_grid_result.txt"
	out.write_text(json.dumps({"timerange": timerange, "rows": rows}, indent=2))
	print(f"\nSaved {out}")
	if ok:
		best = ok[0]
		print("\nBEST params:", {k: best[k] for k in keys})
		# 写回最优 default
		text = set_defaults(orig, {k: best[k] for k in keys})
		# 保持最优周期
		text2 = text
		text2 = re.sub(r'^(\ttimeframe = ).*$', r'\g<1>"1h"', text2, count=1, flags=re.M)
		text2 = re.sub(
			r'^(\tstructure_timeframe = ).*$', r'\g<1>"4h"', text2, count=1, flags=re.M
		)
		text2 = re.sub(
			r'^(\tbias_timeframe: Optional\[str\] = ).*$',
			r'\g<1>"8h"',
			text2,
			count=1,
			flags=re.M,
		)
		STRAT_PATH.write_text(text2)
		print("Wrote best defaults into Wyckoff_BTC.py")
		# 长周期验证
		print("\nValidate 20230101- ...", flush=True)
		res = run_one("1h", "20230101-")
		print(res)


if __name__ == "__main__":
	main()