| {{ lv.label }} |
{% if lv.price is not none %}{{ price_fmt(sym, lv.price) }}{% else %}—{% endif %} |
{% if lv.contracts is not none %}{{ amt_disp(sym, lv.contracts) }}{% else %}—{% endif %} |
{% if lv.avg_entry is not none %}{{ price_fmt(sym, lv.avg_entry) }}{% else %}—{% endif %} |
- {% if lv.take_profit is not none %}{{ price_fmt(sym, lv.take_profit) }}{% else %}—{% endif %} |
- {% if lv.stop_loss is not none %}{{ price_fmt(sym, lv.stop_loss) }}{% else %}—{% endif %} |
+ {% if lv.profit_u is not none %}{{ mf(lv.profit_u) }}{% else %}—{% endif %} |
+ {% if lv.risk_u is not none %}{{ mf(lv.risk_u) }}{% else %}—{% endif %} |
+ {% if lv.rr is not none %}{{ '%.2f'|format(lv.rr) }}{% else %}—{% endif %} |
{{ lv.status_label }} |
{% endfor %}
diff --git a/strategy_trend_lib.py b/strategy_trend_lib.py
index 8305a55..aaf8d88 100644
--- a/strategy_trend_lib.py
+++ b/strategy_trend_lib.py
@@ -241,6 +241,48 @@ def calc_take_profit_for_rr(
return None
+def calc_risk_budget_usdt(snapshot_usdt: float, risk_percent: float) -> Optional[float]:
+ """计划止损金额 U = 可用快照 × 风险比例。"""
+ try:
+ snap = float(snapshot_usdt)
+ rp = float(risk_percent)
+ if snap <= 0 or rp <= 0:
+ return None
+ return round(snap * rp / 100.0, 4)
+ except (TypeError, ValueError):
+ return None
+
+
+def calc_money_reward_risk_ratio(profit_u: float, risk_u: float) -> Optional[float]:
+ """金额盈亏比 = 止盈盈利 U / 止损金额 U。"""
+ try:
+ r = float(risk_u)
+ p = float(profit_u)
+ if r <= 0:
+ return None
+ return round(p / r, 4)
+ except (TypeError, ValueError):
+ return None
+
+
+def calc_tp_profit_usdt(
+ direction: str,
+ avg_entry: float,
+ take_profit_price: float,
+ contracts: float,
+ contract_size: float = 1.0,
+) -> Optional[float]:
+ """到达止盈价时，按累计张数与加仓后均价的盈利 U。"""
+ try:
+ from hub_position_metrics import estimate_linear_swap_upnl_usdt
+
+ return estimate_linear_swap_upnl_usdt(
+ direction, float(avg_entry), float(take_profit_price), float(contracts), float(contract_size)
+ )
+ except (TypeError, ValueError):
+ return None
+
+
def weighted_avg_entry(legs: list[tuple[float, float]]) -> Optional[float]:
"""按 (成交价, 张数) 加权均价。"""
total = 0.0
@@ -262,7 +304,7 @@ def weighted_avg_entry(legs: list[tuple[float, float]]) -> Optional[float]:
def build_trend_preview_level_rows(preview: dict) -> tuple[dict, list[dict]]:
"""
- 预览：参考价首仓止盈 + 每档补仓后止盈；止损统一为计划止损（加仓后最大止损）。
+ 预览：表单止盈价下每档累计持仓的盈利 U；止损金额 = 快照×风险；盈亏比按金额对比。
返回 (增强后的 preview 字段, 表格行列表，含首仓行)。
"""
p = dict(preview or {})
@@ -272,16 +314,24 @@ def build_trend_preview_level_rows(preview: dict) -> tuple[dict, list[dict]]:
sl = float(p.get("stop_loss"))
user_tp = float(p.get("take_profit"))
first_amt = float(p.get("first_order_amount"))
+ snapshot = float(p.get("snapshot_available_usdt"))
+ risk_percent = float(p.get("risk_percent"))
except (TypeError, ValueError):
return p, []
- rr = calc_planned_reward_risk_ratio(direction, ref, sl, user_tp)
- if rr is None:
+ risk_u = calc_risk_budget_usdt(snapshot, risk_percent)
+ if risk_u is None or risk_u <= 0:
return p, []
- first_tp = calc_take_profit_for_rr(direction, ref, sl, rr)
- p["preview_target_rr"] = rr
- p["preview_first_take_profit"] = first_tp
+ try:
+ contract_size = float(p.get("contract_size") or 1.0)
+ if contract_size <= 0:
+ contract_size = 1.0
+ except (TypeError, ValueError):
+ contract_size = 1.0
+
+ p["preview_risk_amount_u"] = risk_u
+ p["preview_take_profit_price"] = user_tp
p["preview_unified_stop_loss"] = sl
try:
@@ -297,45 +347,81 @@ def build_trend_preview_level_rows(preview: dict) -> tuple[dict, list[dict]]:
except Exception:
leg_amounts = []
- rows: list[dict] = [
- {
- "i": 0,
- "label": "首仓",
- "price": ref,
- "contracts": first_amt,
- "avg_entry": ref,
- "take_profit": first_tp,
- "stop_loss": sl,
- "is_first": True,
+ def _row_dict(
+ *,
+ i: int,
+ label: str,
+ price: float,
+ leg_contracts: float,
+ cum_contracts: float,
+ avg: float,
+ is_first: bool,
+ ) -> dict:
+ profit_u = calc_tp_profit_usdt(direction, avg, user_tp, cum_contracts, contract_size)
+ rr_money = calc_money_reward_risk_ratio(profit_u, risk_u) if profit_u is not None else None
+ return {
+ "i": i,
+ "label": label,
+ "price": price,
+ "contracts": leg_contracts,
+ "cum_contracts": cum_contracts,
+ "avg_entry": avg,
+ "take_profit_price": user_tp,
+ "profit_u": profit_u,
+ "risk_u": risk_u,
+ "rr": rr_money,
+ "stop_loss_price": sl,
+ "take_profit": profit_u,
+ "stop_loss": risk_u,
+ "is_first": is_first,
}
+
+ cum_contracts = first_amt
+ first_profit = calc_tp_profit_usdt(direction, ref, user_tp, cum_contracts, contract_size)
+ first_rr = calc_money_reward_risk_ratio(first_profit, risk_u) if first_profit is not None else None
+ p["preview_first_profit_u"] = first_profit
+ p["preview_target_rr"] = first_rr
+ p["preview_first_take_profit"] = user_tp
+
+ rows: list[dict] = [
+ _row_dict(
+ i=0,
+ label="首仓",
+ price=ref,
+ leg_contracts=first_amt,
+ cum_contracts=cum_contracts,
+ avg=ref,
+ is_first=True,
+ )
]
accumulated: list[tuple[float, float]] = [(ref, first_amt)]
for i, pair in enumerate(zip(grid, leg_amounts), 1):
try:
price = float(pair[0])
- contracts = float(pair[1])
+ leg_contracts = float(pair[1])
except (TypeError, ValueError):
continue
- accumulated.append((price, contracts))
+ accumulated.append((price, leg_contracts))
avg = weighted_avg_entry(accumulated)
- tp_after = calc_take_profit_for_rr(direction, avg, sl, rr) if avg is not None else None
+ if avg is None:
+ continue
+ cum_contracts += leg_contracts
rows.append(
- {
- "i": i,
- "label": f"补仓{i}",
- "price": price,
- "contracts": contracts,
- "avg_entry": avg,
- "take_profit": tp_after,
- "stop_loss": sl,
- "is_first": False,
- }
+ _row_dict(
+ i=i,
+ label=f"补仓{i}",
+ price=price,
+ leg_contracts=leg_contracts,
+ cum_contracts=cum_contracts,
+ avg=avg,
+ is_first=False,
+ )
)
return p, rows
def enrich_trend_dca_levels_with_tp(plan: dict, levels: list[dict]) -> list[dict]:
- """运行中计划：为 dca_levels 补充加仓后均价、止盈、统一止损。"""
+ """运行中计划：为 dca_levels 补充加仓后均价、止盈盈利 U、止损金额 U、金额盈亏比。"""
if not levels:
return levels
p = plan or {}
@@ -344,9 +430,15 @@ def enrich_trend_dca_levels_with_tp(plan: dict, levels: list[dict]) -> list[dict
sl = float(p.get("stop_loss"))
user_tp = float(p.get("take_profit"))
first_amt = float(p.get("first_order_amount"))
+ snapshot = float(p.get("snapshot_available_usdt"))
+ risk_percent = float(p.get("risk_percent"))
except (TypeError, ValueError):
return levels
+ risk_u = calc_risk_budget_usdt(snapshot, risk_percent)
+ if risk_u is None or risk_u <= 0:
+ return levels
+
ref_raw = p.get("live_price_ref")
if ref_raw in (None, ""):
ref_raw = p.get("avg_entry_price")
@@ -355,12 +447,16 @@ def enrich_trend_dca_levels_with_tp(plan: dict, levels: list[dict]) -> list[dict
except (TypeError, ValueError):
return levels
- rr = calc_planned_reward_risk_ratio(direction, ref, sl, user_tp)
- if rr is None:
- return levels
+ try:
+ contract_size = float(p.get("contract_size") or 1.0)
+ if contract_size <= 0:
+ contract_size = 1.0
+ except (TypeError, ValueError):
+ contract_size = 1.0
out: list[dict] = []
accumulated: list[tuple[float, float]] = []
+ cum_contracts = 0.0
for lv in levels:
row = dict(lv)
is_first = row.get("leg_key") == "first" or row.get("label") == "首仓" or row.get("i") == 0
@@ -371,21 +467,32 @@ def enrich_trend_dca_levels_with_tp(plan: dict, levels: list[dict]) -> list[dict
except (TypeError, ValueError):
amt_f = first_amt
accumulated = [(ref, amt_f)]
+ cum_contracts = amt_f
row["avg_entry"] = ref
- row["take_profit"] = calc_take_profit_for_rr(direction, ref, sl, rr)
- row["stop_loss"] = sl
else:
price = row.get("price")
contracts = row.get("contracts")
if price is not None and contracts is not None:
try:
- accumulated.append((float(price), float(contracts)))
+ leg_contracts = float(contracts)
+ accumulated.append((float(price), leg_contracts))
avg = weighted_avg_entry(accumulated)
if avg is not None:
row["avg_entry"] = avg
- row["take_profit"] = calc_take_profit_for_rr(direction, avg, sl, rr)
- row["stop_loss"] = sl
+ cum_contracts += leg_contracts
except (TypeError, ValueError):
pass
+ avg_entry = row.get("avg_entry")
+ if avg_entry is not None:
+ profit_u = calc_tp_profit_usdt(
+ direction, float(avg_entry), user_tp, cum_contracts, contract_size
+ )
+ row["take_profit_price"] = user_tp
+ row["profit_u"] = profit_u
+ row["risk_u"] = risk_u
+ row["rr"] = calc_money_reward_risk_ratio(profit_u, risk_u) if profit_u is not None else None
+ row["take_profit"] = profit_u
+ row["stop_loss"] = risk_u
+ row["stop_loss_price"] = sl
out.append(row)
return out
diff --git a/strategy_trend_register.py b/strategy_trend_register.py
index 8984f0e..a6605c1 100644
--- a/strategy_trend_register.py
+++ b/strategy_trend_register.py
@@ -249,6 +249,7 @@ def parse_trend_plan(cfg: dict, form_dict) -> tuple[Optional[dict], Optional[str
leg_list = json.loads(leg_json)
except Exception:
leg_list = []
+ contract_size = float(market.get("contractSize") or 1)
return {
"symbol": symbol,
"exchange_symbol": exchange_symbol,
@@ -271,6 +272,7 @@ def parse_trend_plan(cfg: dict, form_dict) -> tuple[Optional[dict], Optional[str
"leg_amounts_json": leg_json,
"grid": grid,
"leg_amounts": leg_list,
+ "contract_size": contract_size,
}, None
@@ -486,6 +488,12 @@ def enrich_trend_plan(cfg: dict, row) -> dict:
d["floating_mark"] = None
else:
d["floating_pnl"] = d["floating_mark"] = None
+ get_cs = getattr(m, "get_contract_size", None)
+ if callable(get_cs):
+ try:
+ d["contract_size"] = float(get_cs(ex_sym))
+ except (TypeError, ValueError):
+ pass
from strategy_snapshot_lib import attach_trend_dca_levels
d = attach_trend_dca_levels(d)
@@ -1097,6 +1105,14 @@ def load_trend_page_context(conn, request_obj, cfg: dict) -> dict[str, Any]:
trend_preview = _row(cfg, pr)
preview_expires_ms = int(pr["expires_at_ms"])
+ get_cs = getattr(m, "get_contract_size", None)
+ if callable(get_cs) and not trend_preview.get("contract_size"):
+ try:
+ trend_preview["contract_size"] = float(
+ get_cs(trend_preview.get("exchange_symbol") or trend_preview.get("symbol") or "")
+ )
+ except (TypeError, ValueError):
+ pass
trend_preview, trend_preview_levels = build_trend_preview_level_rows(trend_preview)
elif pr:
trend_preview_expired = True
diff --git a/tests/test_trend_preview_tp.py b/tests/test_trend_preview_tp.py
index 2d3f65f..368bf8c 100644
--- a/tests/test_trend_preview_tp.py
+++ b/tests/test_trend_preview_tp.py
@@ -1,4 +1,4 @@
-"""趋势回调预览：参考价首仓止盈与补仓后止盈。"""
+"""趋势回调预览：止盈盈利 U、止损金额 U、金额盈亏比。"""
from __future__ import annotations
import json
@@ -11,41 +11,47 @@ sys.path.insert(0, str(ROOT))
from strategy_trend_lib import ( # noqa: E402
build_trend_preview_level_rows,
- calc_planned_reward_risk_ratio,
- calc_take_profit_for_rr,
+ calc_money_reward_risk_ratio,
+ calc_risk_budget_usdt,
+ calc_tp_profit_usdt,
)
class TestTrendPreviewTp(unittest.TestCase):
- def test_short_ref_price_first_tp_matches_form_tp(self):
- ref, sl, tp = 72.6, 75.5, 65.0
- rr = calc_planned_reward_risk_ratio("short", ref, sl, tp)
- self.assertIsNotNone(rr)
- first_tp = calc_take_profit_for_rr("short", ref, sl, rr)
- self.assertAlmostEqual(first_tp, tp, places=2)
+ def test_risk_budget_from_snapshot(self):
+ self.assertAlmostEqual(calc_risk_budget_usdt(110.73, 5), 5.5365, places=2)
- def test_preview_levels_include_first_and_dca_tp(self):
+ def test_short_profit_at_form_take_profit(self):
+ profit = calc_tp_profit_usdt("short", 72.53, 66.0, 1114, 0.00167)
+ self.assertIsNotNone(profit)
+ self.assertGreater(profit, 0)
+ rr = calc_money_reward_risk_ratio(profit, 5.5365)
+ self.assertIsNotNone(rr)
+ self.assertGreater(rr, 1.5)
+
+ def test_preview_levels_use_money_rr(self):
preview = {
"direction": "short",
- "live_price_ref": 72.6,
+ "live_price_ref": 72.53,
"stop_loss": 75.5,
- "take_profit": 65.0,
- "first_order_amount": 1113,
+ "take_profit": 66.0,
+ "first_order_amount": 1114,
+ "snapshot_available_usdt": 110.73,
+ "risk_percent": 5,
+ "contract_size": 0.00167,
"grid_prices_json": json.dumps([73.42, 73.83]),
"leg_amounts_json": json.dumps([222, 222]),
}
enriched, rows = build_trend_preview_level_rows(preview)
- self.assertEqual(enriched["preview_unified_stop_loss"], 75.5)
- self.assertAlmostEqual(enriched["preview_first_take_profit"], 65.0, places=1)
+ self.assertAlmostEqual(enriched["preview_risk_amount_u"], 5.5365, places=2)
+ self.assertEqual(enriched["preview_take_profit_price"], 66.0)
self.assertEqual(len(rows), 3)
self.assertEqual(rows[0]["label"], "首仓")
- self.assertAlmostEqual(rows[0]["take_profit"], 65.0, places=1)
- self.assertEqual(rows[0]["stop_loss"], 75.5)
- self.assertIsNotNone(rows[1]["avg_entry"])
- self.assertIsNotNone(rows[1]["take_profit"])
- self.assertEqual(rows[1]["stop_loss"], 75.5)
- # 做空：补仓价上移 → 均价上移 → 同等 RR 下止盈价上移
- self.assertGreater(rows[2]["take_profit"], rows[1]["take_profit"])
+ self.assertEqual(rows[0]["risk_u"], enriched["preview_risk_amount_u"])
+ self.assertIsNotNone(rows[0]["profit_u"])
+ self.assertAlmostEqual(rows[0]["rr"], rows[0]["profit_u"] / 5.5365, places=2)
+ self.assertEqual(rows[1]["risk_u"], enriched["preview_risk_amount_u"])
+ self.assertGreater(rows[2]["profit_u"], rows[1]["profit_u"])
if __name__ == "__main__":