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)) DECISION_SOURCE_PREFIX = "CASE_ANALYSIS_ANALYST_MANUAL_BUY_POINT_CHART_REVIEW_EXTERNAL_DRAFT_BATCH" VALID_ACTIONS = {"BUY", "REVIEW_HELD"} 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 write_csv(df: pd.DataFrame, name: str) -> Path: path = ROOT / name df.to_csv(path, index=False, encoding="utf-8-sig") return path def write_json(obj: dict, name: str) -> Path: path = ROOT / name path.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8") return path def require_nonblank(df: pd.DataFrame, columns: list[str]) -> None: for col in columns: if col not in df.columns: raise RuntimeError(f"missing required column: {col}") if df[col].isna().any() or df[col].astype(str).str.strip().eq("").any(): raise RuntimeError(f"blank required column: {col}") def validate_source(decisions: pd.DataFrame, template: pd.DataFrame) -> None: required = [ "external_decision_id", "candidate_id", "case_id", "symbol", "signal_trade_date", "entry_trade_date", "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "review_input_chart_path", "review_input_chart_sha256", "manual_draft_path", "manual_draft_sha256", ] require_nonblank(decisions, required) if len(decisions) != len(template): raise RuntimeError(f"manual decision row count mismatch: {len(decisions)} != {len(template)}") if decisions["external_decision_id"].duplicated().any(): raise RuntimeError("duplicate external_decision_id") if set(decisions["external_decision_id"]) != set(template["external_decision_id"]): raise RuntimeError("manual decision external_decision_id set mismatch") if not decisions["human_decision_action"].isin(VALID_ACTIONS).all(): raise RuntimeError("invalid manual action") if not decisions["decision_source"].str.startswith(DECISION_SOURCE_PREFIX).all(): raise RuntimeError("unexpected decision_source") decision_times = pd.to_datetime(decisions["decision_time"], utc=True, errors="raise") now_utc = pd.Timestamp(datetime.now(TZ)).tz_convert("UTC") if decision_times.gt(now_utc).any(): future_count = int(decision_times.gt(now_utc).sum()) raise RuntimeError(f"manual decision_time is in the future: {future_count}") for row in decisions.itertuples(index=False): chart = ROOT / row.review_input_chart_path if not chart.exists() or sha256_file(chart) != row.review_input_chart_sha256: raise RuntimeError(f"chart hash mismatch: {row.candidate_id}") draft = ROOT / row.manual_draft_path if not draft.exists() or sha256_file(draft) != row.manual_draft_sha256: raise RuntimeError(f"manual draft hash mismatch: {row.external_decision_id}") def build_manifest() -> pd.DataFrame: rows = [] for path in sorted(ROOT.rglob("*")): if path.is_file() and path.name not in {"manifest.csv", "manifest.json"}: rows.append( { "path": path.relative_to(ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path), } ) return pd.DataFrame(rows) def make_boundary_rows(held_rows: pd.DataFrame, scope: pd.DataFrame) -> list[dict]: rows: list[dict] = [] for row in held_rows.itertuples(index=False): rows.append( { "case_id": row.case_id, "candidate_id": row.candidate_id, "symbol": row.symbol, "boundary_type": "BUY_POINT_REVIEW_HELD", "boundary_reason": row.human_decision_reason_cn, "external_decision_id": row.external_decision_id, } ) closed_scope = scope[scope["buy_point_replay_status"].eq("NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY")] for row in closed_scope.itertuples(index=False): rows.append( { "case_id": row.case_id, "candidate_id": row.candidate_id, "symbol": row.symbol, "boundary_type": "NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY", "boundary_reason": "market gate closed; no new BUY generated", "external_decision_id": "", } ) gap_scope = scope[scope["buy_point_replay_status"].eq("BUY_POINT_MINUTE_DATA_GAP_HELD")] for row in gap_scope.itertuples(index=False): rows.append( { "case_id": row.case_id, "candidate_id": row.candidate_id, "symbol": row.symbol, "boundary_type": "BUY_POINT_MINUTE_DATA_GAP_HELD", "boundary_reason": "entry-day minute data gap; no manual BUY generated", "external_decision_id": "", } ) return rows def main() -> None: decisions = pd.read_csv(ROOT / "manual_buy_decision_external_source_ledger.csv", encoding="utf-8-sig") template = pd.read_csv(ROOT / "manual_buy_decision_external_template.csv", encoding="utf-8-sig") candidates = pd.read_csv(ROOT / "strict_note_buy_point_review_candidate_ledger.csv", encoding="utf-8-sig") scope = pd.read_csv(ROOT / "strict_note_buy_replay_scope.csv", encoding="utf-8-sig") validate_source(decisions, template) joined = candidates.merge( decisions, on=["case_id", "candidate_id", "symbol", "signal_trade_date", "entry_trade_date"], how="left", suffixes=("", "_decision"), ) if joined["human_decision_action"].isna().any(): raise RuntimeError("candidate without manual decision") buy_rows = joined[joined["human_decision_action"].eq("BUY")].copy() held_rows = joined[joined["human_decision_action"].eq("REVIEW_HELD")].copy() order_rows = [] lot_rows = [] for i, row in enumerate(buy_rows.itertuples(index=False), start=1): order_id = f"ORD-{RUN_ID}-{i:05d}" lot_id = f"LOT-{RUN_ID}-{i:05d}" order_rows.append( { "order_id": order_id, "case_id": row.case_id, "candidate_id": row.candidate_id, "external_decision_id": row.external_decision_id, "symbol": row.symbol, "trade_date": row.entry_trade_date, "trade_time": "MANUAL_BUY_POINT_REVIEW_CONFIRMED", "action": "BUY", "price_source": "ENTRY_DAY_MANUAL_REVIEW_CLOSE_PROXY_PENDING_MINUTE_POINT", "price": row.close_price, "position_delta_pct": 0.04, "tranche_index": 1, "order_status": "STRICT_BUY_CONFIRMED_BY_EXTERNAL_MANUAL_DECISION", "decision_source": row.decision_source, "decision_reason_cn": row.human_decision_reason_cn, "evidence_image_path": row.review_input_chart_path, } ) lot_rows.append( { "lot_id": lot_id, "open_order_id": order_id, "case_id": row.case_id, "candidate_id": row.candidate_id, "external_decision_id": row.external_decision_id, "symbol": row.symbol, "entry_trade_date": row.entry_trade_date, "entry_price": row.close_price, "position_pct": 0.04, "lot_status": "OPEN_PENDING_STRICT_SELL_REPLAY", "decision_source": row.decision_source, } ) orders = pd.DataFrame(order_rows) lots = pd.DataFrame(lot_rows) boundary = pd.DataFrame(make_boundary_rows(held_rows, scope)) write_csv(orders, "strict_note_buy_order_ledger.csv") write_csv(lots, "strict_note_buy_lot_ledger.csv") write_csv(boundary, "strict_note_buy_boundary_table.csv") case_summary = ( scope.groupby("case_id") .agg(selected_candidates=("candidate_id", "count")) .reset_index() .merge(orders.groupby("case_id").size().rename("buy_orders").reset_index(), on="case_id", how="left") .merge(boundary.groupby("case_id").size().rename("boundary_rows").reset_index(), on="case_id", how="left") ) case_summary["buy_orders"] = case_summary["buy_orders"].fillna(0).astype(int) case_summary["boundary_rows"] = case_summary["boundary_rows"].fillna(0).astype(int) case_summary["case_status"] = case_summary["buy_orders"].gt(0).map( {True: "STRICT_BUY_CONFIRMED_PENDING_SELL_REPLAY", False: "NO_BUY_OR_BOUNDARY_ONLY"} ) write_csv(case_summary, "strict_note_buy_case_summary.csv") closed_scope = scope[scope["buy_point_replay_status"].eq("NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY")] gap_scope = scope[scope["buy_point_replay_status"].eq("BUY_POINT_MINUTE_DATA_GAP_HELD")] generated_at = datetime.now(TZ).isoformat(timespec="seconds") items = [ ("MANUAL_SOURCE_ROWS_MATCH_TEMPLATE", len(decisions) == len(template), f"decisions={len(decisions)}, template={len(template)}"), ("ALL_READY_CANDIDATES_DECIDED", len(joined) == len(decisions), f"ready={len(joined)}"), ("MARKET_CLOSED_NO_BUY", set(closed_scope["candidate_id"]).isdisjoint(set(orders["candidate_id"])) if not orders.empty else True, f"market_closed={len(closed_scope)}"), ("DATA_GAP_NO_BUY", set(gap_scope["candidate_id"]).isdisjoint(set(orders["candidate_id"])) if not orders.empty else True, f"data_gap={len(gap_scope)}"), ("BUY_ORDER_HAS_EXTERNAL_DECISION", orders["external_decision_id"].astype(str).str.len().gt(0).all() if not orders.empty else True, f"buy_orders={len(orders)}"), ("LOTS_MATCH_BUY_ORDERS", len(lots) == len(orders), f"lots={len(lots)}, orders={len(orders)}"), ("BOUNDARY_ROWS_COMPLETE", len(boundary) == len(held_rows) + len(closed_scope) + len(gap_scope), f"boundary={len(boundary)}"), ] self_items = pd.DataFrame([{"item": k, "status": "PASS" if ok else "FAIL", "detail": detail} for k, ok, detail in items]) write_csv(self_items, "buy_apply_self_check_items.csv") status = "PASS_FOR_STRICT_BUY_DECISION_APPLICATION_REVIEW_READY" if self_items["status"].eq("PASS").all() else "FAIL" write_json( { "run_id": RUN_ID, "generated_at": generated_at, "stage": status, "pass_count": int(self_items["status"].eq("PASS").sum()), "fail_count": int(self_items["status"].eq("FAIL").sum()), }, "buy_apply_self_check.json", ) write_json( { "run_id": RUN_ID, "generated_at": generated_at, "stage": status, "counts": { "manual_decisions": int(len(decisions)), "buy_orders": int(len(orders)), "open_lots_pending_sell_replay": int(len(lots)), "buy_review_held": int(len(held_rows)), "market_gate_closed_boundary": int(len(closed_scope)), "minute_data_gap_boundary": int(len(gap_scope)), "boundary_rows_total": int(len(boundary)), }, "boundary": "This package applies external BUY decisions only; sell/trend/rolling replay and returns are not generated yet.", }, "buy_apply_summary.json", ) (ROOT / "buy_apply_summary.md").write_text( "# Strict Note BUY Decision Application Summary\n\n" f"- generated_at: {generated_at}\n" f"- manual decisions: {len(decisions)}\n" f"- BUY orders: {len(orders)}\n" f"- open lots pending sell replay: {len(lots)}\n" f"- review-held among ready candidates: {len(held_rows)}\n" f"- market-gate-closed boundary: {len(closed_scope)}\n" f"- minute-data-gap boundary: {len(gap_scope)}\n" f"- boundary rows total: {len(boundary)}\n\n" "Boundary: this package applies external BUY decisions only. It does not generate SELL orders, returns, win rate, drawdown, or strategy-effectiveness conclusions.\n", encoding="utf-8", ) manifest = build_manifest() write_csv(manifest, "manifest.csv") write_json( { "run_id": RUN_ID, "generated_at": generated_at, "file_count": int(len(manifest)), "files": manifest.to_dict(orient="records"), }, "manifest.json", ) if __name__ == "__main__": main()