"""中控历史测算：趋势回调 / 滚仓，以损定仓（无交易所精度，张数按公式估算）。""" from __future__ import annotations from typing import Any, Callable, Optional, Tuple from strategy_roll_lib import 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 calc_roll_calculator( *, direction: str, capital_usdt: float, risk_percent: float, qty_existing: float, entry_existing: float, take_profit: float, add_price: float, new_stop_loss: float, legs_done: int = 0, ) -> Tuple[Optional[dict[str, Any]], Optional[str]]: preview, err = preview_roll( direction=direction, symbol="CALC", qty_existing=qty_existing, entry_existing=entry_existing, initial_take_profit=take_profit, add_mode="market", new_stop_loss=new_stop_loss, risk_percent=risk_percent, capital_base_usdt=capital_usdt, add_price=add_price, legs_done=legs_done, ) if err: return None, err if not preview: return None, "计算失败" 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 rr = None loss = preview.get("loss_at_sl_usdt") reward = preview.get("reward_at_tp_usdt") try: if loss and float(loss) > 0 and reward is not None: rr = round(float(reward) / float(loss), 4) except (TypeError, ValueError): rr = None return { "direction": preview.get("direction"), "capital_usdt": _f(capital_usdt), "risk_percent": _f(risk_percent, 2), "risk_budget_u": _f(preview.get("risk_budget_usdt")), "qty_existing": _f(qty_existing, 8), "entry_existing": _f(entry_existing, 8), "take_profit": _f(take_profit, 8), "add_price": _f(preview.get("add_price"), 8), "new_stop_loss": _f(new_stop_loss, 8), "add_contracts": _f(preview.get("add_amount_raw"), 8), "qty_after": _f(preview.get("qty_after"), 8), "avg_entry_after": _f(preview.get("avg_entry_after"), 8), "loss_at_sl_u": _f(loss), "profit_at_tp_u": _f(reward), "rr": rr, "leg_index_next": preview.get("leg_index_next"), }, None