dekun
361 lines
12 KiB
Python
361 lines
12 KiB
Python
"""中控历史测算：趋势回调 / 滚仓，以损定仓（无交易所精度，张数按公式估算）。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Callable, Optional, Tuple
|
|
|
|
from strategy_roll_lib import max_roll_legs, preview_roll
|
|
from strategy_trend_lib import (
|
|
build_trend_preview_level_rows,
|
|
calc_risk_fraction,
|
|
compute_trend_plan_core,
|
|
validate_trend_bounds,
|
|
)
|
|
|
|
DEFAULT_DCA_LEGS = 5
|
|
DEFAULT_CONTRACT_SIZE = 1.0
|
|
MARGIN_BUFFER = 0.95
|
|
|
|
|
|
def _identity_amount_precise(_symbol: str, amount: float) -> Optional[float]:
|
|
try:
|
|
v = float(amount)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if v <= 0:
|
|
return None
|
|
return round(v, 8)
|
|
|
|
|
|
def amount_from_margin(
|
|
margin_capital: float,
|
|
leverage: int,
|
|
price: float,
|
|
contract_size: float = DEFAULT_CONTRACT_SIZE,
|
|
) -> Optional[float]:
|
|
try:
|
|
margin = float(margin_capital)
|
|
lev = int(leverage)
|
|
px = float(price)
|
|
cs = float(contract_size) if contract_size else DEFAULT_CONTRACT_SIZE
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if margin <= 0 or lev <= 0 or px <= 0 or cs <= 0:
|
|
return None
|
|
notional = margin * lev
|
|
return notional / (px * cs)
|
|
|
|
|
|
def calc_trend_calculator(
|
|
*,
|
|
direction: str,
|
|
capital_usdt: float,
|
|
risk_percent: float,
|
|
leverage: int,
|
|
entry_price: float,
|
|
stop_loss: float,
|
|
add_upper: float,
|
|
take_profit: float,
|
|
dca_legs: int = DEFAULT_DCA_LEGS,
|
|
contract_size: float = DEFAULT_CONTRACT_SIZE,
|
|
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
|
|
direction = (direction or "long").strip().lower()
|
|
if direction not in ("long", "short"):
|
|
return None, "方向须为 long 或 short"
|
|
try:
|
|
capital = float(capital_usdt)
|
|
rp = float(risk_percent)
|
|
lev = int(leverage)
|
|
entry = float(entry_price)
|
|
sl = float(stop_loss)
|
|
upper = float(add_upper)
|
|
tp = float(take_profit)
|
|
legs = max(1, int(dca_legs))
|
|
cs = float(contract_size) if contract_size else DEFAULT_CONTRACT_SIZE
|
|
except (TypeError, ValueError):
|
|
return None, "参数格式错误"
|
|
if capital <= 0 or rp <= 0 or lev <= 0 or entry <= 0 or sl <= 0 or upper <= 0 or tp <= 0:
|
|
return None, "资金、风险、杠杆与价格须大于 0"
|
|
|
|
bound_err = validate_trend_bounds(direction, sl, upper)
|
|
if bound_err:
|
|
return None, bound_err
|
|
|
|
rf = calc_risk_fraction(direction, upper, sl)
|
|
if rf is None or rf <= 0:
|
|
return None, "止损与补仓区间边界组合无法计算风险比例"
|
|
|
|
risk_budget = capital * (rp / 100.0)
|
|
notional = risk_budget / rf
|
|
margin_plan = min(notional / float(lev), capital * MARGIN_BUFFER)
|
|
if margin_plan <= 0:
|
|
return None, "计划保证金过小"
|
|
|
|
target_amt = amount_from_margin(margin_plan, lev, entry, cs)
|
|
if target_amt is None or target_amt <= 0:
|
|
return None, "无法计算计划张数，请检查入场价与杠杆"
|
|
|
|
payload, err = compute_trend_plan_core(
|
|
direction=direction,
|
|
stop_loss=sl,
|
|
add_upper=upper,
|
|
risk_percent=rp,
|
|
snapshot_usdt=capital,
|
|
leverage=lev,
|
|
live_price=entry,
|
|
target_order_amount=target_amt,
|
|
exchange_symbol="CALC",
|
|
dca_legs=legs,
|
|
amount_precise=_identity_amount_precise,
|
|
min_amount=0.0,
|
|
full_margin_buffer_ratio=MARGIN_BUFFER,
|
|
)
|
|
if err:
|
|
return None, err
|
|
|
|
payload["take_profit"] = tp
|
|
payload["leverage"] = lev
|
|
payload["contract_size"] = cs
|
|
preview, rows = build_trend_preview_level_rows(payload)
|
|
|
|
def _f(v: Any, nd: int = 4) -> Any:
|
|
if v is None:
|
|
return None
|
|
try:
|
|
return round(float(v), nd)
|
|
except (TypeError, ValueError):
|
|
return v
|
|
|
|
table = []
|
|
for row in rows:
|
|
table.append(
|
|
{
|
|
"label": row.get("label"),
|
|
"price": _f(row.get("price"), 8),
|
|
"contracts": _f(row.get("contracts"), 8),
|
|
"avg_entry": _f(row.get("avg_entry"), 8),
|
|
"profit_u": _f(row.get("profit_u")),
|
|
"risk_u": _f(row.get("risk_u")),
|
|
"rr": _f(row.get("rr"), 4),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"direction": direction,
|
|
"capital_usdt": _f(capital),
|
|
"risk_percent": _f(rp, 2),
|
|
"risk_budget_u": _f(preview.get("preview_risk_amount_u")),
|
|
"leverage": lev,
|
|
"entry_price": _f(entry, 8),
|
|
"stop_loss": _f(sl, 8),
|
|
"add_upper": _f(upper, 8),
|
|
"take_profit": _f(tp, 8),
|
|
"plan_margin_u": _f(preview.get("plan_margin_capital")),
|
|
"target_contracts": _f(preview.get("target_order_amount"), 8),
|
|
"first_contracts": _f(preview.get("first_order_amount"), 8),
|
|
"dca_legs": int(preview.get("dca_legs") or legs),
|
|
"first_profit_u": _f(preview.get("preview_first_profit_u")),
|
|
"first_rr": _f(preview.get("preview_target_rr"), 4),
|
|
"rows": table,
|
|
}, None
|
|
|
|
|
|
def _round(v: Any, nd: int = 4) -> Any:
|
|
if v is None:
|
|
return None
|
|
try:
|
|
return round(float(v), nd)
|
|
except (TypeError, ValueError):
|
|
return v
|
|
|
|
|
|
def _money_rr(profit_u: Optional[float], risk_u: Optional[float]) -> Optional[float]:
|
|
try:
|
|
if risk_u is None or float(risk_u) <= 0 or profit_u is None:
|
|
return None
|
|
return round(float(profit_u) / float(risk_u), 4)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def calc_initial_roll_qty(
|
|
direction: str,
|
|
entry_price: float,
|
|
stop_loss: float,
|
|
risk_budget_usdt: float,
|
|
) -> Tuple[Optional[float], Optional[str]]:
|
|
"""首仓以损定仓：打到初始止损亏损 = 风险预算。"""
|
|
try:
|
|
entry = float(entry_price)
|
|
sl = float(stop_loss)
|
|
budget = float(risk_budget_usdt)
|
|
except (TypeError, ValueError):
|
|
return None, "参数格式错误"
|
|
if entry <= 0 or sl <= 0 or budget <= 0:
|
|
return None, "入场价、止损与风险预算须大于 0"
|
|
direction = (direction or "long").strip().lower()
|
|
if direction == "short":
|
|
per_unit = sl - entry
|
|
if per_unit <= 0:
|
|
return None, "做空：止损价须高于首仓入场价"
|
|
else:
|
|
per_unit = entry - sl
|
|
if per_unit <= 0:
|
|
return None, "做多：止损价须低于首仓入场价"
|
|
return budget / per_unit, None
|
|
|
|
|
|
def calc_roll_calculator(
|
|
*,
|
|
direction: str,
|
|
capital_usdt: float,
|
|
risk_percent: float,
|
|
entry_price: float,
|
|
stop_loss: float,
|
|
take_profit: float,
|
|
add_legs: list[dict[str, float]] | None = None,
|
|
legs_done: int = 0,
|
|
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
|
|
"""
|
|
滚仓历史测算：首仓自动以损定仓；止盈锁定首仓价；最多 3 次滚仓加仓。
|
|
add_legs: [{add_price, new_stop_loss}, ...]，按顺序链式计算。
|
|
legs_done: 已完成滚仓次数（仅标记，仍参与链式状态推进）。
|
|
"""
|
|
direction = (direction or "long").strip().lower()
|
|
if direction not in ("long", "short"):
|
|
return None, "方向须为 long 或 short"
|
|
try:
|
|
capital = float(capital_usdt)
|
|
rp = float(risk_percent)
|
|
entry = float(entry_price)
|
|
initial_sl = float(stop_loss)
|
|
tp = float(take_profit)
|
|
done = max(0, int(legs_done))
|
|
except (TypeError, ValueError):
|
|
return None, "参数格式错误"
|
|
if capital <= 0 or rp <= 0 or entry <= 0 or initial_sl <= 0 or tp <= 0:
|
|
return None, "资金、风险与价格须大于 0"
|
|
if done > max_roll_legs(direction):
|
|
return None, f"已完成滚仓次数不能超过 {max_roll_legs(direction)} 次"
|
|
|
|
legs_in: list[dict[str, float]] = []
|
|
for raw in add_legs or []:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
try:
|
|
ap = float(raw.get("add_price"))
|
|
nsl = float(raw.get("new_stop_loss"))
|
|
except (TypeError, ValueError):
|
|
return None, "加仓价与新止损须为有效数字"
|
|
if ap <= 0 or nsl <= 0:
|
|
return None, "加仓价与新止损须大于 0"
|
|
legs_in.append({"add_price": ap, "new_stop_loss": nsl})
|
|
|
|
if done + len(legs_in) > max_roll_legs(direction):
|
|
return None, f"已完成 {done} 次 + 待测算 {len(legs_in)} 次，合计不能超过 {max_roll_legs(direction)} 次滚仓"
|
|
|
|
if direction == "long":
|
|
if tp <= entry:
|
|
return None, "做多：止盈价须高于首仓入场价"
|
|
else:
|
|
if tp >= entry:
|
|
return None, "做空：止盈价须低于首仓入场价"
|
|
|
|
risk_budget = capital * (rp / 100.0)
|
|
qty, err = calc_initial_roll_qty(direction, entry, initial_sl, risk_budget)
|
|
if err:
|
|
return None, err
|
|
if qty is None or qty <= 0:
|
|
return None, "无法计算首仓张数"
|
|
|
|
qty_f = float(qty)
|
|
avg = entry
|
|
rows: list[dict[str, Any]] = []
|
|
|
|
if direction == "long":
|
|
first_loss = (avg - initial_sl) * qty_f
|
|
first_profit = (tp - avg) * qty_f
|
|
else:
|
|
first_loss = (initial_sl - avg) * qty_f
|
|
first_profit = (avg - tp) * qty_f
|
|
|
|
rows.append(
|
|
{
|
|
"label": "首仓",
|
|
"leg_index": 0,
|
|
"already_done": False,
|
|
"entry_or_add_price": _round(entry, 8),
|
|
"stop_loss": _round(initial_sl, 8),
|
|
"add_contracts": _round(qty_f, 8),
|
|
"total_contracts": _round(qty_f, 8),
|
|
"avg_entry": _round(avg, 8),
|
|
"take_profit": _round(tp, 8),
|
|
"loss_at_sl_u": _round(first_loss),
|
|
"profit_at_tp_u": _round(first_profit),
|
|
"rr": _money_rr(first_profit, first_loss),
|
|
}
|
|
)
|
|
|
|
current_qty = qty_f
|
|
current_avg = avg
|
|
|
|
for i, leg in enumerate(legs_in):
|
|
leg_no = i + 1
|
|
preview, err = preview_roll(
|
|
direction=direction,
|
|
symbol="CALC",
|
|
qty_existing=current_qty,
|
|
entry_existing=current_avg,
|
|
initial_take_profit=tp,
|
|
add_mode="market",
|
|
new_stop_loss=leg["new_stop_loss"],
|
|
risk_percent=rp,
|
|
capital_base_usdt=capital,
|
|
add_price=leg["add_price"],
|
|
legs_done=i,
|
|
)
|
|
if err:
|
|
return None, f"滚仓第 {leg_no} 次：{err}"
|
|
if not preview:
|
|
return None, f"滚仓第 {leg_no} 次计算失败"
|
|
|
|
current_qty = float(preview["qty_after"])
|
|
current_avg = float(preview["avg_entry_after"])
|
|
loss = preview.get("loss_at_sl_usdt")
|
|
reward = preview.get("reward_at_tp_usdt")
|
|
rows.append(
|
|
{
|
|
"label": f"滚仓{leg_no}",
|
|
"leg_index": leg_no,
|
|
"already_done": leg_no <= done,
|
|
"entry_or_add_price": _round(preview.get("add_price"), 8),
|
|
"stop_loss": _round(preview.get("new_stop_loss"), 8),
|
|
"add_contracts": _round(preview.get("add_amount_raw"), 8),
|
|
"total_contracts": _round(current_qty, 8),
|
|
"avg_entry": _round(current_avg, 8),
|
|
"take_profit": _round(tp, 8),
|
|
"loss_at_sl_u": _round(loss),
|
|
"profit_at_tp_u": _round(reward),
|
|
"rr": _money_rr(reward, loss),
|
|
}
|
|
)
|
|
|
|
last = rows[-1]
|
|
return {
|
|
"direction": direction,
|
|
"capital_usdt": _round(capital),
|
|
"risk_percent": _round(rp, 2),
|
|
"risk_budget_u": _round(risk_budget),
|
|
"entry_price": _round(entry, 8),
|
|
"stop_loss": _round(initial_sl, 8),
|
|
"take_profit": _round(tp, 8),
|
|
"legs_done": done,
|
|
"roll_legs_planned": len(legs_in),
|
|
"first_contracts": _round(qty_f, 8),
|
|
"final_contracts": last.get("total_contracts"),
|
|
"final_avg_entry": last.get("avg_entry"),
|
|
"final_loss_at_sl_u": last.get("loss_at_sl_u"),
|
|
"final_profit_at_tp_u": last.get("profit_at_tp_u"),
|
|
"final_rr": last.get("rr"),
|
|
"rows": rows,
|
|
}, None
|