#!/usr/bin/env python3 """ OKX 币圈做T 回测工具 (v2.0.0) 基于历史 K 线模拟策略, 验证 buy/sell 价位参数 """ import os, json, sys, argparse, datetime sys.path.insert(0, os.path.dirname(__file__)) # 加载凭证 okx_creds = {} with open(os.path.expanduser('~/.bashrc')) as f: import re for line in f: m = re.match(r'export\s+(OKX_\w+)=(.*)', line.strip()) if m: okx_creds[m.group(1)] = m.group(2).strip().strip('"').strip("'") def fetch_history_klines(sym, bar='1H', days=30): """拉 OKX 历史 K 线 (OKX 限制单次 100 根, 多页拉) 用 OKX 的 'after' 参数翻页 (传毫秒时间戳) """ import subprocess import hmac, base64, hashlib all_data = [] # OKX 时间戳 (毫秒) cur_ts = int(datetime.datetime.utcnow().timestamp() * 1000) # 计算需要多少页 (1H K线, 24 根/天) pages = max(1, (days * 24 + 99) // 100) for page in range(pages): path = f"/api/v5/market/history-candles?instId={sym}-USDT-SWAP&bar={bar}&limit=100&after={cur_ts}" msg = datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.') + \ f"{datetime.datetime.utcnow().microsecond // 1000:03d}Z" + 'GET' + path signature = base64.b64encode( hmac.new(okx_creds['OKX_SECRET'].encode(), msg.encode(), hashlib.sha256).digest() ).decode() ts_str = msg[:30] # YYYY-MM-DDTHH:MM:SS.sssZ (但实际上 ms 只有 3 位 + Z) # 修正: 用 'Z' 结尾的后 24 字节 curl_cmd = [ 'curl', '-s', '--proxy', 'http://127.0.0.1:7890', '-H', f'OK-ACCESS-KEY: {okx_creds["OKX_API_KEY"]}', '-H', f'OK-ACCESS-SIGN: {signature}', '-H', f'OK-ACCESS-TIMESTAMP: {ts_str}', '-H', f'OK-ACCESS-PASSPHRASE: {okx_creds["OKX_PASSPHRASE"]}', f'https://www.okx.com{path}' ] try: r = subprocess.run(curl_cmd, capture_output=True, text=True, timeout=20) data = json.loads(r.stdout) if data.get('code') == '0': klines = data.get('data', []) if not klines: break all_data.extend(klines) # 翻页: after 是上一个数据最小时间戳 - 1 cur_ts = int(klines[-1][0]) - 1 if len(klines) < 100: break else: print(f"⚠️ Page {page} code={data.get('code')} msg={data.get('msg')}") break except Exception as e: print(f"⚠️ Page {page} failed: {e}") break print(f"📥 拉到 {len(all_data)} 根 K 线") return all_data def calc_atr(klines, period=14): """ATR 计算""" if len(klines) < period + 1: return None closes = [float(k[4]) for k in klines] highs = [float(k[2]) for k in klines] lows = [float(k[3]) for k in klines] trs = [] for i in range(1, len(closes)): tr = max(highs[i] - lows[i], abs(highs[i] - closes[i-1]), abs(lows[i] - closes[i-1])) trs.append(tr) return sum(trs[-period:]) / period def simulate_strategy(klines, atr_multiplier=0.5, t_qty=0.05, leverage=25, ct_val=0.1, initial_usdt=1000, threshold=0.003): """基于历史 K 线模拟做T策略 每小时检查价位: - 跌到 buy1/buy2 → 买入 - 涨到 sell1/sell2 → 卖出 持仓同步变化 (跟 okx_t_monitor 一致) """ trades = [] position = 0 avg_cost = 0 last_trade_ts = None for i in range(20, len(klines)): row = klines[i] ts = row[0] high = float(row[2]) low = float(row[3]) close = float(row[4]) # 计算过去 14 根 K 线的 ATR past = klines[i-20:i] atr = calc_atr(past, 14) if not atr: continue buy1 = close - atr * atr_multiplier * 0.5 buy2 = close - atr * atr_multiplier sell1 = close + atr * atr_multiplier * 0.5 sell2 = close + atr * atr_multiplier # 检查是否触及价位 (用 high/low 比对 close) if last_trade_ts == ts: continue # 优先 sell1 > buy1 (趋势方向) if position > 0 and (high >= sell2 or (high >= sell1 and position > 0)): # 卖出 sell_price = sell2 if high >= sell2 else sell1 pnl = (sell_price - avg_cost) * position trades.append(('sell', sell_price, position, pnl, ts)) position = 0 avg_cost = 0 last_trade_ts = ts elif position == 0 and (low <= buy2 or low <= buy1): buy_price = buy2 if low <= buy2 else buy1 position = t_qty avg_cost = buy_price trades.append(('buy', buy_price, position, None, ts)) last_trade_ts = ts # 统计 total_pnl = sum(t[3] for t in trades if t[3] is not None) buy_count = sum(1 for t in trades if t[0] == 'buy') sell_count = sum(1 for t in trades if t[0] == 'sell') win_trades = [t for t in trades if t[3] and t[3] > 0] win_rate = len(win_trades) / sell_count * 100 if sell_count > 0 else 0 return { 'trades': trades, 'total_pnl': total_pnl, 'buy_count': buy_count, 'sell_count': sell_count, 'win_rate': win_rate, 'final_position': position, 'final_avg_cost': avg_cost, } def main(): parser = argparse.ArgumentParser(description='币圈做T回测 (v2.0.0)') parser.add_argument('symbol', help='币种 (如 ETH)') parser.add_argument('--mode', choices=['short', 'trend'], default='trend', help='short=日内(1H,默认) / trend=趋势(4H,默认短期)') parser.add_argument('--days', type=int, default=30, help='回测天数 (short=7, trend=30)') parser.add_argument('--bar', default=None, help='K 线周期 (覆盖 mode 默认)') parser.add_argument('--atr-multiplier', type=float, default=None, help='ATR 倍数') parser.add_argument('--t-qty', type=float, default=0.05, help='每笔数量 (默认 0.05)') parser.add_argument('--leverage', type=int, default=25, help='杠杆 (默认 25)') parser.add_argument('--ct-val', type=float, default=0.1, help='合约面值 (默认 0.1)') args = parser.parse_args() # Mode-based defaults if args.bar is None: args.bar = '1H' if args.mode == 'short' else '4H' if args.atr_multiplier is None: # Trend: 更宽价位 (ATR × 1.5), 避免被洗 args.atr_multiplier = 0.5 if args.mode == 'short' else 1.5 if args.days == 30: # 如果用户没指定，按 mode args.days = 7 if args.mode == 'short' else 30 print(f"📊 {args.symbol} {args.bar} 回测 ({args.days} 天, mode={args.mode})") print(f" ATR={args.atr_multiplier} t_qty={args.t_qty} lev={args.leverage}x") print() # 拉数据 klines = fetch_history_klines(args.symbol, args.bar, args.days) if not klines: print("❌ 没拉到数据") sys.exit(1) print(f"✅ 拉到 {len(klines)} 根 K 线") print() # 模拟 result = simulate_strategy(klines, args.atr_multiplier, args.t_qty, args.leverage, args.ct_val) # 报告 print(f"📈 回测结果:") print(f" 买入: {result['buy_count']} 次") print(f" 卖出: {result['sell_count']} 次") print(f" 胜率: {result['win_rate']:.1f}%") print(f" 总盈亏: ${result['total_pnl']:.2f}") print(f" 最终仓位: {result['final_position']}张 @ ${result['final_avg_cost']:.2f}" if result['final_position'] > 0 else " 最终仓位: 0 (全平)") # Top 5 交易 closed = [t for t in result['trades'] if t[3] is not None] if closed: print() print(f" Top 5 盈利交易:") for t in sorted(closed, key=lambda x: -x[3])[:5]: print(f" ${t[1]:.2f} | pnl ${t[3]:.2f} | {t[4]}") if __name__ == '__main__': main()