kjh2064
ee3e799de1
주요 변경: - tools/build_rebalance_engine_v1.py: REBALANCE_ENGINE_V1 신규 * account_snapshot 직접 합산(_build_snap_position_map) → 소수주 분리 행 병합 * 레짐 소스 macro.REGIME_PRELIM 최우선 (GAS 와 동일) - src/gas_adapter_parts/gdf_06_rebalance.gs: runRebalanceSheet_() 신규 * Logger.log / getSpreadsheet_() 로 run_all 연동 수정 - src/gas_adapter_parts/gdc_01_fetch_fundamentals.gs * _mergePositionRecord_(): 소수주 중복 행 합산 신규 * parseInt → parseFloat (qty, availQty) - src/gas_adapter_parts/gdf_01_price_metrics.gs * 미보유 종목 SELL_READY → WATCH_EXIT_SIGNAL - spec/41_release_dag.yaml: build_rebalance_sheet 노드 추가 (step_count 63) - spec/51_formula_lifecycle_registry.yaml: REBALANCE_ENGINE_V1 등록 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
95 lines
4.1 KiB
Python
95 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _load_yaml(path: Path) -> dict[str, Any]:
|
|
if not path.exists():
|
|
return {}
|
|
payload = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
return payload if isinstance(payload, dict) else {}
|
|
|
|
|
|
def _domain_for(formula_id: str, formula: dict[str, Any]) -> str:
|
|
fid = formula_id.upper()
|
|
text = " ".join([fid, str(formula.get("purpose", "")), str(formula.get("canonical_ref", "")), str(formula.get("agents_md_ref", ""))]).upper()
|
|
if any(k in text for k in ["CASH", "SHORTFALL", "FLOOR", "RECOVERY", "RAISE", "LIQUIDITY", "SLIPPAGE"]):
|
|
return "cash"
|
|
if any(k in text for k in ["ENTRY", "BREAKOUT", "CHASE", "ALPHA", "TIMING", "FOLLOW_THROUGH", "TRANCHE", "PULLBACK"]):
|
|
return "entry"
|
|
if any(k in text for k in ["EXIT", "SELL", "STOP", "TAKE_PROFIT", "WATERFALL", "REBOUND", "PRESERVATION", "TRAILING"]):
|
|
return "exit"
|
|
if any(k in text for k in ["PORTFOLIO", "POSITION", "HEAT", "BETA", "SECTOR", "REGIME", "WEIGHT", "CONCENTRATION", "DRAWDOWN"]):
|
|
return "portfolio"
|
|
if any(k in text for k in ["REPORT", "RUNTIME", "DASHBOARD", "LEDGER", "AUDIT", "TRACE", "NARRATIVE", "QUALITY", "PROOF", "DECISION"]):
|
|
return "reporting"
|
|
return "risk"
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--in", dest="input_path", default="spec/13_formula_registry.yaml")
|
|
parser.add_argument("--out", dest="out_dir", default="spec/formulas")
|
|
args = parser.parse_args()
|
|
|
|
src = ROOT / args.input_path
|
|
out_dir = ROOT / args.out_dir
|
|
payload = _load_yaml(src)
|
|
formulas = ((payload.get("formula_registry") or {}).get("formulas")) or {}
|
|
|
|
buckets: dict[str, dict[str, Any]] = {k: {"schema_version": "formula_domain.v1", "source": str(src), "domain": k, "formulas": {}} for k in ["risk", "entry", "exit", "cash", "portfolio", "reporting", "fundamental", "smart_money", "macro"]}
|
|
for fid, formula in formulas.items():
|
|
domain = _domain_for(str(fid), formula if isinstance(formula, dict) else {})
|
|
|
|
# Auto-populate required contract fields
|
|
f_dict = dict(formula) if isinstance(formula, dict) else {}
|
|
if "owner" not in f_dict:
|
|
f_dict["owner"] = "quant_team"
|
|
if "lifecycle_state" not in f_dict:
|
|
f_dict["lifecycle_state"] = "active"
|
|
if "input_fields" not in f_dict:
|
|
inputs = f_dict.get("inputs") or []
|
|
f_dict["input_fields"] = [inp["field"] for inp in inputs if isinstance(inp, dict) and "field" in inp]
|
|
if "output_fields" not in f_dict:
|
|
out = f_dict.get("output")
|
|
if isinstance(out, dict) and "field" in out:
|
|
f_dict["output_fields"] = [out["field"]]
|
|
else:
|
|
f_dict["output_fields"] = []
|
|
if "missing_policy" not in f_dict:
|
|
f_dict["missing_policy"] = "DATA_MISSING. 계산 결과를 추정하지 않는다."
|
|
if "golden_cases" not in f_dict:
|
|
f_dict["golden_cases"] = []
|
|
if "activation_threshold" not in f_dict:
|
|
f_dict["activation_threshold"] = {"min_t20_sample": 30}
|
|
if "retirement_condition" not in f_dict:
|
|
f_dict["retirement_condition"] = "performance_degradation"
|
|
|
|
buckets[domain]["formulas"][fid] = f_dict
|
|
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
for domain, doc in buckets.items():
|
|
(out_dir / f"{domain}.yaml").write_text(yaml.safe_dump(doc, sort_keys=False, allow_unicode=True), encoding="utf-8")
|
|
|
|
manifest = {
|
|
"schema_version": "formula_domain_manifest.v1",
|
|
"source": str(src),
|
|
"domains": {domain: f"spec/formulas/{domain}.yaml" for domain in buckets},
|
|
"formula_count": len(formulas),
|
|
}
|
|
(out_dir / "manifest.yaml").write_text(yaml.safe_dump(manifest, sort_keys=False, allow_unicode=True), encoding="utf-8")
|
|
print(yaml.safe_dump(manifest, sort_keys=False, allow_unicode=True).strip())
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
|