dekun
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
from packages.domain.aggregate import aggregate_leverage, summarize_values
|
|
from packages.domain.range import resolve_range
|
|
|
|
|
|
def test_summarize_values():
|
|
s = summarize_values([100, 200, 300, 400], min_leverage=200)
|
|
assert s["n"] == 4
|
|
assert s["mean"] == 250
|
|
assert s["median"] == 250
|
|
assert s["pct_ge_min"] == 0.75
|
|
|
|
|
|
def test_summarize_empty():
|
|
s = summarize_values([], min_leverage=100)
|
|
assert s["n"] == 0
|
|
assert s["mean"] is None
|
|
|
|
|
|
def test_resolve_range_day():
|
|
info = resolve_range("day", "2026-07-31")
|
|
assert info["start_ymd"] == "2026-07-31"
|
|
assert info["end_ymd"] == "2026-07-31"
|
|
assert info["days"] == 1
|
|
assert info["end_ms"] > info["start_ms"]
|
|
|
|
|
|
def test_resolve_range_week():
|
|
info = resolve_range("week", "2026-07-31")
|
|
assert info["start_ymd"] == "2026-07-25"
|
|
assert info["end_ymd"] == "2026-07-31"
|
|
assert info["days"] == 7
|
|
|
|
|
|
def test_resolve_range_month_rolling():
|
|
info = resolve_range("month", "2026-07-31", month_mode="rolling_30")
|
|
assert info["start_ymd"] == "2026-07-02"
|
|
assert info["days"] == 30
|
|
|
|
|
|
def test_resolve_range_month_calendar():
|
|
info = resolve_range("month", "2026-07-31", month_mode="calendar")
|
|
assert info["start_ymd"] == "2026-07-01"
|
|
assert info["end_ymd"] == "2026-07-31"
|
|
|
|
|
|
def test_aggregate_leverage_buckets():
|
|
# 2026-07-31 14:30 Asia/Shanghai
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
sh = ZoneInfo("Asia/Shanghai")
|
|
ts = int(datetime(2026, 7, 31, 14, 30, tzinfo=sh).timestamp() * 1000)
|
|
rows = [
|
|
{"ts_ms": ts, "side": "C", "leverage": 120},
|
|
{"ts_ms": ts, "side": "P", "leverage": 80},
|
|
{"ts_ms": ts, "side": "C", "leverage": 180},
|
|
]
|
|
buckets = aggregate_leverage(rows, bucket_minutes=60, min_leverage=100, side="C")
|
|
assert len(buckets) == 24
|
|
b14 = next(b for b in buckets if b["bucket_start_min"] == 14 * 60)
|
|
assert b14["n"] == 2
|
|
assert b14["mean"] == 150
|
|
assert b14["label"] == "14:00"
|
|
empty = next(b for b in buckets if b["bucket_start_min"] == 0)
|
|
assert empty["n"] == 0
|