| {{ 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 %} |
{{ lv.status_label }} |
{% endfor %}
diff --git a/strategy_trend_lib.py b/strategy_trend_lib.py
index 98b6c21..8305a55 100644
--- a/strategy_trend_lib.py
+++ b/strategy_trend_lib.py
@@ -191,3 +191,201 @@ def compute_trend_plan_core(
"leg_amounts": leg_list,
}
return payload, None
+
+
+def calc_planned_reward_risk_ratio(
+ direction: str, entry_price: float, stop_loss: float, take_profit: float
+) -> Optional[float]:
+ """盈亏比（reward/risk），与四所 calc_rr_ratio 口径一致。"""
+ try:
+ entry = float(entry_price)
+ sl = float(stop_loss)
+ tp = float(take_profit)
+ if entry <= 0 or sl <= 0 or tp <= 0:
+ return None
+ direction = (direction or "long").strip().lower()
+ if direction == "short":
+ risk = sl - entry
+ reward = entry - tp
+ else:
+ risk = entry - sl
+ reward = tp - entry
+ if risk <= 0 or reward <= 0:
+ return None
+ return round(reward / risk, 4)
+ except (TypeError, ValueError):
+ return None
+
+
+def calc_take_profit_for_rr(
+ direction: str, entry_price: float, stop_loss: float, reward_risk_ratio: float
+) -> Optional[float]:
+ """按统一止损与目标 RR 反推止盈价。"""
+ try:
+ entry = float(entry_price)
+ sl = float(stop_loss)
+ rr = float(reward_risk_ratio)
+ if entry <= 0 or sl <= 0 or rr <= 0:
+ return None
+ direction = (direction or "long").strip().lower()
+ if direction == "short":
+ risk = sl - entry
+ if risk <= 0:
+ return None
+ return round(entry - rr * risk, 10)
+ risk = entry - sl
+ if risk <= 0:
+ return None
+ return round(entry + rr * risk, 10)
+ except (TypeError, ValueError):
+ return None
+
+
+def weighted_avg_entry(legs: list[tuple[float, float]]) -> Optional[float]:
+ """按 (成交价, 张数) 加权均价。"""
+ total = 0.0
+ cost = 0.0
+ for price, amount in legs or []:
+ try:
+ p = float(price)
+ a = float(amount)
+ except (TypeError, ValueError):
+ continue
+ if a <= 0:
+ continue
+ total += a
+ cost += p * a
+ if total <= 0:
+ return None
+ return cost / total
+
+
+def build_trend_preview_level_rows(preview: dict) -> tuple[dict, list[dict]]:
+ """
+ 预览：参考价首仓止盈 + 每档补仓后止盈；止损统一为计划止损（加仓后最大止损）。
+ 返回 (增强后的 preview 字段, 表格行列表，含首仓行)。
+ """
+ p = dict(preview or {})
+ direction = (p.get("direction") or "long").strip().lower()
+ try:
+ ref = float(p.get("live_price_ref"))
+ sl = float(p.get("stop_loss"))
+ user_tp = float(p.get("take_profit"))
+ first_amt = float(p.get("first_order_amount"))
+ except (TypeError, ValueError):
+ return p, []
+
+ rr = calc_planned_reward_risk_ratio(direction, ref, sl, user_tp)
+ if rr is None:
+ 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
+ p["preview_unified_stop_loss"] = sl
+
+ try:
+ grid = json.loads(p.get("grid_prices_json") or "[]")
+ if not isinstance(grid, list):
+ grid = []
+ except Exception:
+ grid = []
+ try:
+ leg_amounts = json.loads(p.get("leg_amounts_json") or "[]")
+ if not isinstance(leg_amounts, list):
+ leg_amounts = []
+ 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,
+ }
+ ]
+ 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])
+ except (TypeError, ValueError):
+ continue
+ accumulated.append((price, contracts))
+ avg = weighted_avg_entry(accumulated)
+ tp_after = calc_take_profit_for_rr(direction, avg, sl, rr) if avg is not None else None
+ rows.append(
+ {
+ "i": i,
+ "label": f"补仓{i}",
+ "price": price,
+ "contracts": contracts,
+ "avg_entry": avg,
+ "take_profit": tp_after,
+ "stop_loss": sl,
+ "is_first": False,
+ }
+ )
+ return p, rows
+
+
+def enrich_trend_dca_levels_with_tp(plan: dict, levels: list[dict]) -> list[dict]:
+ """运行中计划：为 dca_levels 补充加仓后均价、止盈、统一止损。"""
+ if not levels:
+ return levels
+ p = plan or {}
+ direction = (p.get("direction") or "long").strip().lower()
+ try:
+ sl = float(p.get("stop_loss"))
+ user_tp = float(p.get("take_profit"))
+ first_amt = float(p.get("first_order_amount"))
+ except (TypeError, ValueError):
+ return levels
+
+ ref_raw = p.get("live_price_ref")
+ if ref_raw in (None, ""):
+ ref_raw = p.get("avg_entry_price")
+ try:
+ ref = float(ref_raw)
+ except (TypeError, ValueError):
+ return levels
+
+ rr = calc_planned_reward_risk_ratio(direction, ref, sl, user_tp)
+ if rr is None:
+ return levels
+
+ out: list[dict] = []
+ accumulated: list[tuple[float, float]] = []
+ for lv in levels:
+ row = dict(lv)
+ is_first = row.get("leg_key") == "first" or row.get("label") == "首仓" or row.get("i") == 0
+ if is_first:
+ amt = row.get("contracts")
+ try:
+ amt_f = float(amt if amt is not None else first_amt)
+ except (TypeError, ValueError):
+ amt_f = first_amt
+ accumulated = [(ref, 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)))
+ 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
+ except (TypeError, ValueError):
+ pass
+ out.append(row)
+ return out
diff --git a/strategy_trend_register.py b/strategy_trend_register.py
index 92d9043..8984f0e 100644
--- a/strategy_trend_register.py
+++ b/strategy_trend_register.py
@@ -1093,15 +1093,11 @@ def load_trend_page_context(conn, request_obj, cfg: dict) -> dict[str, Any]:
).fetchone()
now_ms = int(time.time() * 1000)
if pr and int(pr["expires_at_ms"] or 0) >= now_ms:
+ from strategy_trend_lib import build_trend_preview_level_rows
+
trend_preview = _row(cfg, pr)
preview_expires_ms = int(pr["expires_at_ms"])
- try:
- grid = json.loads(trend_preview.get("grid_prices_json") or "[]")
- legs = json.loads(trend_preview.get("leg_amounts_json") or "[]")
- except Exception:
- grid, legs = [], []
- for i, pair in enumerate(zip(grid, legs), 1):
- trend_preview_levels.append({"i": i, "price": pair[0], "contracts": pair[1]})
+ trend_preview, trend_preview_levels = build_trend_preview_level_rows(trend_preview)
elif pr:
trend_preview_expired = True
return {
diff --git a/tests/test_trend_preview_tp.py b/tests/test_trend_preview_tp.py
new file mode 100644
index 0000000..2d3f65f
--- /dev/null
+++ b/tests/test_trend_preview_tp.py
@@ -0,0 +1,52 @@
+"""趋势回调预览：参考价首仓止盈与补仓后止盈。"""
+from __future__ import annotations
+
+import json
+import sys
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+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,
+)
+
+
+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_preview_levels_include_first_and_dca_tp(self):
+ preview = {
+ "direction": "short",
+ "live_price_ref": 72.6,
+ "stop_loss": 75.5,
+ "take_profit": 65.0,
+ "first_order_amount": 1113,
+ "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.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"])
+
+
+if __name__ == "__main__":
+ unittest.main()