You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
"""Helpers for reading effective baked sleeves from experiment manifests."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_TOP_LEVEL_KEYS = {
|
|
"idle_alpha_sleeve_preset": "idle_alpha",
|
|
"form4_capture_sleeve_preset": "form4_sleeve",
|
|
"ownership_capture_sleeve_preset": "ownership_sleeve",
|
|
"risk_off_alpha_sleeve_preset": "risk_off_sleeve",
|
|
}
|
|
|
|
|
|
def _empty_baked_sleeves() -> dict[str, str | None]:
|
|
return {
|
|
"parking": None,
|
|
"idle_alpha": None,
|
|
"form4_sleeve": None,
|
|
"ownership_sleeve": None,
|
|
"risk_off_sleeve": None,
|
|
}
|
|
|
|
|
|
def _apply_payload_baked_sleeves(target: dict[str, str | None], payload: dict[str, Any]) -> None:
|
|
risk = payload.get("risk")
|
|
if isinstance(risk, dict) and "cash_parking_preset" in risk:
|
|
target["parking"] = risk.get("cash_parking_preset")
|
|
for src_key, dst_key in _TOP_LEVEL_KEYS.items():
|
|
if src_key in payload:
|
|
target[dst_key] = payload.get(src_key)
|
|
|
|
overrides = payload.get("overrides")
|
|
if not isinstance(overrides, dict):
|
|
return
|
|
|
|
ov_risk = overrides.get("risk")
|
|
if isinstance(ov_risk, dict) and "cash_parking_preset" in ov_risk:
|
|
target["parking"] = ov_risk.get("cash_parking_preset")
|
|
for src_key, dst_key in _TOP_LEVEL_KEYS.items():
|
|
if src_key in overrides:
|
|
target[dst_key] = overrides.get(src_key)
|
|
|
|
|
|
def _resolve_base_config_path(base_ref: str, config_path: Path, project_root: Path) -> Path | None:
|
|
candidate = Path(base_ref)
|
|
if candidate.is_absolute():
|
|
return candidate if candidate.exists() else None
|
|
|
|
for path in (config_path.parent / candidate, project_root / candidate):
|
|
if path.exists():
|
|
return path
|
|
return None
|
|
|
|
|
|
def _read_recursive(
|
|
config_path: Path,
|
|
project_root: Path,
|
|
visited: set[Path],
|
|
depth_left: int,
|
|
) -> dict[str, str | None]:
|
|
baked = _empty_baked_sleeves()
|
|
if depth_left < 0 or not config_path.exists():
|
|
return baked
|
|
|
|
resolved_path = config_path.resolve()
|
|
if resolved_path in visited:
|
|
return baked
|
|
visited.add(resolved_path)
|
|
|
|
try:
|
|
payload = json.loads(config_path.read_text())
|
|
except Exception:
|
|
return baked
|
|
if not isinstance(payload, dict):
|
|
return baked
|
|
|
|
base_ref = payload.get("base_config")
|
|
if isinstance(base_ref, str) and base_ref:
|
|
base_path = _resolve_base_config_path(base_ref, config_path, project_root)
|
|
if base_path is not None:
|
|
baked = _read_recursive(base_path, project_root, visited, depth_left - 1)
|
|
|
|
# Current payload wins. Explicit nulls are meaningful because they clear inherited sleeves.
|
|
_apply_payload_baked_sleeves(baked, payload)
|
|
return baked
|
|
|
|
|
|
def read_effective_baked_sleeves(
|
|
config_path: Path,
|
|
project_root: Path,
|
|
max_depth: int = 8,
|
|
) -> dict[str, str | None]:
|
|
"""Resolve effective baked sleeves from a manifest/base_config chain."""
|
|
try:
|
|
return _read_recursive(config_path, project_root, visited=set(), depth_left=max_depth)
|
|
except Exception:
|
|
return _empty_baked_sleeves()
|