dekun
45ae57ed43
Unpack margin_one_lot correctly for fixed-amount sizing, use exchange klines for breakout detection, and notify on pending auto orders. Co-authored-by: Cursor <cursoragent@cursor.com>
271 lines
9.0 KiB
Python
271 lines
9.0 KiB
Python
# 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,
|
|
)
|