93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from trex_core.adapters.broker.ownership import (
|
|
OwnershipAwareBroker,
|
|
StrategyOwnershipStore,
|
|
)
|
|
from trex_core.application.broker_port import (
|
|
BalanceItem,
|
|
BrokerBalanceError,
|
|
CancelOutcome,
|
|
CancelState,
|
|
FillStatus,
|
|
OrderAck,
|
|
)
|
|
|
|
|
|
class FakeBroker:
|
|
def __init__(self, balance, fill=None):
|
|
self.balance = balance
|
|
self.fill = fill or FillStatus(state="accepted")
|
|
|
|
def name(self):
|
|
return "fake"
|
|
|
|
def submit_order(self, req):
|
|
return OrderAck(accepted=True, broker_order_id="1")
|
|
|
|
def poll_fill(self, broker_order_id, *, side, symbol):
|
|
return self.fill
|
|
|
|
def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"):
|
|
return CancelOutcome(CancelState.REQUESTED)
|
|
|
|
def get_balance(self):
|
|
return self.balance
|
|
|
|
def get_current_price(self, symbol, market="J"):
|
|
return Decimal("1")
|
|
|
|
|
|
def test_manual_balance_is_hidden_until_trex_fill(tmp_path):
|
|
store = StrategyOwnershipStore(tmp_path / "ownership.json")
|
|
broker = OwnershipAwareBroker(
|
|
FakeBroker([BalanceItem("005930", 19), BalanceItem("010120", 5)]),
|
|
store,
|
|
)
|
|
assert broker.get_balance() == []
|
|
|
|
|
|
def test_confirmed_fill_is_persisted_once_and_manual_remainder_stays_hidden(tmp_path):
|
|
store = StrategyOwnershipStore(tmp_path / "ownership.json")
|
|
fill = FillStatus(
|
|
state="filled",
|
|
filled_qty=2,
|
|
execution_id="kis:1:2",
|
|
)
|
|
delegate = FakeBroker([BalanceItem("005930", 21)], fill)
|
|
broker = OwnershipAwareBroker(delegate, store)
|
|
|
|
broker.poll_fill("1", side="buy", symbol="005930")
|
|
broker.poll_fill("1", side="buy", symbol="005930")
|
|
|
|
assert broker.get_balance() == [BalanceItem("005930", 2)]
|
|
assert StrategyOwnershipStore(store.path).snapshot()[0] == {"005930": 2}
|
|
|
|
|
|
def test_strategy_quantity_above_raw_broker_total_fails_closed(tmp_path):
|
|
store = StrategyOwnershipStore(tmp_path / "ownership.json")
|
|
store.apply_fill(
|
|
execution_id="buy-1",
|
|
symbol="005930",
|
|
side="buy",
|
|
quantity=4,
|
|
)
|
|
broker = OwnershipAwareBroker(FakeBroker([BalanceItem("005930", 3)]), store)
|
|
|
|
with pytest.raises(BrokerBalanceError) as exc:
|
|
broker.get_balance()
|
|
assert exc.value.code == "ownership_mismatch"
|
|
|
|
|
|
def test_sell_cannot_reduce_below_strategy_owned_zero(tmp_path):
|
|
store = StrategyOwnershipStore(tmp_path / "ownership.json")
|
|
with pytest.raises(BrokerBalanceError):
|
|
store.apply_fill(
|
|
execution_id="sell-1",
|
|
symbol="005930",
|
|
side="sell",
|
|
quantity=1,
|
|
)
|