90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
"""策略实例 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
|