forked from chiki2bum2/SniperGold_ML
118 lines
No EOL
5 KiB
Python
118 lines
No EOL
5 KiB
Python
"""Headless CLI qualification (spec 14): all commands + exit codes, run on
|
|
synthetic fixtures only, completely without any interactive session."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
PY = r"C:\Users\defau\AppData\Local\Python\bin\python.exe"
|
|
CLI = os.path.join(REPO, "sniper-data.py")
|
|
|
|
G01 = os.path.join(REPO, "tests", "golden", "cases", "G01.txt")
|
|
LEGACY_EVIDENCE = os.path.join(REPO, "ml", "p3", "p3_s251_external_ingest")
|
|
|
|
|
|
def run_cli(args, cwd=REPO):
|
|
return subprocess.run([PY, CLI] + args, cwd=cwd, capture_output=True,
|
|
text=True, timeout=300)
|
|
|
|
|
|
def write_json(path, obj):
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
json.dump(obj, fh, indent=2, sort_keys=True)
|
|
|
|
|
|
def run():
|
|
os.makedirs(os.path.join(REPO, "qualification"), exist_ok=True)
|
|
with tempfile.TemporaryDirectory(
|
|
dir=os.path.join(REPO, "qualification")) as runroot:
|
|
cfg = {
|
|
"source_path": os.path.abspath(G01),
|
|
"source_grammar": "dotted_v1",
|
|
"source_tz_offset_minutes": 0,
|
|
"has_volume": False,
|
|
"timeframes": ["M1", "M5", "M15", "M30", "H1"],
|
|
"chunk_bytes_nominal": 25_165_824,
|
|
"workload_bytes_nominal": 536_870_912,
|
|
"output_root": runroot,
|
|
"workers_requested": 1,
|
|
"retry_limit": 2,
|
|
}
|
|
cfg_path = os.path.join(runroot, "engine_config.json")
|
|
write_json(cfg_path, cfg)
|
|
research = {
|
|
"dataset_id": "xau_cli_v1",
|
|
"source_dataset_version": "DS_V1.0.0",
|
|
"timeframe": "M1",
|
|
"prediction_horizon_min": 1,
|
|
"target": {"type": "direction", "horizon": "1m", "from": "close"},
|
|
"features": [{"name": "ret1", "primitive": "return", "window": 1}],
|
|
"split": {"policy": "time_ordered", "train": 0.7, "val": 0.15,
|
|
"test": 0.15, "oos_barrier": False,
|
|
"barrier_gap_min": 1},
|
|
}
|
|
rcfg_path = os.path.join(runroot, "research.json")
|
|
write_json(rcfg_path, research)
|
|
|
|
results = []
|
|
|
|
def check(name, expected_code, args):
|
|
r = run_cli(args)
|
|
results.append((name, r.returncode, expected_code,
|
|
r.stderr.strip()[-300:]))
|
|
|
|
# P3-DE-004 CLI regression: status/resume on a run-less directory must
|
|
# answer the spec exit code 4 (BLOCKED) without a traceback (the
|
|
# P3-DE-003 audit observed exit 1 + TypeError on this path).
|
|
emptyroot = os.path.join(runroot, "empty_runless")
|
|
os.makedirs(emptyroot)
|
|
check("cli_status_runless_dir", 4, ["status", "--out-root", emptyroot])
|
|
check("cli_resume_runless_dir", 4, ["resume", "--out-root", emptyroot])
|
|
check("cli_version", 0, ["--version"])
|
|
check("cli_certify", 0, ["certify", "--source", G01,
|
|
"--out-root", runroot])
|
|
check("cli_init", 0, ["init", "--config", cfg_path])
|
|
check("cli_ingest", 3, ["ingest", "--out-root", runroot])
|
|
check("cli_status_before_finalize", 3,
|
|
["status", "--out-root", runroot])
|
|
check("cli_verify_golden", 0, ["verify", "--mode", "golden"])
|
|
check("cli_verify_dataset", 0,
|
|
["verify", "--mode", "dataset", "--out-root", runroot])
|
|
check("cli_run_complete_exists", 0,
|
|
["verify", "--mode", "run-complete", "--out-root", runroot])
|
|
check("cli_manifest", 0, ["manifest", "--out-root", runroot])
|
|
check("cli_build_dataset", 0,
|
|
["build-dataset", "--config", rcfg_path, "--out-root", runroot])
|
|
check("cli_diff_legacy", 0,
|
|
["diff-legacy", "--legacy-evidence", LEGACY_EVIDENCE,
|
|
"--legacy-checkpoint", os.path.join("output",
|
|
"s251_real_checkpoint.json"),
|
|
"--out-root", runroot])
|
|
check("cli_status_completed", 0, ["status", "--out-root", runroot])
|
|
check("cli_resume_blocked_on_completed", 4,
|
|
["resume", "--out-root", runroot])
|
|
check("cli_pause", 0, ["pause", "--out-root", runroot])
|
|
|
|
run_complete = os.path.join(runroot, "RUN_COMPLETE.json")
|
|
rc_exists = os.path.exists(run_complete)
|
|
verdict = all(code == expected for _n, code, expected, _e in results) \
|
|
and rc_exists
|
|
return {"ok": verdict, "run_complete_created": rc_exists,
|
|
"results": [{"command": n, "code": c, "expected": e,
|
|
"stderr_tail": t}
|
|
for n, c, e, t in results]}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
res = run()
|
|
for r in res["results"]:
|
|
ok = r["code"] == r["expected"]
|
|
print("%-32s code=%d expected=%d %s %s"
|
|
% (r["command"], r["code"], r["expected"], "PASS" if ok else "FAIL",
|
|
r["stderr_tail"]))
|
|
print("run_complete_created", res["run_complete_created"])
|
|
print("OK" if res["ok"] else "FAIL")
|
|
sys.exit(0 if res["ok"] else 1) |