#!/usr/bin/env python3 """ 格蕾丝 10U作战bot — 基于烛龙v1.9多时间框架共振 ================================================== 1H吞没/晨星 + 20周期支撑阻力 + 日线MA20趋势过滤 + 4H/15min共振 SL=1.5% TP=4.5% (v1.9回测: 样本外65笔 胜率56.9% Sharpe5.10) 风控铁律: - 单笔风险 <= 20% 账户 (10U -> 最多亏2U/笔) - 每天最多2笔, 亏损2笔立即停机 - OCO必挂 (止损+止盈同时) - 杠杆: 5x (低杠杆求生存) """ import os, sys, time, hmac, hashlib, base64, json, signal import requests import pandas as pd import numpy as np from datetime import datetime, timezone # ============ 配置 ============ DRY_RUN = "--dry-run" in sys.argv INST = "BTC-USDT-SWAP" LEVERAGE = 5 SL_PCT, TP_PCT = 0.015, 0.045 MAX_TRADES_PER_DAY = 2 MAX_LOSSES_PER_DAY = 2 CHECK_INTERVAL = 300 # 5分钟扫一次 DATA_DIR = "/root/quant_pipeline/data" STATE_FILE = f"{DATA_DIR}/grace10u_state.json" LOG_FILE = f"{DATA_DIR}/grace10u_trades.csv" OKX_KEY = os.getenv("OKX_API_KEY", "") OKX_SECRET = os.getenv("OKX_SECRET_KEY", "") OKX_PASS = os.getenv("OKX_PASSPHRASE", "") BASE = "https://www.okx.com" # ============ OKX 签名 ============ def okx_req(method, path, body=""): ts = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z' sig = base64.b64encode(hmac.new( OKX_SECRET.encode(), (ts+method+path+body).encode(), hashlib.sha256).digest()).decode() h = {"OK-ACCESS-KEY": OKX_KEY, "OK-ACCESS-SIGN": sig, "OK-ACCESS-TIMESTAMP": ts, "OK-ACCESS-PASSPHRASE": OKX_PASS, "Content-Type": "application/json"} r = requests.request(method, BASE+path, headers=h, data=body, timeout=15) return r.json() def get_candles(bar, limit=300): r = requests.get(f"{BASE}/api/v5/market/candles?instId={INST}&bar={bar}&limit={limit}", timeout=15) data = r.json()["data"] df = pd.DataFrame(data, columns=["ts","o","h","l","c","vol","_","_","_"]) for col in ["o","h","l","c","vol"]: df[col] = df[col].astype(float) df["ts"] = pd.to_datetime(df["ts"].astype(int), unit="ms") return df.set_index("ts").iloc[::-1].sort_index() def get_balance(): r = okx_req("GET", "/api/v5/account/balance") if r.get("code") == "0": for d in r["data"][0]["details"]: if d["ccy"] == "USDT": return float(d.get("eq", 0)), float(d.get("availBal", 0)) return 0, 0 # ============ 信号 ============ def load_state(): if os.path.exists(STATE_FILE): with open(STATE_FILE) as f: return json.load(f) return {"day": "", "trades": 0, "losses": 0} def save_state(s): with open(STATE_FILE, "w") as f: json.dump(s, f) def log_trade(row): new = not os.path.exists(LOG_FILE) pd.DataFrame([row]).to_csv(LOG_FILE, mode="a", header=new, index=False) def detect_signal(): """返回 (dirc, score, reason) 或 None""" d1 = get_candles("1H", 120) d4 = get_candles("4H", 60) d15 = get_candles("15m", 120) dd = get_candles("1D", 40) # 日线趋势 dd["ma20"] = dd["c"].rolling(20).mean() trend_up = dd["c"].iloc[-1] > dd["ma20"].iloc[-1] # 4H趋势 d4["ma20"] = d4["c"].rolling(20).mean() trend_4h = d4["c"].iloc[-1] > d4["ma20"].iloc[-1] # 1H信号 o = d1["o"]; h = d1["h"]; l = d1["l"]; c = d1["c"] o1 = o.shift(1); h1 = h.shift(1); l1 = l.shift(1); c1 = c.shift(1) prev_l20 = l.shift(1).rolling(20).min() prev_h20 = h.shift(1).rolling(20).max() near_s = (l - prev_l20).abs() / (prev_l20 + 1e-9) < 0.008 near_r = (h - prev_h20).abs() / (prev_h20 + 1e-9) < 0.008 bull_engulf = (c1 < o1) & (c > o) & (o <= c1) & (c >= o1) swup = (h > h.shift(1).rolling(20).max()) & (c < h.shift(1).rolling(20).max()) next_bull = c.shift(-1) > o.shift(-1) next_bear = c.shift(-1) < o.shift(-1) long_raw = bull_engulf & near_s & next_bull short_raw = swup & near_r & next_bear # 15min同向确认 o15 = d15["o"]; h15 = d15["h"]; l15 = d15["l"]; c15 = d15["c"] prev_l20_15 = l15.shift(1).rolling(20).min() prev_h20_15 = h15.shift(1).rolling(20).max() body15 = (c15 - o15).abs() rk15 = h15 - l15 body_r15 = body15 / (rk15 + 1e-9) bull15 = (c15.shift(1) < o15.shift(1)) & (c15 > o15) & (o15 <= c15.shift(1)) & (c15 >= o15.shift(1)) & \ ((l15 - prev_l20_15).abs() / (prev_l20_15 + 1e-9) < 0.008) bear15 = (c15.shift(1) > o15.shift(1)) & (c15 < o15) & (o15 >= c15.shift(1)) & (c15 <= o15.shift(1)) & \ ((h15 - prev_h20_15).abs() / (prev_h20_15 + 1e-9) < 0.008) nbull15 = c15.shift(-1) > o15.shift(-1) nbear15 = c15.shift(-1) < o15.shift(-1) sig15_long = (bull15 & nbull15).iloc[-2:].any() sig15_short = (bear15 & nbear15).iloc[-2:].any() # 量能 vol_mean20 = d1["vol"].rolling(20).mean() vol_ratio = d1["vol"].iloc[-1] / vol_mean20.iloc[-1] # 决策 if long_raw.iloc[-1] and trend_up: score = 40 reason = f"1H吞没@20支撑+日线多头" if trend_4h: score += 20; reason += "+4H共振" if sig15_long: score += 10; reason += "+15m确认" if vol_ratio > 1.5: score += 15; reason += f"+放量{vol_ratio:.1f}x" elif vol_ratio > 1.2: score += 8; reason += f"+量{vol_ratio:.1f}x" return 1, score, reason if short_raw.iloc[-1] and (not trend_up): score = 40 reason = f"1H假突破20阻力+日线空头" if not trend_4h: score += 20; reason += "+4H共振" if sig15_short: score += 10; reason += "+15m确认" if vol_ratio > 1.5: score += 15; reason += f"+放量{vol_ratio:.1f}x" elif vol_ratio > 1.2: score += 8; reason += f"+量{vol_ratio:.1f}x" return -1, score, reason return None # ============ 下单 ============ def set_leverage(): if DRY_RUN: return body = json.dumps({"instId": INST, "lever": str(LEVERAGE), "mgnMode": "cross"}) okx_req("POST", "/api/v5/account/set-leverage", body) def place_trade(dirc, score, reason, price): """开仓+OCO""" side = "buy" if dirc == 1 else "sell" # 仓位: 10U账户, 单笔风险<=20% => 用2U保证金*5x = 10U名义 sz = round(0.3, 2) # 0.3张 (ctVal=0.01 BTC => 0.003BTC ≈ 192U名义, 太高) # 重新计算: 10U账户, 5x杠杆, 单笔投入2U => 名义10U => 10/64000 = 0.000156BTC = 0.0156张 sz = round(max(0.01, 2 * LEVERAGE / price), 4) sl_px = round(price * (1 - SL_PCT) if dirc == 1 else price * (1 + SL_PCT), 1) tp_px = round(price * (1 + TP_PCT) if dirc == 1 else price * (1 - TP_PCT), 1) print(f"\n📡 信号: {'做多' if dirc==1 else '做空'} 评分{score} | {reason}") print(f" 价格 {price} | 仓位 {sz}张 | SL {sl_px} | TP {tp_px}") if DRY_RUN: log_trade({"ts": datetime.now().isoformat(), "mode": "DRY", "dirc": dirc, "score": score, "reason": reason, "price": price, "sz": sz, "sl": sl_px, "tp": tp_px, "result": "SIGNAL"}) return True # 下单 body = json.dumps({"instId": INST, "tdMode": "cross", "side": side, "ordType": "market", "sz": str(sz)}) r = okx_req("POST", "/api/v5/trade/order", body) if r.get("code") != "0": print(f"❌ 下单失败: {r.get('msg')}") return False ord_id = r["data"][0]["ordId"] time.sleep(1) # 挂OCO oco = json.dumps({"instId": INST, "tdMode": "cross", "side": "sell" if dirc == 1 else "buy", "sz": str(sz), "tpTriggerPx": str(tp_px), "tpOrdPx": "-1", "slTriggerPx": str(sl_px), "slOrdPx": "-1"}) r2 = okx_req("POST", "/api/v5/trade/order-algo", oco) if r2.get("code") != "0": print(f"❌ OCO挂单失败: {r2.get('msg')} — 立即平仓!") close = json.dumps({"instId": INST, "tdMode": "cross", "side": "sell" if dirc == 1 else "buy", "ordType": "market", "sz": str(sz)}) okx_req("POST", "/api/v5/trade/close-position", close) return False print(f"✅ 已开仓 {ord_id} + OCO已挂") log_trade({"ts": datetime.now().isoformat(), "mode": "LIVE", "dirc": dirc, "score": score, "reason": reason, "price": price, "sz": sz, "sl": sl_px, "tp": tp_px, "ord_id": ord_id, "result": "OPEN"}) return True # ============ 主循环 ============ def main(): print("=" * 50) print(f"格蕾丝 10U作战bot | {'DRY-RUN 纸交易' if DRY_RUN else 'LIVE 实盘'}") print(f"策略: v1.9多时间框架共振 | 杠杆{LEVERAGE}x | SL{SL_PCT:.1%} TP{TP_PCT:.1%}") print("=" * 50) while True: try: eq, avail = get_balance() if not DRY_RUN else (10.0, 10.0) state = load_state() today = datetime.now().strftime("%Y-%m-%d") if state.get("day") != today: state = {"day": today, "trades": 0, "losses": 0} save_state(state) print(f"\n[{datetime.now().strftime('%H:%M:%S')}] 余额={eq:.2f}U 今日{state['trades']}笔 亏损{state['losses']}次") if state["trades"] >= MAX_TRADES_PER_DAY: print(" 已达今日上限, 停机等明天") time.sleep(600); continue if state["losses"] >= MAX_LOSSES_PER_DAY: print(" 今日亏损2次, 按纪律停机") time.sleep(600); continue if eq < 7: print(" 余额<7U, 触发保护, 停止交易") time.sleep(3600); continue sig = detect_signal() if sig is None: print(" 无信号, 继续观察") time.sleep(CHECK_INTERVAL); continue dirc, score, reason = sig if score < 70: print(f" 信号评分{score} < 70, 不进场 ({reason})") time.sleep(CHECK_INTERVAL); continue px = float(get_candles("1m", 1)["c"].iloc[-1]) if place_trade(dirc, score, reason, px): state["trades"] += 1 save_state(state) time.sleep(3600) # 开仓后1小时不再开新单 except Exception as e: print(f" ⚠️ 异常: {e}") time.sleep(CHECK_INTERVAL) if __name__ == "__main__": main()