from __future__ import annotations import csv import hashlib import json from collections import Counter, defaultdict from datetime import datetime from pathlib import Path from typing import Any import pandas as pd RUN_ID = "RUN-ANA-WUJI-NOTE-FIT-AUDIT-20260615-001" PACKAGE_ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = PACKAGE_ROOT.parents[2] RESULT_ROOT = PROJECT_ROOT / "ana-data" / "result" STRICT_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001" FRONT_AUDIT_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-FRONTDATA-RULE-AUDIT-20260613-001" LIFECYCLE_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-001" LEDGER_FIELDS = [ "audit_row_id", "run_id", "case_id", "symbol", "subject_type", "subject_id", "rule_id", "rule_category", "audit_status", "automation_level", "evidence_level", "trade_date", "trade_time", "source_file", "source_key", "evidence_summary", "evidence_path", "needs_manual_review", "notes", ] SUMMARY_FIELDS = [ "case_id", "ledger_rows", "pass_count", "weak_pass_count", "proxy_pass_count", "fail_count", "data_gap_count", "manual_required_count", "not_applicable_count", "case_note_fit_status", ] PILOT_FIELDS = [ "pilot_group", "case_id", "account_return_closed_lots", "v1_return_scope", "case_note_fit_status", "fail_count", "manual_required_count", "suggested_review_focus", "case_image_board", "case_stock_lifecycle_board", ] SELF_CHECK_FIELDS = ["check_id", "status", "details"] MANIFEST_FIELDS = ["path", "size_bytes", "sha256"] RULE_CATEGORY = { "MKT_OPEN_NEW_UP_3000": "market", "SEL_PRIOR30_LIMITUP_STRICT_BOARD": "selection", "SEL_VOLUME_MULTIPLE": "selection", "SEL_LONG_UPPER_SHADOW_TOP50": "selection", "SEL_BOTTOM_SUPPORT": "selection", "SEL_SIDEWAYS_60D": "selection", "SEL_TOUCH_PREV_HIGH_VOLUME_GT": "selection", "BUY_TIME_WINDOW_NORMAL": "buy", "BUY_OPEN_RUSH_PULLBACK_OPEN_SHRINK_SUPPORT": "buy", "BUY_OPEN_RUSH_PULLBACK_MA20_SHRINK_SUPPORT": "buy", "BUY_NO_CHASE_INTRADAY_SPIKE": "buy", "BUY_ROLLING_LOW_BUY_MA5_EXCEPTION": "buy", "SELL_THREE_DAY_HIGH_NOT_RISING": "sell", "POS_TRANCHE_5_PARTS": "position", "CAP_T1": "lifecycle", } def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def read_csv(path: Path) -> pd.DataFrame: if not path.exists(): return pd.DataFrame() return pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig") def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: with path.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def write_json(path: Path, data: dict[str, Any]) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def bval(value: Any) -> bool | None: text = str(value).strip().lower() if text in {"true", "1", "yes", "y"}: return True if text in {"false", "0", "no", "n"}: return False return None def fnum(value: Any) -> float | None: text = str(value).strip() if text == "": return None try: return float(text) except ValueError: return None def status_from_bool(value: bool | None) -> str: if value is True: return "PASS" if value is False: return "FAIL" return "DATA_GAP" def status_from_proxy_bool(value: bool | None) -> str: if value is True: return "PROXY_PASS" if value is False: return "FAIL" return "DATA_GAP" def source_rel(path: Path) -> str: try: return path.relative_to(PROJECT_ROOT).as_posix() except ValueError: return str(path) def key(*parts: Any) -> str: return "|".join(str(p) for p in parts) class LedgerBuilder: def __init__(self) -> None: self.rows: list[dict[str, Any]] = [] def add( self, *, case_id: str, symbol: str, subject_type: str, subject_id: str, rule_id: str, audit_status: str, automation_level: str, evidence_level: str, trade_date: str = "", trade_time: str = "", source_file: str = "", source_key: str = "", evidence_summary: str = "", evidence_path: str = "", needs_manual_review: bool = False, notes: str = "", ) -> None: self.rows.append( { "audit_row_id": f"NFA-{len(self.rows) + 1:06d}", "run_id": RUN_ID, "case_id": case_id, "symbol": symbol, "subject_type": subject_type, "subject_id": subject_id, "rule_id": rule_id, "rule_category": RULE_CATEGORY.get(rule_id, ""), "audit_status": audit_status, "automation_level": automation_level, "evidence_level": evidence_level, "trade_date": trade_date, "trade_time": trade_time, "source_file": source_file, "source_key": source_key, "evidence_summary": evidence_summary, "evidence_path": evidence_path, "needs_manual_review": str(bool(needs_manual_review)).lower(), "notes": notes, } ) def load_inputs() -> dict[str, pd.DataFrame]: return { "source_buy": read_csv(FRONT_AUDIT_ROOT / "source_buy_trace_audit.csv"), "daily": read_csv(FRONT_AUDIT_ROOT / "daily_candidate_rule_recalc.csv"), "limitup": read_csv(FRONT_AUDIT_ROOT / "limitup_prior30_user_rule_audit.csv"), "rolling": read_csv(FRONT_AUDIT_ROOT / "rolling_buy_trace_audit.csv"), "sell3": read_csv(FRONT_AUDIT_ROOT / "sell_three_day_high_recalc.csv"), "sell3_5m": read_csv(FRONT_AUDIT_ROOT / "three_day_high_5m_proxy_audit.csv"), "orders": read_csv(STRICT_ROOT / "strict_order_ledger.csv"), "lots": read_csv(STRICT_ROOT / "strict_position_lot_ledger.csv"), "strict_case_summary": read_csv(STRICT_ROOT / "strict_case_summary.csv"), "lifecycle": read_csv(LIFECYCLE_ROOT / "stock_lifecycle_index.csv"), } def index_by(df: pd.DataFrame, *cols: str) -> dict[str, dict[str, Any]]: if df.empty or any(col not in df.columns for col in cols): return {} out: dict[str, dict[str, Any]] = {} for row in df.to_dict("records"): out[key(*(row.get(col, "") for col in cols))] = row return out def add_source_buy_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None: source_buy = data["source_buy"] daily_by_order = index_by(data["daily"], "strict_order_id") limitup_by_candidate = index_by(data["limitup"], "case_id", "symbol", "candidate_id") source_path = source_rel(FRONT_AUDIT_ROOT / "source_buy_trace_audit.csv") daily_path = source_rel(FRONT_AUDIT_ROOT / "daily_candidate_rule_recalc.csv") limitup_path = source_rel(FRONT_AUDIT_ROOT / "limitup_prior30_user_rule_audit.csv") for row in source_buy.to_dict("records"): case_id = row.get("case_id", "") symbol = row.get("symbol", "") order_id = row.get("strict_order_id", "") candidate_id = row.get("candidate_id", "") trade_date = row.get("entry_trade_date", "") trade_time = row.get("entry_time", "") source_key = key(order_id, case_id, symbol) builder.add( case_id=case_id, symbol=symbol, subject_type="SOURCE_BUY", subject_id=order_id, rule_id="MKT_OPEN_NEW_UP_3000", audit_status=status_from_bool(bval(row.get("old_up_count_ge3000", ""))), automation_level="AUTO_STRICT", evidence_level="LEDGER_ONLY", trade_date=trade_date, trade_time=trade_time, source_file=source_path, source_key=source_key, evidence_summary=f"old_up_count={row.get('old_up_count', '')}; market_gate_open_flag={row.get('market_gate_open_flag', '')}", evidence_path=row.get("source_evidence_image_path", ""), ) builder.add( case_id=case_id, symbol=symbol, subject_type="SOURCE_BUY", subject_id=order_id, rule_id="BUY_TIME_WINDOW_NORMAL", audit_status=status_from_bool(bval(row.get("normal_buy_time_window_ok", ""))), automation_level="AUTO_STRICT", evidence_level="EXACT", trade_date=trade_date, trade_time=trade_time, source_file=source_path, source_key=source_key, evidence_summary=f"entry_time={trade_time}; normal_buy_time_window_ok={row.get('normal_buy_time_window_ok', '')}", evidence_path=row.get("source_evidence_image_path", ""), ) rank = fnum(row.get("candidate_rank", "")) rank_status = "DATA_GAP" if rank is None else ("PASS" if rank <= 50 else "FAIL") builder.add( case_id=case_id, symbol=symbol, subject_type="SOURCE_BUY", subject_id=order_id, rule_id="SEL_LONG_UPPER_SHADOW_TOP50", audit_status=rank_status, automation_level="AUTO_PROXY", evidence_level="LEDGER_ONLY", trade_date=trade_date, trade_time=trade_time, source_file=source_path, source_key=source_key, evidence_summary=f"candidate_rank={row.get('candidate_rank', '')}; note requires top50 long-upper-shadow pool", evidence_path=row.get("source_evidence_image_path", ""), needs_manual_review=True, notes="排名字段可自动核对，但长上影形态和放量质量仍需图证复核。", ) daily = daily_by_order.get(order_id, {}) builder.add( case_id=case_id, symbol=symbol, subject_type="SOURCE_BUY", subject_id=order_id, rule_id="SEL_VOLUME_MULTIPLE", audit_status=status_from_bool(bval(daily.get("recalc_volume_ratio_ge2", ""))), automation_level="AUTO_STRICT", evidence_level="EXACT", trade_date=daily.get("signal_trade_date", trade_date), trade_time="", source_file=daily_path, source_key=source_key, evidence_summary=( f"recalc_volume_ratio={daily.get('recalc_volume_ratio', '')}; " f"recalc_volume_ratio_ge2={daily.get('recalc_volume_ratio_ge2', '')}; " f"old_volume_ratio={daily.get('old_volume_ratio', row.get('old_volume_ratio', ''))}" ), evidence_path=daily.get("daily_file_path", ""), notes="当前严格倍量先按 >=2.0 打标；V1 历史参考口径可能低于 2.0，需分列解释。", ) touch = bval(daily.get("recalc_touch_prev_high_flag", "")) pass_prev = bval(daily.get("recalc_prev_high_volume_pass_flag", "")) if touch is False: prev_status = "NOT_APPLICABLE" elif touch is True: prev_status = status_from_bool(pass_prev) else: prev_status = "DATA_GAP" builder.add( case_id=case_id, symbol=symbol, subject_type="SOURCE_BUY", subject_id=order_id, rule_id="SEL_TOUCH_PREV_HIGH_VOLUME_GT", audit_status=prev_status, automation_level="AUTO_STRICT", evidence_level="EXACT", trade_date=daily.get("signal_trade_date", trade_date), source_file=daily_path, source_key=source_key, evidence_summary=( f"touch_prev_high={daily.get('recalc_touch_prev_high_flag', '')}; " f"prev_high_volume_pass={daily.get('recalc_prev_high_volume_pass_flag', '')}; " f"prev_high_ref_date={daily.get('recalc_prev60_high_ref_date', '')}" ), evidence_path=daily.get("daily_file_path", ""), ) limit = limitup_by_candidate.get(key(case_id, symbol, candidate_id), {}) builder.add( case_id=case_id, symbol=symbol, subject_type="SOURCE_BUY", subject_id=order_id, rule_id="SEL_PRIOR30_LIMITUP_STRICT_BOARD", audit_status=status_from_bool(bval(limit.get("user_strict_close_limitup_prior30_excl_signal", ""))), automation_level="AUTO_STRICT", evidence_level="EXACT", trade_date=limit.get("signal_trade_date", daily.get("signal_trade_date", "")), source_file=limitup_path, source_key=key(case_id, symbol, candidate_id), evidence_summary=( f"board_rule={limit.get('board_rule', '')}; threshold_pct={limit.get('threshold_pct', '')}; " f"strict_close_prior30={limit.get('user_strict_close_limitup_prior30_excl_signal', '')}; " f"last_limitup={limit.get('user_strict_close_last_limitup_date', '')}" ), evidence_path=daily.get("daily_file_path", ""), notes="使用用户确认口径：信号日前30个交易日，不含信号日，按板块阈值。", ) for rule_id, note in [ ("SEL_BOTTOM_SUPPORT", "底部承接力需要人工看图确认。"), ("SEL_SIDEWAYS_60D", "60日横盘或横盘涨停后回调需要日线图形确认。"), ("BUY_OPEN_RUSH_PULLBACK_OPEN_SHRINK_SUPPORT", "买入属于回踩开盘价还是其他形态，需要旧买入图确认。"), ("BUY_OPEN_RUSH_PULLBACK_MA20_SHRINK_SUPPORT", "买入是否回踩20日均线附近缩量支撑，需要旧买入图确认。"), ("BUY_NO_CHASE_INTRADAY_SPIKE", "是否盘中急拉放量追高，需要买入前分时图确认。"), ]: builder.add( case_id=case_id, symbol=symbol, subject_type="SOURCE_BUY", subject_id=order_id, rule_id=rule_id, audit_status="MANUAL_REQUIRED", automation_level="MANUAL_IMAGE", evidence_level="CHART_REQUIRED", trade_date=trade_date, trade_time=trade_time, source_file=source_path, source_key=source_key, evidence_summary=note, evidence_path=row.get("source_evidence_image_path", ""), needs_manual_review=True, notes=row.get("old_decision_reason_cn", ""), ) def add_rolling_buy_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None: rolling = data["rolling"] source_path = source_rel(FRONT_AUDIT_ROOT / "rolling_buy_trace_audit.csv") for row in rolling.to_dict("records"): case_id = row.get("case_id", "") symbol = row.get("symbol", "") order_id = row.get("strict_order_id", "") near_ma5 = fnum(row.get("near_ma5_pct", "")) time_ok = bval(row.get("rolling_window_1040_1440_ok", "")) ma5_status = "DATA_GAP" if near_ma5 is None else ("PASS" if near_ma5 <= 0.012 else "FAIL") combined_status = "PASS" if time_ok is True and ma5_status == "PASS" else ("FAIL" if time_ok is False or ma5_status == "FAIL" else "DATA_GAP") builder.add( case_id=case_id, symbol=symbol, subject_type="ROLLING_BUY", subject_id=order_id, rule_id="BUY_ROLLING_LOW_BUY_MA5_EXCEPTION", audit_status=combined_status, automation_level="AUTO_STRICT+MANUAL_IMAGE", evidence_level="LEDGER_ONLY", trade_date=row.get("rolling_trade_date", ""), trade_time=row.get("rolling_time", ""), source_file=source_path, source_key=key(order_id, row.get("rolling_signal_id", "")), evidence_summary=( f"rolling_window_1040_1440_ok={row.get('rolling_window_1040_1440_ok', '')}; " f"near_ma5_pct={row.get('near_ma5_pct', '')}; " f"volume_ratio_vs_prev20m={row.get('volume_ratio_vs_prev20m', '')}" ), evidence_path=row.get("evidence_image_path", ""), needs_manual_review=True, notes="时间和MA5附近度可自动判，趋势基础和止跌承接仍需人工看图。", ) def add_sell_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None: sell3 = data["sell3"] proxy_by_signal = index_by(data["sell3_5m"], "signal_id") source_path = source_rel(FRONT_AUDIT_ROOT / "sell_three_day_high_recalc.csv") proxy_path = source_rel(FRONT_AUDIT_ROOT / "three_day_high_5m_proxy_audit.csv") for row in sell3.to_dict("records"): signal_id = row.get("signal_id", "") proxy = proxy_by_signal.get(signal_id, {}) if proxy: status = status_from_proxy_bool(bval(proxy.get("three_highs_not_strictly_rising_5m_proxy", ""))) evidence_level = "PROXY_5M" source_file = proxy_path summary = ( f"verification_status_5m_proxy={proxy.get('verification_status_5m_proxy', '')}; " f"three_highs_not_strictly_rising_5m_proxy={proxy.get('three_highs_not_strictly_rising_5m_proxy', '')}; " f"d-2={proxy.get('d_minus_2_high_front', '')}; d-1={proxy.get('d_minus_1_high_front', '')}; " f"current_5m_scaled={proxy.get('current_high_until_1040_5m_daily_scaled_front', '')}" ) else: safe = bval(row.get("three_highs_not_strictly_rising_decision_safe", "")) daily_proxy = bval(row.get("three_highs_not_strictly_rising_daily_proxy", "")) if safe is not None: status = status_from_bool(safe) evidence_level = "EXACT" else: status = status_from_proxy_bool(daily_proxy) evidence_level = "DAILY_PROXY" source_file = source_path summary = ( f"verification_status={row.get('verification_status', '')}; " f"decision_safe={row.get('three_highs_not_strictly_rising_decision_safe', '')}; " f"daily_proxy={row.get('three_highs_not_strictly_rising_daily_proxy', '')}" ) builder.add( case_id=row.get("case_id", ""), symbol=row.get("symbol", ""), subject_type="SELL_SIGNAL", subject_id=signal_id, rule_id="SELL_THREE_DAY_HIGH_NOT_RISING", audit_status=status, automation_level="AUTO_STRICT_OR_PROXY", evidence_level=evidence_level, trade_date=row.get("observation_trade_date", ""), trade_time=row.get("candidate_time", ""), source_file=source_file, source_key=signal_id, evidence_summary=summary, evidence_path=proxy.get("public_5m_cache_path", ""), notes=row.get("code_evidence_reason_cn", ""), ) def add_order_and_lot_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None: orders = data["orders"] lots = data["lots"] order_path = source_rel(STRICT_ROOT / "strict_order_ledger.csv") lot_path = source_rel(STRICT_ROOT / "strict_position_lot_ledger.csv") for row in orders.to_dict("records"): if row.get("action") != "BUY": continue planned = fnum(row.get("planned_tranche_count", "")) status = "DATA_GAP" if planned is None else ("PASS" if int(planned) == 5 else "FAIL") builder.add( case_id=row.get("case_id", ""), symbol=row.get("symbol", ""), subject_type="ORDER", subject_id=row.get("order_id", ""), rule_id="POS_TRANCHE_5_PARTS", audit_status=status, automation_level="AUTO_STRICT", evidence_level="LEDGER_ONLY", trade_date=row.get("trade_date", ""), trade_time=row.get("trade_time", ""), source_file=order_path, source_key=row.get("order_id", ""), evidence_summary=( f"tranche_index={row.get('tranche_index', '')}; " f"planned_tranche_count={row.get('planned_tranche_count', '')}; " f"position_delta_pct={row.get('position_delta_pct', '')}" ), evidence_path=row.get("evidence_image_path", ""), ) for row in lots.to_dict("records"): entry = row.get("entry_trade_date", "") sellable = row.get("sellable_from_trade_date", "") exit_date = row.get("exit_trade_date", "") if not entry or not sellable: status = "DATA_GAP" elif exit_date and exit_date < sellable: status = "FAIL" else: status = "PASS" builder.add( case_id=row.get("case_id", ""), symbol=row.get("symbol", ""), subject_type="LOT", subject_id=row.get("strict_lot_id", ""), rule_id="CAP_T1", audit_status=status, automation_level="AUTO_STRICT", evidence_level="LEDGER_ONLY", trade_date=entry, trade_time=row.get("entry_time", ""), source_file=lot_path, source_key=row.get("strict_lot_id", ""), evidence_summary=f"entry={entry}; sellable_from={sellable}; exit={exit_date}; lot_status={row.get('lot_status', '')}", ) def build_summary(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: by_case: dict[str, Counter[str]] = defaultdict(Counter) for row in rows: by_case[row["case_id"]][row["audit_status"]] += 1 by_case[row["case_id"]]["_rows"] += 1 out: list[dict[str, Any]] = [] for case_id in sorted(by_case): c = by_case[case_id] if c["FAIL"]: status = "NOTE_MISMATCH_REVIEW" elif c["DATA_GAP"]: status = "DATA_GAP_HELD" elif c["MANUAL_REQUIRED"]: status = "MANUAL_REVIEW_PENDING" elif c["PROXY_PASS"]: status = "NOTE_FIT_PROXY" else: status = "NOTE_FIT_AUTO_PASS" out.append( { "case_id": case_id, "ledger_rows": c["_rows"], "pass_count": c["PASS"], "weak_pass_count": c["WEAK_PASS"], "proxy_pass_count": c["PROXY_PASS"], "fail_count": c["FAIL"], "data_gap_count": c["DATA_GAP"], "manual_required_count": c["MANUAL_REQUIRED"], "not_applicable_count": c["NOT_APPLICABLE"], "case_note_fit_status": status, } ) return out def sample_evenly(rows: list[dict[str, Any]], n: int) -> list[dict[str, Any]]: if len(rows) <= n: return rows if n <= 1: return [rows[0]] indexes = [round(i * (len(rows) - 1) / (n - 1)) for i in range(n)] out: list[dict[str, Any]] = [] seen: set[int] = set() for idx in indexes: if idx not in seen: out.append(rows[idx]) seen.add(idx) return out def build_pilot_queue(case_summary: list[dict[str, Any]], data: dict[str, pd.DataFrame]) -> list[dict[str, Any]]: summary_by_case = {row["case_id"]: row for row in case_summary} strict_case = data["strict_case_summary"] strict_by_case = strict_case.set_index("case_id", drop=False).to_dict("index") if not strict_case.empty else {} lifecycle_cases = set(data["lifecycle"]["case_id"].tolist()) if not data["lifecycle"].empty else set() merged: list[dict[str, Any]] = [] for case_id, note in summary_by_case.items(): strict = strict_by_case.get(case_id, {}) return_value = fnum(strict.get("account_return_closed_lots", "")) case_image = STRICT_ROOT / "cases" / case_id / "case_image_board.md" lifecycle_board = LIFECYCLE_ROOT / "cases" / case_id / "case_stock_lifecycle_board.md" merged.append( { "case_id": case_id, "account_return_closed_lots": "" if return_value is None else f"{return_value:.8f}", "return_float": return_value, "v1_return_scope": strict.get("v1_return_scope", ""), "case_note_fit_status": note.get("case_note_fit_status", ""), "fail_count": note.get("fail_count", "0"), "manual_required_count": note.get("manual_required_count", "0"), "case_image_board": source_rel(case_image) if case_image.exists() else "", "case_stock_lifecycle_board": source_rel(lifecycle_board) if case_id in lifecycle_cases or lifecycle_board.exists() else "", } ) primary = [r for r in merged if r["v1_return_scope"] == "V1_PRIMARY_STRICT_CLOSED_CASE"] winners = sorted([r for r in primary if r["return_float"] is not None and r["return_float"] > 0], key=lambda r: r["case_id"]) losers = sorted([r for r in primary if r["return_float"] is not None and r["return_float"] <= 0], key=lambda r: r["case_id"]) manual_only = sorted( [r for r in merged if r["case_note_fit_status"] == "MANUAL_REVIEW_PENDING"], key=lambda r: r["case_id"], ) out: list[dict[str, Any]] = [] for group, source_rows, focus in [ ("WINNER_SAMPLE", sample_evenly(winners, 10), "盈利样本：检查自动严格差异是否影响成功样本解释。"), ("LOSER_SAMPLE", sample_evenly(losers, 10), "亏损/非正收益样本：检查失败是否来自选股、买点、卖点还是市场环境。"), ("MANUAL_ONLY_SAMPLE", sample_evenly(manual_only, 10), "无自动失败但需要人工看图样本：校准形态类规则。"), ]: for row in source_rows: out.append( { "pilot_group": group, "case_id": row["case_id"], "account_return_closed_lots": row["account_return_closed_lots"], "v1_return_scope": row["v1_return_scope"], "case_note_fit_status": row["case_note_fit_status"], "fail_count": row["fail_count"], "manual_required_count": row["manual_required_count"], "suggested_review_focus": focus, "case_image_board": row["case_image_board"], "case_stock_lifecycle_board": row["case_stock_lifecycle_board"], } ) return out 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 build_manifest() -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for path in sorted(PACKAGE_ROOT.rglob("*")): if not path.is_file(): continue if path.name in {"manifest.csv"}: continue rows.append( { "path": source_rel(path), "size_bytes": path.stat().st_size, "sha256": sha256_file(path), } ) return rows def build_self_check(rows: list[dict[str, Any]], case_summary: list[dict[str, Any]]) -> list[dict[str, Any]]: package_files = [ "README.md", "todo_status.md", "note_rule_mapping.csv", "data_source_inventory.csv", "output_schema.md", "case_note_fit_ledger.csv", "case_note_fit_summary.csv", "manual_review_queue.csv", "rule_exception_inventory.csv", "pilot_case_review_queue.csv", "note_fit_audit_summary.md", "summary.json", ] checks: list[dict[str, Any]] = [] missing = [name for name in package_files if not (PACKAGE_ROOT / name).exists()] checks.append( { "check_id": "PACKAGE_FILES_EXIST", "status": "PASS" if not missing else "FAIL", "details": "all required files exist" if not missing else f"missing={missing}", } ) checks.append( { "check_id": "LEDGER_NON_EMPTY", "status": "PASS" if rows else "FAIL", "details": f"ledger_rows={len(rows)}", } ) bad_pass = [row for row in rows if row["audit_status"] in {"PASS", "PROXY_PASS"} and not row["source_file"]] checks.append( { "check_id": "PASS_ROWS_HAVE_SOURCE", "status": "PASS" if not bad_pass else "FAIL", "details": f"bad_rows={len(bad_pass)}", } ) manual_rows = [row for row in rows if row["audit_status"] == "MANUAL_REQUIRED" or row["needs_manual_review"] == "true"] manual_file_rows = read_csv(PACKAGE_ROOT / "manual_review_queue.csv") checks.append( { "check_id": "MANUAL_QUEUE_MATCHES_LEDGER", "status": "PASS" if len(manual_rows) == len(manual_file_rows) else "FAIL", "details": f"ledger_manual_rows={len(manual_rows)}; file_rows={len(manual_file_rows)}", } ) exception_rows = [row for row in rows if row["audit_status"] == "FAIL"] exception_file_rows = read_csv(PACKAGE_ROOT / "rule_exception_inventory.csv") checks.append( { "check_id": "EXCEPTION_INVENTORY_MATCHES_FAILS", "status": "PASS" if len(exception_rows) == len(exception_file_rows) else "FAIL", "details": f"ledger_fail_rows={len(exception_rows)}; file_rows={len(exception_file_rows)}", } ) checks.append( { "check_id": "CASE_SUMMARY_NON_EMPTY", "status": "PASS" if case_summary else "FAIL", "details": f"case_count={len(case_summary)}", } ) return checks def write_summary_markdown(rows: list[dict[str, Any]], case_summary: list[dict[str, Any]]) -> None: status_counts = Counter(row["audit_status"] for row in rows) rule_counts = Counter(row["rule_id"] for row in rows) case_status_counts = Counter(row["case_note_fit_status"] for row in case_summary) lines = [ "# Note Fit Audit Summary", "", f"generated_at: {now_iso()}", f"run_id: `{RUN_ID}`", "", "## Scope", "", f"- ledger_rows: {len(rows)}", f"- cases: {len(case_summary)}", "", "## Audit Status Counts", "", ] for status, count in status_counts.most_common(): lines.append(f"- {status}: {count}") lines.extend(["", "## Case Status Counts", ""]) for status, count in case_status_counts.most_common(): lines.append(f"- {status}: {count}") lines.extend(["", "## Rule Row Counts", ""]) for rule_id, count in rule_counts.most_common(): lines.append(f"- {rule_id}: {count}") lines.extend( [ "", "## Boundary", "", "This first-pass ledger only handles rules with existing structured evidence. " "Rows marked `MANUAL_REQUIRED` must be reviewed with charts before any strict note-fit conclusion is made.", "", ] ) (PACKAGE_ROOT / "note_fit_audit_summary.md").write_text("\n".join(lines), encoding="utf-8") def main() -> None: data = load_inputs() builder = LedgerBuilder() add_source_buy_rules(builder, data) add_rolling_buy_rules(builder, data) add_sell_rules(builder, data) add_order_and_lot_rules(builder, data) rows = builder.rows write_csv(PACKAGE_ROOT / "case_note_fit_ledger.csv", rows, LEDGER_FIELDS) case_summary = build_summary(rows) write_csv(PACKAGE_ROOT / "case_note_fit_summary.csv", case_summary, SUMMARY_FIELDS) manual_rows = [ row for row in rows if row["audit_status"] == "MANUAL_REQUIRED" or row["needs_manual_review"] == "true" ] write_csv(PACKAGE_ROOT / "manual_review_queue.csv", manual_rows, LEDGER_FIELDS) exception_rows = [row for row in rows if row["audit_status"] == "FAIL"] write_csv(PACKAGE_ROOT / "rule_exception_inventory.csv", exception_rows, LEDGER_FIELDS) pilot_rows = build_pilot_queue(case_summary, data) write_csv(PACKAGE_ROOT / "pilot_case_review_queue.csv", pilot_rows, PILOT_FIELDS) write_summary_markdown(rows, case_summary) write_json( PACKAGE_ROOT / "summary.json", { "generated_at": now_iso(), "run_id": RUN_ID, "ledger_rows": len(rows), "case_count": len(case_summary), "status_counts": dict(Counter(row["audit_status"] for row in rows)), "case_status_counts": dict(Counter(row["case_note_fit_status"] for row in case_summary)), "source_files": { name: len(df) for name, df in data.items() }, }, ) self_check_rows = build_self_check(rows, case_summary) write_csv(PACKAGE_ROOT / "self_check_items.csv", self_check_rows, SELF_CHECK_FIELDS) write_json( PACKAGE_ROOT / "self_check.json", { "generated_at": now_iso(), "run_id": RUN_ID, "status": "PASS" if all(row["status"] == "PASS" for row in self_check_rows) else "FAIL", "items": self_check_rows, }, ) self_check_lines = ["# Self Check", ""] for row in self_check_rows: self_check_lines.append(f"- {row['check_id']}: {row['status']} - {row['details']}") self_check_lines.append("") (PACKAGE_ROOT / "self_check.md").write_text("\n".join(self_check_lines), encoding="utf-8") write_csv(PACKAGE_ROOT / "manifest.csv", build_manifest(), MANIFEST_FIELDS) if __name__ == "__main__": main()