dekun
0cf3756b09
Prevent duplicate opens by atomically claiming an opening slot, verifying exchange perp is flat before live orders, setting leverage from ledger, and preferring exchange position size when closing perps. Co-authored-by: Cursor <cursoragent@cursor.com>
71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
"""实盘防卡状态：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
|
|
assert "opening" 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()
|