feat(strategy): 策略实例 JSON 存储 CRUD(list/get/create/update/delete) [nas]

sanguo_api/instance_store.py

@@ -0,0 +1,89 @@
+"""策略实例 JSON 存储（配置数据，几行；无需 DB 迁移）。
+
+实例 = 代码文件的参数变体（spec §12 三层模型中层）。
+运行态字段 status/last_return 给默认占位，由后续运行关联回填。
+"""
+from __future__ import annotations
+import json
+import os
+import time
+
+_DEFAULT_STATUS = {"backtest": "-", "replay": "-", "paper_live": "-", "live": "-"}
+
+# 路径：env 覆盖 > 仓库根 data/strategy_instances.json。测试用 monkeypatch 改 _STORE_PATH。
+_STORE_PATH: str = os.environ.get(
+    "SANGUO_INSTANCE_STORE",
+    os.path.join(
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+        "data", "strategy_instances.json",
+    ),
+)
+
+
+def _load() -> list[dict]:
+    if not os.path.exists(_STORE_PATH):
+        return []
+    with open(_STORE_PATH, encoding="utf-8") as f:
+        try:
+            return json.load(f)
+        except json.JSONDecodeError:
+            return []
+
+
+def _save(rows: list[dict]) -> None:
+    os.makedirs(os.path.dirname(_STORE_PATH), exist_ok=True)
+    with open(_STORE_PATH, "w", encoding="utf-8") as f:
+        json.dump(rows, f, ensure_ascii=False, indent=2)
+
+
+def list_instances() -> dict:
+    return {"instances": _load()}
+
+
+def get_instance(inst_id: int) -> dict:
+    for r in _load():
+        if r.get("id") == inst_id:
+            return {"instance": r}
+    return {"instance": None}
+
+
+def create_instance(payload: dict) -> int:
+    rows = _load()
+    new_id = (max((r.get("id", 0) for r in rows), default=0) + 1) if rows else 1
+    row = {
+        "id": new_id,
+        "code_file": payload.get("code_file", ""),
+        "name": payload.get("name", ""),
+        "type": payload.get("type", "cta"),
+        "params": payload.get("params", {}),
+        "symbol_or_pool": payload.get("symbol_or_pool", ""),
+        "interval": payload.get("interval", "d"),
+        "match_session": payload.get("match_session", "next_open"),
+        "status": dict(_DEFAULT_STATUS),
+        "last_return": None,
+        "updated_at": time.strftime("%Y-%m-%d"),
+    }
+    rows.append(row)
+    _save(rows)
+    return new_id
+
+
+def update_instance(inst_id: int, payload: dict) -> None:
+    rows = _load()
+    for r in rows:
+        if r.get("id") == inst_id:
+            for k in ("code_file", "name", "type", "params", "symbol_or_pool", "interval", "match_session"):
+                if k in payload:
+                    r[k] = payload[k]
+            r["updated_at"] = time.strftime("%Y-%m-%d")
+            _save(rows)
+            return
+
+
+def delete_instance(inst_id: int) -> bool:
+    rows = _load()
+    new_rows = [r for r in rows if r.get("id") != inst_id]
+    if len(new_rows) == len(rows):
+        return False
+    _save(new_rows)
+    return True

tests/api/test_instance_store.py

@@ -0,0 +1,43 @@
+"""Tests for sanguo_api.instance_store (策略实例 JSON 存储)."""
+from sanguo_api import instance_store
+
+
+def _fresh_store(tmp_path, monkeypatch):
+    db = tmp_path / "instances.json"
+    monkeypatch.setattr(instance_store, "_STORE_PATH", str(db))
+    return str(db)
+
+
+def test_create_get_list_delete(monkeypatch, tmp_path):
+    _fresh_store(tmp_path, monkeypatch)
+    new_id = instance_store.create_instance({
+        "code_file": "double_ma.py", "name": "测试实例", "type": "cta",
+        "params": {"fast_window": 10}, "symbol_or_pool": "600519.SH",
+        "interval": "d", "match_session": "next_open",
+    })
+    assert isinstance(new_id, int) and new_id > 0
+
+    got = instance_store.get_instance(new_id)["instance"]
+    assert got["name"] == "测试实例" and got["params"]["fast_window"] == 10
+
+    items = instance_store.list_instances()["instances"]
+    assert any(i["id"] == new_id for i in items)
+
+    assert instance_store.delete_instance(new_id) is True
+    assert instance_store.get_instance(new_id)["instance"] is None
+
+
+def test_update_instance(monkeypatch, tmp_path):
+    _fresh_store(tmp_path, monkeypatch)
+    nid = instance_store.create_instance({
+        "code_file": "double_ma.py", "name": "u", "type": "cta",
+        "params": {}, "symbol_or_pool": "000001.SZ", "interval": "d", "match_session": "next_open",
+    })
+    instance_store.update_instance(nid, {"name": "改名", "params": {"fast_window": 20}})
+    got = instance_store.get_instance(nid)["instance"]
+    assert got["name"] == "改名" and got["params"]["fast_window"] == 20
+
+
+def test_delete_missing_returns_false(monkeypatch, tmp_path):
+    _fresh_store(tmp_path, monkeypatch)
+    assert instance_store.delete_instance(99999) is False