dekun
b09d1b0886
Semi sizing uses market ask with semi units; Plan panel previews option/perp qty under the form. Co-authored-by: Cursor <cursoragent@cursor.com>
322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""以损定仓纯函数测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from app.strategy.risk_sizing import (
|
||
BASE_EXIT_USDT,
|
||
BASE_OPTION_ETH,
|
||
BASE_PERP_ETH,
|
||
compute_k,
|
||
floor_k_1dp,
|
||
normalize_risk_leverage_basis,
|
||
resolve_sizing_option_ask,
|
||
unit_cost,
|
||
)
|
||
|
||
|
||
def test_floor_k_1dp() -> None:
|
||
assert floor_k_1dp(1.29) == 1.2
|
||
assert floor_k_1dp(0.19) == 0.1
|
||
assert floor_k_1dp(0.09) == 0.0
|
||
assert floor_k_1dp(2.0) == 2.0
|
||
|
||
|
||
def test_compute_k_scales_1_2_15() -> None:
|
||
# I=2000, A=20, fee=0.0005 → unit = 2*20 + 2000*0.0005*3 = 40 + 3 = 43
|
||
# budget=43 → k=1.0
|
||
r = compute_k(budget=43.0, index_px=2000.0, option_ask=20.0, fee_rate=0.0005)
|
||
assert r.ok
|
||
assert r.k == 1.0
|
||
assert r.perp_qty_eth == BASE_PERP_ETH
|
||
assert r.option_qty_eth == BASE_OPTION_ETH
|
||
assert r.net_profit_target == BASE_EXIT_USDT
|
||
assert r.max_loss is not None and r.max_loss <= 43.0 + 1e-6
|
||
assert r.budget == 43.0
|
||
|
||
|
||
def test_compute_k_custom_units() -> None:
|
||
# option_unit=4 → premium unit = 20*4=80; fee=3; cost=83; budget=83 → k=1
|
||
r = compute_k(
|
||
budget=83.0,
|
||
index_px=2000.0,
|
||
option_ask=20.0,
|
||
fee_rate=0.0005,
|
||
perp_unit=0.5,
|
||
option_unit=4.0,
|
||
exit_unit=30.0,
|
||
)
|
||
assert r.ok
|
||
assert r.k == 1.0
|
||
assert r.perp_qty_eth == 0.5
|
||
assert r.option_qty_eth == 4.0
|
||
assert r.net_profit_target == 30.0
|
||
|
||
|
||
def test_money_rounds_2dp() -> None:
|
||
r = compute_k(budget=50.123456, index_px=1900.0, option_ask=18.5, fee_rate=0.0005)
|
||
assert r.ok
|
||
assert r.budget == round(50.123456, 2)
|
||
assert r.max_loss is not None
|
||
assert abs(r.max_loss * 100 - round(r.max_loss * 100)) < 1e-9
|
||
|
||
|
||
def test_compute_k_never_exceeds_budget() -> None:
|
||
r = compute_k(budget=50.0, index_px=1900.0, option_ask=18.5, fee_rate=0.0005)
|
||
assert r.ok
|
||
assert r.k is not None
|
||
assert abs(r.k * 10 - round(r.k * 10)) < 1e-9 # 一位小数
|
||
assert r.max_loss is not None and r.max_loss <= 50.0 + 1e-6
|
||
assert r.perp_qty_eth == round(1.0 * r.k, 4)
|
||
assert r.option_qty_eth == round(2.0 * r.k, 4)
|
||
assert r.net_profit_target == round(15.0 * r.k, 4)
|
||
|
||
|
||
def test_compute_k_too_small() -> None:
|
||
# unit≈43, budget=2 → k_raw≪0.1
|
||
r = compute_k(budget=2.0, index_px=2000.0, option_ask=20.0, fee_rate=0.0005)
|
||
assert not r.ok
|
||
assert "最小仓" in r.detail or "k=" in r.detail
|
||
|
||
|
||
def test_unit_cost() -> None:
|
||
assert abs(unit_cost(index_px=2000, option_ask=20, fee_rate=0.0005) - 43.0) < 1e-9
|
||
|
||
|
||
def test_normalize_risk_leverage_basis() -> None:
|
||
assert normalize_risk_leverage_basis("actual") == "actual"
|
||
assert normalize_risk_leverage_basis("selection") == "selection"
|
||
assert normalize_risk_leverage_basis("min_option_leverage") == "selection"
|
||
assert normalize_risk_leverage_basis("weird", default="selection") == "selection"
|
||
|
||
|
||
def test_resolve_sizing_ask_selection_vs_actual() -> None:
|
||
# 指数 2000、选约杠杆 100 → 隐含卖一 20；实际卖一更便宜 10
|
||
sel_ask, basis = resolve_sizing_option_ask(
|
||
index_px=2000.0,
|
||
option_ask=10.0,
|
||
leverage_basis="selection",
|
||
min_option_leverage=100.0,
|
||
)
|
||
assert basis == "selection"
|
||
assert abs(sel_ask - 20.0) < 1e-9
|
||
|
||
act_ask, basis2 = resolve_sizing_option_ask(
|
||
index_px=2000.0,
|
||
option_ask=10.0,
|
||
leverage_basis="actual",
|
||
min_option_leverage=100.0,
|
||
)
|
||
assert basis2 == "actual"
|
||
assert abs(act_ask - 10.0) < 1e-9
|
||
|
||
|
||
def test_selection_basis_yields_smaller_k_when_ask_cheap() -> None:
|
||
# 预算 43：选约隐含 ask=20 → k=1；若用实际 ask=10 → 单位成本更小 → k 更大
|
||
r_sel = compute_k(budget=43.0, index_px=2000.0, option_ask=20.0, fee_rate=0.0005)
|
||
r_act = compute_k(budget=43.0, index_px=2000.0, option_ask=10.0, fee_rate=0.0005)
|
||
assert r_sel.ok and r_act.ok
|
||
assert r_sel.k == 1.0
|
||
assert r_act.k is not None and r_act.k > r_sel.k
|
||
|
||
|
||
def test_compute_risk_sizing_unit_overrides(tmp_path, monkeypatch) -> None:
|
||
monkeypatch.setenv("MODE", "SIM")
|
||
from app.models.db import Database
|
||
from app.strategy.risk_sizing import compute_risk_sizing
|
||
|
||
db = Database(tmp_path / "risk_units.db")
|
||
db.set_setting("sizing_mode", "risk_based")
|
||
db.set_setting("risk_loss_mode", "absolute")
|
||
db.set_setting("risk_loss_usdt", "83")
|
||
db.set_setting("fee_rate", "0.0005")
|
||
db.set_setting("risk_leverage_basis", "selection")
|
||
db.set_setting("min_option_leverage", "100")
|
||
db.set_setting("risk_perp_unit", "1")
|
||
db.set_setting("risk_option_unit", "2")
|
||
db.set_setting("risk_exit_unit", "15")
|
||
|
||
# 覆盖为单位 0.5:4、出场 5，并强制实际卖一（ask=20 → cost=80+3=83 → k=1）
|
||
r = compute_risk_sizing(
|
||
index_px=2000.0,
|
||
option_ask=20.0,
|
||
db=db,
|
||
perp_unit=0.5,
|
||
option_unit=4.0,
|
||
exit_unit=5.0,
|
||
leverage_basis="actual",
|
||
)
|
||
assert r.ok
|
||
assert r.leverage_basis == "actual"
|
||
assert r.k == 1.0
|
||
assert r.perp_qty_eth == 0.5
|
||
assert r.option_qty_eth == 4.0
|
||
assert r.net_profit_target == 5.0
|
||
db.close()
|
||
|
||
|
||
def test_compute_risk_sizing_respects_basis(tmp_path, monkeypatch) -> None:
|
||
monkeypatch.setenv("MODE", "SIM")
|
||
from app.models.db import Database
|
||
from app.strategy.risk_sizing import compute_risk_sizing
|
||
|
||
db = Database(tmp_path / "risk_basis.db")
|
||
db.set_setting("sizing_mode", "risk_based")
|
||
db.set_setting("risk_loss_mode", "absolute")
|
||
db.set_setting("risk_loss_usdt", "43")
|
||
db.set_setting("fee_rate", "0.0005")
|
||
db.set_setting("min_option_leverage", "100")
|
||
db.set_setting("risk_perp_unit", "1")
|
||
db.set_setting("risk_option_unit", "2")
|
||
db.set_setting("risk_exit_unit", "15")
|
||
|
||
db.set_setting("risk_leverage_basis", "selection")
|
||
r1 = compute_risk_sizing(index_px=2000.0, option_ask=10.0, db=db)
|
||
assert r1.ok
|
||
assert r1.leverage_basis == "selection"
|
||
assert r1.k == 1.0
|
||
assert r1.actual_option_ask == 10.0
|
||
assert r1.option_ask == 20.0
|
||
|
||
db.set_setting("risk_leverage_basis", "actual")
|
||
r2 = compute_risk_sizing(index_px=2000.0, option_ask=10.0, db=db)
|
||
assert r2.ok
|
||
assert r2.leverage_basis == "actual"
|
||
assert r2.k is not None and r2.k > 1.0
|
||
assert r2.option_ask == 10.0
|
||
db.close()
|
||
|
||
|
||
def test_consecutive_loss_days_and_martingale(tmp_path, monkeypatch) -> None:
|
||
monkeypatch.setenv("MODE", "SIM")
|
||
from datetime import datetime, timezone
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from app.models.db import Database
|
||
from app.strategy.risk_sizing import consecutive_loss_days, resolve_martingale
|
||
|
||
db = Database(tmp_path / "mg.db")
|
||
sh = ZoneInfo("Asia/Shanghai")
|
||
|
||
def day_ms(ymd: str, hour: int = 16) -> int:
|
||
dt = datetime.strptime(ymd, "%Y-%m-%d").replace(
|
||
hour=hour, tzinfo=sh
|
||
)
|
||
return int(dt.astimezone(timezone.utc).timestamp() * 1000)
|
||
|
||
# 插入：盈利日打断后连亏 3 天（有成交日序列，跳过无成交日）
|
||
rows = [
|
||
("g1", day_ms("2026-07-28"), 10.0),
|
||
("g2", day_ms("2026-07-29"), -5.0),
|
||
("g3", day_ms("2026-07-30"), -3.0),
|
||
("g4", day_ms("2026-07-31"), -1.0),
|
||
]
|
||
for gid, ms, pnl in rows:
|
||
db.execute(
|
||
"""INSERT INTO groups(
|
||
group_id, status, realized_pnl, close_at_ms, open_at_ms
|
||
) VALUES(?,?,?,?,?)""",
|
||
(gid, "closed", pnl, ms, ms - 3600_000),
|
||
)
|
||
|
||
assert consecutive_loss_days(db) == 3
|
||
|
||
db.set_setting("sizing_mode", "risk_based")
|
||
db.set_setting("risk_loss_mode", "percent")
|
||
db.set_setting("risk_loss_pct", "2")
|
||
db.set_setting("martingale_enabled", "true")
|
||
db.set_setting("martingale_start_after_loss_days", "2")
|
||
db.set_setting("martingale_max_doubles", "3")
|
||
|
||
mg = resolve_martingale(db, base_pct=2.0)
|
||
assert mg["eligible"] is True
|
||
assert mg["loss_days"] == 3
|
||
# 连亏3天、start=2 → doubles = min(3-2+1, 3) = 2 → 2%*4 = 8%
|
||
assert mg["doubles"] == 2
|
||
assert abs(float(mg["effective_pct"]) - 8.0) < 1e-9
|
||
|
||
db.set_setting("risk_loss_pct", "3.1")
|
||
mg2 = resolve_martingale(db, base_pct=3.1)
|
||
assert mg2["eligible"] is False
|
||
assert mg2["doubles"] == 0
|
||
assert abs(float(mg2["effective_pct"]) - 3.1) < 1e-9
|
||
db.close()
|
||
|
||
|
||
def test_martingale_doubles_capped(tmp_path, monkeypatch) -> None:
|
||
monkeypatch.setenv("MODE", "SIM")
|
||
from datetime import datetime, timezone
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from app.models.db import Database
|
||
from app.strategy.risk_sizing import resolve_martingale
|
||
|
||
db = Database(tmp_path / "mg_cap.db")
|
||
sh = ZoneInfo("Asia/Shanghai")
|
||
|
||
def day_ms(ymd: str) -> int:
|
||
dt = datetime.strptime(ymd, "%Y-%m-%d").replace(hour=12, tzinfo=sh)
|
||
return int(dt.astimezone(timezone.utc).timestamp() * 1000)
|
||
|
||
for i, ymd in enumerate(
|
||
["2026-07-26", "2026-07-27", "2026-07-28", "2026-07-29", "2026-07-30"]
|
||
):
|
||
db.execute(
|
||
"""INSERT INTO groups(
|
||
group_id, status, realized_pnl, close_at_ms, open_at_ms
|
||
) VALUES(?,?,?,?,?)""",
|
||
(f"c{i}", "closed", -1.0, day_ms(ymd), day_ms(ymd) - 1000),
|
||
)
|
||
|
||
db.set_setting("sizing_mode", "risk_based")
|
||
db.set_setting("risk_loss_mode", "percent")
|
||
db.set_setting("martingale_enabled", "true")
|
||
db.set_setting("martingale_start_after_loss_days", "2")
|
||
db.set_setting("martingale_max_doubles", "3")
|
||
mg = resolve_martingale(db, base_pct=2.0)
|
||
# 连亏5、start2 → raw=4，cap=3 → 2%*8=16%
|
||
assert mg["doubles"] == 3
|
||
assert abs(float(mg["effective_pct"]) - 16.0) < 1e-9
|
||
db.close()
|
||
|
||
|
||
def test_expiry_settle_counts_as_loss_even_if_profit(tmp_path, monkeypatch) -> None:
|
||
"""到期结算小盈利也按亏损计入倍投连亏日。"""
|
||
monkeypatch.setenv("MODE", "SIM")
|
||
from datetime import datetime, timezone
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from app.models.db import Database
|
||
from app.strategy.risk_sizing import consecutive_loss_days
|
||
|
||
db = Database(tmp_path / "mg_exp.db")
|
||
sh = ZoneInfo("Asia/Shanghai")
|
||
|
||
def day_ms(ymd: str) -> int:
|
||
dt = datetime.strptime(ymd, "%Y-%m-%d").replace(hour=16, tzinfo=sh)
|
||
return int(dt.astimezone(timezone.utc).timestamp() * 1000)
|
||
|
||
# 达标盈利打断；随后两天均为到期小盈利 → 仍计连亏 2
|
||
rows = [
|
||
("e0", day_ms("2026-07-28"), 20.0, "fixed_usdt"),
|
||
("e1", day_ms("2026-07-29"), 3.5, "expiry"),
|
||
("e2", day_ms("2026-07-30"), 1.2, "expiry"),
|
||
]
|
||
for gid, ms, pnl, reason in rows:
|
||
db.execute(
|
||
"""INSERT INTO groups(
|
||
group_id, status, realized_pnl, close_at_ms, open_at_ms, close_reason
|
||
) VALUES(?,?,?,?,?,?)""",
|
||
(gid, "closed", pnl, ms, ms - 3600_000, reason),
|
||
)
|
||
assert consecutive_loss_days(db) == 2
|
||
|
||
# 再来一天达标盈利 → 连亏清零
|
||
db.execute(
|
||
"""INSERT INTO groups(
|
||
group_id, status, realized_pnl, close_at_ms, open_at_ms, close_reason
|
||
) VALUES(?,?,?,?,?,?)""",
|
||
("e3", "closed", 15.0, day_ms("2026-07-31"), day_ms("2026-07-31") - 1000, "fixed_usdt"),
|
||
)
|
||
assert consecutive_loss_days(db) == 0
|
||
db.close()
|