fix: V3全面审阅修复 — 12项问题

P0-1: 风控Fail-Closed(状态文件缺失/过期/异常→拒绝开仓)
P0-2: 1R基准跨模块统一(position_sync+risk_guard从live_config动态读)
P0-3: close_all紧急全平校验返回值+二次验仓
P0-4: Coinbase Premium单位修复(premium_pct/100→比例值)
P1-3: 正向funding计入净PnL(不再只扣负值)
P1-4: 数据新鲜度检查落地(查signal_indicators最新ts)
P1-6: live表DDL补全到SCHEMA_SQL(live_config/live_events/live_trades)
P2-1: _get_risk_usd()加60秒缓存
P2-3: 模拟盘前端*200→从config动态算paper1R
P2-4: XRP/SOL跳过Coinbase Premium采集(无数据源)
P3-2: SQL参数化(fetch_pending_signals用ANY替代f-string)
额外: pnl_r公式修正(gross-fee+funding，funding正负都正确计入)

backend/db.py

@@ -303,6 +303,57 @@ CREATE TABLE IF NOT EXISTS paper_trades (
     score_factors JSONB,
     created_at TIMESTAMP DEFAULT NOW()
 );
+
+-- Live trading tables
+CREATE TABLE IF NOT EXISTS live_config (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL,
+    label TEXT,
+    updated_at TIMESTAMP DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS live_events (
+    id BIGSERIAL PRIMARY KEY,
+    ts BIGINT DEFAULT (EXTRACT(EPOCH FROM NOW()) * 1000)::BIGINT,
+    level TEXT,
+    category TEXT,
+    symbol TEXT,
+    message TEXT,
+    detail JSONB
+);
+
+CREATE TABLE IF NOT EXISTS live_trades (
+    id BIGSERIAL PRIMARY KEY,
+    symbol TEXT NOT NULL,
+    strategy TEXT NOT NULL,
+    direction TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'active',
+    entry_price DOUBLE PRECISION,
+    exit_price DOUBLE PRECISION,
+    entry_ts BIGINT,
+    exit_ts BIGINT,
+    sl_price DOUBLE PRECISION,
+    tp1_price DOUBLE PRECISION,
+    tp2_price DOUBLE PRECISION,
+    tp1_hit BOOLEAN DEFAULT FALSE,
+    score DOUBLE PRECISION,
+    tier TEXT,
+    pnl_r DOUBLE PRECISION,
+    fee_usdt DOUBLE PRECISION DEFAULT 0,
+    funding_fee_usdt DOUBLE PRECISION DEFAULT 0,
+    risk_distance DOUBLE PRECISION,
+    atr_at_entry DOUBLE PRECISION,
+    score_factors JSONB,
+    signal_id BIGINT,
+    binance_order_id TEXT,
+    fill_price DOUBLE PRECISION,
+    slippage_bps DOUBLE PRECISION,
+    protection_gap_ms BIGINT,
+    signal_to_order_ms BIGINT,
+    order_to_fill_ms BIGINT,
+    qty DOUBLE PRECISION,
+    created_at TIMESTAMP DEFAULT NOW()
+);
 """
 
 

backend/live_executor.py

@@ -317,9 +317,15 @@ async def execute_entry(session: aiohttp.ClientSession, signal: dict, db_conn):
                    symbol, {"signal_age_ms": round(signal_age_ms), "score": score})
         return None
 
-    # 0.5 检查风控状态（读risk_guard写的状态文件）
+    # 0.5 检查风控状态（Fail-Closed: 状态异常时拒绝开仓）
+    state_path = "/tmp/risk_guard_state.json"
     try:
-        with open("/tmp/risk_guard_state.json") as f:
+        st = os.stat(state_path)
+        if time.time() - st.st_mtime > 15:
+            logger.error(f"[{symbol}] ❌ 风控状态文件超过15秒未更新，risk_guard可能失联，拒绝开仓")
+            _log_event(db_conn, "critical", "risk", "风控失联(状态文件过期)，拒绝开仓", symbol)
+            return None
+        with open(state_path) as f:
             risk_state = json.load(f)
         if risk_state.get("block_new_entries"):
             logger.warning(f"[{symbol}] ❌ 风控禁止新开仓: {risk_state.get('circuit_break_reason', '未知原因')}")
@@ -328,9 +334,11 @@ async def execute_entry(session: aiohttp.ClientSession, signal: dict, db_conn):
             logger.warning(f"[{symbol}] ❌ 只减仓模式，拒绝开仓")
             return None
     except FileNotFoundError:
-        pass  # risk_guard还没启动，允许交易
+        logger.error(f"[{symbol}] ❌ 风控状态文件不存在，risk_guard未运行，拒绝开仓")
+        return None
     except Exception as e:
-        logger.warning(f"[{symbol}] ⚠ 读取风控状态失败: {e}，继续交易")
+        logger.error(f"[{symbol}] ❌ 读取风控状态失败: {e}，拒绝开仓")
+        return None
 
     # 检查前端紧急操作指令
     try:
@@ -525,21 +533,20 @@ def fetch_pending_signals(conn):
     """查询未处理的新信号"""
     cur = conn.cursor()
     # 查最近60秒内的新信号（score>=阈值、策略匹配、未被live_executor处理过）
-    strategies_str = ",".join(f"'{s}'" for s in ENABLED_STRATEGIES)
+    cur.execute("""
-    cur.execute(f"""
         SELECT si.id, si.symbol, si.signal, si.score, si.ts, si.factors, si.strategy,
                si.price
         FROM signal_indicators si
         WHERE si.signal IS NOT NULL
           AND si.signal != ''
-          AND si.strategy IN ({strategies_str})
+          AND si.strategy = ANY(%s)
           AND si.ts > extract(epoch from now()) * 1000 - 60000
           AND NOT EXISTS (
               SELECT 1 FROM live_trades lt
               WHERE lt.signal_id = si.id AND lt.strategy = si.strategy
           )
         ORDER BY si.ts DESC
-    """)
+    """, (ENABLED_STRATEGIES,))
     rows = cur.fetchall()
     signals = []
     for row in rows:

backend/main.py

@@ -1473,13 +1473,20 @@ async def live_risk_status(user: dict = Depends(get_current_user)):
         return {"status": "unknown", "error": "risk_guard_state.json not found"}
 
 
+_risk_usd_cache = {"v": 2.0, "ts": 0.0}
+
 async def _get_risk_usd() -> float:
     """从live_config读取1R金额，缓存60秒"""
+    now = time.time()
+    if now - _risk_usd_cache["ts"] < 60:
+        return _risk_usd_cache["v"]
     try:
         row = await async_fetchrow("SELECT value FROM live_config WHERE key = $1", "risk_per_trade_usd")
-        return float(row["value"]) if row else 2.0
+        v = float(row["value"]) if row else 2.0
     except Exception:
-        return 2.0
+        v = 2.0
+    _risk_usd_cache.update({"v": v, "ts": now})
+    return v
 
 
 def _require_admin(user: dict):

backend/market_data_collector.py

@@ -113,7 +113,9 @@ class MarketDataCollector:
             "BTCUSDT": "BTC-USD",
             "ETHUSDT": "ETH-USD",
         }
-        coinbase_pair = pair_map[symbol]
+        coinbase_pair = pair_map.get(symbol)
+        if not coinbase_pair:
+            return  # XRP/SOL无Coinbase数据，跳过
 
         binance_url = "https://api.binance.com/api/v3/ticker/price"
         coinbase_url = f"https://api.coinbase.com/v2/prices/{coinbase_pair}/spot"

backend/position_sync.py

@@ -444,8 +444,8 @@ async def check_closed_positions(session, conn):
                     if total_qty > 0:
                         exit_price = sum(float(t["price"]) * float(t["qty"]) for t in close_trades) / total_qty
                 elif trades_data:
-                    # fallback: 不盲目取最后一条（可能是开仓成交），延后本轮结算
+                                        # fallback: 未找到明确平仓成交，延后本轮结算
-                    logger.warning(f"[{symbol}] 未找到明确平仓成交，延后结算")
+                    logger.warning(f"[{symbol}] 未找到明确平仓成交，延后结算")
                     continue
 
                 # 汇总手续费（开仓后200ms起算，避免含其他策略成交）
@@ -464,9 +464,10 @@ async def check_closed_positions(session, conn):
                 tp1_r = abs(lp["tp1_price"] - lp["entry_price"]) / rd
                 gross_pnl_r = 0.5 * tp1_r + 0.5 * gross_pnl_r
 
-            # 手续费(R) — 用实际成交手续费
+            # 手续费(R) — 用实际成交手续费（动态1R）
+            _risk_usd = load_live_risk_usd(conn)
             if actual_fee_usdt > 0:
-                fee_r = actual_fee_usdt / (RISK_PER_TRADE_USD if RISK_PER_TRADE_USD > 0 else rd)
+                fee_r = actual_fee_usdt / (_risk_usd if _risk_usd > 0 else rd)
             else:
                 # fallback: 按0.1%估算(开+平各0.05%)
                 fee_r = 0.001 * lp["entry_price"] / rd
@@ -477,10 +478,11 @@ async def check_closed_positions(session, conn):
             fr_row = cur.fetchone()
             if fr_row:
                 funding_usdt = fr_row[0]
-            funding_r = abs(funding_usdt) / (RISK_PER_TRADE_USD if RISK_PER_TRADE_USD > 0 else rd) if funding_usdt < 0 else 0
+            # P1-3: 正向funding也计入净PnL（正值增加收益，负值减少收益）
+            funding_r = funding_usdt / (_risk_usd if _risk_usd > 0 else rd)
 
-            # 净PnL = gross - fee - funding_cost
+            # 净PnL = gross - fee + funding (funding正值=收入，负值=支出)
-            pnl_r = gross_pnl_r - fee_r - funding_r
+            pnl_r = gross_pnl_r - fee_r + funding_r
 
             # 判断状态
             if pnl_r > 0.5:
@@ -619,6 +621,17 @@ def _log_event(conn, level, category, message, symbol=None, detail=None):
 
 # ============ 主循环 ============
 
+def load_live_risk_usd(conn, default=2.0):
+    """从live_config动态读取1R金额，与live_executor保持一致"""
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT value FROM live_config WHERE key='risk_per_trade_usd'")
+        row = cur.fetchone()
+        return float(row[0]) if row and row[0] else default
+    except Exception:
+        return default
+
+
 def ensure_db_conn(conn):
     """检查DB连接，断线则重连"""
     try:

backend/risk_guard.py

@@ -220,7 +220,7 @@ def check_daily_loss(conn):
     return realized_r, consecutive
 
 
-async def check_unrealized_loss(session):
+async def check_unrealized_loss(session, risk_usd_dynamic=2.0):
     """检查未实现亏损"""
     data, status = await binance_request(session, "GET", "/fapi/v2/positionRisk")
     total_unrealized = 0
@@ -229,7 +229,7 @@ async def check_unrealized_loss(session):
             pnl = float(pos.get("unRealizedProfit", 0))
             total_unrealized += pnl
     # 转为R
-    unrealized_r = total_unrealized / RISK_PER_TRADE_USD if RISK_PER_TRADE_USD > 0 else 0
+    unrealized_r = total_unrealized / risk_usd_dynamic if risk_usd_dynamic > 0 else 0
     risk_state.today_unrealized_r = unrealized_r
     return unrealized_r
 
@@ -245,7 +245,7 @@ async def check_balance(session):
     return 0
 
 
-def check_data_freshness():
+def check_data_freshness(conn=None):
     """检查数据新鲜度"""
     now = time.time()
     issues = []
@@ -254,6 +254,19 @@ def check_data_freshness():
     if api_gap > API_DISCONNECT_THRESHOLD_SEC:
         issues.append(f"API无响应{api_gap:.0f}秒")
 
+    # 检查行情数据新鲜度（signal_indicators最新ts）
+    if conn:
+        try:
+            cur = conn.cursor()
+            cur.execute("SELECT MAX(ts) FROM signal_indicators")
+            row = cur.fetchone()
+            if row and row[0]:
+                market_age = now - (row[0] / 1000)
+                if market_age > MARKET_DATA_STALE_SEC:
+                    issues.append(f"行情数据延迟{market_age:.1f}秒")
+        except Exception:
+            pass
+
     return issues
 
 
@@ -350,11 +363,23 @@ async def trigger_circuit_break(session, conn, reason: str, action: str = "block
                         close_side = "SELL" if amt > 0 else "BUY"
                         qty_prec = SYMBOL_QTY_PRECISION.get(symbol, 3)
                         qty_str = f"{abs(amt):.{qty_prec}f}"
-                        await binance_request(session, "POST", "/fapi/v1/order", {
+                        close_resp, close_status = await binance_request(session, "POST", "/fapi/v1/order", {
                             "symbol": symbol, "side": close_side, "type": "MARKET",
                             "quantity": qty_str, "reduceOnly": "true",
                         })
+                        if close_status != 200:
+                            logger.error(f"[{symbol}] ❌ 紧急平仓失败: {close_resp}")
+                            _log_event(conn, "critical", "risk", f"紧急平仓失败 {symbol}", symbol,
+                                       {"response": str(close_resp)})
+                        else:
                             logger.info(f"[{symbol}] 🔴 紧急平仓 {close_side} qty={qty_str}")
+                            # 二次验仓
+                            verify, v_status = await binance_request(session, "GET", "/fapi/v2/positionRisk", {"symbol": symbol})
+                            if v_status == 200 and isinstance(verify, list):
+                                still_open = any(abs(float(p.get("positionAmt", 0))) > 0 and p.get("symbol") == symbol for p in verify)
+                                if still_open:
+                                    logger.error(f"[{symbol}] ❌ 紧急平仓后仍有仓位! 需人工介入")
+                                    _log_event(conn, "critical", "risk", f"紧急平仓后仍有仓位 {symbol}", symbol)
 
     # 记录历史
     risk_state.breaker_history.append({
@@ -478,6 +503,17 @@ async def check_emergency_commands(session, conn):
 
 # ============ 主循环 ============
 
+def load_live_risk_usd(conn, default=2.0):
+    """从live_config动态读取1R金额，与live_executor保持一致"""
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT value FROM live_config WHERE key='risk_per_trade_usd'")
+        row = cur.fetchone()
+        return float(row[0]) if row and row[0] else default
+    except Exception:
+        return default
+
+
 def ensure_db_conn(conn):
     """检查DB连接，断线则重连"""
     try:
@@ -517,7 +553,8 @@ async def main():
 
                 # 1. 今日亏损检查
                 realized_r, consecutive = check_daily_loss(conn)
-                unrealized_r = await check_unrealized_loss(session)
+                _live_risk_usd = load_live_risk_usd(conn)
+                unrealized_r = await check_unrealized_loss(session, _live_risk_usd)
                 total_r = realized_r + min(unrealized_r, 0)  # 已实现 + 未实现亏损
 
                 if total_r <= DAILY_LOSS_LIMIT_R and risk_state.status != "circuit_break":
@@ -538,7 +575,7 @@ async def main():
                     risk_state.auto_resume_time = time.time() + CONSECUTIVE_LOSS_COOLDOWN_MIN * 60
 
                 # 3. API连接检查
-                freshness_issues = check_data_freshness()
+                freshness_issues = check_data_freshness(conn)
                 if freshness_issues and risk_state.status != "circuit_break":
                     await trigger_circuit_break(
                         session, conn,

backend/signal_engine.py

@@ -151,7 +151,8 @@ def fetch_market_indicators(symbol: str) -> dict:
                 elif ind_type == "open_interest_hist":
                     indicators[ind_type] = float(val.get("sumOpenInterestValue", 0))
                 elif ind_type == "coinbase_premium":
-                    indicators[ind_type] = float(val.get("premium_pct", 0))
+                    # premium_pct存的是百分比(如0.05=0.05%)，转成比例(0.0005)
+                    indicators[ind_type] = float(val.get("premium_pct", 0)) / 100.0
                 elif ind_type == "funding_rate":
                     indicators[ind_type] = float(val.get("fundingRate", val.get("lastFundingRate", 0)))
             return indicators

frontend/app/paper-v52/page.tsx

@@ -294,7 +294,8 @@ function ActivePositions({ strategy }: { strategy: StrategyFilter }) {
           const fullR = riskDist > 0 ? (p.direction === "LONG" ? (currentPrice - entry) / riskDist : (entry - currentPrice) / riskDist) : 0;
           const tp1R = riskDist > 0 ? (p.direction === "LONG" ? ((p.tp1_price || 0) - entry) / riskDist : (entry - (p.tp1_price || 0)) / riskDist) : 0;
           const unrealR = p.tp1_hit ? 0.5 * tp1R + 0.5 * fullR : fullR;
-          const unrealUsdt = unrealR * 200;
+          const paper1R = (config?.initial_balance || 10000) * (config?.risk_per_trade || 0.02);
+          const unrealUsdt = unrealR * paper1R;
           return (
             <div key={p.id} className="px-3 py-2 bg-emerald-50/60">
               <div className="flex items-center justify-between gap-2">

frontend/app/paper/page.tsx

@@ -231,7 +231,8 @@ function ActivePositions() {
           const fullR = riskDist > 0 ? (p.direction === "LONG" ? (currentPrice - entry) / riskDist : (entry - currentPrice) / riskDist) : 0;
           const tp1R = riskDist > 0 ? (p.direction === "LONG" ? ((p.tp1_price || 0) - entry) / riskDist : (entry - (p.tp1_price || 0)) / riskDist) : 0;
           const unrealR = p.tp1_hit ? 0.5 * tp1R + 0.5 * fullR : fullR;
-          const unrealUsdt = unrealR * 200;
+          const paper1R = (config?.initial_balance || 10000) * (config?.risk_per_trade || 0.02);
+          const unrealUsdt = unrealR * paper1R;
           return (
             <div key={p.id} className="px-3 py-2">
               <div className="flex items-center justify-between">