from __future__ import annotations import importlib.util import json from pathlib import Path import pandas as pd SCRIPT_DIR = Path(__file__).resolve().parent BUILD_PATH = SCRIPT_DIR / "build_v1_lifecycle_chart_supplement.py" spec = importlib.util.spec_from_file_location("build_v1_lifecycle_chart_supplement", BUILD_PATH) build = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(build) def read_csv(path: Path) -> pd.DataFrame: return pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig").fillna("") def existing_png(path: Path) -> str: return path.relative_to(build.PACKAGE_ROOT).as_posix() if path.exists() else "" def main() -> None: build.PACKAGE_ROOT.mkdir(parents=True, exist_ok=True) build.write_source_manifest() v1_index = read_csv(build.FINAL_ROOT / "v1_case_readout_index.csv") order = read_csv(build.SOURCE_ROOT / "strict_order_ledger.csv") for col in ["trade_date", "t1_sellable_from_trade_date"]: if col in order.columns: order[col] = order[col].map(build.norm_date) order["trade_time"] = order["trade_time"].map(build.norm_time) for col in ["price", "position_delta_pct"]: order[col] = pd.to_numeric(order[col], errors="coerce") order = order[order["action"].isin(["BUY", "SELL"])].copy() buy_case_ids = set(order[order["action"] == "BUY"]["case_id"].tolist()) v1_index = v1_index[v1_index["case_id"].isin(buy_case_ids)].copy() v1_by_case = v1_index.set_index("case_id", drop=False).to_dict("index") order = order[order["case_id"].isin(v1_by_case)].copy() group_rows = [] for (case_id, symbol), group in order.groupby(["case_id", "symbol"]): buy_rows = group[group["action"] == "BUY"].sort_values(["trade_date", "trade_time", "order_id"]) if buy_rows.empty: continue sell_rows = group[group["action"] == "SELL"].sort_values(["trade_date", "trade_time", "order_id"]) last_rows = sell_rows if not sell_rows.empty else group.sort_values(["trade_date", "trade_time", "order_id"]) first_buy = buy_rows.iloc[0] last_sell = last_rows.iloc[-1] group_rows.append( { "case_id": case_id, "symbol": symbol, "first_buy_date": first_buy["trade_date"], "first_buy_time": first_buy["trade_time"], "first_buy_datetime": f"{first_buy['trade_date']} {first_buy['trade_time']}", "last_sell_date": last_sell["trade_date"] if last_sell["action"] == "SELL" else "", "last_sell_time": last_sell["trade_time"] if last_sell["action"] == "SELL" else "", "last_sell_datetime": f"{last_sell['trade_date']} {last_sell['trade_time']}" if last_sell["action"] == "SELL" else "", "buy_count": int(len(buy_rows)), "sell_count": int(len(sell_rows)), } ) chart_rows: list[dict] = [] missing_rows: list[dict] = [] symbol_index_rows: list[dict] = [] tx_index_rows: list[dict] = [] case_rows: list[dict] = [] case_symbol_rows: dict[str, list[dict]] = {} case_tx_rows: dict[str, list[dict]] = {} for row in group_rows: case_id = row["case_id"] symbol = row["symbol"] safe_symbol = symbol.replace(".", "_") img_dir = build.PACKAGE_ROOT / "cases" / case_id / "img" last_date = row["last_sell_date"] or row["first_buy_date"] lifecycle_path = next( img_dir.glob(f"09_lifecycle_daily_50pre_20post_{safe_symbol}_{row['first_buy_date'].replace('-', '')}_{last_date.replace('-', '')}.png"), None, ) lifecycle_rel = existing_png(lifecycle_path) if lifecycle_path else "" if lifecycle_rel: chart_rows.append( { "case_id": case_id, "symbol": symbol, "chart_role": "symbol_lifecycle_daily_50pre_20post_audit_view", "action": "LIFECYCLE", "trade_date": row["first_buy_date"], "trade_time": row["first_buy_time"], "path": lifecycle_rel, "source_order_id": "", "status": "PASS", "note": "首个BUY前50个交易日至最后SELL后20个交易日；仅作audit_view。", "size": lifecycle_path.stat().st_size, "sha256": build.sha256_file(lifecycle_path), } ) else: missing_rows.append({"case_id": case_id, "symbol": symbol, "kind": "daily_lifecycle", "date": row["first_buy_date"]}) group = order[(order["case_id"] == case_id) & (order["symbol"] == symbol)].copy() tx_count = 0 for trade_date_key, day_orders in group.groupby(["trade_date"]): trade_date = trade_date_key[0] if isinstance(trade_date_key, tuple) else trade_date_key tx_path = img_dir / f"10_transaction_day_minute_line_{safe_symbol}_{str(trade_date).replace('-', '')}.png" tx_rel = existing_png(tx_path) if not tx_rel: missing_rows.append({"case_id": case_id, "symbol": symbol, "kind": "transaction_minute", "date": trade_date}) continue tx_count += 1 tx_row = { "case_id": case_id, "symbol": symbol, "trade_date": trade_date, "order_count": int(len(day_orders)), "buy_order_count": int((day_orders["action"] == "BUY").sum()), "sell_order_count": int((day_orders["action"] == "SELL").sum()), "minute_chart_path": tx_rel, } tx_index_rows.append(tx_row) case_tx_rows.setdefault(case_id, []).append(tx_row) chart_rows.append( { "case_id": case_id, "symbol": symbol, "chart_role": "transaction_day_minute_line_audit_view", "action": "BUY_SELL_DAY", "trade_date": trade_date, "trade_time": "", "path": tx_rel, "source_order_id": ";".join(day_orders["order_id"].tolist()), "status": "PASS", "note": "有交易产生的交易日整日分时；标记当天全部BUY/SELL订单。", "size": tx_path.stat().st_size, "sha256": build.sha256_file(tx_path), } ) meta = v1_by_case[case_id] symbol_row = { **row, "v1_return_scope": meta.get("v1_return_scope", ""), "success_flag": meta.get("success_flag", ""), "account_return_closed_lots": meta.get("account_return_closed_lots", ""), "lifecycle_chart_path": lifecycle_rel, "transaction_day_chart_count": tx_count, "case_lifecycle_board": f"cases/{case_id}/case_lifecycle_board.md", "source_case_image_board": f"{build.SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md", } symbol_index_rows.append(symbol_row) case_symbol_rows.setdefault(case_id, []).append(symbol_row) for case_id, symbols_for_case in case_symbol_rows.items(): build.build_case_board(case_id, pd.Series(v1_by_case[case_id]), symbols_for_case, case_tx_rows.get(case_id, [])) for case_id, meta in v1_by_case.items(): case_orders = order[order["case_id"] == case_id] if case_orders[case_orders["action"] == "BUY"].empty: continue case_rows.append( { "case_id": case_id, "v1_return_scope": meta.get("v1_return_scope", ""), "success_flag": meta.get("success_flag", ""), "account_return_closed_lots": meta.get("account_return_closed_lots", ""), "symbol_count": len(case_symbol_rows.get(case_id, [])), "buy_order_count": int((case_orders["action"] == "BUY").sum()), "sell_order_count": int((case_orders["action"] == "SELL").sum()), "case_lifecycle_board": f"cases/{case_id}/case_lifecycle_board.md", } ) pd.DataFrame(symbol_index_rows).to_csv(build.PACKAGE_ROOT / "lifecycle_symbol_index.csv", index=False, encoding="utf-8-sig") pd.DataFrame(tx_index_rows).to_csv(build.PACKAGE_ROOT / "transaction_day_chart_index.csv", index=False, encoding="utf-8-sig") pd.DataFrame(chart_rows).to_csv(build.PACKAGE_ROOT / "chart_evidence_audit.csv", index=False, encoding="utf-8-sig") pd.DataFrame(missing_rows).to_csv(build.PACKAGE_ROOT / "missing_chart_inputs.csv", index=False, encoding="utf-8-sig") pd.DataFrame(case_rows).to_csv(build.PACKAGE_ROOT / "case_lifecycle_index.csv", index=False, encoding="utf-8-sig") build.build_root_docs(case_rows, symbol_index_rows, tx_index_rows) link_rows = [] for md in sorted(build.PACKAGE_ROOT.rglob("*.md")): link_rows.extend(build.local_link_targets(md)) link_df = pd.DataFrame(link_rows) link_df.to_csv(build.PACKAGE_ROOT / "link_evidence_audit.csv", index=False, encoding="utf-8-sig") expected_lifecycle = len(group_rows) expected_tx = len(order.groupby(["case_id", "symbol", "trade_date"]).size()) lifecycle_actual = int((pd.DataFrame(chart_rows)["chart_role"] == "symbol_lifecycle_daily_50pre_20post_audit_view").sum()) if chart_rows else 0 tx_actual = int((pd.DataFrame(chart_rows)["chart_role"] == "transaction_day_minute_line_audit_view").sum()) if chart_rows else 0 link_missing = 0 if link_df.empty else int((~link_df["target_exists"]).sum()) self_checks = [ ("FINAL_V1_PACKAGE_EXISTS", build.FINAL_ROOT.exists(), str(build.FINAL_ROOT)), ("SOURCE_V1_PACKAGE_EXISTS", build.SOURCE_ROOT.exists(), str(build.SOURCE_ROOT)), ("V1_BUY_CASES_251", len(v1_index) == 251, f"buy_cases={len(v1_index)}"), ("LIFECYCLE_CHARTS_COMPLETE", lifecycle_actual == expected_lifecycle, f"actual={lifecycle_actual}, expected={expected_lifecycle}"), ("TRANSACTION_DAY_CHARTS_COMPLETE", tx_actual == expected_tx, f"actual={tx_actual}, expected={expected_tx}"), ("MISSING_CHART_INPUTS_ZERO", len(missing_rows) == 0, f"missing={len(missing_rows)}"), ("MARKDOWN_LOCAL_LINKS_REACHABLE", link_missing == 0, f"links={len(link_df)}, missing={link_missing}"), ("RETURN_STAT_READY_FALSE_PRESERVED", True, "supplement package does not change V1 return_stat_ready=false"), ] self_df = pd.DataFrame( [{"check_id": check_id, "status": "PASS" if passed else "FAIL", "detail": detail} for check_id, passed, detail in self_checks] ) self_df.to_csv(build.PACKAGE_ROOT / "self_check_items.csv", index=False, encoding="utf-8-sig") fail_count = int((self_df["status"] != "PASS").sum()) self_json = { "run_id": build.RUN_ID, "task_id": build.TASK_ID, "generated_at": build.now_iso(), "source_run_id": build.SOURCE_RUN_ID, "final_run_id": build.FINAL_RUN_ID, "status": "PASS_FOR_V1_LIFECYCLE_CHART_SUPPLEMENT_READY" if fail_count == 0 else "PARTIAL_WITH_MISSING_MARKET_INPUTS", "pass_count": int((self_df["status"] == "PASS").sum()), "fail_count": fail_count, "v1_buy_case_count": len(v1_index), "case_symbol_lifecycle_count": lifecycle_actual, "case_symbol_lifecycle_expected": expected_lifecycle, "transaction_day_chart_count": tx_actual, "transaction_day_chart_expected": expected_tx, "missing_chart_input_count": len(missing_rows), "return_stat_ready": False, } build.write_json(build.PACKAGE_ROOT / "self_check.json", self_json) build.build_root_docs(case_rows, symbol_index_rows, tx_index_rows, self_json) (build.PACKAGE_ROOT / "self_check.md").write_text( "\n".join( [ "# 自检结果", "", f"- 状态：`{self_json['status']}`", f"- PASS：`{self_json['pass_count']}`", f"- FAIL：`{self_json['fail_count']}`", f"- V1 有 BUY case：`{self_json['v1_buy_case_count']}`", f"- 生命周期日线图：`{self_json['case_symbol_lifecycle_count']}` / `{self_json['case_symbol_lifecycle_expected']}`", f"- 交易日分时图：`{self_json['transaction_day_chart_count']}` / `{self_json['transaction_day_chart_expected']}`", f"- 缺失行情输入：`{self_json['missing_chart_input_count']}`", "", "本包只补充人工审阅图片，不改变 V1 账本、收益口径和 `RETURN_STAT_READY=false`。", ] ) + "\n", encoding="utf-8", ) build.write_manifest() print(json.dumps(self_json, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()