"""期权交割账单镜像 + 期期开仓账本。""" from __future__ import annotations from types import SimpleNamespace from app.live.option_settle import ( OptionSettlement, fetch_option_settlement, settlement_to_fill, ) def test_settlement_to_fill_zero_when_not_found() -> None: st = OptionSettlement(found=False) assert settlement_to_fill(st, qty_eth=2.0) == (0.0, 0.0, 0.0) def test_okx_settlement_from_exercise_bills() -> None: class _C: def get_option_settlement_bills(self, inst_id, *, begin_ms, end_ms=None): return [ { "subType": "170", "type": "3", "ts": str(begin_ms + 1000), "balChg": "50.0", "fee": "-0.5", "ccy": "USDT", } ] st = fetch_option_settlement( _C(), exchange="okx", option_inst_id="ETH-USDT-260808-2000-C", qty_eth=2.0, begin_ms=1_700_000_000_000, ) assert st.found assert abs(st.cash - 50.0) < 1e-9 px, fee, notional = settlement_to_fill(st, qty_eth=2.0) assert fee == 0.5 assert abs(notional - 50.5) < 1e-9 assert abs(px - 25.25) < 1e-9 def test_bn_settlement_from_exercise_record() -> None: class _C: def get_option_exercise_records(self, symbol, *, begin_ms, end_ms=None): return [ { "symbol": symbol, "amount": "40", "fee": "0.2", "currency": "USDT", "quantity": "2", } ] st = fetch_option_settlement( _C(), exchange="binance", option_inst_id="ETH-260808-2000-C", qty_eth=2.0, begin_ms=1, ) assert st.found assert abs(st.cash - 39.8) < 1e-9 px, fee, notional = settlement_to_fill(st, qty_eth=2.0) assert abs(notional - 40.0) < 1e-9 assert abs(px - 20.0) < 1e-9 assert abs(fee - 0.2) < 1e-9 def test_okx_otm_expiry_bill_found_zero_cash() -> None: class _C: def get_option_settlement_bills(self, *_a, **_k): return [ { "subType": "172", "type": "3", "ts": "1700000001000", "balChg": "0", "fee": "0", "ccy": "USDT", } ] st = fetch_option_settlement( _C(), exchange="okx", option_inst_id="ETH-OPT", qty_eth=1.0, begin_ms=1_700_000_000_000, ) assert st.found assert st.cash == 0.0 assert settlement_to_fill(st, qty_eth=1.0)[0] == 0.0 def test_oo_open_applies_ledger_cash(monkeypatch, tmp_path) -> None: monkeypatch.setenv("MODE", "LIVE") from app.config import get_settings get_settings.cache_clear() from app.live.executor import OkxLiveExecutor from app.models.db import Database db = Database(tmp_path / "oo.db") ex = OkxLiveExecutor(db) monkeypatch.setattr(ex, "_guard_live", lambda: None) monkeypatch.setattr( "app.live.executor.claim_open_slot", lambda _db: (True, "ok") ) monkeypatch.setattr( "app.live.executor.assert_safe_to_open_live", lambda _e: (True, "ok") ) monkeypatch.setattr( "app.live.executor.stamp_opening_intent", lambda *_a, **_k: None ) monkeypatch.setattr( "app.live.executor.release_open_slot_if_opening", lambda *_a, **_k: None ) monkeypatch.setattr(ex, "_ct_mult", lambda *_a, **_k: 0.01) class _C: def place_market(self, *, inst_id, side, sz, **_k): return SimpleNamespace( avg_px=10.0 if "C" in inst_id or "call" in inst_id.lower() or inst_id.endswith("-C") or "CALL" in inst_id else 8.0, fee=0.1, sz=float(sz), ) # simpler fixed fills fills = [ SimpleNamespace(avg_px=10.0, fee=0.1, sz=100.0), SimpleNamespace(avg_px=8.0, fee=0.05, sz=100.0), ] def place(**_k): return fills.pop(0) monkeypatch.setattr(ex, "_client", lambda: SimpleNamespace(place_market=place)) monkeypatch.setattr( ex.ledger, "get_setting_float", lambda k, d=0: 1.0 if "qty" in k else d, ) before = float(ex.ledger.snapshot()["available"]) r = ex.open_oo_group( group_id="G-oo1", call_inst_id="ETH-CALL", put_inst_id="ETH-PUT", call_strike=2000.0, put_strike=1900.0, entry_index_px=1950.0, expiry_ymd="260810", ) assert r.ok, r.detail after = float(ex.ledger.snapshot()["available"]) # call 10*1 +0.1 + put 8*1 +0.05 = 18.15 assert before - after > 18.0 db.close() get_settings.cache_clear()