"""成交价与手续费: 滑点 = 1×fee_rate.""" from __future__ import annotations import os from dataclasses import asdict, dataclass @dataclass(slots=True) class PriceResult: base_px: float fill_px: float fee: float slip: float notional: float def to_dict(self) -> dict[str, float]: return asdict(self) def sim_fee_rate(override: float | None = None) -> float: if override is not None: return float(override) try: return float(os.getenv("SIM_FEE_RATE") or "0.0005") except (TypeError, ValueError): return 0.0005 def perp_fill( *, side: str, action: str, bid: float, ask: float, qty: float, fee_rate: float, ) -> PriceResult: """ side: long|short action: open|close 开多/平空: 吃卖一 ×(1+f) 开空/平多: 吃买一 ×(1-f) qty: 标的数量(合约张数 × 合约面值) """ f = float(fee_rate) buying = (action == "open" and side == "long") or (action == "close" and side == "short") if buying: base = float(ask) fill = base * (1.0 + f) else: base = float(bid) fill = base * (1.0 - f) notional = abs(fill * float(qty)) fee = notional * f slip = abs(fill - base) * float(qty) return PriceResult(base_px=base, fill_px=fill, fee=fee, slip=slip, notional=notional) def option_fill( *, action: str, bid: float, ask: float, qty: float, fee_rate: float, ) -> PriceResult: """开仓买入吃卖一; 平仓卖出吃买一. qty = sheets × ct_mult.""" f = float(fee_rate) if action == "open": base = float(ask) fill = base * (1.0 + f) else: base = float(bid) fill = base * (1.0 - f) notional = abs(fill * float(qty)) fee = notional * f slip = abs(fill - base) * float(qty) return PriceResult(base_px=base, fill_px=fill, fee=fee, slip=slip, notional=notional) @dataclass(slots=True) class SpotConvertResult: """USDC/USDT 现货兑换: price = USDT per USDC.""" direction: str from_ccy: str to_ccy: str from_amount: float to_amount: float base_px: float fill_px: float fee: float def spot_usdc_usdt_fill( *, direction: str, amount: float, bid: float, ask: float, fee_rate: float, ) -> SpotConvertResult: """ 对齐实盘 USDC-USDT 现货市价: - usdt_to_usdc: 用 USDT 买 USDC, 吃卖一 ×(1+f) - usdc_to_usdt: 卖 USDC 换 USDT, 吃买一 ×(1-f) amount 为付出币种数量. """ f = float(fee_rate) amt = float(amount) d = (direction or "").strip().lower() if d == "usdt_to_usdc": base = float(ask) fill = base * (1.0 + f) if fill <= 0: raise ValueError("无效卖一价") to_amt = amt / fill fee = amt * f return SpotConvertResult( direction=d, from_ccy="USDT", to_ccy="USDC", from_amount=amt, to_amount=to_amt, base_px=base, fill_px=fill, fee=fee, ) if d == "usdc_to_usdt": base = float(bid) fill = base * (1.0 - f) if fill <= 0: raise ValueError("无效买一价") to_amt = amt * fill fee = to_amt * f return SpotConvertResult( direction=d, from_ccy="USDC", to_ccy="USDT", from_amount=amt, to_amount=to_amt, base_px=base, fill_px=fill, fee=fee, ) raise ValueError("direction 须为 usdt_to_usdc 或 usdc_to_usdt") def spot_coin_usdt_fill( *, direction: str, amount: float, bid: float, ask: float, fee_rate: float, coin: str = "ETH", ) -> SpotConvertResult: """ 对齐实盘 ETH-USDT / BTC-USDT 现货市价: - usdt_to_coin: 用 USDT 买币, 吃卖一 ×(1+f); amount=USDT - coin_to_usdt: 卖币换 USDT, 吃买一 ×(1-f); amount=币数量 """ f = float(fee_rate) amt = float(amount) ccy = (coin or "ETH").strip().upper() or "ETH" d = (direction or "").strip().lower() if d in ("usdt_to_coin", "usdt_to_eth", "usdt_to_btc"): base = float(ask) fill = base * (1.0 + f) if fill <= 0: raise ValueError("无效卖一价") to_amt = amt / fill fee = amt * f return SpotConvertResult( direction="usdt_to_coin", from_ccy="USDT", to_ccy=ccy, from_amount=amt, to_amount=to_amt, base_px=base, fill_px=fill, fee=fee, ) if d in ("coin_to_usdt", "eth_to_usdt", "btc_to_usdt"): base = float(bid) fill = base * (1.0 - f) if fill <= 0: raise ValueError("无效买一价") to_amt = amt * fill fee = to_amt * f return SpotConvertResult( direction="coin_to_usdt", from_ccy=ccy, to_ccy="USDT", from_amount=amt, to_amount=to_amt, base_px=base, fill_px=fill, fee=fee, ) raise ValueError("direction 须为 usdt_to_coin 或 coin_to_usdt")