feat(portfolio): 研判留痕模块——七档静态表唯一源+裁决 JSONL(spec §4.7 P3 第一刀) [vps]

2026-09-25 10:09:21 +08:00
parent f0495a093a
commit dc2b2c9282
2 changed files with 124 additions and 0 deletions
@@ -0,0 +1,68 @@
+# sanguo_portfolio/verdict_records.py
+# -*- coding: utf-8 -*-
+"""研判卡裁决留痕——spec §4.7 P3 第一刀之 2(2026-09-25).
+
+七档介入阶梯(决议 C)唯一源:api 拼卡/回放打标/裁决校验三方共用,文字逐字取自 spec。
+裁决 JSONL append-only:月键一条(重复=KeyError→api 层 409),只记录不执行——
+七档全部是已存在配置项的人工触点,零新增自动决策路径。
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+from datetime import datetime
+
+SEVEN_TIERS: list[dict] = [
+    {"level": 0, "action": "不动，继续观察", "meaning": "信任纪律，接受回撤", "cost": "判断错则回撤加深"},
+    {"level": 1, "action": "降总仓位上限", "meaning": "账户层 100%→70% 之类，选股不动", "cost": "反弹少赚"},
+    {"level": 2, "action": "策略间挪钱", "meaning": "挨打策略→防御型/现金", "cost": "挪错两头挨打"},
+    {"level": 3, "action": "降调仓频率/只出不进", "meaning": "跳过调仓日或 N 期只卖不买", "cost": "错过轮动起点"},
+    {"level": 4, "action": "因子族豁免/降权", "meaning": "只对逆风族降权", "cost": "精细判断易错"},
+    {"level": 5, "action": "转研究响应", "meaning": "仓位不动，开假设卡片走 P1/P4", "cost": "研究周期内继续承受"},
+    {"level": 6, "action": "防御模式（最重）", "meaning": "全账户最小暴露/清仓观望", "cost": "大概率卖在低点+再进场时点难题"},
+]
+
+_MONTH_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")
+
+
+def default_verdict_records_path() -> str:
+    """env SANGUO_VERDICT_RECORDS 覆盖;缺省 data/verdict_records.jsonl(相对 cwd=各机数据根)."""
+    return os.environ.get("SANGUO_VERDICT_RECORDS",
+                          os.path.join("data", "verdict_records.jsonl"))
+
+
+def load_rulings(path: str | None = None) -> list[dict]:
+    p = path or default_verdict_records_path()
+    if not os.path.exists(p):
+        return []
+    out: list[dict] = []
+    with open(p, encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                out.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return out
+
+
+def append_ruling(month: str, level: int, reason: str, decided_by: str = "user",
+                  path: str | None = None) -> dict:
+    if not _MONTH_RE.match(month or ""):
+        raise ValueError(f"month 须为 YYYY-MM: {month!r}")
+    if not isinstance(level, int) or isinstance(level, bool) or not 0 <= level <= 6:
+        raise ValueError(f"level 须为 0..6 整数: {level!r}")
+    p = path or default_verdict_records_path()
+    for r in load_rulings(p):
+        if r.get("month") == month:
+            raise KeyError(f"当月已有裁决: {month}(level={r.get('level')})")
+    rec = {"month": month, "level": level, "reason": reason or "",
+           "decided_by": decided_by,
+           "decided_at": datetime.now().isoformat(timespec="seconds")}
+    os.makedirs(os.path.dirname(os.path.abspath(p)), exist_ok=True)
+    with open(p, "a", encoding="utf-8") as f:
+        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+    return rec
@@ -0,0 +1,56 @@
+# tests/portfolio/test_verdict_records.py
+"""研判留痕 TDD——spec §4.7 P3 第一刀之 2(七档唯一源+裁决 JSONL)."""
+import json
+
+import pytest
+
+from sanguo_portfolio import verdict_records as vr
+
+
+def test_seven_tiers_shape():
+    assert [t["level"] for t in vr.SEVEN_TIERS] == list(range(7))
+    for t in vr.SEVEN_TIERS:
+        assert set(t) == {"level", "action", "meaning", "cost"}
+    assert vr.SEVEN_TIERS[6]["action"] == "防御模式（最重）"
+
+
+def test_append_and_load_roundtrip(tmp_path):
+    p = str(tmp_path / "verdict_records.jsonl")
+    rec = vr.append_ruling("2026-09", 2, "小盘逆风先挪钱", path=p)
+    assert rec["month"] == "2026-09" and rec["level"] == 2
+    rows = vr.load_rulings(p)
+    assert len(rows) == 1
+    assert rows[0]["reason"] == "小盘逆风先挪钱"
+    assert rows[0]["decided_at"]
+
+
+def test_duplicate_month_rejected(tmp_path):
+    p = str(tmp_path / "v.jsonl")
+    vr.append_ruling("2026-09", 0, "", path=p)
+    with pytest.raises(KeyError):
+        vr.append_ruling("2026-09", 6, "再来一次", path=p)
+
+
+@pytest.mark.parametrize("month,level", [
+    ("2026-9", 1), ("26-09", 1), ("2026-13", 1), ("", 1),
+    ("2026-09", 7), ("2026-09", -1), ("2026-09", "2"),
+])
+def test_invalid_inputs(month, level):
+    with pytest.raises(ValueError):
+        vr.append_ruling(month, level, "", path="/tmp/never-written.jsonl")
+
+
+def test_load_missing_file_returns_empty(tmp_path):
+    assert vr.load_rulings(str(tmp_path / "nope.jsonl")) == []
+
+
+def test_corrupt_line_skipped(tmp_path):
+    p = tmp_path / "v.jsonl"
+    p.write_text('{"month": "2026-09", "level": 1}\nnot-json\n', encoding="utf-8")
+    rows = vr.load_rulings(str(p))
+    assert len(rows) == 1 and rows[0]["month"] == "2026-09"
+
+
+def test_default_path_env(monkeypatch, tmp_path):
+    monkeypatch.setenv("SANGUO_VERDICT_RECORDS", str(tmp_path / "e.jsonl"))
+    assert vr.default_verdict_records_path() == str(tmp_path / "e.jsonl")