from __future__ import annotations import hashlib import json from datetime import datetime from pathlib import Path import pandas as pd RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001" ROOT = Path(__file__).resolve().parents[1] 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 check(name: str, passed: bool, detail: str, rows: list[dict]) -> None: rows.append( { "check_name": name, "status": "PASS" if passed else "FAIL", "detail": detail, } ) def main() -> None: checks: list[dict] = [] order_ledger = pd.read_csv(ROOT / "order_ledger.csv", encoding="utf-8-sig") lots = pd.read_csv(ROOT / "position_lot_ledger.csv", encoding="utf-8-sig") decisions = pd.read_csv(ROOT / "decision_log.csv", encoding="utf-8-sig") sell_decisions = pd.read_csv(ROOT / "sell_decision_log.csv", encoding="utf-8-sig") exit_resolution = pd.read_csv(ROOT / "exit_resolution_log.csv", encoding="utf-8-sig") account = pd.read_csv(ROOT / "daily_account_ledger.csv", encoding="utf-8-sig") case_summary = pd.read_csv(ROOT / "case_summary.csv", encoding="utf-8-sig") image_manifest = pd.read_csv(ROOT / "image_manifest.csv", encoding="utf-8-sig") action_counts = order_ledger.action.value_counts().to_dict() closed = lots[lots.lot_status == "CLOSED_BY_AI_SELL"].copy() unresolved = lots[lots.lot_status != "CLOSED_BY_AI_SELL"].copy() check( "order_counts", action_counts.get("BUY", 0) == 14 and action_counts.get("SELL", 0) == len(closed), f"{action_counts}; closed_lots={len(closed)}", checks, ) stage_counts = decisions.groupby(["decision_stage", "action_status"]).size().to_dict() check("decision_log_has_entry_and_exit", decisions.decision_stage.isin(["ENTRY_AI_REVIEW", "EXIT_AI_REVIEW"]).all(), str(stage_counts), checks) check("lot_status_counts", len(closed) + len(unresolved) == 14, f"closed={len(closed)}, unresolved={len(unresolved)}", checks) check( "unresolved_status_explicit", set(unresolved.lot_status).issubset({"WINDOW_END_VALUATION_ONLY", "EXIT_DATA_GAP_HELD", "EXIT_REVIEW_HELD"}), ",".join(sorted(set(unresolved.lot_status))), checks, ) check( "exit_resolution_all_lots_explicit", len(exit_resolution) == len(lots) and not exit_resolution.final_action_status.isna().any(), f"resolution_rows={len(exit_resolution)}; statuses={exit_resolution.final_action_status.value_counts().to_dict()}", checks, ) sell_orders = order_ledger[order_ledger.action == "SELL"].copy() merged = sell_orders.merge(lots, left_on="source_lot_id", right_on="trade_lot_id", suffixes=("_order", "_lot")) t1_ok = ( pd.to_datetime(merged.trade_date) >= pd.to_datetime(merged.sellable_from_trade_date) ).all() and ( pd.to_datetime(merged.trade_date) > pd.to_datetime(merged.entry_trade_date) ).all() check("t1_guard_for_sell_orders", bool(t1_ok), f"sell_orders={len(sell_orders)}", checks) recompute_ok = True for _, row in closed.iterrows(): entry = float(row.entry_price) exit_price = float(row.exit_price) position = float(row.position_pct) lot_ret = exit_price / entry - 1 contrib = lot_ret * position recompute_ok = recompute_ok and abs(lot_ret - float(row.lot_return_pct)) < 1e-6 recompute_ok = recompute_ok and abs(contrib - float(row.account_return_contribution_pct)) < 1e-6 check("lot_return_recompute", bool(recompute_ok), f"closed_lots={len(closed)}", checks) case_ok = True for _, row in case_summary.iterrows(): group = lots[lots.case_id == row.case_id] contrib = pd.to_numeric(group.account_return_contribution_pct, errors="coerce").fillna(0).sum() case_ok = case_ok and abs(contrib - float(row.account_return_closed_lots)) < 1e-6 case_ok = case_ok and str(row.strict_baseline_return_ready_flag) == "0" check("case_summary_recompute", bool(case_ok), f"cases={len(case_summary)}", checks) account_ok = True account_direction_ok = True account_balance_ok = True final_open_ok = True sell_open_position_ok = True for case_id, group in account.groupby("case_id"): sorted_group = group.sort_values(["event_date", "event_time", "symbol"]).copy() prev_cash = 1.0 prev_open = 0.0 for _, event in sorted_group.iterrows(): cash = float(event.cash_pct_after_event) open_pos = float(event.open_position_pct_after_event) nav = float(event.account_nav_after_event) realized_delta = float(event.realized_return_delta) account_balance_ok = account_balance_ok and abs(nav - (cash + open_pos)) < 1e-6 if event.action == "BUY": account_direction_ok = account_direction_ok and cash < prev_cash and open_pos > prev_open and abs(realized_delta) < 1e-9 elif event.action == "SELL": account_direction_ok = account_direction_ok and cash > prev_cash and open_pos < prev_open sell_open_position_ok = sell_open_position_ok and open_pos >= -1e-9 prev_cash = cash prev_open = open_pos last = sorted_group.iloc[-1] last_nav = float(last.account_nav_after_event) expected = 1.0 + pd.to_numeric(lots[lots.case_id == case_id].account_return_contribution_pct, errors="coerce").fillna(0).sum() account_ok = account_ok and abs(last_nav - expected) < 1e-6 expected_open = pd.to_numeric( lots[ (lots.case_id == case_id) & (lots.lot_status != "CLOSED_BY_AI_SELL") ].position_pct, errors="coerce", ).fillna(0).sum() final_open_ok = final_open_ok and abs(float(last.open_position_pct_after_event) - expected_open) < 1e-6 check("daily_account_ledger_recompute", bool(account_ok), f"event_rows={len(account)}", checks) check("account_cash_position_direction", bool(account_direction_ok), "BUY cash down/open up; SELL cash up/open down", checks) check("account_nav_equals_cash_plus_open_position", bool(account_balance_ok), "nav equals cash plus open position after each event", checks) check("account_final_open_position_matches_unclosed_lots", bool(final_open_ok), "final open position equals unresolved lot position sum per case", checks) check("sell_reduces_open_position_without_negative_open", bool(sell_open_position_ok), "SELL events reduce open position and never leave negative open position", checks) lookahead_ok = True for df in [order_ledger, decisions, sell_decisions]: if "lookahead_violation_flag" in df.columns: lookahead_ok = lookahead_ok and not df.lookahead_violation_flag.astype(str).str.lower().isin(["true", "1"]).any() check("lookahead_flags_zero", bool(lookahead_ok), "order/decision/sell_decision flags checked", checks) chart_rows = [] image_ok = True for _, row in image_manifest.iterrows(): path = ROOT / str(row.path) exists = path.exists() actual_hash = sha256_file(path) if exists else "" hash_ok = exists and actual_hash == str(row.sha256) image_ok = image_ok and hash_ok chart_rows.append( { "case_id": row.case_id, "symbol": row.symbol, "chart_role": row.chart_role, "path": row.path, "exists": str(exists), "sha256_match": str(hash_ok), "decision_time": row.decision_time, } ) chart_audit = pd.DataFrame(chart_rows) chart_audit.to_csv(ROOT / "chart_evidence_audit.csv", index=False, encoding="utf-8-sig") role_counts = image_manifest.chart_role.value_counts().to_dict() check("image_manifest_hash_match", bool(image_ok), f"images={len(image_manifest)}, roles={role_counts}", checks) expected_images = 35 + 60 + 14 + 14 + action_counts.get("SELL", 0) check("image_manifest_expected_count", len(image_manifest) == expected_images, f"images={len(image_manifest)}, expected={expected_images}", checks) strict_ready = False check("return_stat_not_ready_boundary", not strict_ready and (case_summary.strict_baseline_return_ready_flag.astype(str) == "0").all(), "RETURN_STAT_READY is intentionally false", checks) checks_df = pd.DataFrame(checks) checks_df.to_csv(ROOT / "self_check_items.csv", index=False, encoding="utf-8-sig") fail_count = int((checks_df.status == "FAIL").sum()) summary = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"), "stage": "STRUCTURE_PILOT_EXIT_REVIEW_RESOLVED_SELF_CHECK_DONE", "overall_status": "PASS_FOR_STRUCTURE_PILOT_RETURN_STAT_HELD" if fail_count == 0 else "FAIL", "fail_count": fail_count, "check_count": len(checks_df), "order_counts": action_counts, "lot_status_counts": lots.lot_status.value_counts().to_dict(), "image_role_counts": role_counts, "closed_lot_account_return_sum": float(pd.to_numeric(lots.account_return_contribution_pct, errors="coerce").fillna(0).sum()), "strict_baseline_return_ready_flag": False, "boundary": "Machine self-check passed only for the current structure pilot exit-review-resolved boundary. Execution audit is still required and return statistics are not ready.", "artifacts": { "self_check_items.csv": { "size": (ROOT / "self_check_items.csv").stat().st_size, "sha256": sha256_file(ROOT / "self_check_items.csv"), }, "chart_evidence_audit.csv": { "size": (ROOT / "chart_evidence_audit.csv").stat().st_size, "sha256": sha256_file(ROOT / "chart_evidence_audit.csv"), }, }, } (ROOT / "self_check.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) (ROOT / "self_check.md").write_text( "\n".join( [ "# self_check", "", f"run_id：`{RUN_ID}`", f"状态：`{summary['overall_status']}`", f"检查项：{len(checks_df)}，失败：{fail_count}", "", "关键读数：", f"- BUY：{action_counts.get('BUY', 0)}", f"- SELL：{action_counts.get('SELL', 0)}", f"- 图片：{len(image_manifest)}", f"- 已闭合 lot 账户贡献合计：{summary['closed_lot_account_return_sum']:.4%}", "", "边界：当前只通过结构试点自检，执行审核未通过且未到 RETURN_STAT_READY 前不得引用收益统计。", "", ] ), encoding="utf-8", ) if __name__ == "__main__": main()