INNER CODE UNIT · Python
_load_history_json
simonlin1212/Vibe-Research · calc/cli.py:189
def _load_history_json(spec: dict, run_dir: str | None) -> tuple[list, dict]:
"""从 <run_dir>/raw/ 的 JSON(或 JSONP:取第一个 '{' 到最后一个 '}')确定性加载一张 K 线表:
spec = {raw_ref, rows_path: "data.sz300308.qfqday", columns: {"date": 0, "open": 1, "close": 2, "high": 3, "low": 4} 或 {"date": "date", ...}, where?: {"tradestatus": "1"}}
返回 (rows[list[dict]], 解析记录)。行可以是数组(列给下标)或对象(列给键名);where 只对对象行生效。"""
if not isinstance(spec, dict):
raise ValueError("history_json 必须是对象 {raw_ref, rows_path, columns, where?}")
rel = str(spec.get("raw_ref") or "")
rows_path = str(spec.get("rows_path") or "")
cols = spec.get("columns")
where = spec.get("where") or {}
if not isinstance(cols, dict) or not cols or "date" not in cols:
raise ValueError("history_json.columns 必须是含 date 的映射 {字段名: 下标或键名}")
if not isinstance(where, dict):
raise ValueError("history_json.where 必须是对象")
target = _raw_target(rel, run_dir, "history_json")
data = target.read_bytes()
sha = hashlib.sha256(data).hexdigest()
obj = _parse_json_or_jsonp(data)