dekun
a841c8f31b
Use dual indices for watch/preview; perp entry uses ETH mark; expiry/close convert coin PnL via option index; align legacy long→Call; omit combo U when index missing. Co-authored-by: Cursor <cursoragent@cursor.com>
259 lines
9.4 KiB
Python
259 lines
9.4 KiB
Python
"""永期实例模式 HEDGE_PO_MODE:钳制/定仓/校验/组合U口径."""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from lib.hedge_plan.hedge_plan_option_primary_lib import (
|
||
contracts_from_option_sheets,
|
||
estimate_combo_net_pnl,
|
||
size_from_premium,
|
||
)
|
||
from lib.hedge_plan.hedge_po_mode_lib import (
|
||
MODE_BTC,
|
||
MODE_BTC_ETH,
|
||
MODE_ETH,
|
||
apply_mode_to_start_body,
|
||
clamp_hedge_po_mode_on_startup,
|
||
combo_net_usdt,
|
||
default_option_perp_ratio,
|
||
get_hedge_po_mode,
|
||
normalize_hedge_po_mode,
|
||
option_underlying,
|
||
perp_underlying,
|
||
public_mode_payload,
|
||
size_cross_notional,
|
||
validate_body_matches_mode,
|
||
)
|
||
|
||
|
||
class TestHedgePoMode(unittest.TestCase):
|
||
def test_normalize_aliases(self):
|
||
self.assertEqual(normalize_hedge_po_mode("ETH"), MODE_ETH)
|
||
self.assertEqual(normalize_hedge_po_mode("btc-eth"), MODE_BTC_ETH)
|
||
self.assertEqual(normalize_hedge_po_mode("跨币"), MODE_BTC_ETH)
|
||
self.assertEqual(normalize_hedge_po_mode("nope"), "")
|
||
|
||
def test_underlyings(self):
|
||
self.assertEqual(option_underlying(MODE_ETH), "ETH")
|
||
self.assertEqual(perp_underlying(MODE_ETH), "ETH")
|
||
self.assertEqual(option_underlying(MODE_BTC), "BTC")
|
||
self.assertEqual(perp_underlying(MODE_BTC), "BTC")
|
||
self.assertEqual(option_underlying(MODE_BTC_ETH), "BTC")
|
||
self.assertEqual(perp_underlying(MODE_BTC_ETH), "ETH")
|
||
|
||
def test_cross_ratio_from_k(self):
|
||
with patch.dict(os.environ, {"HEDGE_CROSS_NOTIONAL_K": "0.5"}, clear=False):
|
||
self.assertAlmostEqual(default_option_perp_ratio(MODE_BTC_ETH), 2.0)
|
||
|
||
def test_clamp_startup_and_sync_underly(self):
|
||
with patch.dict(
|
||
os.environ,
|
||
{"HEDGE_PO_MODE": "btc_eth", "OKX_OPTIONS_DEFAULT_UNDERLY": "ETH"},
|
||
clear=False,
|
||
):
|
||
summary = clamp_hedge_po_mode_on_startup()
|
||
self.assertEqual(summary["mode"], MODE_BTC_ETH)
|
||
self.assertEqual(os.environ["HEDGE_PO_MODE"], MODE_BTC_ETH)
|
||
self.assertEqual(os.environ["OKX_OPTIONS_DEFAULT_UNDERLY"], "BTC")
|
||
self.assertEqual(get_hedge_po_mode(), MODE_BTC_ETH)
|
||
|
||
def test_clamp_invalid_falls_back(self):
|
||
with patch.dict(
|
||
os.environ,
|
||
{"HEDGE_PO_MODE": "weird", "OKX_OPTIONS_DEFAULT_UNDERLY": "BTC"},
|
||
clear=False,
|
||
):
|
||
summary = clamp_hedge_po_mode_on_startup()
|
||
self.assertEqual(summary["mode"], MODE_BTC)
|
||
self.assertTrue(summary["warnings"])
|
||
|
||
def test_apply_and_validate_body(self):
|
||
with patch.dict(os.environ, {"HEDGE_PO_MODE": "btc_eth", "HEDGE_CROSS_NOTIONAL_K": "0.5"}):
|
||
clamp_hedge_po_mode_on_startup()
|
||
body = apply_mode_to_start_body(
|
||
{
|
||
"direction": "long",
|
||
"underlying": "ETH",
|
||
"opt_inst_id": "BTC-USD-260831-90000-C",
|
||
"exchange_symbol": "ETH/USDT:USDT",
|
||
"profit_target_usdt": 20,
|
||
}
|
||
)
|
||
self.assertEqual(body["underlying"], "BTC")
|
||
self.assertEqual(body["perp_underlying"], "ETH")
|
||
self.assertEqual(body["exit_mode"], "profit")
|
||
self.assertTrue(body["option_primary"])
|
||
self.assertAlmostEqual(float(body["option_perp_ratio"]), 2.0)
|
||
self.assertIsNone(validate_body_matches_mode(body))
|
||
bad = dict(body, opt_inst_id="ETH-USD-260831-2000-C")
|
||
self.assertIn("BTC", validate_body_matches_mode(bad) or "")
|
||
|
||
def test_size_cross_notional_half(self):
|
||
# 0.1 BTC @ 100k → 10k U 期权名义; k=0.5 → 5k ETH 名义 @ 2500 → 2 ETH → 200 张(cs=0.01)
|
||
sz = size_cross_notional(
|
||
option_coin_qty=0.1,
|
||
option_index=100_000,
|
||
perp_index=2500,
|
||
contract_size=0.01,
|
||
notional_k=0.5,
|
||
)
|
||
self.assertTrue(sz["ok"])
|
||
self.assertAlmostEqual(sz["option_notional_usdt"], 10_000.0)
|
||
self.assertAlmostEqual(sz["perp_notional_usdt"], 5_000.0)
|
||
self.assertAlmostEqual(sz["perp_coin"], 2.0)
|
||
self.assertAlmostEqual(sz["contracts"], 200.0)
|
||
|
||
def test_size_from_premium_cross_matches_notional(self):
|
||
# ask=0.01 BTC → 预算 1 BTC 权利金(币); usable 0.95 → coin=0.95 → sheets
|
||
# 用整张回写后再跨币
|
||
sized = size_from_premium(
|
||
premium_budget=1000, # 若 ask 用币报价则预算单位需一致;此处用大预算保证 ≥1 张
|
||
ask=0.01,
|
||
ct_mult=0.01,
|
||
ratio=2,
|
||
contract_size=0.01,
|
||
option_index=100_000,
|
||
perp_index=2_500,
|
||
)
|
||
self.assertTrue(sized["ok"], sized.get("msg"))
|
||
self.assertTrue(sized.get("cross_asset") or sized.get("option_notional_usdt"))
|
||
# coin_qty = sheets * 0.01; 名义路径
|
||
self.assertGreater(float(sized["contracts"]), 0)
|
||
|
||
def test_contracts_from_sheets_same_vs_cross(self):
|
||
same = contracts_from_option_sheets(
|
||
sheets=10, ct_mult=0.01, ratio=2, contract_size=0.01
|
||
)
|
||
self.assertTrue(same["ok"])
|
||
# 0.1 币 / 2 / 0.01 = 5 张
|
||
self.assertAlmostEqual(same["contracts"], 5.0)
|
||
|
||
cross = contracts_from_option_sheets(
|
||
sheets=10,
|
||
ct_mult=0.01,
|
||
ratio=2,
|
||
contract_size=0.01,
|
||
option_index=100_000,
|
||
perp_index=2_500,
|
||
)
|
||
self.assertTrue(cross["ok"])
|
||
# 0.1 BTC * 100k * 0.5 / 2500 / 0.01 = 200
|
||
self.assertAlmostEqual(cross["contracts"], 200.0)
|
||
|
||
def test_combo_net_usdt_not_mix_units(self):
|
||
net = estimate_combo_net_pnl(
|
||
view_side="long",
|
||
strike=90000,
|
||
index_px=91000,
|
||
ask_open=0.01,
|
||
bid=0.012,
|
||
sheets=1,
|
||
ct_mult=0.01,
|
||
perp_direction="short",
|
||
perp_entry=2500,
|
||
perp_mark=2450,
|
||
contracts=10,
|
||
contract_size=0.01,
|
||
fee=0.0,
|
||
option_fee=0.0,
|
||
)
|
||
# opt_net 为币;perp_net 为 U;禁止直接用 net['net'] 出门控
|
||
u = combo_net_usdt(net, option_index=91000)
|
||
self.assertIsNotNone(u)
|
||
self.assertAlmostEqual(u, float(net["opt_net"]) * 91000 + float(net["perp_net"]), places=4)
|
||
|
||
def test_public_payload(self):
|
||
with patch.dict(os.environ, {"HEDGE_PO_MODE": "eth"}):
|
||
clamp_hedge_po_mode_on_startup()
|
||
p = public_mode_payload()
|
||
self.assertEqual(p["po_mode"], MODE_ETH)
|
||
self.assertEqual(p["exit_mode"], "profit")
|
||
self.assertIn("子账户", p["instance_rule"])
|
||
|
||
def test_resolve_option_index_cross_no_perp_fallback(self):
|
||
# 跨币缺指数时不得用 ETH 永续价折 BTC 权利金(曾导致 0.014×2476≈34U)
|
||
from lib.hedge_plan.hedge_plan_live_positions_lib import resolve_option_index_px
|
||
|
||
with patch(
|
||
"lib.exchange.okx_public_http_lib.fetch_index_px",
|
||
side_effect=RuntimeError("no net"),
|
||
):
|
||
px = resolve_option_index_px(
|
||
{},
|
||
"BTC",
|
||
ex_opt=None,
|
||
cross_asset=True,
|
||
perp_fallback=2476.0,
|
||
)
|
||
self.assertIsNone(px)
|
||
|
||
with patch(
|
||
"lib.exchange.okx_public_http_lib.fetch_index_px",
|
||
return_value=91000.0,
|
||
):
|
||
px = resolve_option_index_px(
|
||
{},
|
||
"BTC",
|
||
cross_asset=True,
|
||
perp_fallback=2476.0,
|
||
)
|
||
self.assertAlmostEqual(px, 91000.0)
|
||
|
||
# 同币允许永续价兜底
|
||
with patch(
|
||
"lib.exchange.okx_public_http_lib.fetch_index_px",
|
||
side_effect=RuntimeError("no net"),
|
||
):
|
||
px = resolve_option_index_px(
|
||
{},
|
||
"ETH",
|
||
cross_asset=False,
|
||
perp_fallback=2476.0,
|
||
)
|
||
self.assertAlmostEqual(px, 2476.0)
|
||
|
||
def test_settle_coin_option_returns_coin_units(self):
|
||
from lib.hedge_plan.hedge_plan_settle_lib import settle_option_leg_at_spot
|
||
|
||
# BTC put ITM: K=90000 S=80000 → 币内在 (90000-80000)/80000=0.125; 1张×0.01=0.00125币
|
||
# 权利金 0.01 币 → 净利 0.00125-0.01 = -0.00875 币(非 ×8万的美元差)
|
||
pnl = settle_option_leg_at_spot(
|
||
{
|
||
"inst_id": "BTC-USD-261018-90000-P",
|
||
"opt_type": "P",
|
||
"strike": 90000,
|
||
"size": 1,
|
||
"ct_mult": 0.01,
|
||
"premium": 0.01,
|
||
},
|
||
80000.0,
|
||
)
|
||
self.assertAlmostEqual(pnl, 0.00125 - 0.01, places=6)
|
||
|
||
def test_size_from_premium_preview_needs_dual_index(self):
|
||
# 无双指数 → 同币路径;有双指数 → 跨币张数
|
||
same = size_from_premium(
|
||
premium_budget=1000,
|
||
ask=0.01,
|
||
ct_mult=0.01,
|
||
ratio=2,
|
||
contract_size=0.01,
|
||
)
|
||
cross = size_from_premium(
|
||
premium_budget=1000,
|
||
ask=0.01,
|
||
ct_mult=0.01,
|
||
ratio=2,
|
||
contract_size=0.01,
|
||
option_index=100_000,
|
||
perp_index=2_500,
|
||
)
|
||
self.assertTrue(same["ok"] and cross["ok"])
|
||
self.assertGreater(float(cross["contracts"]), float(same["contracts"]))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|