dekun
75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
"""中控改委托同步与条件单按角色去重。"""
|
|
|
|
from lib.hub.hub_order_sync_lib import (
|
|
cond_order_role,
|
|
dedupe_conditional_orders_by_role,
|
|
exchange_tpsl_from_cond_orders,
|
|
sync_active_monitor_tpsl_prices,
|
|
)
|
|
from lib.hub.hub_symbol_lib import symbols_match
|
|
|
|
|
|
def test_cond_order_role():
|
|
assert cond_order_role({"label": "止损 84.1"}) == "sl"
|
|
assert cond_order_role({"label": "止盈 76"}) == "tp"
|
|
assert cond_order_role({"label": "市价 买入"}) is None
|
|
|
|
|
|
def test_dedupe_conditional_orders_by_role_keeps_one_sl():
|
|
rows = [
|
|
{"label": "止盈 76", "trigger_price": 76},
|
|
{"label": "止损", "trigger_price": 84.1},
|
|
{"label": "止损 84.1", "trigger_price": 84.1, "id": "x:sl"},
|
|
]
|
|
out = dedupe_conditional_orders_by_role(rows)
|
|
assert len(out) == 2
|
|
sl_rows = [r for r in out if cond_order_role(r) == "sl"]
|
|
assert len(sl_rows) == 1
|
|
assert sl_rows[0]["label"] == "止损 84.1"
|
|
|
|
|
|
def test_exchange_tpsl_from_cond_orders():
|
|
cond = [
|
|
{"label": "止损 84.1", "trigger_price": 84.1, "algo_id": "1"},
|
|
{"label": "止盈 76", "trigger_price": 76, "algo_id": "1"},
|
|
]
|
|
et = exchange_tpsl_from_cond_orders(cond)
|
|
assert et["sl"]["trigger_price"] == 84.1
|
|
assert et["tp"]["trigger_price"] == 76
|
|
|
|
|
|
def test_sync_active_monitor_tpsl_prices_updates_matching_order():
|
|
class Row(dict):
|
|
def __getitem__(self, key):
|
|
return dict.get(self, key)
|
|
|
|
class Conn:
|
|
def __init__(self):
|
|
self.rows = [
|
|
Row(
|
|
id=5,
|
|
symbol="SOL/USDT:USDT",
|
|
exchange_symbol="SOL/USDT:USDT",
|
|
direction="short",
|
|
)
|
|
]
|
|
self.updates = []
|
|
|
|
def execute(self, sql, params=None):
|
|
if "SELECT" in sql:
|
|
return self
|
|
if "UPDATE" in sql and params:
|
|
self.updates.append(params)
|
|
return self
|
|
|
|
def fetchall(self):
|
|
return self.rows
|
|
|
|
conn = Conn()
|
|
out = sync_active_monitor_tpsl_prices(
|
|
conn, "SOL/USDT:USDT", "short", 85.0, 75.0, symbols_match=symbols_match
|
|
)
|
|
assert out["ok"] is True
|
|
assert out["updated"] == 1
|
|
assert conn.updates == [(85.0, 75.0, 5)]
|