import logging import config logger = logging.getLogger(__name__) def check_position_limit(symbol: str, amount_usdt: float, portfolio: dict) -> bool: """Reject if a single position would exceed MAX_POSITION_PCT of total balance.""" total_balance = portfolio.get("total_balance_usdt", 0) if total_balance <= 0: return False current_exposure = portfolio.get("positions", {}).get(symbol, {}).get("value_usdt", 0) new_exposure = current_exposure + amount_usdt limit = total_balance * config.MAX_POSITION_PCT if new_exposure > limit: logger.info( "Position limit: %s would be %.2f USDT (limit %.2f)", symbol, new_exposure, limit, ) return False return True def check_total_exposure(portfolio: dict, additional_usdt: float = 0) -> bool: """Reject if total exposure would exceed MAX_TOTAL_EXPOSURE_PCT.""" total_balance = portfolio.get("total_balance_usdt", 0) if total_balance <= 0: return False current_exposure = sum( pos.get("value_usdt", 0) for pos in portfolio.get("positions", {}).values() ) new_total = current_exposure + additional_usdt limit = total_balance * config.MAX_TOTAL_EXPOSURE_PCT if new_total > limit: logger.info( "Total exposure would be %.2f USDT (limit %.2f)", new_total, limit, ) return False return True def apply_stop_loss_take_profit(positions: dict, current_prices: dict) -> list[dict]: """Check positions for take-profit AND software stop-loss safety net. Returns SELL signals for: - Take-profit hits (pnl >= TAKE_PROFIT_PCT) - Stop-loss safety net (pnl <= -STOP_LOSS_PCT) — second line of defence in case the exchange stop order was cancelled or failed. """ signals = [] for sym, pos in positions.items(): if pos.get("amount", 0) == 0: continue entry = pos.get("entry_price", 0) if entry <= 0: continue current = current_prices.get(sym, 0) if current <= 0: continue pnl_pct = (current - entry) / entry if pnl_pct >= config.TAKE_PROFIT_PCT: signals.append({ "symbol": sym, "action": "SELL", "reason": f"停利 ({pnl_pct:+.2%})", "confidence": 1.0, "suggested_amount_pct": 1.0, }) logger.info("TAKE PROFIT %s: %.6g → %.6g (%+.2f%%)", sym, entry, current, pnl_pct * 100) elif pnl_pct <= -config.STOP_LOSS_PCT: signals.append({ "symbol": sym, "action": "SELL", "reason": f"止損安全網 ({pnl_pct:+.2%})", "confidence": 1.0, "suggested_amount_pct": 1.0, }) logger.warning( "STOP-LOSS SAFETY NET %s: %.6g → %.6g (%+.2f%%) — exchange stop order may have failed", sym, entry, current, pnl_pct * 100, ) return signals def adjust_size_by_volatility(amount_usdt: float, atr: float, price: float) -> float: """Reduce position size when ATR is high relative to price.""" if price <= 0 or atr <= 0: return amount_usdt atr_pct = atr / price # If ATR > 2%, scale down linearly (at 4% ATR → half size) if atr_pct > 0.02: factor = max(0.25, 0.02 / atr_pct) adjusted = amount_usdt * factor logger.info("Volatility adj: %.2f → %.2f USDT (ATR %.2f%%)", amount_usdt, adjusted, atr_pct * 100) return adjusted return amount_usdt def validate_trade(signal: dict, portfolio: dict, indicators_df=None) -> tuple[dict | None, str]: """Run all risk checks. Returns (signal, "") on success or (None, reason) on rejection.""" action = signal.get("action", "HOLD") if action == "HOLD": return None, "HOLD signal" symbol = signal["symbol"] # 基準 = USDT 餘額 + 所有持倉成本 (entry_price × amount) available = portfolio.get("available_usdt", 0) positions_cost = sum( pos.get("entry_price", 0) * pos.get("amount", 0) for pos in portfolio.get("positions", {}).values() ) total_balance = available + positions_cost pct = signal.get("suggested_amount_pct", 0.1) amount_usdt = total_balance * pct if action == "BUY": # Check minimum order if amount_usdt < config.MIN_ORDER_USDT: reason = f"Order too small: {amount_usdt:.2f} < {config.MIN_ORDER_USDT:.2f} USDT" logger.info(reason) return None, reason # Volatility adjustment if indicators_df is not None and not indicators_df.empty: last = indicators_df.iloc[-1] atr = last.get("atr", 0) price = last.get("close", 0) if atr and price: amount_usdt = adjust_size_by_volatility(amount_usdt, atr, price) # Available balance check — cap to available (minus small margin for fees/rounding) available = portfolio.get("available_usdt", 0) - 1 if amount_usdt > available: if available < config.MIN_ORDER_USDT: return None, f"可用餘額不足 ({available:.2f} < {config.MIN_ORDER_USDT:.2f} USDT)" logger.info("金額超出餘額，調整: %.2f → %.2f USDT", amount_usdt, available) amount_usdt = available # Check minimum order amount (exchange-enforced per-symbol minimum) min_amt = config.MIN_ORDER_AMOUNT.get(symbol, 0) if min_amt > 0: # Need price to convert USDT → base currency amount if indicators_df is not None and not indicators_df.empty: est_price = indicators_df.iloc[-1].get("close", 0) else: est_price = 0 if est_price > 0: est_amount = amount_usdt / est_price if est_amount < min_amt: return None, f"低於最低交易量 ({est_amount:.6f} < {min_amt} {symbol})" # Position limit — use same total_balance for consistency current_exposure = portfolio.get("positions", {}).get(symbol, {}).get("value_usdt", 0) limit = total_balance * config.MAX_POSITION_PCT if current_exposure + amount_usdt > limit: room = limit - current_exposure if room < config.MIN_ORDER_USDT: return None, f"持倉已達上限 ({current_exposure:.2f} / {limit:.2f} USDT)" logger.info("持倉上限調整: %.2f → %.2f USDT", amount_usdt, room) amount_usdt = room elif action == "SELL": # For sells, ensure we actually hold the position pos = portfolio.get("positions", {}).get(symbol, {}) if pos.get("amount", 0) <= 0: logger.info("No position to sell for %s", symbol) return None, "無持倉，跳過" # SELL: 一律全部賣出 pos_value = pos.get("value_usdt", 0) if pos_value <= 0: pos_value = pos.get("amount", 0) * pos.get("entry_price", 0) amount_usdt = pos_value # 傳遞實際持倉量，避免 USDT→amount 反算誤差 signal = dict(signal) signal["sell_amount"] = pos["amount"] signal = dict(signal) signal["amount_usdt"] = amount_usdt return signal, ""