lulu/arbitrage-engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: P0第二轮修复 — JWT安全/DB密码/SL紧急平仓reduceOnly/TP1状态守卫/超时精度/跨策略去重 + 硬编码消除

Browse Source 
P0-1: JWT_SECRET生产环境强制配置，测试环境保留默认
P0-2: DB密码生产环境强制从env读，测试环境保留fallback
P0-3: SL三次失败→查真实持仓→reduceOnly平仓→校验结果→写event
P0-4: TP1后SL重挂失败则不推进tp1_hit状态，continue等下轮重试
P0-5: 超时自动平仓用SYMBOL_QTY_PRECISION格式化+校验结果
P0-6: 同币种去重改为不区分策略(币安单向模式共享净仓位)
P1-1: 手续费窗口entry_ts-200→+200(避免纳入开仓前成交)
额外: 模拟盘*200和实盘*2硬编码→从配置动态读取
This commit is contained in:
dev-worker 2026-03-02 16:11:43 +00:00
parent 8694e5cf3a
commit 27a51b4d19
5 changed files with 77 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
backend/auth.py
View File

@ -12,7 +12,11 @@ from pydantic import BaseModel, EmailStr
from db import get_sync_conn from db import get_sync_conn
JWT_SECRET = os.getenv("JWT_SECRET", "arb-engine-jwt-secret-v2-2026") _TRADE_ENV = os.getenv("TRADE_ENV", "testnet")
_jwt_default = "arb-engine-jwt-secret-v2-2026" if _TRADE_ENV == "testnet" else None
JWT_SECRET = os.getenv("JWT_SECRET") or _jwt_default
if not JWT_SECRET or len(JWT_SECRET) < 32:
raise RuntimeError("JWT_SECRET 未配置或长度不足(>=32)，生产环境必须设置环境变量")
ACCESS_TOKEN_HOURS = 24 ACCESS_TOKEN_HOURS = 24
REFRESH_TOKEN_DAYS = 7 REFRESH_TOKEN_DAYS = 7

41
backend/live_executor.py
View File

@ -41,12 +41,17 @@ BINANCE_ENDPOINTS = {
BASE_URL = BINANCE_ENDPOINTS[TRADE_ENV] BASE_URL = BINANCE_ENDPOINTS[TRADE_ENV]
# 数据库 # 数据库
_DB_PASSWORD = os.getenv("DB_PASSWORD", "arb_engine_2026" if TRADE_ENV == "testnet" else "")
if not _DB_PASSWORD:
print("FATAL: DB_PASSWORD 未设置（生产环境必须配置）", file=sys.stderr)
sys.exit(1)
DB_CONFIG = { DB_CONFIG = {
"host": os.getenv("DB_HOST", "10.106.0.3"), "host": os.getenv("DB_HOST", "10.106.0.3"),
"port": int(os.getenv("DB_PORT", "5432")), "port": int(os.getenv("DB_PORT", "5432")),
"dbname": os.getenv("DB_NAME", "arb_engine"), "dbname": os.getenv("DB_NAME", "arb_engine"),
"user": os.getenv("DB_USER", "arb"), "user": os.getenv("DB_USER", "arb"),
"password": os.getenv("DB_PASSWORD", "arb_engine_2026"), "password": _DB_PASSWORD,
} }
# 策略 # 策略
@ -354,10 +359,11 @@ async def execute_entry(session: aiohttp.ClientSession, signal: dict, db_conn):
logger.warning(f"[{symbol}] ❌ 已达最大持仓数 {MAX_POSITIONS}") logger.warning(f"[{symbol}] ❌ 已达最大持仓数 {MAX_POSITIONS}")
return None return None
# 3. 检查是否已有同币种同方向持仓 # 3. 检查是否已有同币种持仓（不区分策略，币安单向模式下同币共享净仓位）
cur.execute("SELECT id FROM live_trades WHERE symbol=%s AND strategy=%s AND status='active'", (symbol, strategy)) cur.execute("SELECT id, strategy FROM live_trades WHERE symbol=%s AND status IN ('active', 'tp1_hit')", (symbol,))
if cur.fetchone(): existing = cur.fetchone()
logger.info(f"[{symbol}] ⏭ 已有活跃持仓，跳过") if existing:
logger.info(f"[{symbol}] ⏭ 已有活跃持仓(id={existing[0]}, strategy={existing[1]})，跳过")
return None return None
# 4. 设置杠杆和逐仓 # 4. 设置杠杆和逐仓
@ -419,9 +425,28 @@ async def execute_entry(session: aiohttp.ClientSession, signal: dict, db_conn):
if sl_status == 200: if sl_status == 200:
break break
if sl_status != 200: if sl_status != 200:
logger.error(f"[{symbol}] ❌ SL 3次全部失败，紧急市价平仓! data={sl_data}") logger.error(f"[{symbol}] ❌ SL 3次全部失败，紧急reduceOnly平仓! data={sl_data}")
await place_market_order(session, symbol, close_side, qty) # 查真实持仓量，用reduceOnly市价平仓（避免反向开仓）
_log_event(db_conn, "critical", "trade", f"SL挂单3次失败，已紧急平仓", symbol, {"sl_data": str(sl_data)}) emergency_pos = await get_position(session, symbol)
if emergency_pos:
emergency_amt = abs(float(emergency_pos.get("positionAmt", 0)))
emergency_prec = SYMBOL_PRECISION.get(symbol, {"qty": 3})
emergency_qty_str = f"{emergency_amt:.{emergency_prec['qty']}f}"
close_data, close_status = await binance_request(session, "POST", "/fapi/v1/order", {
"symbol": symbol, "side": close_side, "type": "MARKET",
"quantity": emergency_qty_str, "reduceOnly": "true",
})
if close_status != 200:
logger.error(f"[{symbol}] ❌ 紧急平仓也失败! close_data={close_data}")
_log_event(db_conn, "critical", "trade",
f"SL失败后紧急平仓也失败，需人工介入", symbol,
{"sl_data": str(sl_data), "close_data": str(close_data)})
else:
logger.info(f"[{symbol}] ✅ 紧急reduceOnly平仓成功 qty={emergency_qty_str}")
_log_event(db_conn, "critical", "trade", f"SL挂单3次失败，已紧急reduceOnly平仓", symbol, {"sl_data": str(sl_data)})
else:
logger.warning(f"[{symbol}] SL失败但币安已无持仓，无需平仓")
_log_event(db_conn, "critical", "trade", f"SL挂单3次失败，但币安无持仓", symbol, {"sl_data": str(sl_data)})
return None return None
t_after_sl = time.time() * 1000 t_after_sl = time.time() * 1000
protection_gap_ms = int(t_after_sl - t_fill) protection_gap_ms = int(t_after_sl - t_fill)

10
backend/main.py
View File

@ -683,8 +683,9 @@ async def paper_summary(
total = len(closed) total = len(closed)
wins = len([r for r in closed if r["pnl_r"] > 0]) wins = len([r for r in closed if r["pnl_r"] > 0])
total_pnl = sum(r["pnl_r"] for r in closed) total_pnl = sum(r["pnl_r"] for r in closed)
total_pnl_usdt = total_pnl * 200 # 1R = $200 paper_1r_usd = paper_config["initial_balance"] * paper_config["risk_per_trade"]
balance = 10000 + total_pnl_usdt total_pnl_usdt = total_pnl * paper_1r_usd
balance = paper_config["initial_balance"] + total_pnl_usdt
win_rate = (wins / total * 100) if total > 0 else 0 win_rate = (wins / total * 100) if total > 0 else 0
gross_profit = sum(r["pnl_r"] for r in closed if r["pnl_r"] > 0) gross_profit = sum(r["pnl_r"] for r in closed if r["pnl_r"] > 0)
gross_loss = abs(sum(r["pnl_r"] for r in closed if r["pnl_r"] <= 0)) gross_loss = abs(sum(r["pnl_r"] for r in closed if r["pnl_r"] <= 0))
@ -760,7 +761,8 @@ async def paper_positions(
else: else:
d["unrealized_pnl_r"] = round((entry - current_price) / rd, 2) d["unrealized_pnl_r"] = round((entry - current_price) / rd, 2)
# 浮动盈亏(USDT) — 假设1R = risk_per_trade # 浮动盈亏(USDT) — 假设1R = risk_per_trade
d["unrealized_pnl_usdt"] = round(d["unrealized_pnl_r"] * 200, 2) # 2% of 10000 paper_1r = paper_config["initial_balance"] * paper_config["risk_per_trade"]
d["unrealized_pnl_usdt"] = round(d["unrealized_pnl_r"] * paper_1r, 2)
else: else:
d["unrealized_pnl_r"] = 0 d["unrealized_pnl_r"] = 0
d["unrealized_pnl_usdt"] = 0 d["unrealized_pnl_usdt"] = 0
@ -1278,7 +1280,7 @@ async def live_positions(
d["unrealized_pnl_r"] = round((current_price - entry) / rd, 4) d["unrealized_pnl_r"] = round((current_price - entry) / rd, 4)
else: else:
d["unrealized_pnl_r"] = round((entry - current_price) / rd, 4) d["unrealized_pnl_r"] = round((entry - current_price) / rd, 4)
d["unrealized_pnl_usdt"] = round(d["unrealized_pnl_r"] * 2, 2) d["unrealized_pnl_usdt"] = round(d["unrealized_pnl_r"] * (await _get_risk_usd()), 2)
else: else:
d["unrealized_pnl_r"] = 0 d["unrealized_pnl_r"] = 0
d["unrealized_pnl_usdt"] = 0 d["unrealized_pnl_usdt"] = 0

25
backend/position_sync.py
View File

@ -34,12 +34,17 @@ BINANCE_ENDPOINTS = {
} }
BASE_URL = BINANCE_ENDPOINTS[TRADE_ENV] BASE_URL = BINANCE_ENDPOINTS[TRADE_ENV]
_DB_PASSWORD = os.getenv("DB_PASSWORD", "arb_engine_2026" if TRADE_ENV == "testnet" else "")
if not _DB_PASSWORD:
print("FATAL: DB_PASSWORD 未设置（生产环境必须配置）", file=sys.stderr)
sys.exit(1)
DB_CONFIG = { DB_CONFIG = {
"host": os.getenv("DB_HOST", "10.106.0.3"), "host": os.getenv("DB_HOST", "10.106.0.3"),
"port": int(os.getenv("DB_PORT", "5432")), "port": int(os.getenv("DB_PORT", "5432")),
"dbname": os.getenv("DB_NAME", "arb_engine"), "dbname": os.getenv("DB_NAME", "arb_engine"),
"user": os.getenv("DB_USER", "arb"), "user": os.getenv("DB_USER", "arb"),
"password": os.getenv("DB_PASSWORD", "arb_engine_2026"), "password": _DB_PASSWORD,
} }
CHECK_INTERVAL = 30 # 对账间隔(秒) CHECK_INTERVAL = 30 # 对账间隔(秒)
@ -368,20 +373,28 @@ async def check_tp1_triggers(session, conn):
else: else:
new_sl = lp["entry_price"] * 0.9995 new_sl = lp["entry_price"] * 0.9995
# 挂新SL（半仓） # 挂新SL（半仓）— 失败则不推进状态
prec = SYMBOL_PRECISION.get(symbol, {"qty": 3, "price": 2}) prec = SYMBOL_PRECISION.get(symbol, {"qty": 3, "price": 2})
ok, _ = await rehang_sl(session, symbol, lp["direction"], new_sl, bp["amount"]) ok, sl_resp = await rehang_sl(session, symbol, lp["direction"], new_sl, bp["amount"])
if not ok:
logger.error(f"[{symbol}] ❌ TP1后重挂SL失败: {sl_resp}，不推进tp1_hit状态")
_log_event(conn, "critical", "trade",
"TP1后重挂SL失败，仓位可能裸奔，需人工确认", symbol,
{"trade_id": lp["id"], "sl_resp": str(sl_resp)})
continue
# 重新挂TP2（半仓） # 重新挂TP2（半仓）
tp2_price = lp["tp2_price"] tp2_price = lp["tp2_price"]
qty_str = f"{abs(bp['amount']):.{prec['qty']}f}" qty_str = f"{abs(bp['amount']):.{prec['qty']}f}"
price_str = f"{tp2_price:.{prec['price']}f}" price_str = f"{tp2_price:.{prec['price']}f}"
await binance_request(session, "POST", "/fapi/v1/order", { tp2_data, tp2_status = await binance_request(session, "POST", "/fapi/v1/order", {
"symbol": symbol, "side": close_side, "type": "TAKE_PROFIT_MARKET", "symbol": symbol, "side": close_side, "type": "TAKE_PROFIT_MARKET",
"stopPrice": price_str, "quantity": qty_str, "reduceOnly": "true", "stopPrice": price_str, "quantity": qty_str, "reduceOnly": "true",
}) })
if tp2_status != 200:
logger.error(f"[{symbol}] ❌ TP2重挂失败: {tp2_data}，SL已挂但TP2缺失")
# 更新DB # SL成功才更新DB
cur.execute(""" cur.execute("""
UPDATE live_trades SET tp1_hit=TRUE, sl_price=%s, status='tp1_hit' UPDATE live_trades SET tp1_hit=TRUE, sl_price=%s, status='tp1_hit'
WHERE id=%s WHERE id=%s
@ -436,7 +449,7 @@ async def check_closed_positions(session, conn):
# 汇总手续费（开仓后200ms起算，避免含其他策略成交） # 汇总手续费（开仓后200ms起算，避免含其他策略成交）
for t in trades_data: for t in trades_data:
t_time = int(t.get("time", 0)) t_time = int(t.get("time", 0))
if t_time >= entry_ts - 200: if t_time >= entry_ts + 200: # 开仓后200ms起算，避免纳入开仓前成交
actual_fee_usdt += abs(float(t.get("commission", 0))) actual_fee_usdt += abs(float(t.get("commission", 0)))
# 计算pnl — gross(不含费) # 计算pnl — gross(不含费)

18
backend/risk_guard.py
View File

@ -39,12 +39,17 @@ BINANCE_ENDPOINTS = {
} }
BASE_URL = BINANCE_ENDPOINTS[TRADE_ENV] BASE_URL = BINANCE_ENDPOINTS[TRADE_ENV]
_DB_PASSWORD = os.getenv("DB_PASSWORD", "arb_engine_2026" if TRADE_ENV == "testnet" else "")
if not _DB_PASSWORD:
print("FATAL: DB_PASSWORD 未设置（生产环境必须配置）", file=sys.stderr)
sys.exit(1)
DB_CONFIG = { DB_CONFIG = {
"host": os.getenv("DB_HOST", "10.106.0.3"), "host": os.getenv("DB_HOST", "10.106.0.3"),
"port": int(os.getenv("DB_PORT", "5432")), "port": int(os.getenv("DB_PORT", "5432")),
"dbname": os.getenv("DB_NAME", "arb_engine"), "dbname": os.getenv("DB_NAME", "arb_engine"),
"user": os.getenv("DB_USER", "arb"), "user": os.getenv("DB_USER", "arb"),
"password": os.getenv("DB_PASSWORD", "arb_engine_2026"), "password": _DB_PASSWORD,
} }
# 风控参数 # 风控参数
@ -279,11 +284,16 @@ async def check_hold_timeout(session, conn):
for p in pos_data: for p in pos_data:
amt = abs(float(p.get("positionAmt", 0))) amt = abs(float(p.get("positionAmt", 0)))
if amt > 0 and p["symbol"] == symbol: if amt > 0 and p["symbol"] == symbol:
await binance_request(session, "POST", "/fapi/v1/order", { qty_prec = SYMBOL_QTY_PRECISION.get(symbol, 3)
qty_str = f"{amt:.{qty_prec}f}"
close_data, close_status = await binance_request(session, "POST", "/fapi/v1/order", {
"symbol": symbol, "side": close_side, "type": "MARKET", "symbol": symbol, "side": close_side, "type": "MARKET",
"quantity": str(amt), "reduceOnly": "true", "quantity": qty_str, "reduceOnly": "true",
}) })
logger.info(f"[{symbol}] 🔴 自动平仓完成 qty={amt}") if close_status != 200:
logger.error(f"[{symbol}] ❌ 超时自动平仓失败: {close_data}")
else:
logger.info(f"[{symbol}] 🔴 超时自动平仓完成 qty={qty_str}")
del risk_state.timeout_queue[trade_id] del risk_state.timeout_queue[trade_id]
elif hold_min >= HOLD_TIMEOUT_RED_MIN: elif hold_min >= HOLD_TIMEOUT_RED_MIN: