接入 SimNow 模拟盘与期货下单、策略及品种推荐功能。
新增 vnpy CTP 桥接、以损定仓/固定张数、趋势回调与滚仓策略、按资金推荐品种及交易风控;模拟盘走 SimNow,实盘预留期货公司配置。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,94 @@
|
||||
"""期货计仓:固定张数 / 以损定仓(不含币圈全仓杠杆模式)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from typing import Optional
|
||||
|
||||
from contract_specs import get_contract_spec
|
||||
|
||||
MODE_FIXED = "fixed"
|
||||
MODE_RISK = "risk"
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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 calc_lots_by_risk(
|
||||
entry: float,
|
||||
stop_loss: float,
|
||||
direction: str,
|
||||
capital: float,
|
||||
risk_percent: float,
|
||||
ths_code: str,
|
||||
*,
|
||||
max_lots: Optional[int] = None,
|
||||
) -> 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, "止损方向与入场价不匹配"
|
||||
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
|
||||
max_by_margin = int(math.floor(cap * 0.85 / margin_per_lot)) if margin_per_lot > 0 else lots
|
||||
if max_by_margin < 1:
|
||||
return None, "可用资金不足以覆盖 1 手保证金"
|
||||
lots = min(lots, max_by_margin)
|
||||
if max_lots is not None:
|
||||
lots = min(lots, int(max_lots))
|
||||
return lots, None
|
||||
|
||||
|
||||
def calc_order_tick_metrics(ths_code: str, lots: float, price: Optional[float] = 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 = round(mark * mult * margin_rate, 2) if mark > 0 else None
|
||||
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,
|
||||
}
|
||||
Reference in New Issue
Block a user