dekun
126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
"""对冲启动:再拉卖一 + 对冲专用预算缓冲重算张数."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest import mock
|
|
from unittest.mock import MagicMock
|
|
|
|
from lib.hedge_plan.hedge_plan_orders_lib import (
|
|
_hedge_budget_buffer,
|
|
execute_options_options_start,
|
|
refresh_oo_sizing_before_start,
|
|
refresh_po_option_quote_before_start,
|
|
)
|
|
|
|
|
|
class TestHedgeStartRefresh(unittest.TestCase):
|
|
def test_hedge_budget_buffer_independent(self):
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{
|
|
"HEDGE_PLAN_BUDGET_BUFFER": "0.9",
|
|
"OKX_OPTIONS_BUDGET_BUFFER": "0.5",
|
|
},
|
|
):
|
|
self.assertAlmostEqual(_hedge_budget_buffer(None), 0.9)
|
|
self.assertAlmostEqual(_hedge_budget_buffer({"budget_buffer": 0.88}), 0.88)
|
|
|
|
def test_refresh_oo_resizes_from_fresh_ask(self):
|
|
def quote(_ex, inst_id):
|
|
if inst_id == "A":
|
|
return {
|
|
"ok": True,
|
|
"ask": 20,
|
|
"ask_sz": 100,
|
|
"ct_mult": 0.01,
|
|
"meta": {"optType": "C"},
|
|
}
|
|
return {
|
|
"ok": True,
|
|
"ask": 20,
|
|
"ask_sz": 100,
|
|
"ct_mult": 0.01,
|
|
"meta": {"optType": "P"},
|
|
}
|
|
|
|
cfg = {
|
|
"exchange_options": object(),
|
|
"quote_option_contract": quote,
|
|
"trade_budget_usdc": 10,
|
|
"budget_buffer": 0.95,
|
|
}
|
|
body = {
|
|
"oo_sheets_mode": "same_sheets",
|
|
"leg_a": {"inst_id": "A", "sheets": 1, "opt_type": "C"},
|
|
"leg_b": {"inst_id": "B", "sheets": 1, "opt_type": "P"},
|
|
}
|
|
# unit cost = 20*0.01=0.2 each → pair 0.4; budget min(100*0.95,10)=10 → n=25
|
|
with mock.patch(
|
|
"lib.exchange.okx_options_lib.fetch_options_trading_usdc",
|
|
return_value=100.0,
|
|
):
|
|
out = refresh_oo_sizing_before_start(cfg, body)
|
|
self.assertTrue(out.get("ok"), out)
|
|
self.assertEqual(body["leg_a"]["sheets"], 25)
|
|
self.assertEqual(body["leg_b"]["sheets"], 25)
|
|
self.assertEqual(body["leg_a"]["ask"], 20)
|
|
self.assertEqual(out["sheets_a"], 25)
|
|
|
|
def test_execute_oo_start_uses_refreshed_sheets(self):
|
|
quote = MagicMock(
|
|
side_effect=lambda _ex, inst_id: {
|
|
"ok": True,
|
|
"ask": 10,
|
|
"ask_sz": 50,
|
|
"ct_mult": 0.01,
|
|
"tick_sz": "0.1",
|
|
"meta": {"optType": "C" if inst_id == "A" else "P"},
|
|
}
|
|
)
|
|
cfg = {
|
|
"exchange_options": object(),
|
|
"quote_option_contract": quote,
|
|
"place_option_limit_order": MagicMock(),
|
|
"td_mode_for_option_buy": lambda x: "isolated",
|
|
"trade_budget_usdc": 4,
|
|
"budget_buffer": 0.95,
|
|
}
|
|
body = {
|
|
"oo_sheets_mode": "same_sheets",
|
|
"leg_a": {"inst_id": "A", "sheets": 1, "opt_type": "C"},
|
|
"leg_b": {"inst_id": "B", "sheets": 1, "opt_type": "P"},
|
|
}
|
|
# cost 0.1+0.1=0.2; budget 4 → 20 sheets each
|
|
with mock.patch(
|
|
"lib.exchange.okx_options_lib.fetch_options_trading_usdc",
|
|
return_value=100.0,
|
|
):
|
|
out = execute_options_options_start(cfg, body, dry_run=True)
|
|
self.assertTrue(out["ok"], out)
|
|
self.assertEqual(body["leg_a"]["sheets"], 20)
|
|
self.assertEqual(out["results"][0]["sheets"], 20)
|
|
self.assertIn("refresh", out)
|
|
self.assertTrue(out["refresh"]["ok"])
|
|
|
|
def test_refresh_po_keeps_sheets(self):
|
|
quote = MagicMock(
|
|
return_value={
|
|
"ok": True,
|
|
"ask": 12.5,
|
|
"ask_sz": 8,
|
|
"ct_mult": 0.01,
|
|
"meta": {"optType": "P"},
|
|
}
|
|
)
|
|
cfg = {"exchange_options": object(), "quote_option_contract": quote}
|
|
body = {"opt_inst_id": "X", "sheets": 3}
|
|
out = refresh_po_option_quote_before_start(cfg, body)
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(body["ask"], 12.5)
|
|
self.assertEqual(body["sheets"], 3)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|