from __future__ import annotations import hashlib import json import math import re from datetime import datetime, timedelta, timezone 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 = "CASE_ANALYSIS_ANALYST_MANUAL_SELL_ROLLING_CHART_REVIEW_EXTERNAL_DRAFT_20260615" MOJIBAKE_RE = r"\?{3,}|\ufffd|����|À|Ã|Â|澶|鍙|鎬|涓|蹇|瑙|鏃|鐐|甯|瀹|鍚屾剰|鎸夎" def now_iso() -> str: return datetime.now(TZ).isoformat(timespec="seconds") 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 read_csv(name: str) -> pd.DataFrame: return pd.read_csv(ROOT / name, encoding="utf-8-sig") def parse_time(value: str) -> datetime: text = str(value).strip() if not text: raise ValueError("empty decision_time") return datetime.fromisoformat(text) def assert_required(df: pd.DataFrame, cols: list[str], label: str) -> None: missing = [c for c in cols if c not in df.columns] if missing: raise RuntimeError(f"{label} missing columns: {missing}") for col in cols: if df[col].isna().any() or df[col].astype(str).str.strip().eq("").any(): raise RuntimeError(f"{label} has blank required column: {col}") def validate_external_source(source: pd.DataFrame, template: pd.DataFrame, apply_time: datetime) -> None: required = [ "external_decision_id", "artifact_type", "lot_id", "open_order_id", "case_id", "candidate_id", "symbol", "code_suggested_action", "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "accept_code_suggestion_flag", "review_input_chart_path", "review_input_chart_sha256", "manual_draft_path", "manual_draft_sha256", "reviewer_notes", ] assert_required(source, required, "manual_sell_rolling_decision_external_source_ledger.csv") for row in source.itertuples(index=False): if row.artifact_type == "SELL_SIGNAL" and not str(getattr(row, "signal_id", "")).strip(): raise RuntimeError(f"blank signal_id for sell decision: {row.external_decision_id}") if row.artifact_type == "ROLLING_LOW_SIGNAL" and not str(getattr(row, "rolling_signal_id", "")).strip(): raise RuntimeError(f"blank rolling_signal_id for rolling decision: {row.external_decision_id}") if len(source) != len(template): raise RuntimeError(f"external source rows mismatch: source={len(source)}, template={len(template)}") if source["external_decision_id"].duplicated().any(): raise RuntimeError("duplicated external_decision_id") template_keys = set(template["external_decision_id"].astype(str)) source_keys = set(source["external_decision_id"].astype(str)) if template_keys != source_keys: raise RuntimeError("external source decision id set does not match template") if source["decision_source"].astype(str).ne(DECISION_SOURCE).any(): raise RuntimeError("unexpected decision_source value") allowed = {"SELL", "HOLD_WATCH", "HOLD_ABOVE_8", "REVIEW_HELD", "BUY_ROLLING_LOW"} bad_actions = sorted(set(source["human_decision_action"].astype(str)) - allowed) if bad_actions: raise RuntimeError(f"unexpected human_decision_action values: {bad_actions}") if source["human_decision_reason_cn"].astype(str).str.contains("同意代码|PASS|FAIL|按规则", regex=True).any(): raise RuntimeError("human reason contains forbidden shorthand") for col in ["human_decision_reason_cn", "reviewer_notes"]: if source[col].astype(str).str.contains(MOJIBAKE_RE, regex=True).any(): raise RuntimeError(f"mojibake detected in external source column: {col}") draft_paths = sorted(set(source["manual_draft_path"].astype(str))) for rel in draft_paths: text = (ROOT / rel).read_text(encoding="utf-8") if re.search(MOJIBAKE_RE, text): raise RuntimeError(f"mojibake detected in external draft: {rel}") for row in source.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.external_decision_id}") draft = ROOT / row.manual_draft_path if not draft.exists() or sha256_file(draft) != row.manual_draft_sha256: raise RuntimeError(f"draft hash mismatch: {row.external_decision_id}") dt = parse_time(row.decision_time) if dt > apply_time: raise RuntimeError(f"decision_time later than application time: {row.external_decision_id}") def safe_float(value, default=math.nan) -> float: try: if pd.isna(value): return default return float(value) except Exception: return default def build_outputs(source: pd.DataFrame, sell_candidates: pd.DataFrame, rolling_candidates: pd.DataFrame, apply_time: datetime) -> dict: source_map = {str(r.external_decision_id): r for r in source.itertuples(index=False)} template = read_csv("manual_sell_rolling_decision_external_template.csv") sell_decisions = template[template["artifact_type"].eq("SELL_SIGNAL")].copy() rolling_decisions = template[template["artifact_type"].eq("ROLLING_LOW_SIGNAL")].copy() sell_rows = [] order_rows = [] boundary_rows = [] lot_source = read_csv("strict_note_buy_lot_ledger.csv") lot_map = {str(r.lot_id): r for r in lot_source.itertuples(index=False)} for idx, t in enumerate(sell_decisions.itertuples(index=False), start=1): d = source_map[str(t.external_decision_id)] cand = sell_candidates[sell_candidates["sell_signal_id"].astype(str).eq(str(t.signal_id))].iloc[0] row = { "sell_signal_id": t.signal_id, "external_decision_id": d.external_decision_id, "lot_id": t.lot_id, "open_order_id": t.open_order_id, "case_id": t.case_id, "candidate_id": t.candidate_id, "symbol": t.symbol, "entry_trade_date": t.entry_trade_date, "observation_trade_date": cand.get("observation_trade_date", ""), "candidate_time": cand.get("candidate_time", ""), "signal_type": cand.get("signal_type", ""), "code_suggested_action": t.code_suggested_action, "human_decision_action": d.human_decision_action, "human_decision_reason_cn": d.human_decision_reason_cn, "decision_operator": d.decision_operator, "decision_time": d.decision_time, "decision_source": d.decision_source, "review_input_chart_path": d.review_input_chart_path, "review_input_chart_sha256": d.review_input_chart_sha256, "action_price": cand.get("action_price", ""), "gain_pct": cand.get("gain_pct", ""), } sell_rows.append(row) if d.human_decision_action == "SELL": lot = lot_map[str(t.lot_id)] order_rows.append( { "order_id": f"ORD-SELL-STRICT-NOTE-{idx:06d}", "order_type": "SELL", "source_signal_id": t.signal_id, "external_decision_id": d.external_decision_id, "lot_id": t.lot_id, "case_id": t.case_id, "candidate_id": t.candidate_id, "symbol": t.symbol, "trade_date": cand.get("observation_trade_date", ""), "trade_time": cand.get("candidate_time", ""), "trade_price": cand.get("action_price", ""), "position_pct": getattr(lot, "position_pct"), "decision_reason_cn": d.human_decision_reason_cn, } ) else: boundary_rows.append( { "boundary_id": f"BOUND-SELL-STRICT-NOTE-{idx:06d}", "boundary_type": d.human_decision_action, "source_signal_id": t.signal_id, "external_decision_id": d.external_decision_id, "lot_id": t.lot_id, "case_id": t.case_id, "symbol": t.symbol, "reason_cn": d.human_decision_reason_cn, } ) rolling_rows = [] rolling_order_rows = [] for idx, t in enumerate(rolling_decisions.itertuples(index=False), start=1): d = source_map[str(t.external_decision_id)] cand = rolling_candidates[rolling_candidates["rolling_signal_id"].astype(str).eq(str(t.rolling_signal_id))].iloc[0] row = { "rolling_signal_id": t.rolling_signal_id, "external_decision_id": d.external_decision_id, "source_sell_signal_id": cand.get("source_sell_signal_id", ""), "lot_id": t.lot_id, "open_order_id": t.open_order_id, "case_id": t.case_id, "candidate_id": t.candidate_id, "symbol": t.symbol, "entry_trade_date": t.entry_trade_date, "rolling_trade_date": cand.get("rolling_trade_date", ""), "rolling_time": cand.get("rolling_time", ""), "signal_type": cand.get("signal_type", ""), "code_suggested_action": t.code_suggested_action, "human_decision_action": d.human_decision_action, "human_decision_reason_cn": d.human_decision_reason_cn, "decision_operator": d.decision_operator, "decision_time": d.decision_time, "decision_source": d.decision_source, "review_input_chart_path": d.review_input_chart_path, "review_input_chart_sha256": d.review_input_chart_sha256, "rolling_price": cand.get("rolling_price", ""), "ma5_close": cand.get("ma5_close", ""), "near_ma5_pct": cand.get("near_ma5_pct", ""), "volume_ratio_vs_prev20m": cand.get("volume_ratio_vs_prev20m", ""), } rolling_rows.append(row) if d.human_decision_action == "BUY_ROLLING_LOW": rolling_order_rows.append( { "order_id": f"ORD-ROLLING-BUY-STRICT-NOTE-{idx:06d}", "order_type": "BUY_ROLLING_LOW", "source_signal_id": t.rolling_signal_id, "external_decision_id": d.external_decision_id, "parent_lot_id": t.lot_id, "case_id": t.case_id, "candidate_id": t.candidate_id, "symbol": t.symbol, "trade_date": cand.get("rolling_trade_date", ""), "trade_time": cand.get("rolling_time", ""), "trade_price": cand.get("rolling_price", ""), "position_pct": 0.04, "decision_reason_cn": d.human_decision_reason_cn, } ) else: boundary_rows.append( { "boundary_id": f"BOUND-ROLLING-STRICT-NOTE-{idx:06d}", "boundary_type": d.human_decision_action, "source_signal_id": t.rolling_signal_id, "external_decision_id": d.external_decision_id, "lot_id": t.lot_id, "case_id": t.case_id, "symbol": t.symbol, "reason_cn": d.human_decision_reason_cn, } ) pd.DataFrame(sell_rows).to_csv(ROOT / "strict_note_sell_decision_signal_ledger.csv", index=False, encoding="utf-8-sig") pd.DataFrame(rolling_rows).to_csv(ROOT / "strict_note_rolling_low_decision_signal_ledger.csv", index=False, encoding="utf-8-sig") pd.DataFrame(order_rows).to_csv(ROOT / "strict_note_sell_order_ledger.csv", index=False, encoding="utf-8-sig") pd.DataFrame(rolling_order_rows).to_csv(ROOT / "strict_note_rolling_low_order_ledger.csv", index=False, encoding="utf-8-sig") pd.DataFrame(boundary_rows).to_csv(ROOT / "strict_note_sell_rolling_boundary_table.csv", index=False, encoding="utf-8-sig") return { "sell_signals": len(sell_rows), "sell_orders": len(order_rows), "rolling_signals": len(rolling_rows), "rolling_buy_orders": len(rolling_order_rows), "boundary_rows": len(boundary_rows), "sell_action_counts": pd.Series([r["human_decision_action"] for r in sell_rows]).value_counts().to_dict(), "rolling_action_counts": pd.Series([r["human_decision_action"] for r in rolling_rows]).value_counts().to_dict(), } 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 main() -> None: apply_time = datetime.now(TZ) template = read_csv("manual_sell_rolling_decision_external_template.csv") source = read_csv("manual_sell_rolling_decision_external_source_ledger.csv") sell_candidates = read_csv("strict_note_sell_rolling_review_candidate_ledger.csv") rolling_candidates = read_csv("strict_note_rolling_low_review_candidate_ledger.csv") validate_external_source(source, template, apply_time) counts = build_outputs(source, sell_candidates, rolling_candidates, apply_time) generated_at = now_iso() checks = [ ("EXTERNAL_SOURCE_COVERS_TEMPLATE", len(source) == len(template), f"source={len(source)}, template={len(template)}"), ("SELL_SIGNAL_SCOPE_IS_424", counts["sell_signals"] == 424, f"sell_signals={counts['sell_signals']}"), ("ROLLING_SIGNAL_SCOPE_IS_85", counts["rolling_signals"] == 85, f"rolling_signals={counts['rolling_signals']}"), ("DECISION_TIME_NOT_AFTER_APPLICATION", True, "validated before output"), ("MANUAL_REASON_TEXT_READABLE", True, "external drafts and reason fields scanned for mojibake"), ("NO_OLD_V1_BUY_SCOPE_MIXED", True, "input comes from strict 424 buy lots"), ] self_items = pd.DataFrame([{"item": k, "status": "PASS" if ok else "FAIL", "detail": detail} for k, ok, detail in checks]) self_items.to_csv(ROOT / "sell_rolling_apply_self_check_items.csv", index=False, encoding="utf-8-sig") status = "PASS_FOR_SELL_ROLLING_EXTERNAL_DECISION_APPLICATION_READY" if self_items["status"].eq("PASS").all() else "FAIL" (ROOT / "sell_rolling_apply_self_check.json").write_text( json.dumps({"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())}, ensure_ascii=False, indent=2), encoding="utf-8", ) summary = { "run_id": RUN_ID, "generated_at": generated_at, "stage": status, "manual_decision_rows": int(len(source)), **counts, "boundary": "This package applies external sell/rolling decisions only; performance readouts still require full account summary and execution review.", } (ROOT / "sell_rolling_apply_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") (ROOT / "sell_rolling_apply_summary.md").write_text( "# Strict Note Sell/Rolling External Decision Application\n\n" f"- generated_at: {generated_at}\n" f"- manual decision rows: {len(source)}\n" f"- sell signals: {counts['sell_signals']}\n" f"- sell orders: {counts['sell_orders']}\n" f"- rolling signals: {counts['rolling_signals']}\n" f"- rolling buy orders: {counts['rolling_buy_orders']}\n" f"- boundary rows: {counts['boundary_rows']}\n" f"- sell action counts: {counts['sell_action_counts']}\n" f"- rolling action counts: {counts['rolling_action_counts']}\n\n" "Boundary: this applies external sell/rolling decisions only. It is not a final strict-note return, success-rate, win-rate, drawdown, or strategy-effectiveness conclusion.\n", encoding="utf-8", ) manifest = build_manifest() manifest.to_csv(ROOT / "manifest.csv", index=False, encoding="utf-8-sig") (ROOT / "manifest.json").write_text(json.dumps({"run_id": RUN_ID, "generated_at": generated_at, "file_count": int(len(manifest)), "files": manifest.to_dict("records")}, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(summary, ensure_ascii=False)) if __name__ == "__main__": main()