import argparse import csv import itertools import json from dataclasses import asdict, dataclass from pathlib import Path @dataclass class Bar: ts: int open: float high: float low: float close: float @dataclass class StrategyParams: min_prev_entity_pct: float = 0.1 breakout_buffer_pct: float = 0.03 shadow_threshold_pct: float = 0.15 stop_loss_pct: float = 0.35 take_profit_pct: float = 0.8 trailing_start_pct: float = 0.5 trailing_backoff_pct: float = 0.25 use_atr_dynamic_threshold: bool = True atr_length: int = 14 breakout_buffer_atr_mult: float = 0.12 shadow_threshold_atr_mult: float = 0.18 stop_loss_atr_mult: float = 0.45 take_profit_atr_mult: float = 0.95 trailing_start_atr_mult: float = 0.6 trailing_backoff_atr_mult: float = 0.3 @dataclass class BacktestResult: score: float total_return_pct: float max_drawdown_pct: float win_rate_pct: float profit_factor: float trades: int wins: int params: StrategyParams @dataclass class RobustEvalResult: robust_score: float consistency_score: float overfit_gap: float train_risk_adj: float valid_risk_adj: float train: BacktestResult valid: BacktestResult full: BacktestResult params: StrategyParams def load_csv_bars(path: Path) -> list[Bar]: bars: list[Bar] = [] with path.open("r", encoding="utf-8") as f: reader = csv.DictReader(f) required = {"id", "open", "high", "low", "close"} if not required.issubset(reader.fieldnames or set()): raise ValueError(f"CSV missing columns: {required}") for row in reader: try: bars.append( Bar( ts=int(float(row["id"])), open=float(row["open"]), high=float(row["high"]), low=float(row["low"]), close=float(row["close"]), ) ) except (ValueError, TypeError): continue bars.sort(key=lambda x: x.ts) return bars def resample_to_minutes(bars: list[Bar], minutes: int) -> list[Bar]: if not bars: return [] bucket_sec = minutes * 60 grouped: list[Bar] = [] cur_bucket = None cur_open = cur_high = cur_low = cur_close = None for bar in bars: b = bar.ts // bucket_sec if cur_bucket is None or b != cur_bucket: if cur_bucket is not None: grouped.append( Bar( ts=cur_bucket * bucket_sec, open=cur_open, high=cur_high, low=cur_low, close=cur_close, ) ) cur_bucket = b cur_open = bar.open cur_high = bar.high cur_low = bar.low cur_close = bar.close else: cur_high = max(cur_high, bar.high) cur_low = min(cur_low, bar.low) cur_close = bar.close if cur_bucket is not None: grouped.append( Bar( ts=cur_bucket * bucket_sec, open=cur_open, high=cur_high, low=cur_low, close=cur_close, ) ) return grouped def compute_atr_series(bars: list[Bar], length: int) -> list[float | None]: atr: list[float | None] = [None] * len(bars) if len(bars) < length + 1: return atr tr_list: list[float] = [0.0] * len(bars) for i in range(1, len(bars)): high = bars[i].high low = bars[i].low prev_close = bars[i - 1].close tr = max(high - low, abs(high - prev_close), abs(low - prev_close)) tr_list[i] = tr for i in range(length, len(bars)): window = tr_list[i - length + 1:i + 1] atr[i] = sum(window) / length return atr def resolve_dynamic_distance( base_price: float, fixed_pct: float, atr_value: float | None, atr_mult: float, use_atr_dynamic_threshold: bool, ) -> float: fixed_distance = base_price * fixed_pct / 100 if use_atr_dynamic_threshold and atr_value and atr_value > 0: return max(fixed_distance, atr_value * atr_mult) return fixed_distance def kline_entity_abs(bar: Bar) -> float: return abs(bar.close - bar.open) def kline_entity_edges(bar: Bar) -> tuple[float, float]: return max(bar.open, bar.close), min(bar.open, bar.close) def upper_shadow_abs(bar: Bar) -> float: return max(0.0, bar.high - max(bar.open, bar.close)) def lower_shadow_abs(bar: Bar) -> float: return max(0.0, min(bar.open, bar.close) - bar.low) def close_position( equity: float, side: int, entry_price: float, exit_price: float, fee_rate: float, ) -> tuple[float, float]: net_ret = side * (exit_price - entry_price) / entry_price - 2 * fee_rate equity *= (1 + net_ret) return equity, net_ret def backtest_strategy( bars: list[Bar], params: StrategyParams, fee_rate: float = 0.0004, min_trades: int = 20, ) -> BacktestResult: atr_series = compute_atr_series(bars, params.atr_length) position = 0 entry_price = None max_favorable_price = None min_favorable_price = None equity = 1.0 peak_equity = 1.0 max_drawdown = 0.0 trades = 0 wins = 0 gross_profit = 0.0 gross_loss = 0.0 for i in range(1, len(bars)): current = bars[i] current_price = current.close atr_value = atr_series[i - 1] prev_idx = None for j in range(i - 1, -1, -1): prev_bar = bars[j] entity = kline_entity_abs(prev_bar) entity_pct = (entity / prev_bar.open * 100) if prev_bar.open else 0 if entity_pct > params.min_prev_entity_pct: prev_idx = j break if prev_idx is None: continue prev = bars[prev_idx] prev_entity = kline_entity_abs(prev) prev_entity_upper, prev_entity_lower = kline_entity_edges(prev) prev_is_bullish_for_calc = prev.close > prev.open prev_is_bearish_for_calc = prev.close < prev.open current_open_above_prev_close = current.open > prev.close current_open_below_prev_close = current.open < prev.close use_current_open_as_base = ( (prev_is_bullish_for_calc and current_open_above_prev_close) or (prev_is_bearish_for_calc and current_open_below_prev_close) ) if use_current_open_as_base: calc_lower = current.open calc_upper = current.open long_trigger = calc_lower + prev_entity / 3 short_trigger = calc_upper - prev_entity / 3 long_breakout = calc_upper + prev_entity / 3 short_breakout = calc_lower - prev_entity / 3 else: long_trigger = prev_entity_lower + prev_entity / 3 short_trigger = prev_entity_upper - prev_entity / 3 long_breakout = prev_entity_upper + prev_entity / 3 short_breakout = prev_entity_lower - prev_entity / 3 breakout_buffer = max( prev_entity * 0.1, current_price * params.breakout_buffer_pct / 100, (atr_value * params.breakout_buffer_atr_mult) if (params.use_atr_dynamic_threshold and atr_value and atr_value > 0) else 0, ) long_breakout_effective = long_breakout + breakout_buffer short_breakout_effective = short_breakout - breakout_buffer prev_is_bearish = prev.close < prev.open current_is_bullish = current.close > current.open skip_short_by_upper_third = prev_is_bearish and current_is_bullish prev_is_bullish = prev.close > prev.open current_is_bearish = current.close < current.open skip_long_by_lower_third = prev_is_bullish and current_is_bearish if position != 0 and entry_price is not None: sl_distance = resolve_dynamic_distance( base_price=entry_price, fixed_pct=params.stop_loss_pct, atr_value=atr_value, atr_mult=params.stop_loss_atr_mult, use_atr_dynamic_threshold=params.use_atr_dynamic_threshold, ) tp_distance = resolve_dynamic_distance( base_price=entry_price, fixed_pct=params.take_profit_pct, atr_value=atr_value, atr_mult=params.take_profit_atr_mult, use_atr_dynamic_threshold=params.use_atr_dynamic_threshold, ) trail_start_distance = resolve_dynamic_distance( base_price=entry_price, fixed_pct=params.trailing_start_pct, atr_value=atr_value, atr_mult=params.trailing_start_atr_mult, use_atr_dynamic_threshold=params.use_atr_dynamic_threshold, ) trail_backoff_distance = resolve_dynamic_distance( base_price=entry_price, fixed_pct=params.trailing_backoff_pct, atr_value=atr_value, atr_mult=params.trailing_backoff_atr_mult, use_atr_dynamic_threshold=params.use_atr_dynamic_threshold, ) should_close = False if position == 1: max_favorable_price = max(max_favorable_price or entry_price, current_price) profit_distance = current_price - entry_price loss_distance = entry_price - current_price if loss_distance >= sl_distance: should_close = True elif profit_distance >= tp_distance: should_close = True elif ( profit_distance >= trail_start_distance and (max_favorable_price - current_price) >= trail_backoff_distance ): should_close = True else: min_favorable_price = min(min_favorable_price or entry_price, current_price) profit_distance = entry_price - current_price loss_distance = current_price - entry_price if loss_distance >= sl_distance: should_close = True elif profit_distance >= tp_distance: should_close = True elif ( profit_distance >= trail_start_distance and (current_price - min_favorable_price) >= trail_backoff_distance ): should_close = True if should_close: equity, net_ret = close_position( equity=equity, side=position, entry_price=entry_price, exit_price=current_price, fee_rate=fee_rate, ) trades += 1 if net_ret > 0: wins += 1 gross_profit += net_ret else: gross_loss += net_ret position = 0 entry_price = None max_favorable_price = None min_favorable_price = None peak_equity = max(peak_equity, equity) max_drawdown = max(max_drawdown, (peak_equity - equity) / peak_equity) continue signal = None if position == 0: if current_price >= long_breakout_effective and not skip_long_by_lower_third: signal = "long" elif current_price <= short_breakout_effective and not skip_short_by_upper_third: signal = "short" elif position == 1: if current_price <= short_trigger and not skip_short_by_upper_third: signal = "reverse_short" else: upper_abs = upper_shadow_abs(prev) upper_thr = resolve_dynamic_distance( base_price=max(prev.open, prev.close), fixed_pct=params.shadow_threshold_pct, atr_value=atr_value, atr_mult=params.shadow_threshold_atr_mult, use_atr_dynamic_threshold=params.use_atr_dynamic_threshold, ) if upper_abs > upper_thr and current_price <= prev_entity_lower: signal = "reverse_short" elif position == -1: if current_price >= long_trigger and not skip_long_by_lower_third: signal = "reverse_long" else: lower_abs = lower_shadow_abs(prev) lower_thr = resolve_dynamic_distance( base_price=min(prev.open, prev.close), fixed_pct=params.shadow_threshold_pct, atr_value=atr_value, atr_mult=params.shadow_threshold_atr_mult, use_atr_dynamic_threshold=params.use_atr_dynamic_threshold, ) if lower_abs > lower_thr and current_price >= prev_entity_upper: signal = "reverse_long" if signal == "long": position = 1 entry_price = current_price max_favorable_price = current_price min_favorable_price = None elif signal == "short": position = -1 entry_price = current_price min_favorable_price = current_price max_favorable_price = None elif signal == "reverse_long": equity, net_ret = close_position( equity=equity, side=position, entry_price=entry_price, exit_price=current_price, fee_rate=fee_rate, ) trades += 1 if net_ret > 0: wins += 1 gross_profit += net_ret else: gross_loss += net_ret position = 1 entry_price = current_price max_favorable_price = current_price min_favorable_price = None peak_equity = max(peak_equity, equity) max_drawdown = max(max_drawdown, (peak_equity - equity) / peak_equity) elif signal == "reverse_short": equity, net_ret = close_position( equity=equity, side=position, entry_price=entry_price, exit_price=current_price, fee_rate=fee_rate, ) trades += 1 if net_ret > 0: wins += 1 gross_profit += net_ret else: gross_loss += net_ret position = -1 entry_price = current_price min_favorable_price = current_price max_favorable_price = None peak_equity = max(peak_equity, equity) max_drawdown = max(max_drawdown, (peak_equity - equity) / peak_equity) if position != 0 and entry_price is not None: equity, net_ret = close_position( equity=equity, side=position, entry_price=entry_price, exit_price=bars[-1].close, fee_rate=fee_rate, ) trades += 1 if net_ret > 0: wins += 1 gross_profit += net_ret else: gross_loss += net_ret peak_equity = max(peak_equity, equity) max_drawdown = max(max_drawdown, (peak_equity - equity) / peak_equity) total_return_pct = (equity - 1) * 100 win_rate_pct = (wins / trades * 100) if trades else 0.0 loss_abs = abs(gross_loss) profit_factor = (gross_profit / loss_abs) if loss_abs > 1e-12 else 999.0 score = total_return_pct - max_drawdown * 100 * 0.5 if trades < min_trades: score -= (min_trades - trades) * 0.8 return BacktestResult( score=score, total_return_pct=total_return_pct, max_drawdown_pct=max_drawdown * 100, win_rate_pct=win_rate_pct, profit_factor=profit_factor, trades=trades, wins=wins, params=params, ) def split_train_valid( bars: list[Bar], train_ratio: float = 0.7, gap_bars: int = 0, ) -> tuple[list[Bar], list[Bar]]: """ 时间序列分离：前段训练、后段验证。 gap_bars 用于在训练与验证之间留空，降低相邻样本泄漏。 """ if not bars: return [], [] train_ratio = min(max(train_ratio, 0.5), 0.9) split_idx = int(len(bars) * train_ratio) split_idx = max(1, min(split_idx, len(bars) - 1)) valid_start = min(len(bars), split_idx + max(0, gap_bars)) train_bars = bars[:split_idx] valid_bars = bars[valid_start:] return train_bars, valid_bars def risk_adjusted_return(result: BacktestResult) -> float: """简单风险调整收益：收益 - 0.6 * 回撤。""" return result.total_return_pct - 0.6 * result.max_drawdown_pct def compute_robust_score( train_result: BacktestResult, valid_result: BacktestResult, min_train_trades: int = 20, min_valid_trades: int = 10, ) -> tuple[float, float, float, float, float]: """ 稳健性分数（越高越稳健）： - 以验证集风险调整收益为主 - 奖励训练/验证一致性 - 惩罚过拟合（训练好、验证差） - 惩罚验证成交次数过少 """ train_ra = risk_adjusted_return(train_result) valid_ra = risk_adjusted_return(valid_result) overfit_gap = abs(train_ra - valid_ra) denom = abs(train_ra) + abs(valid_ra) + 1e-9 consistency = max(0.0, 1.0 - overfit_gap / denom) train_trade_penalty = max(0, min_train_trades - train_result.trades) * 0.4 valid_trade_penalty = max(0, min_valid_trades - valid_result.trades) * 1.2 pf_bonus = min(valid_result.profit_factor, 3.0) * 2.0 win_bonus = max(0.0, (valid_result.win_rate_pct - 50.0) * 0.08) direction_penalty = 0.0 if train_ra > 0 and valid_ra <= 0: direction_penalty += 12.0 if train_result.total_return_pct > 0 and valid_result.total_return_pct < 0: direction_penalty += 8.0 robust_score = ( 0.75 * valid_ra + 0.25 * train_ra + 10.0 * consistency + pf_bonus + win_bonus - 0.2 * overfit_gap - train_trade_penalty - valid_trade_penalty - direction_penalty ) return robust_score, consistency, overfit_gap, train_ra, valid_ra def evaluate_param_set( train_bars: list[Bar], valid_bars: list[Bar], full_bars: list[Bar], params: StrategyParams, fee_rate: float, min_train_trades: int, min_valid_trades: int, ) -> RobustEvalResult: train_result = backtest_strategy( bars=train_bars, params=params, fee_rate=fee_rate, min_trades=0, ) valid_result = backtest_strategy( bars=valid_bars, params=params, fee_rate=fee_rate, min_trades=0, ) full_result = backtest_strategy( bars=full_bars, params=params, fee_rate=fee_rate, min_trades=0, ) robust_score, consistency, overfit_gap, train_ra, valid_ra = compute_robust_score( train_result=train_result, valid_result=valid_result, min_train_trades=min_train_trades, min_valid_trades=min_valid_trades, ) return RobustEvalResult( robust_score=robust_score, consistency_score=consistency, overfit_gap=overfit_gap, train_risk_adj=train_ra, valid_risk_adj=valid_ra, train=train_result, valid=valid_result, full=full_result, params=params, ) def quick_grid() -> dict[str, list[float | int]]: return { "atr_length": [10, 14, 20], "breakout_buffer_atr_mult": [0.08, 0.12], "shadow_threshold_atr_mult": [0.14, 0.2], "stop_loss_atr_mult": [0.35, 0.5], "take_profit_atr_mult": [0.8, 1.0], "trailing_start_atr_mult": [0.45, 0.65], "trailing_backoff_atr_mult": [0.2, 0.3], } def full_grid() -> dict[str, list[float | int]]: return { "atr_length": [10, 14, 20], "breakout_buffer_atr_mult": [0.08, 0.12, 0.16], "shadow_threshold_atr_mult": [0.12, 0.18, 0.24], "stop_loss_atr_mult": [0.35, 0.5, 0.65], "take_profit_atr_mult": [0.8, 1.0, 1.2], "trailing_start_atr_mult": [0.45, 0.65, 0.85], "trailing_backoff_atr_mult": [0.2, 0.3, 0.4], } def iter_param_combos(base: StrategyParams, grid: dict[str, list[float | int]]): keys = list(grid.keys()) values = [grid[k] for k in keys] for combo in itertools.product(*values): data = asdict(base) for k, v in zip(keys, combo): data[k] = v yield StrategyParams(**data) def result_metrics_dict(result: BacktestResult) -> dict[str, float | int]: return { "score": result.score, "total_return_pct": result.total_return_pct, "max_drawdown_pct": result.max_drawdown_pct, "win_rate_pct": result.win_rate_pct, "profit_factor": result.profit_factor, "trades": result.trades, "wins": result.wins, } def save_best_params( best: RobustEvalResult, out_path: Path, train_ratio: float, split_gap_bars: int, ): payload = { "apply_live": False, "selection_metric": "robust_score", "robust_score": best.robust_score, "consistency_score": best.consistency_score, "overfit_gap": best.overfit_gap, "train_risk_adj": best.train_risk_adj, "valid_risk_adj": best.valid_risk_adj, "split": { "train_ratio": train_ratio, "split_gap_bars": split_gap_bars, }, "train_metrics": result_metrics_dict(best.train), "valid_metrics": result_metrics_dict(best.valid), "full_metrics": result_metrics_dict(best.full), "params_for_trade_py": asdict(best.params), } out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def main(): parser = argparse.ArgumentParser(description="ATR dynamic parameter optimizer for bitmart strategy") parser.add_argument( "--csv", default=str(Path(__file__).resolve().parent / "数据" / "kline_1.csv"), help="path to csv with id/open/high/low/close", ) parser.add_argument("--interval-min", type=int, default=5, help="resample interval minutes") parser.add_argument("--mode", choices=["quick", "full"], default="quick", help="grid size") parser.add_argument("--limit-bars", type=int, default=30000, help="use latest N bars after resample") parser.add_argument("--train-ratio", type=float, default=0.7, help="train split ratio in time order") parser.add_argument("--split-gap-bars", type=int, default=0, help="gap bars between train and valid") parser.add_argument("--min-trades", type=int, default=20, help="deprecated: kept for compatibility") parser.add_argument("--min-train-trades", type=int, default=20, help="min train trades for robustness") parser.add_argument("--min-valid-trades", type=int, default=10, help="min valid trades for robustness") parser.add_argument("--fee-rate", type=float, default=0.0004, help="round-trip half fee per side") parser.add_argument("--top-n", type=int, default=10, help="print top N results") parser.add_argument( "--out-json", default=str(Path(__file__).resolve().parent / "atr_best_params.json"), help="best result json output path", ) args = parser.parse_args() csv_path = Path(args.csv).resolve() if not csv_path.exists(): raise FileNotFoundError(f"CSV not found: {csv_path}") bars = load_csv_bars(csv_path) bars = resample_to_minutes(bars, args.interval_min) if args.limit_bars and args.limit_bars > 0: bars = bars[-args.limit_bars:] if len(bars) < 400: raise ValueError("not enough bars for optimization") train_bars, valid_bars = split_train_valid( bars=bars, train_ratio=args.train_ratio, gap_bars=args.split_gap_bars, ) if len(train_bars) < 200 or len(valid_bars) < 100: raise ValueError( f"insufficient split bars: train={len(train_bars)} valid={len(valid_bars)}; " "consider increasing --limit-bars or adjusting --train-ratio" ) base = StrategyParams() grid = quick_grid() if args.mode == "quick" else full_grid() combos = list(iter_param_combos(base, grid)) print( f"bars={len(bars)} train={len(train_bars)} valid={len(valid_bars)} " f"| combos={len(combos)} | mode={args.mode} | train_ratio={args.train_ratio:.2f} gap={args.split_gap_bars}" ) results: list[RobustEvalResult] = [] for idx, params in enumerate(combos, 1): result = evaluate_param_set( train_bars=train_bars, valid_bars=valid_bars, full_bars=bars, params=params, fee_rate=args.fee_rate, min_train_trades=args.min_train_trades, min_valid_trades=args.min_valid_trades, ) results.append(result) if idx % 50 == 0 or idx == len(combos): print(f"progress {idx}/{len(combos)}") results.sort(key=lambda x: x.robust_score, reverse=True) top_n = max(1, args.top_n) top = results[:top_n] for i, r in enumerate(top, 1): p = r.params print( f"[{i}] robust={r.robust_score:.2f} consistency={r.consistency_score:.3f} gap={r.overfit_gap:.2f} | " f"train(ret={r.train.total_return_pct:.2f}% dd={r.train.max_drawdown_pct:.2f}% trades={r.train.trades}) | " f"valid(ret={r.valid.total_return_pct:.2f}% dd={r.valid.max_drawdown_pct:.2f}% trades={r.valid.trades}) | " f"full(ret={r.full.total_return_pct:.2f}% dd={r.full.max_drawdown_pct:.2f}% trades={r.full.trades}) | " f"atr={p.atr_length} brk={p.breakout_buffer_atr_mult:.2f} shadow={p.shadow_threshold_atr_mult:.2f} " f"sl={p.stop_loss_atr_mult:.2f} tp={p.take_profit_atr_mult:.2f} " f"ts={p.trailing_start_atr_mult:.2f} tb={p.trailing_backoff_atr_mult:.2f}" ) out_path = Path(args.out_json).resolve() save_best_params( top[0], out_path, train_ratio=args.train_ratio, split_gap_bars=args.split_gap_bars, ) print(f"saved best params -> {out_path}") print("note: set apply_live=true in json when you want 交易.py to auto-load it") if __name__ == "__main__": main()