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()
|
