dekun/eth_hedge_sim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
eth_hedge_sim/backend/tests/test_residual_premium_close.py
T

261 lines
9.1 KiB
Python
Raw Permalink Blame History

"""残留期权：权利金回升达标后中途平。"""
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_manual_close_skips_premium_ratio(tmp_path, monkeypatch) -> None:
monkeypatch.setenv("MODE", "SIM")
db = Database(tmp_path / "manual.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=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)
monkeypatch.setattr(m, "_ct_mult", lambda _id: 0.01)
assert m.try_close_one_residual(m.list_residual_options()[0]) is None
enriched = m.list_residual_options_enriched()
assert len(enriched) == 1
assert enriched[0]["liquidity_ok"] is True
assert enriched[0]["bid_px"] == 5.0
assert enriched[0]["bid_sz"] == 10_000.0
assert abs(float(enriched[0]["bid_sz_eth"]) - 100.0) < 1e-9 # 10000*0.01
assert abs(float(enriched[0]["current_premium"]) - 10.0) < 1e-9 # 5*2
assert abs(float(enriched[0]["recovery_pct"]) - 10.0) < 1e-9 # 10/100*100
assert float(enriched[0]["recovery_pct"]) < 20.0
r = m.close_residual_manual("G-res")
assert r.ok is True
row = db.fetchone(
"SELECT status FROM residual_options WHERE group_id=?", ("G-res",)
)
assert row is not None and row["status"] == "settled"
db.close()
def test_manual_close_liquidity_still_required(tmp_path, monkeypatch) -> None:
monkeypatch.setenv("MODE", "SIM")
db = Database(tmp_path / "manual_liq.db")
db.set_setting("residual_min_premium_pct", "20")
_seed_residual(db, initial_premium=100.0, qty=2.0)
m = Matcher(db)
oq = SimpleNamespace(bid=5.0, ask=5.5, bid_sz=1.0, mark_px=5.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)
r = m.close_residual_manual("G-res")
assert r.ok is False
assert "liquidity" in (r.detail or "")
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_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()
Reference in New Issue View Git Blame Copy Permalink