dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix amp-stats perp PnL to exit at daily profit target.

Browse Source 
Hit A/B via open-to-high/low; day PnL equals target when touched, otherwise settle at close.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-28 14:41:29 +08:00
parent 2ce67da8e8
commit 90be23e845
5 changed files with 222 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/hub/amp_stats_lib.py
+169 -46
View File
@@ -275,15 +275,16 @@ def enrich_rows_pnl(
item["take_profit_hit"] = hit
item["profit"] = round(move - prem, 4) if prem is not None else None
if hedge is not None:
item["perp_hedge_pnl"] = perp_hedge_day_pnl(
change=float(item.get("change") or 0),
open_px=float(item.get("open") or 0),
close_px=float(item.get("close") or 0),
option_leverage=float(hedge["option_leverage"]),
opt_coins=float(hedge["opt_coins"]),
)
day = perp_hedge_day_result(item, hedge)
item["perp_hedge_pnl"] = day["pnl"]
item["perp_hedge_exit"] = day["exit"]
item["perp_hedge_hit_a"] = day["hit_a"]
item["perp_hedge_hit_b"] = day["hit_b"]
else:
item["perp_hedge_pnl"] = None
item["perp_hedge_exit"] = None
item["perp_hedge_hit_a"] = False
item["perp_hedge_hit_b"] = False
out.append(item)
return out
@@ -336,7 +337,37 @@ def perp_hedge_day_premium(*, open_px: float, option_leverage: float, opt_coins:
return coins * (o / lev)
def perp_hedge_day_pnl(
def perp_hedge_required_moves(
*,
open_px: float,
hedge: dict[str, float],
) -> tuple[Optional[float], Optional[float], Optional[str]]:
"""按当日开盘推 A/B 达目标盈利所需点数."""
from lib.hub.hub_perp_options_calc_lib import calc_perp_options_points
spot = float(open_px or 0)
if spot <= 0:
return None, None, "开盘价无效"
points_data, points_err = calc_perp_options_points(
base="ETH",
spot=spot,
capital_usdt=max(spot / hedge["perp_leverage"] * 2, 1000.0),
target_profit_u=hedge["target_profit_u"],
perp_leverage=hedge["perp_leverage"],
option_leverage=hedge["option_leverage"],
ratio_perp=hedge["ratio_perp"],
ratio_opt=hedge["ratio_opt"],
ct_mult=hedge["ct_mult"],
)
if not points_data:
return None, None, points_err
move_a = float((points_data.get("case_a") or {}).get("move_points") or 0) or None
mb = (points_data.get("case_b") or {}).get("move_points_portfolio")
move_b = float(mb) if mb is not None else None
return move_a, move_b, points_err
def perp_hedge_day_pnl_eod(
*,
change: float,
open_px: float,
@@ -344,7 +375,7 @@ def perp_hedge_day_pnl(
option_leverage: float,
opt_coins: float,
) -> float:
"""单日组合净利(永续多1币 + 买期权);入场/权利金按当日开盘.
"""未触达目标时按收盘结算的组合净利.
上涨: change − 权利金 − 永续开平手续费
下跌: |change|×(opt_coins−1) − 权利金
@@ -361,20 +392,114 @@ def perp_hedge_day_pnl(
if open_px and close_px and open_px > 0 and close_px > 0:
fee = estimate_roundtrip_fee_usdt(open_px, close_px, qty=1.0, contract_size=1.0)
return round(chg - prem - fee, 4)
# 下跌: 永续亏 chg(负), 期权内在 |chg|*coins
return round(abs(chg) * (coins - 1.0) - prem, 4)
def perp_hedge_day_result(row: dict[str, Any], hedge: dict[str, float]) -> dict[str, Any]:
"""单日永期结果:触达目标点数则按目标盈利出场,否则收盘结算.
A: 开→高 ≥ move_a → 出场净利 = 目标盈利
B: 开→低 ≥ move_b → 出场净利 = 目标盈利
两边都触达时仍按目标盈利(路径未知,任一边出场均约为目标).
"""
open_px = float(row.get("open") or 0)
close_px = float(row.get("close") or 0)
change = float(row.get("change") or 0)
up_pts = float(row.get("up_points") or 0)
down_pts = float(row.get("down_points") or 0)
target = float(hedge["target_profit_u"])
move_a, move_b, _err = perp_hedge_required_moves(open_px=open_px, hedge=hedge)
hit_a = bool(move_a is not None and move_a > 0 and up_pts >= move_a)
hit_b = bool(move_b is not None and move_b > 0 and down_pts >= move_b)
if hit_a or hit_b:
if hit_a and hit_b:
exit_tag = "target_both"
elif hit_a:
exit_tag = "target_a"
else:
exit_tag = "target_b"
return {
"pnl": round(target, 4),
"exit": exit_tag,
"hit_a": hit_a,
"hit_b": hit_b,
"move_a": move_a,
"move_b": move_b,
}
return {
"pnl": perp_hedge_day_pnl_eod(
change=change,
open_px=open_px,
close_px=close_px,
option_leverage=float(hedge["option_leverage"]),
opt_coins=float(hedge["opt_coins"]),
),
"exit": "eod",
"hit_a": False,
"hit_b": False,
"move_a": move_a,
"move_b": move_b,
}
# 兼容旧名:默认按「目标出场」完整日结果取 pnl
def perp_hedge_day_pnl(
*,
change: float,
open_px: float,
close_px: float,
option_leverage: float,
opt_coins: float,
up_points: Optional[float] = None,
down_points: Optional[float] = None,
target_profit_u: Optional[float] = None,
perp_leverage: float = 10.0,
ratio_perp: float = 1.0,
ratio_opt: float = 2.0,
ct_mult: float = 0.01,
) -> float:
"""单日盈亏.若给了目标与开→高/低,触达则按目标出场;否则收盘结算."""
if target_profit_u is None or up_points is None or down_points is None:
return perp_hedge_day_pnl_eod(
change=change,
open_px=open_px,
close_px=close_px,
option_leverage=option_leverage,
opt_coins=opt_coins,
)
hedge = {
"target_profit_u": float(target_profit_u),
"perp_leverage": float(perp_leverage),
"option_leverage": float(option_leverage),
"ratio_perp": float(ratio_perp),
"ratio_opt": float(ratio_opt),
"ct_mult": float(ct_mult),
"opt_coins": float(opt_coins),
"opt_sheets": float(opt_coins) / float(ct_mult),
}
return float(
perp_hedge_day_result(
{
"open": open_px,
"close": close_px,
"change": change,
"up_points": up_points,
"down_points": down_points,
},
hedge,
)["pnl"]
)
def perp_hedge_stats(
rows: list[dict[str, Any]],
hedge: dict[str, float],
) -> dict[str, Any]:
"""永期对冲:所需点数达标 + 按日组合盈亏汇总.
日盈亏权利金按当日开盘;推所需点数用样本开盘中位数作入场参照.
达标看开→高/开→低是否触达当日入场推得的 A/B 点数;
触达则日盈亏=目标盈利,否则收盘结算.汇总展示点数用样本开盘中位.
"""
from lib.hub.hub_perp_options_calc_lib import calc_perp_options_points
opens = [float(r.get("open") or 0) for r in (rows or []) if float(r.get("open") or 0) > 0]
spot_ref = statistics.median(opens) if opens else None
prem_ref = (
@@ -392,40 +517,24 @@ def perp_hedge_stats(
move_b = None
points_err = None
if spot_ref is not None and spot_ref > 0:
points_data, points_err = calc_perp_options_points(
base="ETH",
spot=spot_ref,
capital_usdt=max(spot_ref / hedge["perp_leverage"] * 2, 1000.0),
target_profit_u=hedge["target_profit_u"],
perp_leverage=hedge["perp_leverage"],
option_leverage=hedge["option_leverage"],
ratio_perp=hedge["ratio_perp"],
ratio_opt=hedge["ratio_opt"],
ct_mult=hedge["ct_mult"],
)
if points_data:
move_a = float((points_data.get("case_a") or {}).get("move_points") or 0) or None
mb = (points_data.get("case_b") or {}).get("move_points_portfolio")
move_b = float(mb) if mb is not None else None
move_a, move_b, points_err = perp_hedge_required_moves(open_px=spot_ref, hedge=hedge)
else:
points_err = "样本无有效开盘价,无法推所需点数"
# 按日开盘重算盈亏(不沿用固定权利金)
work: list[dict[str, Any]] = []
for r in rows or []:
item = dict(r)
item["perp_hedge_pnl"] = perp_hedge_day_pnl(
change=float(item.get("change") or 0),
open_px=float(item.get("open") or 0),
close_px=float(item.get("close") or 0),
option_leverage=float(hedge["option_leverage"]),
opt_coins=float(hedge["opt_coins"]),
)
day = perp_hedge_day_result(item, hedge)
item["perp_hedge_pnl"] = day["pnl"]
item["perp_hedge_exit"] = day["exit"]
item["perp_hedge_hit_a"] = day["hit_a"]
item["perp_hedge_hit_b"] = day["hit_b"]
work.append(item)
n = len(work)
empty = {
"enabled": True,
"entry": "open",
"exit": "target_or_eod",
"spot": None if spot_ref is None else round(spot_ref, 4),
"target_profit_u": round(hedge["target_profit_u"], 4),
"perp_leverage": round(hedge["perp_leverage"], 4),
@@ -445,6 +554,9 @@ def perp_hedge_stats(
"hit_a_ratio": None,
"hit_b_days": 0,
"hit_b_ratio": None,
"target_exit_days": 0,
"target_exit_ratio": None,
"eod_days": 0,
"pnl_total": None,
"pnl_avg": None,
"win_days": 0,
@@ -459,12 +571,10 @@ def perp_hedge_stats(
if n <= 0:
return empty
hit_a = 0
hit_b = 0
if move_a is not None and move_a > 0:
hit_a = sum(1 for r in work if float(r.get("change") or 0) >= move_a)
if move_b is not None and move_b > 0:
hit_b = sum(1 for r in work if float(r.get("change") or 0) <= -move_b)
hit_a = sum(1 for r in work if r.get("perp_hedge_hit_a"))
hit_b = sum(1 for r in work if r.get("perp_hedge_hit_b"))
target_exits = sum(1 for r in work if str(r.get("perp_hedge_exit") or "").startswith("target"))
eod_days = sum(1 for r in work if r.get("perp_hedge_exit") == "eod")
pnls = [float(r["perp_hedge_pnl"]) for r in work if r.get("perp_hedge_pnl") is not None]
win = sum(1 for p in pnls if p > 0)
@@ -476,9 +586,12 @@ def perp_hedge_stats(
empty.update(
{
"hit_a_days": hit_a,
"hit_a_ratio": round(hit_a / n, 4) if move_a else None,
"hit_a_ratio": round(hit_a / n, 4),
"hit_b_days": hit_b,
"hit_b_ratio": round(hit_b / n, 4) if move_b else None,
"hit_b_ratio": round(hit_b / n, 4),
"target_exit_days": target_exits,
"target_exit_ratio": round(target_exits / n, 4),
"eod_days": eod_days,
"pnl_total": round(sum(pnls), 4) if pnls else None,
"pnl_avg": round(statistics.fmean(pnls), 4) if pnls else None,
"win_days": win,
@@ -1043,7 +1156,7 @@ def build_export_csv(payload: dict[str, Any]) -> str:
[
"A所需点数",
ph.get("move_a"),
"A达标天",
"A达标天(开→高)",
ph.get("hit_a_days"),
"占比",
ph.get("hit_a_ratio"),
@@ -1053,12 +1166,22 @@ def build_export_csv(payload: dict[str, Any]) -> str:
[
"B所需点数(组合)",
ph.get("move_b"),
"B达标天",
"B达标天(开→低)",
ph.get("hit_b_days"),
"占比",
ph.get("hit_b_ratio"),
]
)
w.writerow(
[
"目标出场天",
ph.get("target_exit_days"),
"收盘结算天",
ph.get("eod_days"),
"目标盈利",
ph.get("target_profit_u"),
]
)
w.writerow(
[
"组合盈亏合计",