230 lines
8.3 KiB
Python
230 lines
8.3 KiB
Python
"""Tests for B3 预算硬限制(spec §B3):create/update 校验 + /live/budget-info。
|
|
|
|
直调路由函数(与 test_portfolio_live.py 同风格,免 auth)。
|
|
核心断言:剩余=现金−Σ其他实例预算;超限/快照过期 → 400 fail-closed。
|
|
"""
|
|
import sqlite3
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from sanguo_api import routes_live as rl
|
|
from sanguo_live import persistence as lp
|
|
|
|
|
|
@pytest.fixture()
|
|
def live_db(tmp_path):
|
|
db = str(tmp_path / "live.db")
|
|
rl.set_db_path(db)
|
|
return db
|
|
|
|
|
|
def _snapshot(db, account="66639661", cash=1_000_000.0, age_min=0):
|
|
lp.upsert_account_snapshot(
|
|
db, account, cash=cash, market_value=0.0, total=cash, positions=[])
|
|
if age_min:
|
|
old = (datetime.now(timezone.utc)
|
|
- timedelta(minutes=age_min)).isoformat()
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute(
|
|
"UPDATE qmt_account_snapshot SET updated_at=? WHERE account=?",
|
|
(old, account))
|
|
|
|
|
|
def _create(db, capital=500_000.0, account="66639661"):
|
|
req = rl.LiveCreateRequest(
|
|
account=account, strategy_name="p1", strategy_type="portfolio",
|
|
strategy_class="all_weather", initial_capital=capital)
|
|
return rl.create_live(req)["account_id"]
|
|
|
|
|
|
# ---------------- budget-info ----------------
|
|
|
|
def test_budget_info_basic(live_db):
|
|
_snapshot(live_db, cash=1_000_000)
|
|
_create(live_db, capital=300_000) # 已分配 30 万
|
|
st = rl.get_budget_info("66639661")
|
|
assert st["fresh"] is True
|
|
assert st["account_cash"] == 1_000_000
|
|
assert st["allocated"] == 300_000
|
|
assert st["remaining"] == 700_000
|
|
|
|
|
|
def test_budget_info_includes_stopped(live_db):
|
|
"""停止实例持仓仍占资金 → Σ 不分状态。"""
|
|
_snapshot(live_db, cash=1_000_000)
|
|
aid = _create(live_db, capital=300_000)
|
|
lp.update_account_status(live_db, aid, "stopped")
|
|
assert rl.get_budget_info("66639661")["allocated"] == 300_000
|
|
|
|
|
|
def test_budget_info_stale_fail_closed(live_db):
|
|
_snapshot(live_db, cash=1_000_000, age_min=30)
|
|
st = rl.get_budget_info("66639661")
|
|
assert st["fresh"] is False
|
|
assert st["account_cash"] is None
|
|
assert st["remaining"] is None
|
|
|
|
|
|
def test_budget_info_missing_snapshot(live_db):
|
|
st = rl.get_budget_info("NO_SUCH")
|
|
assert st["fresh"] is False
|
|
assert st["remaining"] is None
|
|
|
|
|
|
# ---------------- create 校验 ----------------
|
|
|
|
def test_create_within_budget_ok(live_db):
|
|
_snapshot(live_db, cash=1_000_000)
|
|
_create(live_db, capital=300_000)
|
|
_create(live_db, capital=700_000) # 恰好用满
|
|
assert rl.get_budget_info("66639661")["remaining"] == 0
|
|
|
|
|
|
def test_create_over_budget_400_with_remaining(live_db):
|
|
_snapshot(live_db, cash=1_000_000)
|
|
_create(live_db, capital=300_000)
|
|
with pytest.raises(HTTPException) as e:
|
|
_create(live_db, capital=800_000)
|
|
assert e.value.status_code == 400
|
|
assert "700000" in e.value.detail # 报文带剩余数
|
|
assert "剩余可分配" in e.value.detail
|
|
|
|
|
|
def test_create_snapshot_missing_400(live_db):
|
|
"""快照不存在 → fail-closed(不猜数)。"""
|
|
with pytest.raises(HTTPException) as e:
|
|
_create(live_db)
|
|
assert e.value.status_code == 400
|
|
assert "快照不可用" in e.value.detail
|
|
|
|
|
|
def test_create_snapshot_stale_400(live_db):
|
|
_snapshot(live_db, cash=1_000_000, age_min=30)
|
|
with pytest.raises(HTTPException) as e:
|
|
_create(live_db)
|
|
assert e.value.status_code == 400
|
|
|
|
|
|
def test_create_budget_only_checks_same_qmt_account(live_db):
|
|
"""Σ 只算同 QMT 账号——别的账号的实例不挤本账户池。"""
|
|
_snapshot(live_db, account="A1", cash=500_000)
|
|
_snapshot(live_db, account="A2", cash=900_000)
|
|
_create(live_db, account="A1", capital=400_000)
|
|
# A1 剩 10 万,但 A2 独立
|
|
assert rl.get_budget_info("A2")["remaining"] == 900_000
|
|
|
|
|
|
# ---------------- update 校验 ----------------
|
|
|
|
def _make_editable(live_db):
|
|
_snapshot(live_db, cash=1_000_000) # create 需新鲜快照
|
|
return _create(live_db, capital=300_000)
|
|
|
|
|
|
def _age_snapshot(db, account="66639661"):
|
|
old = (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat()
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute(
|
|
"UPDATE qmt_account_snapshot SET updated_at=? WHERE account=?",
|
|
(old, account))
|
|
|
|
|
|
def test_update_budget_over_400_excludes_self(live_db):
|
|
"""改预算超限 → 400;Σ 不含自身旧值(300 万自身不重复计)。"""
|
|
aid = _make_editable(live_db) # 快照 100万
|
|
# 另一实例占了 50 万 → 剩余 50 万;自己 30 万不计入
|
|
_create(live_db, capital=500_000)
|
|
with pytest.raises(HTTPException) as e:
|
|
rl.update_live(aid, rl.LiveUpdateRequest(initial_capital=600_000))
|
|
assert e.value.status_code == 400
|
|
assert "500000" in e.value.detail
|
|
|
|
|
|
def test_update_budget_within_ok(live_db):
|
|
aid = _make_editable(live_db)
|
|
_create(live_db, capital=500_000)
|
|
rl.update_live(aid, rl.LiveUpdateRequest(initial_capital=500_000))
|
|
acc = lp.get_account(live_db, aid)
|
|
assert acc["initial_capital"] == 500_000
|
|
|
|
|
|
def test_update_rename_skips_budget_check(live_db):
|
|
"""仅改名不触发校验——快照过期也能改名(不是预算事件)。"""
|
|
aid = _make_editable(live_db)
|
|
_age_snapshot(live_db)
|
|
rl.update_live(aid, rl.LiveUpdateRequest(name="new_name"))
|
|
assert lp.get_account(live_db, aid)["name"] == "new_name"
|
|
|
|
|
|
def test_update_budget_change_blocked_when_stale(live_db):
|
|
"""快照过期时改预算 → 400(改预算必须看到新鲜现金)。"""
|
|
aid = _make_editable(live_db)
|
|
_age_snapshot(live_db)
|
|
with pytest.raises(HTTPException) as e:
|
|
rl.update_live(aid, rl.LiveUpdateRequest(initial_capital=100_000))
|
|
assert e.value.status_code == 400
|
|
assert "快照不可用" in e.value.detail
|
|
|
|
|
|
def test_delete_frees_budget(live_db):
|
|
"""删除实例释放预算(重建流程依赖)。"""
|
|
_snapshot(live_db, cash=1_000_000)
|
|
aid = _create(live_db, capital=300_000)
|
|
rl.delete_live(aid)
|
|
assert rl.get_budget_info("66639661")["remaining"] == 1_000_000
|
|
|
|
|
|
# ---------------- account-snapshot(B4 第三层数据源) ----------------
|
|
|
|
def test_account_snapshot_breakdown(live_db):
|
|
"""全局快照 + Σ实例分解:unattributed = 全账户市值 − Σ实例市值。"""
|
|
lp.upsert_account_snapshot(
|
|
live_db, "66639661", cash=400_000.0, market_value=600_000.0,
|
|
total=1_000_000.0,
|
|
positions=[{"symbol": "600036.SH", "volume": 1000, "can_use": 1000,
|
|
"avg_price": 38.0, "mv": 38_000.0}])
|
|
aid = _create(live_db, capital=300_000)
|
|
lp.save_balance(live_db, aid, "2026-08-19 15:00:00",
|
|
100_000.0, 200_000.0, 300_000.0)
|
|
r = rl.get_account_snapshot_route("66639661")
|
|
assert r["fresh"] is True
|
|
assert r["cash"] == 400_000.0
|
|
assert r["market_value"] == 600_000.0
|
|
assert len(r["positions"]) == 1
|
|
assert len(r["instances"]) == 1
|
|
assert r["instances"][0]["market_value"] == 200_000.0
|
|
assert r["instance_mv_total"] == 200_000.0
|
|
assert r["unattributed_mv"] == 400_000.0
|
|
|
|
|
|
def test_account_snapshot_no_snapshot_all_none(live_db):
|
|
"""快照缺失 → fresh=False 数值 None,但实例列表仍返回(有实例无快照)。"""
|
|
_snapshot(live_db, cash=1_000_000)
|
|
aid = _create(live_db, capital=300_000) # 播种快照后建,再删快照
|
|
with sqlite3.connect(live_db) as conn:
|
|
conn.execute("DELETE FROM qmt_account_snapshot")
|
|
r = rl.get_account_snapshot_route("66639661")
|
|
assert r["fresh"] is False
|
|
assert r["cash"] is None
|
|
assert r["unattributed_mv"] is None
|
|
assert len(r["instances"]) == 1
|
|
|
|
|
|
def test_account_snapshot_filters_other_accounts(live_db):
|
|
"""实例只归同 QMT 账号——别的账号实例不进分解。"""
|
|
_snapshot(live_db, cash=1_000_000)
|
|
_snapshot(live_db, account="OTHER", cash=500_000)
|
|
aid1 = _create(live_db, capital=100_000)
|
|
lp.save_balance(live_db, aid1, "d1", 0, 50_000.0, 50_000.0)
|
|
# OTHER 账号的实例
|
|
req = rl.LiveCreateRequest(
|
|
account="OTHER", strategy_name="p2", strategy_type="portfolio",
|
|
strategy_class="all_weather", initial_capital=100_000)
|
|
aid2 = rl.create_live(req)["account_id"]
|
|
lp.save_balance(live_db, aid2, "d1", 0, 70_000.0, 70_000.0)
|
|
r = rl.get_account_snapshot_route("66639661")
|
|
assert [i["id"] for i in r["instances"]] == [aid1]
|
|
assert r["instance_mv_total"] == 50_000.0
|