dekun
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
"""中控统计曲线:各机累计已实现盈亏叠加合计."""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
class TestStatsRealizedPnlCurve(unittest.TestCase):
|
|
def test_build_stats_curve_sums_machines(self):
|
|
from lib.stats_agg_lib import build_stats_curve
|
|
|
|
t0 = 1_700_000_040_000 # 已对齐到分钟桶
|
|
t1 = t0 + 60_000
|
|
t2 = t0 + 120_000
|
|
fake_rows = [
|
|
{
|
|
"id": "a",
|
|
"name": "A",
|
|
"online": True,
|
|
"error": "",
|
|
"points": [{"t": t0, "v": 10.0}, {"t": t2, "v": 15.0}],
|
|
"trade_count": 2,
|
|
"latest": 15.0,
|
|
"start_ms": t0,
|
|
"end_ms": t2,
|
|
},
|
|
{
|
|
"id": "b",
|
|
"name": "B",
|
|
"online": True,
|
|
"error": "",
|
|
"points": [{"t": t1, "v": 5.0}],
|
|
"trade_count": 1,
|
|
"latest": 5.0,
|
|
"start_ms": t1,
|
|
"end_ms": t2,
|
|
},
|
|
]
|
|
with patch("lib.stats_agg_lib.pull_all_pnl_curves", return_value=fake_rows):
|
|
out = build_stats_curve("all")
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out["metric"], "realized_pnl")
|
|
self.assertEqual(len(out["series"]), 2)
|
|
totals = {p["t"]: p["v"] for p in out["total"]}
|
|
# t0: only A=10; t1: A=10 + B=5; t2: A=15 + B=5
|
|
self.assertAlmostEqual(totals[t0], 10.0)
|
|
self.assertAlmostEqual(totals[t1], 15.0)
|
|
self.assertAlmostEqual(totals[t2], 20.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|