"""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