"""§12.6 策略实例做实：绑定/发起即建档/同步/删除保护/读时聚合/全景。""" import json import os import sqlite3 from fastapi.testclient import TestClient from sanguo_api import instance_store from sanguo_api.app import create_app from sanguo_api.auth import create_token, set_jwt_config from sanguo_api.routes_paper import set_db_path as set_paper_db from sanguo_api.routes_live import set_db_path as set_live_db def _setup(tmp_path, monkeypatch): # 档案 JSON 隔离到临时文件（否则读写仓库 data/ 互相污染） monkeypatch.setattr(instance_store, "_STORE_PATH", str(tmp_path / "inst.json")) set_jwt_config(secret="t", expire_minutes=60) pdb = os.path.join(str(tmp_path), "paper.db") ldb = os.path.join(str(tmp_path), "live.db") app = create_app(db_path=pdb) set_paper_db(pdb) set_live_db(ldb) return TestClient(app), create_token("admin") def _auth(token): return {"Authorization": f"Bearer {token}"} _LIVE_BODY = { "mode": "live", "symbols": ["600000"], "strategies": [{"name": "DoubleMaStrategy", "symbol": "600000", "params": {"fast_window": 5}, "match_session": "next_open"}], "start": "2024-01-01", "end": "", "initial_capital": 1_000_000, } def test_auto_create_instance_on_launch(tmp_path, monkeypatch): """发起即建档：不带 instance_id 创建模拟盘 → 自动建档案并绑定。""" c, token = _setup(tmp_path, monkeypatch) r = c.post("/api/v1/paper/create", json=_LIVE_BODY, headers=_auth(token)) assert r.status_code == 200 insts = c.get("/api/v1/strategy/instances").json()["instances"] assert len(insts) == 1 iid = insts[0]["id"] assert insts[0]["params"] == {"fast_window": 5} # 账户行带上 instance_id row = sqlite3.connect(set_paper_db.__self__ if False else _paper_db_path(c)).execute( "SELECT instance_id FROM paper_accounts WHERE id=1").fetchone() assert row[0] == iid def _paper_db_path(c): from sanguo_api.routes_paper import _db_path return _db_path["path"] def test_bind_existing_instance(tmp_path, monkeypatch): """带合法 instance_id → 直接绑定，不新建档案。""" c, token = _setup(tmp_path, monkeypatch) iid = instance_store.create_instance({ "code_file": "double_ma.py", "name": "双均线·浦发", "type": "cta", "params": {"fast_window": 10}, "symbol_or_pool": "600000", "interval": "d", }) body = {**_LIVE_BODY, "instance_id": iid} r = c.post("/api/v1/paper/create", json=body, headers=_auth(token)) assert r.status_code == 200 # 发起时快照（D1）：账户参数=档案当时的参数（fast=10），即使 body 里另有参数 raw = sqlite3.connect(_paper_db_path(c)).execute( "SELECT strategies, instance_id FROM paper_accounts WHERE id=1").fetchone() assert raw[1] == iid assert json.loads(raw[0])[0]["params"] == {"fast_window": 10} def test_sync_params_to_running_accounts(tmp_path, monkeypatch): """D2/D3：档案改参 → 同步接口批量刷运行中模拟账户（实走+影子一起）。""" c, token = _setup(tmp_path, monkeypatch) iid = instance_store.create_instance({ "code_file": "", "name": "x", "type": "cta", "params": {"fast_window": 5}, "symbol_or_pool": "600000", "interval": "d", }) c.post("/api/v1/paper/create", json={**_LIVE_BODY, "instance_id": iid}, headers=_auth(token)) c.post("/api/v1/paper/create", json={ **_LIVE_BODY, "instance_id": iid, "mode": "shadow", }, headers=_auth(token)) # 档案改参 → 漂移出现 instance_store.update_instance(iid, {"params": {"fast_window": 20}}) enriched = c.get("/api/v1/strategy/instances/enriched").json()["instances"] me = [i for i in enriched if i["id"] == iid][0] assert me["drift"] is True # 同步 → 两个模拟账户参数齐刷 r = c.post(f"/api/v1/paper/sync/{iid}", headers=_auth(token)) assert r.status_code == 200 assert r.json()["synced"] == 2 rows = sqlite3.connect(_paper_db_path(c)).execute( "SELECT strategies FROM paper_accounts WHERE instance_id=?", (iid,)).fetchall() for (raw,) in rows: assert json.loads(raw)[0]["params"] == {"fast_window": 20} def test_delete_protection_with_running_account(tmp_path, monkeypatch): """D5：档案下有运行中账户 → 删除被拦（409）。""" c, token = _setup(tmp_path, monkeypatch) r = c.post("/api/v1/paper/create", json=_LIVE_BODY, headers=_auth(token)) assert r.status_code == 200 iid = c.get("/api/v1/strategy/instances").json()["instances"][0]["id"] resp = c.delete(f"/api/v1/strategy/instances/{iid}", headers=_auth(token)) assert resp.status_code == 409 assert "运行中" in resp.json()["detail"] # 无运行账户的档案可删 iid2 = instance_store.create_instance({"name": "free", "params": {}}) assert c.delete(f"/api/v1/strategy/instances/{iid2}", headers=_auth(token)).status_code == 200 def test_enriched_and_overview(tmp_path, monkeypatch): """读时聚合（四格实走格 running）+ P1 全景（runs + 合并持仓归因）。""" c, token = _setup(tmp_path, monkeypatch) r = c.post("/api/v1/paper/create", json=_LIVE_BODY, headers=_auth(token)) iid = c.get("/api/v1/strategy/instances").json()["instances"][0]["id"] enriched = c.get("/api/v1/strategy/instances/enriched").json()["instances"] me = [i for i in enriched if i["id"] == iid][0] assert me["status"]["paper_live"] == "running" assert len(me["running_accounts"]) == 1 # 写持仓快照 → 全景归因出现 db = _paper_db_path(c) from sanguo_trader.persistence import init_db, save_positions init_db(db) save_positions(db, 1, "live", {"600000": {"volume": 100, "frozen": 0, "avg_price": 10.5}}, date="2026-08-15") ov = c.get(f"/api/v1/strategy/instances/{iid}/overview").json() assert ov["instance"]["name"] assert len(ov["runs"]) == 1 and ov["runs"][0]["kind"] == "paper" assert ov["positions"] and ov["positions"][0]["symbol"] == "600000" assert ov["positions"][0]["volume"] == 100 def test_live_create_auto_instance(tmp_path, monkeypatch): """实盘发起同样绑档案（发起即建档）。""" c, token = _setup(tmp_path, monkeypatch) r = c.post("/api/v1/live/create", json={ "name": "live1", "account": "8886000519", "strategy_class": "AShareDoubleMaStrategy", "strategy_name": "dm", "setting": {"fast_window": 5}, }, headers=_auth(token)) assert r.status_code == 200 from sanguo_live.persistence import list_accounts ldb = os.path.join(str(tmp_path), "live.db") accs = list_accounts(ldb) assert accs and accs[0].get("instance_id") insts = c.get("/api/v1/strategy/instances").json()["instances"] assert insts[0]["id"] == accs[0]["instance_id"] def test_update_instance_run_event_kinds(tmp_path, monkeypatch): """事件型回写只认 backtest/replay；paper_live/live 拒绝（读时聚合负责）。""" monkeypatch.setattr(instance_store, "_STORE_PATH", str(tmp_path / "i.json")) iid = instance_store.create_instance({"name": "a", "params": {}}) assert instance_store.update_instance_run(iid, "backtest", "done", 0.15) assert instance_store.update_instance_run(iid, "replay", "failed") assert not instance_store.update_instance_run(iid, "paper_live", "running") inst = instance_store.get_instance(iid)["instance"] assert inst["status"]["backtest"] == "done" assert inst["status"]["replay"] == "failed" assert inst["last_return"] == 0.15 # 不存在的档案静默丢弃 assert not instance_store.update_instance_run(999, "backtest", "done")