"""残留期权：权利金回升达标后中途平。""" from __future__ import annotations from types import SimpleNamespace from app.models.db import Database from app.sim.matcher import Matcher def _seed_residual( db: Database, *, group_id: str = "G-res", initial_premium: float = 100.0, qty: float = 2.0, entry_px: float = 50.0, ) -> None: now = 1_700_000_000_000 with db._lock: db._conn.execute( """INSERT INTO groups( group_id, status, bias, option_side, perp_side, option_inst_id, strike, expiry_ymd, initial_premium, open_at_ms, close_at_ms, close_reason, realized_pnl, fees, slip_cost ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "option_residual", "test", "call", "short", "ETH-USD_UM-260801-2000-C", 2000.0, "260801", initial_premium, now - 10_000, now - 5_000, "target_perp_only", 10.0, 1.0, 0.0, ), ) db._conn.execute( """INSERT INTO residual_options( group_id, option_inst_id, option_side, option_qty_eth, option_qty_contracts, option_entry_px, strike, expiry_ymd, expiry_ms, entry_index_px, initial_premium, status, created_at_ms, note ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "ETH-USD_UM-260801-2000-C", "call", qty, 200.0, entry_px, 2000.0, "260801", now + 86_400_000, 1900.0, initial_premium, "pending", now - 5_000, "test residual", ), ) db._conn.commit() def test_residual_premium_below_threshold_skips(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") db = Database(tmp_path / "below.db") db.set_setting("residual_min_premium_pct", "20") _seed_residual(db, initial_premium=100.0, qty=2.0) m = Matcher(db) # bid=5 → premium=10 < 20 oq = SimpleNamespace(bid=5.0, ask=5.5, bid_sz=10_000.0, mark_px=5.0) monkeypatch.setattr(m, "_quote_held_option", lambda _id: oq) monkeypatch.setattr(m, "_close_spot_px", lambda _snap: 1900.0) assert m.try_close_one_residual(m.list_residual_options()[0]) is None row = db.fetchone( "SELECT status FROM residual_options WHERE group_id=?", ("G-res",) ) assert row is not None and row["status"] == "pending" db.close() def test_residual_premium_above_threshold_closes(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") db = Database(tmp_path / "above.db") db.set_setting("residual_min_premium_pct", "20") _seed_residual(db, initial_premium=100.0, qty=2.0, entry_px=50.0) m = Matcher(db) # bid=15 → premium=30 >= 20 oq = SimpleNamespace(bid=15.0, ask=15.5, bid_sz=10_000.0, mark_px=15.0) monkeypatch.setattr(m, "_quote_held_option", lambda _id: oq) monkeypatch.setattr(m, "_close_spot_px", lambda _snap: 1900.0) monkeypatch.setattr(m, "_ct_mult", lambda _id: 0.01) out = m.try_close_one_residual(m.list_residual_options()[0]) assert out is not None assert out["reason"] == "residual_premium_close" row = db.fetchone( "SELECT status, settle_px FROM residual_options WHERE group_id=?", ("G-res",) ) assert row is not None and row["status"] == "settled" g = db.fetchone("SELECT status FROM groups WHERE group_id=?", ("G-res",)) assert g is not None and g["status"] == "closed" db.close() def test_residual_liquidity_fail_skips(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") db = Database(tmp_path / "liq.db") db.set_setting("residual_min_premium_pct", "20") _seed_residual(db, initial_premium=100.0, qty=2.0) m = Matcher(db) # premium ok but depth tiny oq = SimpleNamespace(bid=15.0, ask=15.5, bid_sz=1.0, mark_px=15.0) monkeypatch.setattr(m, "_quote_held_option", lambda _id: oq) monkeypatch.setattr(m, "_close_spot_px", lambda _snap: 1900.0) monkeypatch.setattr(m, "_ct_mult", lambda _id: 0.01) assert m.try_close_one_residual(m.list_residual_options()[0]) is None row = db.fetchone( "SELECT status FROM residual_options WHERE group_id=?", ("G-res",) ) assert row is not None and row["status"] == "pending" db.close() def test_residual_recheck_bid_drop_skips(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") db = Database(tmp_path / "recheck.db") db.set_setting("residual_min_premium_pct", "20") _seed_residual(db, initial_premium=100.0, qty=2.0) m = Matcher(db) good = SimpleNamespace(bid=15.0, ask=15.5, bid_sz=10_000.0, mark_px=15.0) bad = SimpleNamespace(bid=5.0, ask=5.5, bid_sz=10_000.0, mark_px=5.0) quotes = iter([good, bad]) monkeypatch.setattr(m, "_quote_held_option", lambda _id: next(quotes)) monkeypatch.setattr(m, "_close_spot_px", lambda _snap: 1900.0) monkeypatch.setattr(m, "_ct_mult", lambda _id: 0.01) assert m.try_close_one_residual(m.list_residual_options()[0]) is None row = db.fetchone( "SELECT status FROM residual_options WHERE group_id=?", ("G-res",) ) assert row is not None and row["status"] == "pending" db.close() def test_residual_book_pending_guard(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") db = Database(tmp_path / "guard.db") _seed_residual(db, initial_premium=100.0, qty=2.0) m = Matcher(db) row = m.list_residual_options()[0] first = m._book_residual_market_close( row, fill_px=15.0, fee=0.01, notional=30.0, slip=0.0, now_ms=1_700_000_100_000, note="first", filled_contracts=200.0, remaining_contracts=0.0, ) assert first is not None and first.get("fully_done") is True second = m._book_residual_market_close( row, fill_px=15.0, fee=0.01, notional=30.0, slip=0.0, now_ms=1_700_000_200_000, note="second", filled_contracts=200.0, remaining_contracts=0.0, ) assert second is None db.close() def test_settings_exposes_residual_min_premium_pct(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") from app.api import settings as settings_api from app.models.db import set_db d = Database(tmp_path / "set.db") set_db(d) try: d.set_setting("residual_min_premium_pct", "35") payload = settings_api._read_settings() assert float(payload["residual_min_premium_pct"]) == 35.0 finally: set_db(None) d.close()