策略库
-
策略代码 → 实例（参数变体）→ 回测 / 模拟盘 / 实盘 全生命周期
+
代码 → 实例档案 → 回测 / 模拟 / 实盘 · 全生命周期
-
-
-
-
-
-
-
- {{ g.file.name }}
- {{ typeLabel(g.file.type) }}
- {{ g.file.class_name }}
-
-
-
-
-
-
-
-
-
- 实例
- 参数
- 标的/池
- 最近收益
- 回测
- 回放
- 实走
- 实盘
- 操作
-
-
- {{ i.name }}
- {{ i.interval }} · {{ paramSummary(i.params) }}
- {{ i.symbol_or_pool }}
- {{ pct(i.last_return) }}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
暂无实例，编辑代码后可创建参数实例
-
+
+
+
+
+
+
+
{{ stats.files }}策略代码
+
+ {{ stats.insts }}实例档案
+
+
+ {{ stats.running }}在跑运行 · {{ inspection ? '返回' : '巡检' }}
+
+
{{ stats.drift }}参数漂移
+
+
+
+
+
+
+
+
+
+
+ 在跑运行
+ 跨全部档案
+ 实时巡检 · 点左侧统计条「在跑运行」返回
+
+
+
+ {{ kindLabel[row.run.kind] }}
+ {{ row.inst.name }}
+ {{ row.run.label }}
+ {{ pct(row.run.ret) }}
+
+
+
当前没有在跑的运行
+
+
+
+
+
+ {{ selectedMeta?.name || '—' }}
+
+ {{ selectedMeta?.type === 'portfolio' ? '组合' : 'CTA' }}
+
+ {{ selectedMeta?.class_name }}
+
+
+
+
+
+
+
+ {{ i.name }}
+ 参数已漂移
+ #{{ i.id }} · {{ i.interval }} · 更新 {{ i.updated_at }}
+
+
{{ paramSummary(i.params) }}
+
{{ i.symbol_or_pool }}
+
{{ pct(latestRet(i)) }}{{ retSub(i) }}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ 该代码还没有实例档案 — 点右上「+ 新实例」创建，或直接发起运行（发起即建档）
+
+
+
+
+
+
diff --git a/sanguo_api/instance_store.py b/sanguo_api/instance_store.py
index ddb9498..1d9a6d2 100644
--- a/sanguo_api/instance_store.py
+++ b/sanguo_api/instance_store.py
@@ -1,7 +1,8 @@
"""策略实例 JSON 存储（配置数据，几行；无需 DB 迁移）。
-实例 = 代码文件的参数变体（spec §12 三层模型中层）。
-运行态字段 status/last_return 给默认占位，由后续运行关联回填。
+实例 = 代码文件的参数变体（spec §12 三层模型中层）= 策略档案（§12.6 做实设计）。
+status/last_return：事件型运行（回测/回放）完成时落盘回写；持续型运行
+（实走/影子/实盘）由 API 读时聚合覆盖（见 routes_strategy.enrich_instances）。
"""
from __future__ import annotations
import json
@@ -87,3 +88,70 @@ def delete_instance(inst_id: int) -> bool:
return False
_save(new_rows)
return True
+
+
+# ===== §12.6 做实：事件型运行回写 + 参数漂移（§12.6 P0）=====
+
+_KINDS = ("backtest", "replay")
+
+
+def update_instance_run(inst_id: int, kind: str, status: str,
+ ret: float | None = None) -> bool:
+ """事件型运行（回测/回放）完成/失败时回写档案。
+
+ kind: backtest/replay（paper_live/live 是持续运行，读时聚合不落盘）。
+ ret: 小数收益（0.1548=+15.48%），None 则保留旧值。
+ 返回 False = 实例不存在（如已删，静默丢弃）。
+ """
+ if kind not in _KINDS:
+ return False
+ rows = _load()
+ for r in rows:
+ if r.get("id") == inst_id:
+ r.setdefault("status", dict(_DEFAULT_STATUS))
+ r["status"][kind] = status
+ if ret is not None:
+ r["last_return"] = ret
+ r.setdefault("run_returns", {})
+ r["run_returns"][kind] = ret
+ r["updated_at"] = time.strftime("%Y-%m-%d")
+ _save(rows)
+ return True
+ return False
+
+
+def get_instance_params_snapshot(inst_id: int) -> dict | None:
+ """档案当前参数（发起时快照用，D1）。不存在返回 None。"""
+ for r in _load():
+ if r.get("id") == inst_id:
+ return {
+ "params": dict(r.get("params") or {}),
+ "symbol_or_pool": r.get("symbol_or_pool", ""),
+ "interval": r.get("interval", "d"),
+ "match_session": r.get("match_session", "next_open"),
+ "code_file": r.get("code_file", ""),
+ "type": r.get("type", "cta"),
+ "name": r.get("name", ""),
+ }
+ return None
+
+
+def account_params_drifted(snapshot_params: dict, account: dict) -> bool:
+ """参数漂移检测（D2）：账户 strategies 参数 vs 档案当前参数。
+
+ account 为 paper_accounts 行（strategies 是 JSON 字符串或已解析 list）。
+ 只比对参数键值（标的/周期改了也算漂移——都影响行为）。
+ """
+ import json as _json
+ raw = account.get("strategies")
+ if isinstance(raw, str):
+ try:
+ strats = _json.loads(raw)
+ except (ValueError, TypeError):
+ return False
+ else:
+ strats = raw or []
+ if not strats:
+ return False
+ live_params = dict(strats[0].get("params") or {})
+ return live_params != dict(snapshot_params.get("params") or {})
diff --git a/sanguo_api/routes.py b/sanguo_api/routes.py
index e6c257f..07ec6fb 100644
--- a/sanguo_api/routes.py
+++ b/sanguo_api/routes.py
@@ -111,6 +111,7 @@ async def submit_cta(req: CtaBacktestRequest):
capital=req.capital,
position_pct=req.position_pct,
interval=req.interval,
+ instance_id=req.instance_id,
)
return {"task_id": tid}
diff --git a/sanguo_api/routes_live.py b/sanguo_api/routes_live.py
index 14ce5f3..fc7aa92 100644
--- a/sanguo_api/routes_live.py
+++ b/sanguo_api/routes_live.py
@@ -9,6 +9,7 @@ GET 查询持仓/成交/账户/状态。runner(supervisor) 是独立常驻进程
from __future__ import annotations
import os
+import time
from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel
@@ -53,6 +54,21 @@ class LiveCreateRequest(BaseModel):
pool: str = ""
max_pool: int = 0
benchmark: str = ""
+ # §12.6 实例做实：账户绑档案；空=发起即建档
+ instance_id: int | None = None
+
+
+def _resolve_file_by_class(class_name: str) -> str:
+ """类名/策略名 → 策略文件名（发起即建档反查 code_file；查不到返回空）。"""
+ try:
+ from .strategy_registry import list_strategy_files
+
+ for f in list_strategy_files()["files"]:
+ if f.get("class_name") == class_name or f.get("name", "").removesuffix(".py") == class_name:
+ return f.get("name", "")
+ except Exception:
+ pass
+ return ""
def _normalize_vt_symbol(code: str) -> str:
@@ -74,6 +90,20 @@ def create_live(req: LiveCreateRequest):
db = _db_path["path"] or ":memory:"
init_db(db)
+ # §12.6 发起绑档案（合法 instance_id 用之；否则发起即建档）
+ from . import instance_store
+
+ if not (req.instance_id and instance_store.get_instance_params_snapshot(req.instance_id)):
+ sym = req.pool if req.strategy_type == "portfolio" else req.vt_symbol
+ req.instance_id = instance_store.create_instance({
+ "code_file": _resolve_file_by_class(req.strategy_class),
+ "name": f"{req.strategy_class}·live·{time.strftime('%m%d')}",
+ "type": req.strategy_type,
+ "params": dict(req.setting or {}),
+ "symbol_or_pool": sym,
+ "interval": req.interval or "d",
+ "match_session": "next_open",
+ })
payload = req.model_dump()
if payload.get("strategy_type") == "portfolio":
# 组合实盘:vt_symbol 占位为池名;setting 存组合参数(supervisor 转发 env)
diff --git a/sanguo_api/routes_paper.py b/sanguo_api/routes_paper.py
index a5ddea0..f35c4c4 100644
--- a/sanguo_api/routes_paper.py
+++ b/sanguo_api/routes_paper.py
@@ -6,11 +6,13 @@ create 建 paper_account（持久化配置）；GET 查询净值/成交/状态
"""
import json
import sqlite3
+import time
from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel
from .auth import verify_token as verify_token_impl
+from . import instance_store
from .validation import validate_backtest_range
router = APIRouter()
@@ -62,6 +64,41 @@ class PaperCreateRequest(BaseModel):
pool: str = "hs300_subset"
max_pool: int = 30
benchmark: str = "000300.XSHG"
+ # §12.6 实例做实：账户绑档案；空=发起即建档（自动创建实例再发起）
+ instance_id: int | None = None
+
+
+def _resolve_file_by_class(class_name: str) -> str:
+ """类名 → 策略文件名（发起即建档时反查 code_file；查不到返回空）。"""
+ try:
+ from .strategy_registry import list_strategy_files
+
+ for f in list_strategy_files()["files"]:
+ if f.get("class_name") == class_name:
+ return f.get("name", "")
+ except Exception:
+ pass
+ return ""
+
+
+def _ensure_instance_for_paper(req: PaperCreateRequest) -> int:
+ """§12.6 D1/D5：发起时绑档案。带合法 instance_id 用之；否则发起即建档。"""
+ from . import instance_store
+
+ if req.instance_id and instance_store.get_instance_params_snapshot(req.instance_id):
+ return req.instance_id
+ strat = req.strategies[0] if req.strategies else None
+ cls = strat.name if strat else ""
+ sym = req.pool if req.strategy_type == "portfolio" else ",".join(req.symbols)
+ return instance_store.create_instance({
+ "code_file": _resolve_file_by_class(cls),
+ "name": f"{cls}·{req.mode}·{time.strftime('%m%d')}",
+ "type": req.strategy_type,
+ "params": dict(strat.params) if strat else {},
+ "symbol_or_pool": sym,
+ "interval": req.interval,
+ "match_session": strat.match_session if strat else "next_open",
+ })
@router.post("/paper/create", dependencies=[Depends(verify_token)])
@@ -71,6 +108,13 @@ def create_paper(req: PaperCreateRequest):
db = _db_path["path"] or ":memory:"
init_db(db)
+ # 实例绑定先于日期归一(model_dump 要带上 instance_id 落库)
+ req.instance_id = _ensure_instance_for_paper(req)
+ # §12.6 D1 发起时快照：绑已有档案 → 账户参数用档案当时的参数（复印件），
+ # 后续改档案不影响本账户（漂移可见，手动同步）
+ snap = instance_store.get_instance_params_snapshot(req.instance_id) if req.instance_id else None
+ if snap and req.strategies:
+ req.strategies[0].params = dict(snap.get("params") or {})
# 实走/影子是开放账户:起止日期无意义,开始=创建当天(组合日终重放依赖 start_date,
# 空值会崩),结束留空;仅回放保留用户填的历史区间
if req.mode in ("live", "shadow"):
@@ -103,11 +147,22 @@ def create_paper(req: PaperCreateRequest):
if req.mode == "replay": # 回放后台线程跑，create 立即返回（避免阻塞 worker 502）
def _bg():
from sanguo_trader.persistence import update_account_status
+ from .instance_store import update_instance_run
try:
_run_replay(db, aid, req)
update_account_status(db, aid, "done")
+ # §12.6 回放完成回写档案（收益=末次净值/初始-1）
+ if req.instance_id:
+ from sanguo_trader.persistence import load_last_balance
+ last = load_last_balance(db, aid)
+ ret = None
+ if last and req.initial_capital:
+ ret = (last.get("total_equity", 0) - req.initial_capital) / req.initial_capital
+ update_instance_run(req.instance_id, "replay", "done", ret)
except Exception as e: # noqa: BLE001
update_account_status(db, aid, "failed", str(e))
+ if req.instance_id:
+ update_instance_run(req.instance_id, "replay", "failed")
threading.Thread(target=_bg, daemon=True).start()
status = "running"
elif req.mode in ("live", "shadow"): # 实走:每日 20:30 step;影子:等 VPS 影子柜台进程接管
@@ -117,6 +172,44 @@ def create_paper(req: PaperCreateRequest):
return {"account_id": aid, "status": status}
+@router.post("/paper/sync/{instance_id}", dependencies=[Depends(verify_token)])
+def sync_instance_params(instance_id: int):
+ """§12.6 D2/D3 参数同步：档案当前参数 → 该档案全部运行中模拟账户。
+
+ 实走+影子一起换（影子与实盘/对照账户参数必须锁死，否则双轨对账失效）。
+ 实盘不在此列（D2：实盘不提供在线改参，停了重发）。
+ 注意：仅改 params；在跑进程的内存参数于当日结算/重启后生效（快照在结算时重读）。
+ """
+ from . import instance_store
+
+ snap = instance_store.get_instance_params_snapshot(instance_id)
+ if snap is None:
+ raise HTTPException(404, "实例不存在")
+ db = _db_path["path"] or ":memory:"
+ synced = 0
+ with sqlite3.connect(db) as conn:
+ rows = conn.execute(
+ "SELECT id, strategies FROM paper_accounts "
+ "WHERE instance_id=? AND status='running'",
+ (instance_id,),
+ ).fetchall()
+ for aid, raw in rows:
+ try:
+ strats = json.loads(raw) if isinstance(raw, str) else (raw or [])
+ except (ValueError, TypeError):
+ continue
+ if not strats:
+ continue
+ strats[0]["params"] = dict(snap["params"])
+ conn.execute(
+ "UPDATE paper_accounts SET strategies=?, updated_at=? WHERE id=?",
+ (json.dumps(strats, ensure_ascii=False), time.strftime("%Y-%m-%d %H:%M:%S"), aid),
+ )
+ synced += 1
+ conn.commit()
+ return {"synced": synced}
+
+
@router.get("/paper", dependencies=[Depends(verify_token)])
def list_papers():
"""模拟盘列表（启用聚宽级模拟交易列表页）。每行带最新净值 + 收益率。"""
diff --git a/sanguo_api/routes_portfolio.py b/sanguo_api/routes_portfolio.py
index 960df8f..8bfa1f0 100644
--- a/sanguo_api/routes_portfolio.py
+++ b/sanguo_api/routes_portfolio.py
@@ -45,6 +45,7 @@ class PortfolioBacktestRequest(BaseModel):
min_commission: float = Field(default=5.0, description="单笔最低佣金(元)")
slippage: float = Field(default=0.0, description="滑点比率(万10=0.001,0=不加)")
interval: str = Field(default="d", description="K线周期:组合回放暂仅日线(d)")
+ instance_id: int | None = Field(default=None, description="§12.6 可选关联策略档案(带了则结果回写)")
@router.post("/portfolio/backtest", dependencies=[Depends(verify_token)])
@@ -66,6 +67,7 @@ async def run_portfolio_backtest(req: PortfolioBacktestRequest):
min_commission=req.min_commission,
slippage=req.slippage,
interval=req.interval,
+ instance_id=req.instance_id,
)
return {"task_id": tid}
diff --git a/sanguo_api/routes_strategy.py b/sanguo_api/routes_strategy.py
index 3846f47..71f6b46 100644
--- a/sanguo_api/routes_strategy.py
+++ b/sanguo_api/routes_strategy.py
@@ -81,6 +81,32 @@ def post_instance(req: InstancePayload):
return {"id": new_id}
+@router.get("/strategy/instances/enriched")
+def get_instances_enriched():
+ """实例列表 + 读时聚合：持续运行覆盖四格的实走/实盘格 + 在跑账户 + 漂移标记。
+
+ 注意必须注册在 /strategy/instances/{inst_id} 之前（否则 enriched 被当 inst_id 解析）。
+ """
+ rt = _instance_runtime()
+ out = []
+ for inst in instance_store.list_instances()["instances"]:
+ ent = rt.get(inst["id"], {"running_accounts": [], "drift": False})
+ paper_runs = [a for a in ent["running_accounts"] if a["kind"] in ("paper", "shadow")]
+ live_runs = [a for a in ent["running_accounts"] if a["kind"] == "live"]
+ status = dict(inst.get("status") or {})
+ if paper_runs:
+ status["paper_live"] = "running"
+ if live_runs:
+ status["live"] = "running"
+ out.append({
+ **inst,
+ "status": status,
+ "running_accounts": ent["running_accounts"],
+ "drift": ent["drift"],
+ })
+ return {"instances": out}
+
+
@router.get("/strategy/instances/{inst_id}")
def get_instance(inst_id: int):
return instance_store.get_instance(inst_id)
@@ -94,7 +120,170 @@ def put_instance(inst_id: int, req: InstancePayload):
@router.delete("/strategy/instances/{inst_id}")
def del_instance(inst_id: int):
+ # §12.6 D5 删除保护：有运行中账户（模拟/实盘）的档案禁止删
+ running = _instance_runtime().get(inst_id, {}).get("running_accounts", [])
+ if running:
+ names = "、".join(a["label"] for a in running[:3])
+ raise HTTPException(
+ status_code=409,
+ detail=f"该档案下有运行中的账户（{names}…），请先停止再删除",
+ )
ok = instance_store.delete_instance(inst_id)
if not ok:
raise HTTPException(status_code=404, detail="实例不存在")
return {"ok": True}
+
+
+# ===== §12.6 做实：读时聚合（持续型运行）+ P1 实例全景 =====
+
+def _instance_runtime() -> dict[int, dict]:
+ """读时聚合每个档案的模拟/实盘账户（持续型运行的"回写"）。
+
+ 返回 {instance_id: {running_accounts: [...], drift: bool}}；DB 不可达返回 {}
+ （策略库退化为纯 JSON 状态展示，不阻塞）。
+ """
+ import json as _json
+ import sqlite3
+
+ out: dict[int, dict] = {}
+ try:
+ from .routes_paper import _db_path as paper_db
+
+ pdb = paper_db["path"]
+ if pdb:
+ with sqlite3.connect(pdb) as conn:
+ conn.row_factory = sqlite3.Row
+ rows = conn.execute(
+ "SELECT id, name, mode, engine, status, strategies, instance_id, "
+ "initial_capital FROM paper_accounts WHERE instance_id IS NOT NULL"
+ ).fetchall()
+ from sanguo_trader.persistence import load_last_balance
+
+ for r in rows:
+ ent = out.setdefault(r["instance_id"], {"running_accounts": [], "drift": False})
+ ret = None
+ last = load_last_balance(pdb, r["id"])
+ if last and r["initial_capital"]:
+ ret = (last.get("total_equity", 0) - r["initial_capital"]) / r["initial_capital"]
+ snapshot = instance_store.get_instance_params_snapshot(r["instance_id"])
+ drifted = bool(snapshot and instance_store.account_params_drifted(snapshot, dict(r)))
+ if drifted:
+ ent["drift"] = True
+ if r["status"] == "running":
+ ent["running_accounts"].append({
+ "kind": "shadow" if r["engine"] == "shadow" else "paper",
+ "aid": r["id"],
+ "label": r["name"] or f"paper#{r['id']}",
+ "ret": ret,
+ "drifted": drifted,
+ })
+ except Exception:
+ pass
+ try:
+ from .routes_live import _db_path as live_db
+ from sanguo_live.persistence import get_first_balance, get_last_balance, list_accounts
+
+ ldb = live_db["path"]
+ if ldb:
+ for a in list_accounts(ldb):
+ iid = a.get("instance_id")
+ if not iid:
+ continue
+ ent = out.setdefault(iid, {"running_accounts": [], "drift": False})
+ ret = None
+ first, last = get_first_balance(ldb, a["id"]), get_last_balance(ldb, a["id"])
+ if first and last and first.get("total"):
+ ret = (last["total"] - first["total"]) / first["total"]
+ if a.get("status") == "running":
+ ent["running_accounts"].append({
+ "kind": "live",
+ "aid": a["id"],
+ "label": a.get("name") or f"live#{a['id']}",
+ "ret": ret,
+ "drifted": False,
+ })
+ except Exception:
+ pass
+ return out
+
+
+@router.get("/strategy/instances/{inst_id}/overview")
+def get_instance_overview(inst_id: int):
+ """P1 实例全景：档案 + 全部运行账户（含净值尾部）+ 合并持仓归因。"""
+ import sqlite3
+
+ snap = instance_store.get_instance_params_snapshot(inst_id)
+ if snap is None:
+ raise HTTPException(status_code=404, detail="实例不存在")
+ runs: list[dict] = []
+ positions: dict[str, dict] = {}
+
+ def _absorb(symbol: str, volume: float, avg: float, label: str) -> None:
+ p = positions.setdefault(symbol, {"symbol": symbol, "volume": 0.0,
+ "avg_price": avg, "accounts": []})
+ p["volume"] += volume
+ p["accounts"].append({"label": label, "volume": volume})
+
+ try:
+ from .routes_paper import _db_path as paper_db
+
+ pdb = paper_db["path"]
+ if pdb:
+ from sanguo_trader.persistence import load_last_balance, load_positions
+
+ with sqlite3.connect(pdb) as conn:
+ conn.row_factory = sqlite3.Row
+ rows = conn.execute(
+ "SELECT * FROM paper_accounts WHERE instance_id=? ORDER BY id", (inst_id,)
+ ).fetchall()
+ for r in rows:
+ last = load_last_balance(pdb, r["id"])
+ ret = None
+ if last and r["initial_capital"]:
+ ret = (last["total_equity"] - r["initial_capital"]) / r["initial_capital"]
+ with sqlite3.connect(pdb) as conn:
+ curve = conn.execute(
+ "SELECT date, total_equity FROM paper_daily_balance "
+ "WHERE account_id=? ORDER BY date DESC LIMIT 120",
+ (r["id"],),
+ ).fetchall()
+ runs.append({
+ "kind": "shadow" if r["engine"] == "shadow" else "paper",
+ "aid": r["id"], "label": r["name"], "mode": r["mode"],
+ "status": r["status"], "ret": ret,
+ "equity": [{"date": c[0], "equity": c[1]} for c in reversed(curve)],
+ })
+ if r["status"] == "running":
+ scope = "shadow" if r["engine"] == "shadow" else r["mode"]
+ for sym, p in load_positions(pdb, r["id"], scope).items():
+ _absorb(sym, p.get("volume", 0), p.get("avg_price", 0),
+ r["name"] or f"paper#{r['id']}")
+ except Exception:
+ pass
+ try:
+ from .routes_live import _db_path as live_db
+
+ ldb = live_db["path"]
+ if ldb:
+ from sanguo_live.persistence import (
+ get_first_balance, get_last_balance, list_accounts, load_positions,
+ )
+
+ for a in list_accounts(ldb):
+ if a.get("instance_id") != inst_id:
+ continue
+ first, last = get_first_balance(ldb, a["id"]), get_last_balance(ldb, a["id"])
+ ret = None
+ if first and last and first.get("total"):
+ ret = (last["total"] - first["total"]) / first["total"]
+ runs.append({
+ "kind": "live", "aid": a["id"], "label": a.get("name"),
+ "status": a.get("status"), "ret": ret, "equity": [],
+ })
+ if a.get("status") == "running":
+ for p in load_positions(ldb, a["id"]):
+ _absorb(p["symbol"], p.get("volume", 0), p.get("avg_price", 0),
+ a.get("name") or f"live#{a['id']}")
+ except Exception:
+ pass
+ return {"instance": snap, "runs": runs, "positions": list(positions.values())}
diff --git a/sanguo_api/schemas.py b/sanguo_api/schemas.py
index 587b39b..e5d583c 100644
--- a/sanguo_api/schemas.py
+++ b/sanguo_api/schemas.py
@@ -22,6 +22,8 @@ class CtaBacktestRequest(BaseModel):
stamp_duty_rate: float = 0.0005 # 卖方 0.05%
transfer_fee_rate: float = 0.00001 # 沪市 0.001%
slippage: float = 0.0 # 滑点（比率，万10=0.001）；0=不加滑点
+ # §12.6 实例做实：可选关联档案（带了则结果回写档案四格）
+ instance_id: int | None = None
class OptimizeRequest(BaseModel):
diff --git a/sanguo_live/persistence.py b/sanguo_live/persistence.py
index bb6f368..39af907 100644
--- a/sanguo_live/persistence.py
+++ b/sanguo_live/persistence.py
@@ -88,6 +88,7 @@ def init_db(db_path: str) -> None:
("pool", "TEXT"),
("max_pool", "INTEGER"),
("benchmark", "TEXT"),
+ ("instance_id", "INTEGER"), # §12.6 实例做实:账户绑档案
):
try:
conn.execute(f"ALTER TABLE live_accounts ADD COLUMN {col} {ddl}")
@@ -106,8 +107,8 @@ def save_account(db_path: str, account: dict[str, Any]) -> int:
(name, account, vt_symbol, strategy_class, strategy_name, setting,
status, interval, initial_capital, connect_wait_sec, init_wait_sec,
mini_path, created_at, updated_at,
- strategy_type, pool, max_pool, benchmark)
- VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+ strategy_type, pool, max_pool, benchmark, instance_id)
+ VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
account.get("name", "live"),
account.get("account", ""),
@@ -126,6 +127,7 @@ def save_account(db_path: str, account: dict[str, Any]) -> int:
account.get("pool", ""),
int(account.get("max_pool", 0) or 0),
account.get("benchmark", ""),
+ account.get("instance_id"),
),
)
conn.commit()
diff --git a/sanguo_orchestrator/runner.py b/sanguo_orchestrator/runner.py
index 4809ec9..1476d5b 100644
--- a/sanguo_orchestrator/runner.py
+++ b/sanguo_orchestrator/runner.py
@@ -44,7 +44,7 @@ class Orchestrator:
async def submit_cta(self, strategy_class, symbol: str, params: dict,
start: str, end: str, cfg, benchmark: str = "hs300",
capital: float = 1_000_000, position_pct: float = 0.95,
- interval: str = "d") -> str:
+ interval: str = "d", instance_id: int | None = None) -> str:
"""Submit a CTA backtesting task asynchronously"""
# Stable uuid up front → reused as the persisted DB task_id, so runner-id ==
# DB task_id (durable across restarts; previously used id(params) memory addr).
@@ -62,6 +62,7 @@ class Orchestrator:
capital=capital,
position_pct=position_pct,
interval=interval,
+ instance_id=instance_id,
)
await self._notify_stage(task_id, "排队中")
@@ -142,7 +143,8 @@ class Orchestrator:
stamp_duty_rate: float = 0.001,
min_commission: float = 5.0,
slippage: float = 0.0,
- interval: str = "d") -> str:
+ interval: str = "d",
+ instance_id: int | None = None) -> str:
"""Submit a portfolio backtest task asynchronously.
Runs runner_backtest as a subprocess (3600s hard cap) inside the
@@ -169,6 +171,7 @@ class Orchestrator:
interval=interval,
db_path=self.db_path,
file_dir=file_dir,
+ instance_id=instance_id,
)
await self._notify_stage(task_id, "排队中")
@@ -197,8 +200,30 @@ class Orchestrator:
task = self.pool.get_task(task_id)
if task:
task.fail(f"{type(e).__name__}: {e}")
+ self._write_back_instance(task_id, "failed")
await self._notify_stage(task_id, "失败")
+ def _write_back_instance(self, task_id: str, status: str, result=None) -> None:
+ """§12.6 回测完成/失败回写档案（事件型运行；仅提交时带了 instance_id 的任务）。"""
+ spec = self._pending.get(task_id) or {}
+ inst_id = spec.get("instance_id")
+ if not inst_id:
+ return
+ try:
+ from sanguo_api.instance_store import update_instance_run
+
+ ret = None
+ if result is not None and status == "done":
+ stats = getattr(result, "statistics", None) or {}
+ m = stats.get("metrics") if isinstance(stats.get("metrics"), dict) else stats
+ ret = m.get("total_return")
+ update_instance_run(int(inst_id), "backtest", status, ret)
+ except Exception:
+ import logging
+ logging.getLogger(__name__).warning(
+ "instance write-back failed for %s", task_id, exc_info=True
+ )
+
async def _on_done(self, task_id: str, result) -> None:
"""Handle task completion (with None-guard for unknown tasks)"""
task = self.pool.get_task(task_id)
@@ -211,6 +236,7 @@ class Orchestrator:
# load_result(result.id). FactorReport (no .id) falls back to None until S2.
task.complete(result_id=getattr(result, "id", None))
task.raw_result = result # S2: keep in-memory result (FactorReport) for ic-summary/report
+ self._write_back_instance(task_id, "done", result) # §12.6 回测完成回写档案
# S2: persist factor result so it appears in task list & survives restart
if getattr(result, "ic_summary", None) and getattr(result, "factor_names", None) is not None:
self._persist_factor(task_id, result)
diff --git a/sanguo_trader/persistence.py b/sanguo_trader/persistence.py
index 406e7cf..d979b48 100644
--- a/sanguo_trader/persistence.py
+++ b/sanguo_trader/persistence.py
@@ -79,6 +79,11 @@ def init_db(db_path: str) -> None:
conn.execute("ALTER TABLE paper_accounts ADD COLUMN engine TEXT DEFAULT 'eod_replay'")
except sqlite3.OperationalError:
pass # 列已存在
+ # 迁移：老库补 instance_id 列（§12.6 实例做实：账户绑档案）
+ try:
+ conn.execute("ALTER TABLE paper_accounts ADD COLUMN instance_id INTEGER")
+ except sqlite3.OperationalError:
+ pass # 列已存在
conn.execute("PRAGMA journal_mode=WAL")
conn.commit()
@@ -90,8 +95,8 @@ def save_account(db_path: str, account: dict[str, Any]) -> int:
(task_id, owner_id, name, strategy_type, mode, interval, symbols, strategies,
initial_capital, rate, slippage, size, pricetick,
stamp_duty_rate, transfer_fee_rate, min_commission,
- status, start_date, end_date, engine, created_at, updated_at)
- VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+ status, start_date, end_date, engine, instance_id, created_at, updated_at)
+ VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
account.get("task_id"), account.get("owner_id", "admin"),
account.get("name"), account.get("strategy_type", "cta"),
@@ -108,6 +113,7 @@ def save_account(db_path: str, account: dict[str, Any]) -> int:
account.get("start_date") or account.get("start"),
account.get("end_date") or account.get("end"),
account.get("engine", "eod_replay"),
+ account.get("instance_id"),
_now(), _now(),
),
)
diff --git a/tests/api/test_instance_binding.py b/tests/api/test_instance_binding.py
new file mode 100644
index 0000000..f5dd5eb
--- /dev/null
+++ b/tests/api/test_instance_binding.py
@@ -0,0 +1,171 @@
+"""§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")