""" 最终回测 - Top 3 盈利参数 + 不同仓位大小 Top 3 profitable combos from scan: 1. EMA(8/21/120) ATR>0.30% SL=0.4% MH=1800s → +2.88% 2. EMA(30/80/200) ATR>0.20% SL=0.8% MH=3600s → +2.53% 3. EMA(8/21/120) ATR>0.20% SL=0.8% MH=1800s → +1.94% 每组参数测试 risk_pct = [0.005, 0.01, 0.02, 0.03, 0.05] 输出详细月度报告和交易明细 """ import sys, time, datetime, sqlite3, statistics from pathlib import Path from dataclasses import dataclass from typing import List @dataclass class Trade: open_time: datetime.datetime close_time: datetime.datetime direction: str open_price: float close_price: float size: float pnl: float pnl_pct: float fee: float rebate: float hold_seconds: float close_reason: str class EMA: __slots__ = ('k', 'v') def __init__(self, p): self.k = 2.0 / (p + 1); self.v = None def update(self, x): self.v = x if self.v is None else x * self.k + self.v * (1 - self.k) return self.v def load(): db = Path(__file__).parent.parent / 'models' / 'database.db' s = int(datetime.datetime(2025,1,1).timestamp()) * 1000 e = int(datetime.datetime(2026,1,1).timestamp()) * 1000 conn = sqlite3.connect(str(db)) rows = conn.cursor().execute( "SELECT id,open,high,low,close FROM bitmart_eth_1m WHERE id>=? AND id AP + 1: s = 0.0 for j in range(-AP, 0): tr = H[j] - L[j]; d1 = abs(H[j] - C[j-1]); d2 = abs(L[j] - C[j-1]) if d1 > tr: tr = d1 if d2 > tr: tr = d2 s += tr atr_pct = s / (AP * c) if c > 0 else 0 cu = pf is not None and pf <= pslow and fast > slow cd = pf is not None and pf >= pslow and fast < slow pf = fast; pslow = slow if pos != 0 and ot_: pp = (c - op_) / op_ if pos == 1 else (op_ - c) / op_ hsec = (dt - ot_).total_seconds() if -pp >= hsl: _close(c, dt, f"hard_SL({pp*100:+.2f}%)") continue if hsec >= min_hold: dc = False; reason = "" if -pp >= sl_pct: dc = True; reason = f"SL({pp*100:+.2f}%)" elif hsec >= mh: dc = True; reason = f"timeout({hsec:.0f}s)" elif pos == 1 and cd: dc = True; reason = "cross_rev" elif pos == -1 and cu: dc = True; reason = "cross_rev" elif pend == 'cl' and pos == 1: dc = True; reason = "delayed_cross" elif pend == 'cs' and pos == -1: dc = True; reason = "delayed_cross" if dc: _close(c, dt, reason) if atr_pct >= atr_min: if (cd or fast < slow) and c < big: _open('S', c, dt) elif (cu or fast > slow) and c > big: _open('L', c, dt) continue else: if pos == 1 and cd: pend = 'cl' elif pos == -1 and cu: pend = 'cs' if pos == 0 and atr_pct >= atr_min: if cu and c > big: _open('L', c, dt) elif cd and c < big: _open('S', c, dt) # equity tracking every hour if i % 60 == 0: eq = bal if pos != 0 and op_ > 0: pp = (c - op_) / op_ if pos == 1 else (op_ - c) / op_ eq += ps_ * pp eq_curve.append((dt, eq)) if eq > peak: peak = eq dd = (peak - eq) / peak if peak > 0 else 0 if dd > max_dd: max_dd = dd if pos != 0: _close(data[-1][4], data[-1][0], "backtest_end") return trades, bal, max_dd, eq_curve def print_report(name, trades, balance, max_dd, init=1000.0): if not trades: print(f"\n[{name}] No trades.\n", flush=True) return n = len(trades) wins = [t for t in trades if t.pnl > 0] losses = [t for t in trades if t.pnl <= 0] wr = len(wins) / n * 100 net = balance - init tot_pnl = sum(t.pnl for t in trades) tot_fee = sum(t.fee for t in trades) tot_reb = sum(t.rebate for t in trades) avg_hold = statistics.mean([t.hold_seconds for t in trades]) vol = sum(t.size for t in trades) * 2 pf_n = sum(t.pnl for t in wins) if wins else 0 pf_d = abs(sum(t.pnl for t in losses)) if losses else 0 pf = pf_n / pf_d if pf_d > 0 else float('inf') long_t = [t for t in trades if t.direction == 'long'] short_t = [t for t in trades if t.direction == 'short'] long_wr = len([t for t in long_t if t.pnl > 0]) / len(long_t) * 100 if long_t else 0 short_wr = len([t for t in short_t if t.pnl > 0]) / len(short_t) * 100 if short_t else 0 reasons = {} for t in trades: r = t.close_reason.split('(')[0] reasons[r] = reasons.get(r, 0) + 1 print(f"\n{'='*70}", flush=True) print(f" [{name}]", flush=True) print(f"{'='*70}", flush=True) print(f" Initial: {init:>10.2f} USDT", flush=True) print(f" Final: {balance:>10.2f} USDT", flush=True) print(f" Net P&L: {net:>+10.2f} USDT ({net/init*100:+.2f}%)", flush=True) print(f" Max Drawdown: {max_dd*100:>9.2f}%", flush=True) print(f"\n Trades: {n:>6} (Long {len(long_t)} WR={long_wr:.0f}% | Short {len(short_t)} WR={short_wr:.0f}%)", flush=True) print(f" Win Rate: {wr:>5.1f}%", flush=True) print(f" Profit Factor:{pf:>6.2f}", flush=True) print(f" Avg Hold: {avg_hold:>5.0f}s ({avg_hold/60:.1f}min)", flush=True) print(f"\n Dir P&L: {tot_pnl:>+10.2f}", flush=True) print(f" Total Fee: {tot_fee:>10.2f}", flush=True) print(f" Rebate(90%): {tot_reb:>+10.2f}", flush=True) print(f" Net Fee: {tot_fee - tot_reb:>10.2f}", flush=True) print(f" Volume: {vol:>10.0f}", flush=True) if wins: print(f"\n Avg Win: {statistics.mean([t.pnl for t in wins]):>+10.4f}", flush=True) if losses: print(f" Avg Loss: {statistics.mean([t.pnl for t in losses]):>+10.4f}", flush=True) print(f" Best: {max(t.pnl for t in trades):>+10.4f}", flush=True) print(f" Worst: {min(t.pnl for t in trades):>+10.4f}", flush=True) print(f"\n Close Reasons:", flush=True) for r, c in sorted(reasons.items(), key=lambda x: -x[1]): print(f" {r:<20} {c:>5} ({c/n*100:.1f}%)", flush=True) # Monthly print(f"\n {'Month':<8} {'Trd':>5} {'DirPnL':>9} {'Rebate':>9} {'Net':>9} {'WR':>6}", flush=True) print(f" {'-'*50}", flush=True) monthly = {} for t in trades: k = t.close_time.strftime('%Y-%m') if k not in monthly: monthly[k] = {'n': 0, 'pnl': 0, 'reb': 0, 'fee': 0, 'w': 0} monthly[k]['n'] += 1 monthly[k]['pnl'] += t.pnl monthly[k]['reb'] += t.rebate monthly[k]['fee'] += t.fee if t.pnl > 0: monthly[k]['w'] += 1 for m in sorted(monthly.keys()): d = monthly[m] net_m = d['pnl'] - d['fee'] + d['reb'] wr_m = d['w'] / d['n'] * 100 if d['n'] > 0 else 0 print(f" {m:<8} {d['n']:>5} {d['pnl']:>+9.2f} {d['reb']:>9.2f} {net_m:>+9.2f} {wr_m:>5.1f}%", flush=True) print(f"{'='*70}", flush=True) def main(): print("Loading data...", flush=True) data = load() print(f"{len(data)} bars\n", flush=True) # Top 3 parameter combos configs = [ {"name": "A", "fp": 8, "sp": 21, "bp": 120, "atr": 0.003, "sl": 0.004, "mh": 1800}, {"name": "B", "fp": 30, "sp": 80, "bp": 200, "atr": 0.002, "sl": 0.008, "mh": 3600}, {"name": "C", "fp": 8, "sp": 21, "bp": 120, "atr": 0.002, "sl": 0.008, "mh": 1800}, ] risk_levels = [0.005, 0.01, 0.02, 0.03, 0.05] print(f"{'='*100}", flush=True) print(f" RISK LEVEL COMPARISON", flush=True) print(f"{'='*100}", flush=True) print(f" {'Config':<50} {'Risk%':>6} {'Net%':>7} {'Net$':>9} {'Trades':>7} {'MaxDD':>7}", flush=True) print(f" {'-'*96}", flush=True) best_result = None best_net = -9999 for cfg in configs: for rp in risk_levels: label = f"EMA({cfg['fp']}/{cfg['sp']}/{cfg['bp']}) ATR>{cfg['atr']*100:.1f}% SL={cfg['sl']*100:.1f}% MH={cfg['mh']}" trades, bal, mdd, eq = run_detailed( data, cfg['fp'], cfg['sp'], cfg['bp'], cfg['atr'], cfg['sl'], cfg['mh'], risk_pct=rp ) net = bal - 1000.0 mk = " <<<" if net > 0 else "" print(f" {label:<50} {rp*100:>5.1f}% {net/10:>+6.2f}% {net:>+8.2f} {len(trades):>7} {mdd*100:>6.2f}%{mk}", flush=True) if net > best_net: best_net = net best_result = (cfg, rp, trades, bal, mdd, eq) print(f"{'='*100}\n", flush=True) # Detailed report for best if best_result: cfg, rp, trades, bal, mdd, eq = best_result label = f"EMA({cfg['fp']}/{cfg['sp']}/{cfg['bp']}) ATR>{cfg['atr']*100:.1f}% SL={cfg['sl']*100:.1f}% MH={cfg['mh']} Risk={rp*100:.1f}%" print_report(f"BEST: {label}", trades, bal, mdd) # Save trades CSV csv = Path(__file__).parent.parent / 'best_trades.csv' with open(csv, 'w', encoding='utf-8-sig') as f: f.write("open_time,close_time,dir,open_px,close_px,size,pnl,pnl_pct,fee,rebate,hold_sec,reason\n") for t in trades: f.write(f"{t.open_time},{t.close_time},{t.direction}," f"{t.open_price:.2f},{t.close_price:.2f},{t.size:.2f}," f"{t.pnl:.4f},{t.pnl_pct*100:.4f}%,{t.fee:.4f},{t.rebate:.4f}," f"{t.hold_seconds:.0f},{t.close_reason}\n") print(f"\nBest trades saved: {csv}", flush=True) # Save equity curve eq_csv = Path(__file__).parent.parent / 'best_equity.csv' with open(eq_csv, 'w', encoding='utf-8-sig') as f: f.write("time,equity\n") for dt, e in eq: f.write(f"{dt},{e:.2f}\n") print(f"Equity curve saved: {eq_csv}", flush=True) if __name__ == '__main__': main()