"""实盘防卡状态:half_open / option_closed_perp_pending。""" from __future__ import annotations from app.sim.ledger import Ledger from app.sim.matcher import BLOCKING_STATUSES, Matcher def test_blocking_statuses_include_repair_states() -> None: assert "half_open" in BLOCKING_STATUSES assert "option_closed_perp_pending" in BLOCKING_STATUSES assert "open" in BLOCKING_STATUSES def test_has_open_position_blocks_half_open(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") from app.models.db import Database db = Database(tmp_path / "t.db") m = Matcher(db) assert m.has_open_position() is False with db._lock: db._conn.execute( """UPDATE positions SET group_id=?, option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?, option_entry_px=?, status=? WHERE id=1""", ("G-test", "ETH-OPT", "call", 2.0, 200.0, 10.0, "half_open"), ) db._conn.commit() assert m.has_open_position() is True assert m.position_status() == "half_open" with db._lock: db._conn.execute( "UPDATE positions SET status='option_closed_perp_pending' WHERE id=1" ) db._conn.commit() assert m.has_open_position() is True with db._lock: db._conn.execute( """UPDATE positions SET group_id=NULL, option_inst_id=NULL, status='flat' WHERE id=1""" ) db._conn.commit() assert m.has_open_position() is False db.close() def test_ledger_allow_negative(tmp_path) -> None: from app.models.db import Database db = Database(tmp_path / "l.db") ledger = Ledger(db) # 掏空 snap = ledger.snapshot() ledger.apply_cash(-snap["available"], kind="drain", note="drain") try: ledger.apply_cash(-1.0, kind="fail", note="should fail") assert False, "expected RuntimeError" except RuntimeError: pass # LIVE 镜像允许透支 bal = ledger.apply_cash(-1.0, kind="live", note="ok", allow_negative=True) assert bal < 0 db.close()