dekun
142 lines
5.5 KiB
Python
142 lines
5.5 KiB
Python
"""中控监控区今日统计聚合."""
|
|
import unittest
|
|
|
|
from lib.hub.hub_monitor_totals_lib import aggregate_monitor_board_totals
|
|
from lib.hub.hub_trades_lib import summarize_trades
|
|
|
|
|
|
class TestHubMonitorTotals(unittest.TestCase):
|
|
def test_aggregate_monitor_board_totals_sums_rows(self):
|
|
rows = [
|
|
{
|
|
"day_stats": {
|
|
"ok": True,
|
|
"opens_today": 2,
|
|
"trade_stats": {
|
|
"closed_count": 1,
|
|
"win_count": 1,
|
|
"loss_count": 0,
|
|
"win_pnl_u": 5.5,
|
|
"loss_pnl_u": 0,
|
|
},
|
|
},
|
|
"agent": {"positions": [{"contracts": 1}], "total_unrealized_pnl": 1.2},
|
|
},
|
|
{
|
|
"day_stats": {
|
|
"ok": True,
|
|
"opens_today": 1,
|
|
"trade_stats": {
|
|
"closed_count": 2,
|
|
"win_count": 0,
|
|
"loss_count": 2,
|
|
"win_pnl_u": 0,
|
|
"loss_pnl_u": -3.0,
|
|
},
|
|
},
|
|
"agent": {"positions": [], "total_unrealized_pnl": 0},
|
|
},
|
|
]
|
|
out = aggregate_monitor_board_totals(rows, trading_day="2026-07-04", reset_hour=8)
|
|
self.assertEqual(out["open_count"], 3)
|
|
self.assertEqual(out["closed_count"], 3)
|
|
self.assertEqual(out["win_count"], 1)
|
|
self.assertEqual(out["loss_count"], 2)
|
|
self.assertEqual(out["win_pnl_u"], 5.5)
|
|
self.assertEqual(out["loss_pnl_u"], -3.0)
|
|
self.assertEqual(out["open_position_count"], 1)
|
|
self.assertEqual(out["float_pnl_u"], 1.2)
|
|
|
|
def test_aggregate_monitor_board_totals_includes_options(self):
|
|
rows = [
|
|
{
|
|
"capabilities": ["options"],
|
|
"options": {
|
|
"ok": True,
|
|
"enabled": True,
|
|
"positions": [{"inst_id": "X"}, {"inst_id": "Y"}],
|
|
"upl_total_usdc": 1.5,
|
|
},
|
|
"agent": {"positions": [], "total_unrealized_pnl": 0},
|
|
}
|
|
]
|
|
out = aggregate_monitor_board_totals(rows, trading_day="2026-07-04", reset_hour=8)
|
|
self.assertEqual(out["options_open_position_count"], 2)
|
|
self.assertEqual(out["open_position_count"], 2)
|
|
self.assertEqual(out["options_float_pnl_u"], 1.5)
|
|
self.assertEqual(out["float_pnl_u"], 1.5)
|
|
|
|
def test_aggregate_monitor_board_totals_coin_options_to_usdt(self):
|
|
rows = [
|
|
{
|
|
"capabilities": ["options"],
|
|
"options": {
|
|
"ok": True,
|
|
"enabled": True,
|
|
"options_margin_mode": "coin",
|
|
"options_index_px": 2000.0,
|
|
"positions": [
|
|
{
|
|
"inst_id": "ETH-USD-260822-2250-C",
|
|
"margin_mode": "coin",
|
|
"premium_ccy": "ETH",
|
|
"upl": 0.002,
|
|
"idx_px": 2000.0,
|
|
}
|
|
],
|
|
"upl_total_usdc": 0.002,
|
|
},
|
|
"agent": {"positions": [], "total_unrealized_pnl": 10.0},
|
|
}
|
|
]
|
|
out = aggregate_monitor_board_totals(rows, trading_day="2026-08-20", reset_hour=8)
|
|
self.assertEqual(out["options_float_pnl_u"], 4.0)
|
|
self.assertEqual(out["float_pnl_u"], 14.0)
|
|
|
|
def test_aggregate_excludes_option_like_agent_positions(self):
|
|
"""子代理误把期权当永续上报时:不算进持仓数,浮盈只用期权 snap."""
|
|
rows = [
|
|
{
|
|
"capabilities": ["options"],
|
|
"options": {
|
|
"ok": True,
|
|
"enabled": True,
|
|
"positions": [{"inst_id": "ETH-USD_UM-260806-1875-C"}],
|
|
"upl_total_usdc": -0.4,
|
|
},
|
|
"agent": {
|
|
"positions": [
|
|
{
|
|
"symbol": "ETH/USD:USD-260806-1875-C",
|
|
"contracts": 66,
|
|
"unrealized_pnl": 27.6,
|
|
},
|
|
{
|
|
"symbol": "BTC/USDT:USDT",
|
|
"contracts": 1,
|
|
"unrealized_pnl": -4.66,
|
|
},
|
|
],
|
|
"total_unrealized_pnl": 22.94,
|
|
},
|
|
}
|
|
]
|
|
out = aggregate_monitor_board_totals(rows, trading_day="2026-08-05", reset_hour=8)
|
|
self.assertEqual(out["open_position_count"], 2)
|
|
self.assertEqual(out["options_open_position_count"], 1)
|
|
self.assertEqual(out["options_float_pnl_u"], -0.4)
|
|
self.assertEqual(out["float_pnl_u"], round(-4.66 + (-0.4), 4))
|
|
|
|
def test_summarize_trades_win_loss_amounts(self):
|
|
stats = summarize_trades(
|
|
[{"pnl_amount": 2.5}, {"pnl_amount": -1.0}, {"pnl_amount": 0}]
|
|
)
|
|
self.assertEqual(stats["win_count"], 1)
|
|
self.assertEqual(stats["loss_count"], 1)
|
|
self.assertEqual(stats["win_pnl_u"], 2.5)
|
|
self.assertEqual(stats["loss_pnl_u"], -1.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|