feat: 计仓改为固定手数/固定金额,推荐过滤与CTP保证金,下单与持仓UI优化

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-25 15:31:34 +08:00
parent c302e1f3ca
commit 9772f3d986
11 changed files with 387 additions and 119 deletions
+70 -32
View File
@@ -1,4 +1,4 @@
"""期货计仓:固定数 / 以损定仓(不含币圈全仓杠杆模式)"""
"""期货计仓:固定数 / 固定金额"""
from __future__ import annotations
import math
@@ -7,15 +7,17 @@ from typing import Optional
from contract_specs import get_contract_spec
MODE_FIXED = "fixed"
MODE_RISK = "risk"
MODE_AMOUNT = "amount"
MODE_RISK = "amount" # 兼容旧配置「以损定仓」
# 单笔报单手数上限(防止以损定仓在止损过近时算出超大手数)
DEFAULT_MAX_ORDER_LOTS = 50
def normalize_sizing_mode(raw: str) -> str:
m = (raw or MODE_RISK).strip().lower()
return m if m in (MODE_FIXED, MODE_RISK) else MODE_RISK
m = (raw or MODE_FIXED).strip().lower()
if m == "risk":
m = MODE_AMOUNT
return m if m in (MODE_FIXED, MODE_AMOUNT) else MODE_FIXED
def price_precision_from_tick(tick_size: float) -> int:
@@ -27,6 +29,62 @@ def price_precision_from_tick(tick_size: float) -> int:
return len(s.split(".")[1])
def _per_lot_risk(entry: float, stop_loss: float, direction: str, ths_code: str) -> tuple[float, Optional[str]]:
spec = get_contract_spec(ths_code)
mult = spec["mult"]
d = (direction or "long").strip().lower()
if d == "short":
per_lot = (stop_loss - entry) * mult
else:
per_lot = (entry - stop_loss) * mult
if per_lot <= 0:
return 0.0, "止损方向与入场价不匹配"
return per_lot, None
def calc_lots_by_amount(
entry: float,
stop_loss: float,
direction: str,
amount: float,
ths_code: str,
*,
capital: float = 0.0,
max_lots: Optional[int] = None,
max_margin_pct: float = 30.0,
) -> tuple[Optional[int], Optional[str]]:
"""固定金额:按止损距离将金额换算为手数。"""
try:
entry_f = float(entry)
sl_f = float(stop_loss)
budget = float(amount)
cap = float(capital or 0)
except (TypeError, ValueError):
return None, "参数格式错误"
if entry_f <= 0 or budget <= 0:
return None, "入场价或固定金额无效"
per_lot_risk, err = _per_lot_risk(entry_f, sl_f, direction, ths_code)
if err:
return None, err
lots = int(math.floor(budget / per_lot_risk))
if lots < 1:
return None, f"按固定金额 {budget:.0f} 元,当前止损距离下不足 1 手"
if cap > 0:
spec = get_contract_spec(ths_code)
margin_per_lot = entry_f * spec["mult"] * spec["margin_rate"]
margin_cap = max(1.0, min(100.0, float(max_margin_pct or 30.0)))
max_by_margin = (
int(math.floor(cap * margin_cap / 100.0 / margin_per_lot))
if margin_per_lot > 0 else lots
)
if max_by_margin < 1:
return None, f"按保证金上限 {margin_cap:g}%,当前不足 1 手"
lots = min(lots, max_by_margin)
cap_lots = max_lots if max_lots is not None else DEFAULT_MAX_ORDER_LOTS
lots = min(lots, cap_lots)
return lots, None
def calc_lots_by_risk(
entry: float,
stop_loss: float,
@@ -38,39 +96,19 @@ def calc_lots_by_risk(
max_lots: Optional[int] = None,
max_margin_pct: float = 30.0,
) -> tuple[Optional[int], Optional[str]]:
"""以损定仓:返回 (手数, 错误信息)"""
"""策略等场景:按权益百分比风险预算换算手数"""
try:
entry_f = float(entry)
sl_f = float(stop_loss)
cap = float(capital)
rp = float(risk_percent)
except (TypeError, ValueError):
return None, "参数格式错误"
if entry_f <= 0 or cap <= 0 or rp <= 0:
return None, "入场价、资金或风险比例无效"
spec = get_contract_spec(ths_code)
mult = spec["mult"]
d = (direction or "long").strip().lower()
if d == "short":
per_lot_risk = (sl_f - entry_f) * mult
else:
per_lot_risk = (entry_f - sl_f) * mult
if per_lot_risk <= 0:
return None, "止损方向与入场价不匹配"
if cap <= 0 or rp <= 0:
return None, "资金或风险比例无效"
budget = cap * rp / 100.0
lots = int(math.floor(budget / per_lot_risk))
if lots < 1:
return None, f"{rp}% 风险预算,当前止损距离下不足 1 手"
margin_rate = spec["margin_rate"]
margin_per_lot = entry_f * mult * margin_rate
margin_cap = max(1.0, min(100.0, float(max_margin_pct or 30.0)))
max_by_margin = int(math.floor(cap * margin_cap / 100.0 / margin_per_lot)) if margin_per_lot > 0 else lots
if max_by_margin < 1:
return None, f"按保证金上限 {margin_cap:g}%,当前不足 1 手"
lots = min(lots, max_by_margin)
cap_lots = max_lots if max_lots is not None else DEFAULT_MAX_ORDER_LOTS
lots = min(lots, cap_lots)
return lots, None
return calc_lots_by_amount(
entry, stop_loss, direction, budget, ths_code,
capital=cap, max_lots=max_lots, max_margin_pct=max_margin_pct,
)
def calc_order_tick_metrics(ths_code: str, lots: float, price: Optional[float] = None) -> dict: