kjh2064
af1236202d
- F14: late_chase_risk_score 검증 * GAS가 유일한 생산처 (Python canonical 없음) * migration_action: KEEP_IN_GAS로 정정, status: DONE - F02/F03/F04/F06: priceBasis 로직 포팅 * formulas/price_basis_v1.py: select_price_basis_tier2/tier1 구현 * tests/parity/test_price_basis_parity_v1.py: 8 parity 테스트 (모두 PASS) * GAS Number.isFinite() 의미론 정확히 재현 (math.isfinite 사용) * 모든 테스트 112/112 PASS 남은 작업 (4개): - F05: decision_logic (action assignment) - F07: score_logic (threshold addition) - F10: routing decision - F15: late_chase_gate Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
def main() -> int:
|
|
graph_path = ROOT / "spec" / "routing" / "decision_graph.yaml"
|
|
if not graph_path.exists():
|
|
print(f"Decision graph spec missing: {graph_path}")
|
|
return 1
|
|
|
|
try:
|
|
graph_data = yaml.safe_load(graph_path.read_text(encoding="utf-8")) or {}
|
|
except Exception as e:
|
|
print(f"Failed to parse decision graph: {e}")
|
|
return 1
|
|
|
|
nodes = graph_data.get("nodes", [])
|
|
edges = graph_data.get("edges", [])
|
|
|
|
# Build adjacency list
|
|
adj = {}
|
|
for node in nodes:
|
|
nid = node.get("id")
|
|
adj[nid] = []
|
|
|
|
for edge in edges:
|
|
if len(edge) == 2:
|
|
u, v = edge[0], edge[1]
|
|
if u in adj and v in adj:
|
|
adj[u].append(v)
|
|
else:
|
|
# If nodes are not declared, dynamically add them
|
|
if u not in adj:
|
|
adj[u] = []
|
|
if v not in adj:
|
|
adj[v] = []
|
|
adj[u].append(v)
|
|
|
|
errors = []
|
|
# Check topological sort order
|
|
in_degree = {n: 0 for n in adj}
|
|
for u in adj:
|
|
for v in adj[u]:
|
|
in_degree[v] += 1
|
|
|
|
# Find nodes with 0 in-degree
|
|
queue = [n for n in adj if in_degree[n] == 0]
|
|
topo_order = []
|
|
while queue:
|
|
curr = queue.pop(0)
|
|
topo_order.append(curr)
|
|
for v in adj.get(curr, []):
|
|
in_degree[v] -= 1
|
|
if in_degree[v] == 0:
|
|
queue.append(v)
|
|
|
|
# If topological sort is not successful (has cycle), fail
|
|
if len(topo_order) != len(adj):
|
|
errors.append("Decision graph contains a cycle")
|
|
gate_passed = False
|
|
else:
|
|
anti_chase_idx = -1
|
|
if "anti_chase" in topo_order:
|
|
anti_chase_idx = topo_order.index("anti_chase")
|
|
else:
|
|
errors.append("anti_chase node not found in graph")
|
|
|
|
target_nodes = ["regime", "sector_beta", "style", "sizing", "execution"]
|
|
if anti_chase_idx != -1:
|
|
for t in target_nodes:
|
|
if t in topo_order:
|
|
t_idx = topo_order.index(t)
|
|
if anti_chase_idx >= t_idx:
|
|
errors.append(f"anti_chase (index {anti_chase_idx}) does not precede {t} (index {t_idx})")
|
|
else:
|
|
# Missing target node is a failure
|
|
errors.append(f"Target node {t} not found in topological order")
|
|
|
|
gate_passed = len(errors) == 0
|
|
|
|
result = {
|
|
"formula_id": "DECISION_GRAPH_PRECEDENCE_VALIDATOR_V1",
|
|
"topo_order": topo_order,
|
|
"errors": errors,
|
|
"gate": "PASS" if gate_passed else "FAIL"
|
|
}
|
|
|
|
# Write output to Temp
|
|
out_dir = ROOT / "Temp"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
out_path = out_dir / "decision_graph_precedence_validation_v1.json"
|
|
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
print(json.dumps(result, ensure_ascii=True, indent=2))
|
|
return 0 if gate_passed else 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|