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>
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
"""claim_open_slot / release_open_slot 单元测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.live.reconcile import claim_open_slot, release_open_slot_if_opening
|
|
from app.sim.matcher import BLOCKING_STATUSES, Matcher
|
|
|
|
|
|
def test_blocking_statuses_include_opening() -> None:
|
|
assert "opening" in BLOCKING_STATUSES
|
|
|
|
|
|
def test_claim_open_slot_from_flat(tmp_path) -> None:
|
|
from app.models.db import Database
|
|
|
|
db = Database(tmp_path / "claim.db")
|
|
m = Matcher(db)
|
|
assert m.has_open_position() is False
|
|
|
|
ok, msg = claim_open_slot(db)
|
|
assert ok is True
|
|
assert msg == "ok"
|
|
assert m.position_status() == "opening"
|
|
assert m.has_open_position() is True
|
|
|
|
ok2, _ = claim_open_slot(db)
|
|
assert ok2 is False
|
|
|
|
release_open_slot_if_opening(db)
|
|
assert m.position_status() == "flat"
|
|
assert m.has_open_position() is False
|
|
|
|
ok3, _ = claim_open_slot(db)
|
|
assert ok3 is True
|
|
release_open_slot_if_opening(db)
|
|
db.close()
|
|
|
|
|
|
def test_claim_rejects_blocking_states(tmp_path) -> None:
|
|
from app.models.db import Database
|
|
|
|
db = Database(tmp_path / "block.db")
|
|
for st in ("open", "half_open", "option_closed_perp_pending"):
|
|
with db._lock:
|
|
db._conn.execute(
|
|
"UPDATE positions SET status=?, group_id=? WHERE id=1",
|
|
(st, "G-test"),
|
|
)
|
|
db._conn.commit()
|
|
ok, msg = claim_open_slot(db)
|
|
assert ok is False
|
|
assert st in msg
|
|
with db._lock:
|
|
db._conn.execute(
|
|
"UPDATE positions SET status='flat', group_id=NULL WHERE id=1"
|
|
)
|
|
db._conn.commit()
|
|
db.close()
|
|
|
|
|
|
def test_release_only_when_opening(tmp_path) -> None:
|
|
from app.models.db import Database
|
|
|
|
db = Database(tmp_path / "rel.db")
|
|
with db._lock:
|
|
db._conn.execute("UPDATE positions SET status='open' WHERE id=1")
|
|
db._conn.commit()
|
|
release_open_slot_if_opening(db)
|
|
row = db.fetchone("SELECT status FROM positions WHERE id=1")
|
|
assert row is not None
|
|
assert row["status"] == "open"
|
|
db.close()
|