fix(portfolio): P2 终审修正——注册表运行副本落 data/ 首触自举+CLI --stage 断链+周报 NULL/平盘修 [vps] [no-doc]

M1 运行副本自举:promote.sh 每次 push 都 rsync config/,运行期转移(毕业/退回/
延长)写 config/ 会被下次部署静默回滚——新增 ensure_runtime_file/
ensure_runtime_registry(config/ 种子→data/ 运行副本,首触拷贝,已存在幂等不覆盖),
CLI/weekly_report/routes_pipeline 默认全改 data/(strategy_registry/
strategy_events/factor_registry/registry_events 四件,env 名不变);
factor verdict 端点经同款 ensure 自举;graduate 证据模板路径同步。
M2 register --stage 直落初档(此前只在 transition 生效=断链)。
M4 load_actual_daily_returns 剔 NULL total_equity 行(不折算 0 毒化收益链)。
M5 daily_returns 平盘日照记 0.0(剔除会低估 TE)。
T5-①删 set_pipeline_paths 死注入(app.py 同步);T5-②ladder/todos 阈值只取
一次传 gate;T5-③梯子 te/fill/vs 任一 None→周报待算(去 or 0.0 硬凑);
T5-⑤reviewIssue 非整数 422;T5-⑦graduate 尾巴(事件/模板/gitea)失败不 500,
200 带 warning 提示人工补。
测试:种子自举幂等/CLI --stage/NULL 剔行/平盘 0.0/verdict 首触自举端到端;
tests/portfolio 661 绿+tests/factor 396 绿。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

sanguo_api/app.py

@@ -8,7 +8,7 @@ from .routes_live import router as live_router, set_db_path as set_live_db_path
 from .routes_portfolio import router as portfolio_router
 from .routes_strategy import router as strategy_router
 from .routes_factor import router as factor_eval_router, set_eval_db_path as set_factor_eval_db_path
-from .routes_pipeline import router as pipeline_router, set_pipeline_paths
+from .routes_pipeline import router as pipeline_router
 from .auth import set_jwt_config
 from .ws import manager
 from sanguo_orchestrator.runner import Orchestrator, resolve_eval_db
@@ -48,7 +48,6 @@ def create_app(db_path: str, file_dir=None, auth_config=None, max_workers: int =
     set_paper_db_path(db_path)
     set_live_db_path(db_path)
     set_factor_eval_db_path(resolve_eval_db(db_path))
-    set_pipeline_paths(db_path)
 
     @app.on_event("startup")
     def _register_live_step():

sanguo_api/routes_pipeline.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import glob
 import json
+import logging
 import os
 from dataclasses import asdict
 from datetime import date, datetime
@@ -20,17 +21,12 @@ from sanguo_api.routes import verify_token
 
 router = APIRouter(dependencies=[Depends(verify_token)])
 
-_paths: dict[str, Any] = {"db_path": None}
-
-
-def set_pipeline_paths(db_path: str) -> None:
-    """app.create_app 注入主库路径(routes_paper.set_db_path 同款)."""
-    _paths["db_path"] = db_path
+logger = logging.getLogger(__name__)
 
 
 def _registry_path() -> str:
     return os.environ.get("SANGUO_STRATEGY_REGISTRY",
-                          os.path.join("config", "strategy_registry.yaml"))
+                          os.path.join("data", "strategy_registry.yaml"))
 
 
 def _pipeline_db() -> str:
@@ -39,8 +35,9 @@ def _pipeline_db() -> str:
 
 
 def _load_strategy_registry() -> dict:
-    from sanguo_portfolio.strategy_registry import load_registry
-    return load_registry(_registry_path())
+    from sanguo_portfolio.strategy_registry import (
+        ensure_runtime_registry, load_registry)
+    return load_registry(ensure_runtime_registry(_registry_path()))
 
 
 def _fmt_vs(v: float | None) -> str:
@@ -56,6 +53,7 @@ def ladder() -> dict:
         shadow_total_weeks, shadow_weeks)
     reg = _load_strategy_registry()
     today = date.today().isoformat()
+    thresholds = pipeline_store.get_effective_thresholds(_pipeline_db())
     items = []
     for name in sorted(reg["strategies"]):
         e = reg["strategies"][name]
@@ -68,15 +66,16 @@ def ladder() -> dict:
             item["shadowWeeks"] = shadow_weeks(e, today)
             item["shadowTotalWeeks"] = shadow_total_weeks(e)
             w = pipeline_store.latest_weekly(_pipeline_db(), name)
-            if w:
+            if w and w.get("te_annual") is not None \
+                    and w.get("fill_rate") is not None \
+                    and w.get("vs_backtest") is not None:
                 item["weeklyReport"] = {
-                    "trackingError": w.get("te_annual") or 0.0,
-                    "fillRate": w.get("fill_rate") or 0.0,
-                    "vsBacktest": _fmt_vs(w.get("vs_backtest"))}
+                    "trackingError": w["te_annual"],
+                    "fillRate": w["fill_rate"],
+                    "vsBacktest": _fmt_vs(w["vs_backtest"])}
                 gate = evaluate_shadow_gate(
-                    item["shadowWeeks"], w.get("te_annual") or 1.0,
-                    w.get("fill_rate") or 0.0,
-                    abs(w.get("vs_backtest") or 1.0))
+                    item["shadowWeeks"], w["te_annual"],
+                    w["fill_rate"], abs(w["vs_backtest"]), thresholds)
                 item["note"] = (item["note"] + " " if item["note"] else "") + \
                     ("gate ✓" if gate.passed else "gate ✗ 未毕业")
             else:
@@ -125,7 +124,7 @@ def graduate(body: dict) -> dict:
     from sanguo_portfolio import pipeline_store
     from sanguo_portfolio.promotion_gate import evaluate_shadow_gate
     from sanguo_portfolio.strategy_registry import (
-        append_event, load_registry, render_graduate_issue, save_registry,
+        append_event, render_graduate_issue, save_registry,
         shadow_weeks, shadow_total_weeks, transition)
 
     name = body.get("name") or ""
@@ -135,7 +134,7 @@ def graduate(body: dict) -> dict:
         raise HTTPException(503, "未配置 SANGUO_PROMOTION_PASSPHRASE(首班用户定)")
     if passphrase != expect:
         raise HTTPException(403, "口令不符")
-    reg = load_registry(_registry_path())
+    reg = _load_strategy_registry()
     entry = reg["strategies"].get(name)
     if entry is None:
         raise HTTPException(404, f"策略未注册: {name}")
@@ -168,26 +167,37 @@ def graduate(body: dict) -> dict:
     extra: dict[str, Any] = {"gate_snapshot": snap, "capital": capital,
                              "at": today}
     if body.get("reviewIssue"):
-        extra["review_issue"] = int(body["reviewIssue"])
+        try:
+            extra["review_issue"] = int(body["reviewIssue"])
+        except (TypeError, ValueError):
+            raise HTTPException(422, "reviewIssue 须为整数")
     new_entry = transition(reg, name, "live", **extra)
     save_registry(_registry_path(), reg)
-    events_path = os.environ.get(
-        "SANGUO_STRATEGY_EVENTS",
-        os.path.join("config", "strategy_events.jsonl"))
-    append_event(events_path, new_entry, "shadow", "live", capital=capital)
     issue_no = None
-    title, body_text = render_graduate_issue(name, new_entry)
-    issue_dir = os.environ.get("SANGUO_REGISTRY_ISSUE_DIR",
-                               os.path.join("reports", "registry_events"))
-    os.makedirs(issue_dir, exist_ok=True)
-    with open(os.path.join(issue_dir, f"{name}_graduate_issue.md"), "w",
-              encoding="utf-8") as f:
-        f.write(f"# {title}\n\n{body_text}\n")
-    token = os.environ.get("SANGUO_GITEA_TOKEN")
-    if token:
-        from sanguo_portfolio.strategy_registry import open_gitea_issue
-        issue_no = open_gitea_issue(title, body_text, token)
-    return {"ok": True, "issue": issue_no}
+    warning = None
+    try:
+        events_path = os.environ.get(
+            "SANGUO_STRATEGY_EVENTS",
+            os.path.join("data", "strategy_events.jsonl"))
+        append_event(events_path, new_entry, "shadow", "live", capital=capital)
+        title, body_text = render_graduate_issue(name, new_entry)
+        issue_dir = os.environ.get("SANGUO_REGISTRY_ISSUE_DIR",
+                                   os.path.join("reports", "registry_events"))
+        os.makedirs(issue_dir, exist_ok=True)
+        with open(os.path.join(issue_dir, f"{name}_graduate_issue.md"), "w",
+                  encoding="utf-8") as f:
+            f.write(f"# {title}\n\n{body_text}\n")
+        token = os.environ.get("SANGUO_GITEA_TOKEN")
+        if token:
+            from sanguo_portfolio.strategy_registry import open_gitea_issue
+            issue_no = open_gitea_issue(title, body_text, token)
+    except Exception as exc:  # noqa: BLE001 转移已落档,尾巴失败不回滚不 500
+        logger.warning("graduate %s 转移已落档,事件/模板落盘失败: %s", name, exc)
+        warning = "转移已落档,事件/模板落盘失败需人工补"
+    resp: dict[str, Any] = {"ok": True, "issue": issue_no}
+    if warning:
+        resp["warning"] = warning
+    return resp
 
 
 @router.get("/pipeline/todos")
@@ -199,6 +209,7 @@ def todos() -> dict:
         shadow_weeks, shadow_total_weeks)
     items: list[dict[str, Any]] = []
     now = datetime.now().isoformat(timespec="seconds")
+    thresholds = pipeline_store.get_effective_thresholds(_pipeline_db())
     # ① 毕业候选(shadow 满 total 周且 gate 过)
     reg = _load_strategy_registry()
     today = date.today().isoformat()
@@ -213,7 +224,7 @@ def todos() -> dict:
         if w and w.get("te_annual") is not None:
             gate_ok = evaluate_shadow_gate(
                 weeks, w["te_annual"], w.get("fill_rate") or 0.0,
-                abs(w.get("vs_backtest") or 1.0)).passed
+                abs(w.get("vs_backtest") or 1.0), thresholds).passed
         if gate_ok:
             items.append({"id": f"grad-{name}", "touchpoint": "promotion",
                           "title": f"{name} 影子期满且 gate 过,等口令+初始分配",
@@ -269,15 +280,18 @@ def factor_verdict(name: str, body: dict) -> dict:
     """晋级评审三按钮(触点②):promote/graveyard 走 version_registry 状态机;
     revise=留观(不改状态只留评审事件——升版本是 factor 域档流程,控制台不越权)."""
     from sanguo_factor import version_registry as vr
+    from sanguo_portfolio.strategy_registry import ensure_runtime_registry
     verdict = body.get("verdict")
     answers = body.get("answers") or []
     note = body.get("note") or ""
     if verdict not in ("promote", "revise", "graveyard"):
         raise HTTPException(422, "verdict 须为 promote/revise/graveyard")
-    reg_path = os.environ.get("SANGUO_FACTOR_REGISTRY",
-                              os.path.join("config", "factor_registry.yaml"))
+    reg_path = ensure_runtime_registry(
+        os.environ.get("SANGUO_FACTOR_REGISTRY",
+                       os.path.join("data", "factor_registry.yaml")),
+        os.path.join("config", "factor_registry.yaml"))
     events_path = os.environ.get("SANGUO_FACTOR_EVENTS",
-                                 os.path.join("config", "registry_events.jsonl"))
+                                 os.path.join("data", "registry_events.jsonl"))
     reg = vr.load_registry(reg_path)
     if name not in reg["factors"]:
         raise HTTPException(404, f"因子未注册: {name}")

sanguo_portfolio/strategy_registry.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import json
 import os
+import shutil
 import sys
 import tempfile
 from datetime import datetime, timezone
@@ -38,6 +39,25 @@ _GATING: dict[str, list[str]] = {
 }
 
 
+def ensure_runtime_file(path: str, seed_path: str) -> str:
+    """运行副本自举(git 种子→data/ 运行副本);已存在=幂等不覆盖."""
+    if not os.path.exists(path) and os.path.exists(seed_path):
+        d = os.path.dirname(os.path.abspath(path))
+        os.makedirs(d, exist_ok=True)
+        shutil.copyfile(seed_path, path)
+    return path
+
+
+def ensure_runtime_registry(path: str, seed_path: str = os.path.join(
+        "config", "strategy_registry.yaml")) -> str:
+    """运行副本自举:config/ 种子(git 随码走)→data/ 运行副本(部署不 clobber).
+
+    promote.sh 每次 push 都 rsync config/;运行期转移若写 config/ 会被下次部署
+    静默回滚——运行态一律落 data/ 首触从种子拷贝(M1 终审裁定).
+    """
+    return ensure_runtime_file(path, seed_path)
+
+
 def load_registry(path: str) -> dict[str, Any]:
     """读注册表;文件不存在→空表(首跑自举)."""
     if not os.path.exists(path):
@@ -195,7 +215,7 @@ def render_graduate_issue(name: str, entry: dict[str, Any]) -> tuple[str, str]:
         "1. 实盘实例由用户手动起(promotion 执行=用户动作,档案只记录)\n"
         "2. 资金分配落账户台账\n"
         "3. 毕业后首周周报三指标观察(TE/fillRate/vsBacktest)\n"
-        "证据: config/strategy_events.jsonl + config/strategy_registry.yaml\n")
+        "证据: data/strategy_events.jsonl + data/strategy_registry.yaml(种子在 config/)\n")
     return title, body
 
 
@@ -223,15 +243,16 @@ def main(argv: list[str] | None = None) -> int:
     ap = argparse.ArgumentParser(description="策略孵化梯注册表 CLI(决议 L 六件之 1)")
     ap.add_argument("cmd", choices=["register", "transition", "extend-shadow"])
     ap.add_argument("name")
-    ap.add_argument("--registry", default="config/strategy_registry.yaml")
-    ap.add_argument("--events", default="config/strategy_events.jsonl")
+    ap.add_argument("--registry", default=os.path.join("data", "strategy_registry.yaml"))
+    ap.add_argument("--events", default=os.path.join("data", "strategy_events.jsonl"))
     ap.add_argument("--hypothesis", default=None)
     ap.add_argument("--composite-version", default=None)
     ap.add_argument("--backtest-run", default=None)
     ap.add_argument("--paper-account-id", type=int, default=None,
                     help="paper_accounts.id(周报三指标与梯子端点按它拉本机数据)")
     ap.add_argument("--backfilled", action="store_true")
-    ap.add_argument("--stage", default=None, help="transition 目标 stage")
+    ap.add_argument("--stage", default=None,
+                    help="stage(register 初档/transition 目标)")
     ap.add_argument("--cause", default=None)
     ap.add_argument("--at", default=None)
     ap.add_argument("--review-issue", type=int, default=None)
@@ -244,6 +265,7 @@ def main(argv: list[str] | None = None) -> int:
     ap.add_argument("--gitea-open", action="store_true",
                     help="毕业时尝试调 Gitea API 开 issue(需 env SANGUO_GITEA_TOKEN)")
     args = ap.parse_args(argv)
+    ensure_runtime_registry(args.registry)
     reg = load_registry(args.registry)
     if args.cmd == "register":
         if not args.hypothesis or not args.composite_version:
@@ -252,6 +274,7 @@ def main(argv: list[str] | None = None) -> int:
             return 1
         entry = upsert_strategy(reg, args.name, hypothesis=args.hypothesis,
                                 composite_version=args.composite_version,
+                                stage=args.stage or "backtest",
                                 backfilled=args.backfilled,
                                 backtest_run=args.backtest_run,
                                 paper_account_id=args.paper_account_id)

sanguo_portfolio/weekly_report.py

@@ -28,11 +28,14 @@ def _week_start(d: date) -> date:
 
 
 def daily_returns(series: list[tuple[str, float]]) -> dict[str, float]:
-    """[(date, equity)] → {date: 当日收益率};首日无前值跳过,前值 0 防除零."""
+    """[(date, equity)] → {date: 当日收益率};首日无前值跳过,前值 0 防除零.
+
+    平盘日(收益 0.0)照记不剔除——剔除会低估 TE(M5 终审裁定).
+    """
     out: dict[str, float] = {}
     prev: float | None = None
     for d, eq in sorted(series):
-        if prev is not None and prev != 0 and eq != prev:
+        if prev is not None and prev != 0:
             out[d] = eq / prev - 1.0
         prev = eq
     return out
@@ -46,7 +49,7 @@ def load_actual_daily_returns(db_path: str, account_id: int,
             "SELECT date, total_equity FROM paper_daily_balance "
             "WHERE account_id=? AND date>=? AND date<=? ORDER BY date",
             (account_id, start, end)).fetchall()
-    return daily_returns([(r[0], float(r[1] or 0.0)) for r in rows])
+    return daily_returns([(r[0], float(r[1])) for r in rows if r[1] is not None])
 
 
 def load_backtest_daily_returns(db_path: str,
@@ -159,9 +162,10 @@ def build_weekly_reports(as_of: str, db_path: str, pipeline_db_path: str,
 
 def run_daily_recompute(db_path: str) -> int:
     """eod 后入口(20:45 job/CLI 同源):今日 as_of 全量幂等重算."""
-    from sanguo_portfolio import pipeline_store
+    from sanguo_portfolio import pipeline_store, strategy_registry
     registry = os.environ.get("SANGUO_STRATEGY_REGISTRY",
-                              os.path.join("config", "strategy_registry.yaml"))
+                              os.path.join("data", "strategy_registry.yaml"))
+    strategy_registry.ensure_runtime_registry(registry)
     as_of = date.today().isoformat()
     rows = build_weekly_reports(as_of, db_path,
                                 pipeline_store.default_pipeline_db_path(),
@@ -179,9 +183,10 @@ def main(argv: list[str] | None = None) -> int:
     args = ap.parse_args(argv)
     db_path = args.db or os.environ.get(
         "SANGUO_DB_PATH", os.path.join("data", "backtest_results.db"))
-    from sanguo_portfolio import pipeline_store
+    from sanguo_portfolio import pipeline_store, strategy_registry
     registry = os.environ.get("SANGUO_STRATEGY_REGISTRY",
-                              os.path.join("config", "strategy_registry.yaml"))
+                              os.path.join("data", "strategy_registry.yaml"))
+    strategy_registry.ensure_runtime_registry(registry)
     as_of = args.as_of or date.today().isoformat()
     rows = build_weekly_reports(as_of, db_path,
                                 pipeline_store.default_pipeline_db_path(),

tests/portfolio/test_routes_pipeline.py

@@ -144,7 +144,6 @@ def test_graduate_success_flow(client, token, env):
             "weeks_counted": 5,
             "computed_at": "2026-10-10T20:45:00"})
     # registry 影子起算日挪到 5 周前使 shadow_weeks>=4(不写死日期)
-    from datetime import date, timedelta
     with open(env["registry"], encoding="utf-8") as f:
         reg = yaml.safe_load(f)
     reg["strategies"]["s_shadow"]["stage_history"][0]["at"] = (
@@ -217,3 +216,23 @@ def test_factor_verdict_routes(client, token, env, tmp_path, monkeypatch):
     assert r2.status_code == 200
     with open(tmp_path / "factor_registry.yaml", encoding="utf-8") as f:
         assert yaml.safe_load(f)["factors"]["fa_gross_margin"]["status"] == "promoted"
+
+
+def test_factor_verdict_bootstraps_runtime_registry(client, token, tmp_path,
+                                                    monkeypatch):
+    # M1:未设 env 时 verdict 端点默认写 data/ 运行副本,首触从 config/ 种子自举
+    monkeypatch.chdir(tmp_path)  # 默认路径全 cwd 相对,隔离到 tmp
+    os.makedirs("config", exist_ok=True)
+    with open(os.path.join("config", "factor_registry.yaml"), "w",
+              encoding="utf-8") as f:
+        yaml.safe_dump({"factors": {"fa_x": {
+            "name": "fa_x", "hypothesis": "H-1", "status": "assessable",
+            "versions": []}}}, f, allow_unicode=True)
+    r = client.post("/api/v1/pipeline/factors/fa_x/verdict", headers=_h(token),
+                    json={"verdict": "promote", "answers": [True] * 5,
+                          "note": "首触自举后直写运行副本"})
+    assert r.status_code == 200
+    with open(os.path.join("data", "factor_registry.yaml"),
+              encoding="utf-8") as f:
+        assert yaml.safe_load(f)["factors"]["fa_x"]["status"] == "promoted"
+    assert os.path.exists(os.path.join("data", "registry_events.jsonl"))

tests/portfolio/test_strategy_registry.py

@@ -14,6 +14,7 @@ import yaml
 from sanguo_portfolio.strategy_registry import (
     VALID_STAGES,
     append_event,
+    ensure_runtime_registry,
     extend_shadow,
     load_registry,
     save_registry,
@@ -132,3 +133,27 @@ def test_seed_yaml_loads_and_is_backfilled():
     assert e["hypothesis"] and e["composite_version"]
     assert any(h["to"] == "live" for h in e["stage_history"])
     assert e.get("retired_cause") in (None, "")
+
+
+def test_ensure_runtime_registry_bootstraps_from_seed(tmp_path):
+    seed = tmp_path / "config" / "strategy_registry.yaml"
+    seed.parent.mkdir()
+    seed.write_text("strategies: {}\n", encoding="utf-8")
+    run = str(tmp_path / "data" / "strategy_registry.yaml")
+    ensure_runtime_registry(run, str(seed))
+    assert os.path.exists(run)
+    ensure_runtime_registry(run, str(seed))  # 已存在=幂等不覆盖
+    run_file = open(run, encoding="utf-8")
+    assert run_file.read() == "strategies: {}\n"
+    run_file.close()
+
+
+def test_cli_register_stage(tmp_path, monkeypatch):
+    # M2:register --stage 直落初档(此前 --stage 只在 transition 生效=断链)
+    from sanguo_portfolio.strategy_registry import main
+    monkeypatch.chdir(tmp_path)  # cwd 无 config/ 种子→ensure 不自举,注册表纯净
+    tmp_yaml = str(tmp_path / "reg.yaml")
+    rc = main(["register", "x9", "--hypothesis", "H-9", "--composite-version",
+               "v1", "--stage", "live", "--registry", tmp_yaml])
+    assert rc == 0
+    assert load_registry(tmp_yaml)["strategies"]["x9"]["stage"] == "live"

tests/portfolio/test_weekly_report.py

@@ -16,6 +16,7 @@ from sanguo_portfolio.weekly_report import (
     compute_fill_rate,
     compute_tracking_error,
     daily_returns,
+    load_actual_daily_returns,
     load_backtest_daily_returns,
     week_return,
 )
@@ -73,6 +74,13 @@ def test_daily_returns_pct_chain():
     assert "2026-09-07" not in r  # 首日无前值
 
 
+def test_daily_returns_keeps_flat_days():
+    # M5:平盘日照记 0.0,剔除会低估 TE
+    r = daily_returns([("a", 100.0), ("b", 100.0), ("c", 101.0)])
+    assert r["b"] == 0.0
+    assert r["c"] == pytest.approx(0.01)
+
+
 def test_tracking_error_aligned_and_none():
     a = {"d1": 0.01, "d2": -0.01, "d3": 0.01}
     b = {"d1": 0.01, "d2": -0.01, "d3": 0.01}
@@ -103,6 +111,23 @@ def test_load_backtest_daily_returns(main_db):
     assert load_backtest_daily_returns(main_db, "nope") is None
 
 
+def test_load_actual_daily_returns_skips_null_equity(tmp_path):
+    # M4:total_equity NULL 行剔除(不折算成 0 毒化收益链)
+    p = str(tmp_path / "null_eq.db")
+    conn = sqlite3.connect(p)
+    conn.execute("CREATE TABLE paper_daily_balance (account_id INTEGER, "
+                 "date TEXT, cash REAL, market_value REAL, total_equity REAL)")
+    conn.executemany("INSERT INTO paper_daily_balance VALUES(1,?,?,?,?)",
+                     [("2026-09-07", 50.0, 50.0, 100.0),
+                      ("2026-09-08", None, None, None),
+                      ("2026-09-09", 50.5, 50.5, 101.0)])
+    conn.commit()
+    conn.close()
+    r = load_actual_daily_returns(p, 1, "2026-09-01", "2026-09-30")
+    assert "2026-09-08" not in r
+    assert r["2026-09-09"] == pytest.approx(0.01)
+
+
 def test_build_weekly_reports_upserts_weeks_with_data(main_db, tmp_path, monkeypatch):
     pipeline_db = str(tmp_path / "pipeline.db")
     monkeypatch.setenv("SANGUO_PIPELINE_DB", pipeline_db)