Files
QuantEngineByItz/tools/validate_gas_call_arity.py
T
kjh2064 ee3e799de1 feat: 리밸런싱 엔진 V1 + GAS 버그 수정 (2026-06-13)
주요 변경:
- 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>
2026-06-13 13:20:14 +09:00

236 lines
6.6 KiB
Python

from __future__ import annotations
import re
import sys
from dataclasses import dataclass
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
GAS_FILES = sorted(ROOT.glob("gas*.gs"))
@dataclass
class FunctionDef:
name: str
min_arity: int
max_arity: int
path: Path
line: int
@dataclass
class CallSite:
name: str
arity: int
path: Path
line: int
def sanitize_js(text: str) -> str:
out: list[str] = []
i = 0
n = len(text)
state = "normal"
while i < n:
ch = text[i]
nxt = text[i + 1] if i + 1 < n else ""
if state == "normal":
if ch == "/" and nxt == "/":
out.extend(" ")
i += 2
state = "line_comment"
continue
if ch == "/" and nxt == "*":
out.extend(" ")
i += 2
state = "block_comment"
continue
if ch == "'":
out.append(" ")
i += 1
state = "single_quote"
continue
if ch == '"':
out.append(" ")
i += 1
state = "double_quote"
continue
if ch == "`":
out.append(" ")
i += 1
state = "template"
continue
out.append(ch)
i += 1
continue
if state == "line_comment":
if ch == "\n":
out.append("\n")
state = "normal"
else:
out.append(" ")
i += 1
continue
if state == "block_comment":
if ch == "*" and nxt == "/":
out.extend(" ")
i += 2
state = "normal"
else:
out.append("\n" if ch == "\n" else " ")
i += 1
continue
if state in {"single_quote", "double_quote", "template"}:
if ch == "\\":
out.append("S")
if i + 1 < n:
out.append("S" if text[i + 1] != "\n" else "\n")
i += 2
continue
if (state == "single_quote" and ch == "'") or (
state == "double_quote" and ch == '"'
) or (state == "template" and ch == "`"):
out.append(" ")
i += 1
state = "normal"
else:
out.append("\n" if ch == "\n" else "S")
i += 1
continue
return "".join(out)
def line_of(text: str, index: int) -> int:
return text.count("\n", 0, index) + 1
def find_matching(text: str, start: int, opener: str, closer: str) -> int:
depth = 0
for idx in range(start, len(text)):
ch = text[idx]
if ch == opener:
depth += 1
elif ch == closer:
depth -= 1
if depth == 0:
return idx
return -1
def count_arguments(chunk: str) -> int:
if not chunk.strip():
return 0
depth_paren = 0
depth_brace = 0
depth_bracket = 0
count = 1
for ch in chunk:
if ch == "(":
depth_paren += 1
elif ch == ")":
depth_paren -= 1
elif ch == "{":
depth_brace += 1
elif ch == "}":
depth_brace -= 1
elif ch == "[":
depth_bracket += 1
elif ch == "]":
depth_bracket -= 1
elif ch == "," and depth_paren == 0 and depth_brace == 0 and depth_bracket == 0:
count += 1
return count
def parse_param_bounds(params: str) -> tuple[int, int]:
tokens = [p.strip() for p in params.split(",") if p.strip()]
if not tokens:
return 0, 0
min_arity = 0
max_arity = len(tokens)
for token in tokens:
if "=" not in token and not token.startswith("..."):
min_arity += 1
return min_arity, max_arity
def extract_function_defs(path: Path) -> list[FunctionDef]:
raw = path.read_text(encoding="utf-8")
text = sanitize_js(raw)
defs: list[FunctionDef] = []
for match in re.finditer(r"\bfunction\s+([A-Za-z_$][\w$]*)\s*\(", text):
name = match.group(1)
open_idx = text.find("(", match.start())
close_idx = find_matching(text, open_idx, "(", ")")
if close_idx == -1:
continue
params = text[open_idx + 1 : close_idx]
min_arity, max_arity = parse_param_bounds(params)
defs.append(
FunctionDef(
name=name,
min_arity=min_arity,
max_arity=max_arity,
path=path,
line=line_of(raw, match.start()),
)
)
return defs
def extract_calls(path: Path, function_names: set[str]) -> list[CallSite]:
raw = path.read_text(encoding="utf-8")
text = sanitize_js(raw)
calls: list[CallSite] = []
for name in sorted(function_names):
pattern = re.compile(rf"\b{re.escape(name)}\s*\(")
for match in pattern.finditer(text):
prefix = text[max(0, match.start() - 16) : match.start()]
if re.search(r"\bfunction\s+$", prefix):
continue
open_idx = text.find("(", match.start())
close_idx = find_matching(text, open_idx, "(", ")")
if close_idx == -1:
continue
args = text[open_idx + 1 : close_idx]
calls.append(CallSite(name=name, arity=count_arguments(args), path=path, line=line_of(raw, match.start())))
return calls
def main() -> int:
if not GAS_FILES:
print("No gas*.gs files found.")
return 1
defs: list[FunctionDef] = []
for path in GAS_FILES:
defs.extend(extract_function_defs(path))
def_map = {item.name: item for item in defs}
errors: list[str] = []
for path in GAS_FILES:
for call in extract_calls(path, set(def_map)):
defined = def_map.get(call.name)
if defined is None:
continue
if not (defined.min_arity <= call.arity <= defined.max_arity):
errors.append(
f"{path.name}:{call.line}: {call.name} call arity={call.arity} "
f"outside definition range={defined.min_arity}..{defined.max_arity} "
f"at {defined.path.name}:{defined.line}"
)
if errors:
print("GAS CALL ARITY VALIDATION FAILED")
for err in errors:
print(f"- {err}")
return 1
print(f"GAS CALL ARITY OK: checked {len(def_map)} functions across {len(GAS_FILES)} files")
return 0
if __name__ == "__main__":
sys.exit(main())