dekun
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""粉尘阈值 / 卖回封顶 / 买入增量逻辑相关单测."""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from lib.options.options_spot_bridge_lib import (
|
|
display_sellable_coin,
|
|
is_sellable_spot_coin,
|
|
spot_min_sell_sz,
|
|
)
|
|
|
|
|
|
class TestSpotDust(unittest.TestCase):
|
|
def test_min_sz_fallback(self):
|
|
self.assertAlmostEqual(spot_min_sell_sz(None, "ETH"), 0.0001)
|
|
self.assertAlmostEqual(spot_min_sell_sz(None, "BTC"), 0.00001)
|
|
|
|
def test_dust_not_sellable(self):
|
|
self.assertFalse(is_sellable_spot_coin(None, "ETH", 0.000004))
|
|
self.assertFalse(is_sellable_spot_coin(None, "BTC", 0.000002))
|
|
self.assertEqual(display_sellable_coin(None, "ETH", 0.000004), 0.0)
|
|
|
|
def test_above_min_sellable(self):
|
|
self.assertTrue(is_sellable_spot_coin(None, "ETH", 0.0001))
|
|
self.assertTrue(is_sellable_spot_coin(None, "BTC", 0.00001))
|
|
self.assertGreater(display_sellable_coin(None, "ETH", 0.001), 0)
|
|
|
|
|
|
class TestSellCapLogic(unittest.TestCase):
|
|
def test_excess_cap_math(self):
|
|
bought, keep = 0.01, 0.004
|
|
excess = max(0.0, bought - keep)
|
|
self.assertAlmostEqual(excess, 0.006)
|
|
sell_amt = min(0.02, excess)
|
|
self.assertAlmostEqual(sell_amt, 0.006)
|
|
|
|
|
|
class TestMaybeSellAfterCloseGuard(unittest.TestCase):
|
|
"""平仓后卖回:持仓查询失败/空响应绝不卖币."""
|
|
|
|
def test_none_positions_no_sell(self):
|
|
from lib.options.options_coin_open_lib import maybe_sell_spot_after_close
|
|
|
|
called = {"sell": 0}
|
|
|
|
def _fetch(_ex):
|
|
return None
|
|
|
|
cfg = {
|
|
"fetch_option_positions": _fetch,
|
|
"get_db": lambda: (_ for _ in ()).throw(AssertionError("不应开库卖币")),
|
|
}
|
|
out = maybe_sell_spot_after_close(
|
|
cfg, object(), inst_id="ETH-USD-260701-2500-C"
|
|
)
|
|
self.assertIsInstance(out, dict)
|
|
self.assertFalse(out.get("ok"))
|
|
self.assertTrue(out.get("skipped"))
|
|
self.assertIn("空响应", str(out.get("msg") or ""))
|
|
self.assertEqual(called["sell"], 0)
|
|
|
|
def test_fetch_raises_no_sell(self):
|
|
from lib.options.options_coin_open_lib import maybe_sell_spot_after_close
|
|
|
|
def _fetch(_ex):
|
|
raise RuntimeError("network down")
|
|
|
|
cfg = {
|
|
"fetch_option_positions": _fetch,
|
|
"get_db": lambda: (_ for _ in ()).throw(AssertionError("不应开库卖币")),
|
|
}
|
|
out = maybe_sell_spot_after_close(
|
|
cfg, object(), inst_id="ETH-USD-260701-2500-C"
|
|
)
|
|
self.assertIsInstance(out, dict)
|
|
self.assertFalse(out.get("ok"))
|
|
self.assertTrue(out.get("skipped"))
|
|
self.assertIn("持仓查询失败", str(out.get("msg") or ""))
|
|
|
|
def test_still_open_skips_sell(self):
|
|
from lib.options.options_coin_open_lib import maybe_sell_spot_after_close
|
|
|
|
inst = "ETH-USD-260701-2500-C"
|
|
|
|
def _fetch(_ex):
|
|
return [{"instId": inst, "pos": "1"}]
|
|
|
|
cfg = {
|
|
"fetch_option_positions": _fetch,
|
|
"get_db": lambda: (_ for _ in ()).throw(AssertionError("有仓不应卖币")),
|
|
}
|
|
out = maybe_sell_spot_after_close(cfg, object(), inst_id=inst)
|
|
self.assertTrue(out.get("ok"))
|
|
self.assertTrue(out.get("skipped"))
|
|
self.assertIn("仍有持仓", str(out.get("msg") or ""))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|