# Copyright (c) 2025-2026 马建军. All rights reserved. # 专有软件 — 未经授权禁止复制、传播、转售。 # 严禁用于:带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。 # 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md """期货计仓:固定手数 / 固定金额。""" from __future__ import annotations import math from typing import Optional from contract_specs import get_contract_spec, margin_one_lot MODE_FIXED = "fixed" MODE_AMOUNT = "amount" MODE_RISK = "amount" # 兼容旧配置「以损定仓」 DEFAULT_MAX_ORDER_LOTS = 50 def normalize_sizing_mode(raw: str) -> str: 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: if tick_size <= 0: return 0 s = f"{tick_size:.10f}".rstrip("0").rstrip(".") if "." not in s: return 0 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, trading_mode: str | None = None, ) -> tuple[Optional[int], Optional[str], dict]: """固定金额:先按止损距离算手数,再按保证金上限收紧。返回 (手数, 错误, 详情)。""" info: dict = { "lots_by_risk": 0, "lots_by_margin": None, "capped_by": None, } try: entry_f = float(entry) sl_f = float(stop_loss) budget = float(amount) cap = float(capital or 0) except (TypeError, ValueError): return None, "参数格式错误", info if entry_f <= 0 or budget <= 0: return None, "入场价或固定金额无效", info per_lot_risk, err = _per_lot_risk(entry_f, sl_f, direction, ths_code) if err: return None, err, info lots = int(math.floor(budget / per_lot_risk)) info["lots_by_risk"] = lots if lots < 1: return None, f"按固定金额 {budget:.0f} 元,当前止损距离下不足 1 手", info if cap > 0: margin_per_lot, _src, _spec = margin_one_lot( ths_code, entry_f, direction=direction, trading_mode=trading_mode, ) if margin_per_lot <= 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 ) info["lots_by_margin"] = max_by_margin info["margin_per_lot"] = round(margin_per_lot, 2) info["max_margin_pct"] = margin_cap if max_by_margin < 1: return None, f"按保证金上限 {margin_cap:g}%,当前不足 1 手", info if max_by_margin < lots: info["capped_by"] = "margin" lots = min(lots, max_by_margin) cap_lots = max_lots if max_lots is not None else DEFAULT_MAX_ORDER_LOTS if lots > cap_lots: lots = cap_lots info["capped_by"] = info.get("capped_by") or "max_lots" info["lots"] = lots return lots, None, info def calc_lots_by_risk( entry: float, stop_loss: float, direction: str, capital: float, risk_percent: float, ths_code: str, *, max_lots: Optional[int] = None, max_margin_pct: float = 30.0, trading_mode: str | None = None, ) -> tuple[Optional[int], Optional[str]]: """策略等场景:按权益百分比风险预算换算手数。""" try: cap = float(capital) rp = float(risk_percent) except (TypeError, ValueError): return None, "参数格式错误" if cap <= 0 or rp <= 0: return None, "资金或风险比例无效" budget = cap * rp / 100.0 lots, err, info = calc_lots_by_amount( entry, stop_loss, direction, budget, ths_code, capital=cap, max_lots=max_lots, max_margin_pct=max_margin_pct, trading_mode=trading_mode, ) return lots, err def calc_order_tick_metrics( ths_code: str, lots: float, price: Optional[float] = None, *, direction: str = "long", trading_mode: str | None = None, ) -> dict: """下单区展示:最小变动价位、每跳盈亏、保证金等。""" spec = get_contract_spec(ths_code) mult = int(spec["mult"]) tick = float(spec.get("tick_size") or 1.0) margin_rate = float(spec["margin_rate"]) lots_i = max(1, int(lots or 1)) tick_value_per_lot = round(tick * mult, 4) tick_value_total = round(tick_value_per_lot * lots_i, 2) prec = price_precision_from_tick(tick) mark = float(price) if price else 0.0 margin_per_lot = None margin_source = "estimate" if mark > 0: margin_per_lot, margin_source, spec_used = margin_one_lot( ths_code, mark, direction=direction, trading_mode=trading_mode, ) if spec_used.get("mult"): mult = int(spec_used["mult"]) if spec_used.get("tick_size"): tick = float(spec_used["tick_size"]) tick_value_per_lot = round(tick * mult, 4) tick_value_total = round(tick_value_per_lot * lots_i, 2) prec = price_precision_from_tick(tick) if margin_per_lot <= 0: margin_per_lot = round(mark * mult * margin_rate, 2) margin_source = "estimate" margin_total = round(margin_per_lot * lots_i, 2) if margin_per_lot else None return { "mult": mult, "tick_size": tick, "price_precision": prec, "tick_value_per_lot": tick_value_per_lot, "tick_value_total": tick_value_total, "lots": lots_i, "margin_per_lot": margin_per_lot, "margin_total": margin_total, "margin_rate": margin_rate, "margin_source": margin_source, } def calc_margin_usage_pct( positions: list[dict], capital: float, *, extra_symbol: str = "", extra_lots: int = 0, extra_price: float = 0, extra_direction: str = "long", trading_mode: str | None = None, ) -> float: """当前持仓 + 拟开仓占权益的保证金比例(%)。""" cap = float(capital or 0) if cap <= 0: return 999.0 total = 0.0 for p in positions: lots = int(p.get("lots") or 0) if lots <= 0: continue ctp_margin = float(p.get("margin") or 0) if ctp_margin > 0: total += ctp_margin continue sym = (p.get("symbol") or p.get("symbol_code") or "").strip() entry = float(p.get("avg_price") or p.get("entry_price") or 0) direction = (p.get("direction") or "long").strip().lower() if entry <= 0 or not sym: continue per_lot, _, _ = margin_one_lot( sym, entry, direction=direction, trading_mode=trading_mode, ) if per_lot <= 0: spec = get_contract_spec(sym) per_lot = entry * spec["mult"] * spec["margin_rate"] total += per_lot * lots if extra_symbol and extra_lots > 0 and extra_price > 0: per_lot, _, _ = margin_one_lot( extra_symbol, extra_price, direction=extra_direction, trading_mode=trading_mode, ) if per_lot <= 0: spec = get_contract_spec(extra_symbol) per_lot = extra_price * spec["mult"] * spec["margin_rate"] total += per_lot * extra_lots return round(total / cap * 100.0, 2) def cap_lots_for_margin_budget( positions: list[dict], capital: float, symbol: str, direction: str, price: float, desired_lots: int, max_margin_pct: float, trading_mode: str | None = None, ) -> tuple[int, float]: """在保证金上限内,返回可加仓手数及占用比例。""" desired = max(0, int(desired_lots or 0)) if desired <= 0: return 0, calc_margin_usage_pct(positions, capital, trading_mode=trading_mode) for lots in range(desired, 0, -1): usage = calc_margin_usage_pct( positions, capital, extra_symbol=symbol, extra_lots=lots, extra_price=price, extra_direction=direction, trading_mode=trading_mode, ) if usage <= max_margin_pct: return lots, usage return 0, calc_margin_usage_pct( positions, capital, extra_symbol=symbol, extra_lots=desired, extra_price=price, extra_direction=direction, trading_mode=trading_mode, )