Add HEDGE_PO_MODE instance modes with shared sim/live path.
Lock one mode per instance/subaccount, cross-asset BTC option/ETH perp sizing, combo-U exit, and ship docs/tests/audit. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
@@ -0,0 +1,177 @@
|
||||
"""永期实例模式 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"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user