dekun
97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
"""权益曲线采样:现金+现货币+权利金+浮盈."""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import unittest
|
|
|
|
|
|
class EquitySampleTests(unittest.TestCase):
|
|
def test_equity_fills_open_cliff(self):
|
|
from lib.instance.instance_cash_sample_lib import equity_usdt_from_snapshot
|
|
|
|
# 开仓后:USDT 所剩无几,BTC 缓冲 + 权利金仓位
|
|
bal = {
|
|
"trading_usdt": 50.0,
|
|
"funding_usdt": 0.0,
|
|
"trading_btc": 0.001,
|
|
"funding_btc": 0.0,
|
|
}
|
|
positions = [
|
|
{
|
|
"premium_paid": 0.01,
|
|
"premium_ccy": "BTC",
|
|
"open_index_px": 84000.0,
|
|
"idx_px": 84000.0,
|
|
"net_pnl": 0.0,
|
|
"mark": 0.01,
|
|
}
|
|
]
|
|
# cash 50 + spot 84 + premium 840 = 974; 浮盈约 0
|
|
eq = equity_usdt_from_snapshot(bal, positions, btc_px=84000.0, eth_px=None)
|
|
self.assertIsNotNone(eq)
|
|
self.assertGreater(float(eq), 900.0)
|
|
# 纯现金会只有 50
|
|
from lib.instance.instance_cash_sample_lib import cash_usdt_from_balances
|
|
|
|
self.assertAlmostEqual(float(cash_usdt_from_balances(bal)), 50.0, places=2)
|
|
|
|
def test_query_prefers_equity_column(self):
|
|
from lib.instance.instance_cash_sample_lib import (
|
|
init_cash_sample_table,
|
|
query_cash_series_with_db,
|
|
record_cash_sample,
|
|
)
|
|
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
init_cash_sample_table(conn)
|
|
record_cash_sample(conn, 100.0, equity_usdt=500.0, ts_ms=1_700_000_000_000)
|
|
record_cash_sample(conn, 120.0, equity_usdt=None, ts_ms=1_700_000_060_000)
|
|
# 旧点无 equity 列时回退 cash:手动插一条仅 cash
|
|
conn.execute(
|
|
"INSERT INTO instance_cash_samples(ts_ms, cash_usdt) VALUES (?, ?)",
|
|
(1_700_000_120_000, 130.0),
|
|
)
|
|
conn.commit()
|
|
out = query_cash_series_with_db(conn, range_key="all")
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out["metric"], "equity")
|
|
pts = out["points"]
|
|
self.assertEqual(len(pts), 3)
|
|
self.assertAlmostEqual(pts[0]["v"], 500.0, places=2)
|
|
self.assertAlmostEqual(pts[1]["v"], 120.0, places=2) # equity 默认写成 cash
|
|
self.assertAlmostEqual(pts[2]["v"], 130.0, places=2) # 历史无 equity → cash
|
|
conn.close()
|
|
|
|
def test_sample_once_dict_snapshot(self):
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from lib.instance.instance_cash_sample_lib import (
|
|
query_cash_series_with_db,
|
|
sample_once,
|
|
)
|
|
|
|
tmp = Path(tempfile.mkdtemp()) / "eq.db"
|
|
|
|
def get_db():
|
|
c = sqlite3.connect(str(tmp))
|
|
c.row_factory = sqlite3.Row
|
|
return c
|
|
|
|
ok = sample_once(
|
|
get_db,
|
|
lambda: {"cash_usdt": 200.0, "equity_usdt": 950.0},
|
|
)
|
|
self.assertTrue(ok)
|
|
conn = get_db()
|
|
try:
|
|
out = query_cash_series_with_db(conn, range_key="all")
|
|
self.assertAlmostEqual(float(out["latest"]), 950.0, places=2)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|