from __future__ import annotations
|
|
import hashlib
|
import json
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
TZ = timezone(timedelta(hours=8))
|
|
|
def sha256_file(path: Path) -> str:
|
h = hashlib.sha256()
|
with path.open("rb") as f:
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
h.update(chunk)
|
return h.hexdigest()
|
|
|
def main() -> None:
|
template = pd.read_csv(ROOT / "manual_buy_decision_external_template.csv", encoding="utf-8-sig")
|
chart_audit = pd.read_csv(ROOT / "chart_evidence_audit.csv", encoding="utf-8-sig")
|
required = {
|
"external_decision_id",
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"signal_trade_date",
|
"entry_trade_date",
|
"code_suggested_action",
|
"review_input_chart_path",
|
"review_input_chart_sha256",
|
}
|
missing = sorted(required - set(template.columns))
|
if missing:
|
raise RuntimeError(f"manual template missing columns: {missing}")
|
|
workbench = template.copy()
|
for col in [
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"accept_code_suggestion_flag",
|
"reviewer_notes",
|
]:
|
workbench[col] = ""
|
|
for row in workbench.itertuples(index=False):
|
chart = ROOT / row.review_input_chart_path
|
if not chart.exists():
|
raise RuntimeError(f"missing chart: {row.review_input_chart_path}")
|
if sha256_file(chart) != row.review_input_chart_sha256:
|
raise RuntimeError(f"chart hash mismatch: {row.review_input_chart_path}")
|
|
workbench.to_csv(ROOT / "manual_buy_decision_workbench_blank.csv", index=False, encoding="utf-8-sig")
|
summary = {
|
"run_id": RUN_ID,
|
"generated_at": datetime.now(TZ).isoformat(timespec="seconds"),
|
"rows": int(len(workbench)),
|
"chart_rows": int(len(chart_audit)),
|
"purpose": "blank workbench only; final human decisions are not generated here",
|
"forbidden_fields_left_blank": [
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"accept_code_suggestion_flag",
|
"reviewer_notes",
|
],
|
}
|
(ROOT / "manual_buy_decision_workbench_summary.json").write_text(
|
json.dumps(summary, ensure_ascii=False, indent=2),
|
encoding="utf-8",
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
