dekun
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""instance_live_pnl_lib 单元测试。"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from lib.instance.instance_live_pnl_lib import (
|
|
position_row_contracts,
|
|
resolve_instance_unrealized_pnl,
|
|
sum_unrealized_pnl_from_metrics,
|
|
sum_unrealized_pnl_from_positions,
|
|
)
|
|
|
|
|
|
class TestInstanceLivePnlLib(unittest.TestCase):
|
|
def test_position_row_contracts_from_info(self):
|
|
pos = {"contracts": 0, "info": {"positionAmt": "12.5"}}
|
|
self.assertAlmostEqual(position_row_contracts(pos), 12.5)
|
|
|
|
def test_sum_from_positions_binance_style(self):
|
|
positions = [
|
|
{"unrealizedPnl": -0.14, "info": {"positionAmt": "100"}},
|
|
]
|
|
self.assertEqual(sum_unrealized_pnl_from_positions(positions), -0.14)
|
|
|
|
def test_sum_from_metrics_fallback(self):
|
|
rows = [{"exchange_symbol": "DOGE/USDT:USDT", "symbol": "DOGE/USDT", "direction": "long"}]
|
|
|
|
def _metrics(ex_sym, direction):
|
|
self.assertEqual(direction, "long")
|
|
return {"unrealized_pnl": -0.14}
|
|
|
|
self.assertEqual(sum_unrealized_pnl_from_metrics(rows, _metrics), -0.14)
|
|
|
|
def test_resolve_prefers_bulk_positions(self):
|
|
def _fetch():
|
|
return [{"unrealizedPnl": 1.2, "contracts": 1}]
|
|
|
|
def _metrics(_ex, _d):
|
|
raise AssertionError("should not call metrics when bulk works")
|
|
|
|
total = resolve_instance_unrealized_pnl(_fetch, [], _metrics)
|
|
self.assertEqual(total, 1.2)
|
|
|
|
def test_resolve_falls_back_to_metrics(self):
|
|
def _fetch():
|
|
raise RuntimeError("api down")
|
|
|
|
rows = [{"exchange_symbol": "BTC/USDT:USDT", "symbol": "BTC/USDT", "direction": "short"}]
|
|
|
|
def _metrics(_ex, direction):
|
|
return {"unrealized_pnl": -2.5} if direction == "short" else None
|
|
|
|
total = resolve_instance_unrealized_pnl(_fetch, rows, _metrics)
|
|
self.assertEqual(total, -2.5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|