from __future__ import annotations
|
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
ROLE_TITLES = {
|
"candidate_daily_100d_decision_view": "1. 选股日K图（约100交易日）",
|
"entry_1m_morning_review_view": "2. 买点早盘1分钟复核图",
|
"entry_1m_late_review_view": "3. 买点尾盘1分钟复核图",
|
"entry_1m_buy_decision_view": "4. 买入裁决1分钟图",
|
"exit_daily_signal_review_view": "5. 卖点 / 持仓日K信号图",
|
"exit_1m_sell_decision_view": "6. 卖出裁决1分钟图",
|
}
|
|
|
def safe_read_csv(name: str) -> pd.DataFrame:
|
path = ROOT / name
|
return pd.read_csv(path, encoding="utf-8-sig") if path.exists() else pd.DataFrame()
|
|
|
def rel_for_case(case_id: str, path: str) -> str:
|
return Path(path).relative_to(f"cases/{case_id}").as_posix()
|
|
|
def clean(value: object, default: str = "") -> str:
|
text = str(value).strip()
|
return default if text.lower() in ["", "nan", "none"] else text
|
|
|
def has_value(value: object) -> bool:
|
return clean(value) != ""
|
|
|
def write_case_image_board(case_id: str, manifest: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame) -> None:
|
case_dir = ROOT / "cases" / case_id
|
rows = manifest[manifest.case_id == case_id].copy()
|
summary = case_summary[case_summary.case_id == case_id]
|
lines = [
|
f"# {case_id} 图片审核板",
|
"",
|
"用途：给人工审核员按图复核本案例从选股、买入、卖点信号到卖出裁决的全链路。",
|
"",
|
]
|
if not summary.empty:
|
s = summary.iloc[0]
|
lines.extend(
|
[
|
"## 案例读数",
|
"",
|
f"- 买入 lot：{s.buy_lot_count}",
|
f"- 已闭合 lot：{s.closed_lot_count}",
|
f"- 未闭合 / 待审 lot：{s.unresolved_lot_count}",
|
f"- 当前收益口径：{s.return_boundary}",
|
"",
|
]
|
)
|
case_lots = lots[lots.case_id == case_id].copy()
|
if not case_lots.empty:
|
lines.extend(["## 持仓状态", ""])
|
for _, lot in case_lots.iterrows():
|
lines.append(
|
f"- {lot.symbol}：{lot.lot_status}；买入 {lot.entry_trade_date} {lot.entry_time} @ {float(lot.entry_price):.2f}"
|
+ (f"；卖出 {lot.exit_trade_date} {lot.exit_time} @ {float(lot.exit_price):.2f}" if has_value(lot.exit_price) else "")
|
)
|
lines.append("")
|
|
for role, title in ROLE_TITLES.items():
|
group = rows[rows.chart_role == role].copy()
|
if group.empty:
|
continue
|
lines.extend([f"## {title}", ""])
|
for _, row in group.iterrows():
|
rel = rel_for_case(case_id, row.path)
|
decision = str(row.decision_time) if "decision_time" in row else ""
|
note = str(row.note) if "note" in row else ""
|
status = str(row.status) if "status" in row else ""
|
lines.extend(
|
[
|
f"### {row.symbol} {decision}",
|
"",
|
f"",
|
"",
|
f"- 图状态：`{status}`",
|
f"- 说明：{note}",
|
"",
|
]
|
)
|
lines.extend(
|
[
|
"## 审核边界",
|
"",
|
"- 本板是图片第一入口，CSV 和 JSON 是反查材料。",
|
"- 当前仍为结构试点；退出复核执行审核已通过，但 `RETURN_STAT_READY=false`，不得引用完整收益、成功率、胜率或回撤。",
|
"",
|
]
|
)
|
(case_dir / "case_image_board.md").write_text("\n".join(lines), encoding="utf-8")
|
|
|
def write_case_story_board(case_id: str, manifest: pd.DataFrame, lots: pd.DataFrame, decisions: pd.DataFrame, case_summary: pd.DataFrame) -> None:
|
case_dir = ROOT / "cases" / case_id
|
summary = case_summary[case_summary.case_id == case_id]
|
case_lots = lots[lots.case_id == case_id].copy()
|
case_decisions = decisions[decisions.case_id == case_id].copy()
|
lines = [
|
f"# {case_id} 一页式故事板",
|
"",
|
f"- 图片审核板：`case_image_board.md`",
|
"- 反查账本：`candidate_ledger.csv`、`decision_log.csv`、`order_ledger.csv`、`position_lot_ledger.csv`、`image_manifest.csv`",
|
"",
|
]
|
if not summary.empty:
|
s = summary.iloc[0]
|
lines.extend(
|
[
|
"## 当前结论边界",
|
"",
|
f"- 买入 lot：{s.buy_lot_count}",
|
f"- 已闭合 lot：{s.closed_lot_count}",
|
f"- 未闭合 / 待审 lot：{s.unresolved_lot_count}",
|
f"- 闭合 lot 账户贡献合计：{float(s.account_return_closed_lots):.4%}（只用于账本复算，不是最终收益）",
|
f"- RETURN_STAT_READY：{s.strict_baseline_return_ready_flag}",
|
"",
|
]
|
)
|
lines.extend(["## 操作时间线", ""])
|
if case_decisions.empty:
|
lines.append("- 本案例无买卖裁决，仅保留候选 / 不开仓证据。")
|
else:
|
for _, row in case_decisions.iterrows():
|
decision_time = clean(row.decision_time, "无裁决时间")
|
price = f" @ {float(row.price):.2f}" if has_value(row.price) else ""
|
image = f"；图：`{row.evidence_image_path}`" if has_value(row.evidence_image_path) else ""
|
lines.append(
|
f"- {row.decision_stage} / {row.action_status}：{row.symbol} {decision_time}{price}；{row.decision_reason_cn}{image}"
|
)
|
lines.append("")
|
if not case_lots.empty:
|
lines.extend(["## Lot 收口", ""])
|
for _, lot in case_lots.iterrows():
|
contribution = "" if str(lot.account_return_contribution_pct).strip() in ["", "nan"] else f"；账户贡献 {float(lot.account_return_contribution_pct):.4%}"
|
lines.append(
|
f"- {lot.trade_lot_id} / {lot.symbol}：{lot.lot_status}；买入 {lot.entry_trade_date} {lot.entry_time} @ {float(lot.entry_price):.2f}"
|
+ (f"；卖出 {lot.exit_trade_date} {lot.exit_time} @ {float(lot.exit_price):.2f}" if has_value(lot.exit_price) else "")
|
+ contribution
|
)
|
lines.append("")
|
role_counts = manifest[manifest.case_id == case_id].chart_role.value_counts().to_dict()
|
lines.extend(
|
[
|
"## 图片清单概览",
|
"",
|
*[f"- {ROLE_TITLES.get(role, role)}：{count} 张" for role, count in role_counts.items()],
|
"",
|
"## 审核提示",
|
"",
|
"- 先看 `case_image_board.md` 的图，再回查账本。",
|
"- 若看到日 K / 分钟线触发不一致或数据缺口，应按待审项处理，不得自行补收益。",
|
"",
|
]
|
)
|
(case_dir / "case_story_board.md").write_text("\n".join(lines), encoding="utf-8")
|
|
|
def write_root_story_board(manifest: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame) -> None:
|
lines = [
|
f"# {RUN_ID} 案例故事板总入口",
|
"",
|
"用途：给人工审核员从一个入口进入 7 个小样本案例；每个案例优先看图片审核板，再回查账本。",
|
"",
|
"## 总体边界",
|
"",
|
"- 当前阶段：`STRUCTURE_PILOT_EXIT_REVIEW_RESOLVED_SELF_CHECK_DONE`",
|
"- 执行审核：尚未提交",
|
"- `RETURN_STAT_READY=false`；不得引用完整 baseline 收益率、成功率、胜率或回撤。",
|
"",
|
"## 案例入口",
|
"",
|
"| case_id | 买入lot | 已闭合 | 待审/未闭合 | 图片板 | 故事板 |",
|
"|---|---:|---:|---:|---|---|",
|
]
|
for _, row in case_summary.sort_values("case_id").iterrows():
|
lines.append(
|
f"| {row.case_id} | {row.buy_lot_count} | {row.closed_lot_count} | {row.unresolved_lot_count} | [图片板](cases/{row.case_id}/case_image_board.md) | [故事板](cases/{row.case_id}/case_story_board.md) |"
|
)
|
no_trade_cases = sorted(set(manifest.case_id.unique()) - set(case_summary.case_id.unique()))
|
for case_id in no_trade_cases:
|
lines.append(f"| {case_id} | 0 | 0 | 0 | [图片板](cases/{case_id}/case_image_board.md) | [故事板](cases/{case_id}/case_story_board.md) |")
|
lines.extend(
|
[
|
"",
|
"## 图片角色统计",
|
"",
|
]
|
)
|
for role, count in manifest.chart_role.value_counts().items():
|
lines.append(f"- {ROLE_TITLES.get(role, role)}：{count} 张")
|
lines.extend(
|
[
|
"",
|
"## Lot 状态统计",
|
"",
|
]
|
)
|
for status, count in lots.lot_status.value_counts().items():
|
lines.append(f"- {status}：{count}")
|
lines.append("")
|
(ROOT / "case_story_board.md").write_text("\n".join(lines), encoding="utf-8")
|
|
|
def write_root_image_board(manifest: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame) -> None:
|
non_real_statuses = ["WINDOW_END_VALUATION_ONLY", "EXIT_DATA_GAP_HELD", "SELL_REVIEW_DATA_MISMATCH_HELD", "EXIT_REVIEW_HELD"]
|
non_real_counts = lots[lots.lot_status.isin(non_real_statuses)].lot_status.value_counts()
|
if non_real_counts.empty:
|
non_real_summary = "当前没有非真实 SELL lot 保留。"
|
else:
|
parts = [f"{status} {count} 笔" for status, count in non_real_counts.items()]
|
non_real_summary = f"当前仍有 {int(non_real_counts.sum())} 笔非真实 SELL lot 保留：{'；'.join(parts)}；不纳入完整收益统计。"
|
lines = [
|
"# RUN 图片审核入口",
|
"",
|
f"run_id：`{RUN_ID}`",
|
"阶段：`STRUCTURE_PILOT_EXIT_REVIEW_RESOLVED_SELF_CHECK_DONE`",
|
"",
|
"本入口用于人工审核图片链路。当前已生成候选日 K、买入 1 分钟复核、买入裁决、卖点日 K 信号、卖出 1 分钟裁决、订单账本和账户账本；退出复核执行审核已通过，但 `RETURN_STAT_READY=false`，不得引用完整收益结论。",
|
"",
|
"## 总体边界",
|
"",
|
"- `RETURN_STAT_READY=false`",
|
f"- {non_real_summary}",
|
"- 根入口只做导航；逐图审核请进入各案例 `case_image_board.md`。",
|
"",
|
"## 案例入口",
|
"",
|
"| case_id | 买入lot | 已闭合 | 待审/未闭合 | 图片板 | 故事板 |",
|
"|---|---:|---:|---:|---|---|",
|
]
|
for _, row in case_summary.sort_values("case_id").iterrows():
|
lines.append(
|
f"| {row.case_id} | {row.buy_lot_count} | {row.closed_lot_count} | {row.unresolved_lot_count} | [图片板](cases/{row.case_id}/case_image_board.md) | [故事板](cases/{row.case_id}/case_story_board.md) |"
|
)
|
no_trade_cases = sorted(set(manifest.case_id.unique()) - set(case_summary.case_id.unique()))
|
for case_id in no_trade_cases:
|
lines.append(f"| {case_id} | 0 | 0 | 0 | [图片板](cases/{case_id}/case_image_board.md) | [故事板](cases/{case_id}/case_story_board.md) |")
|
lines.extend(["", "## 图片角色统计", ""])
|
for role, count in manifest.chart_role.value_counts().items():
|
lines.append(f"- {ROLE_TITLES.get(role, role)}：{count} 张")
|
lines.extend(["", "## Lot 状态统计", ""])
|
for status, count in lots.lot_status.value_counts().items():
|
lines.append(f"- {status}：{count}")
|
lines.append("")
|
(ROOT / "case_image_board.md").write_text("\n".join(lines), encoding="utf-8")
|
|
|
def main() -> None:
|
manifest = safe_read_csv("image_manifest.csv")
|
lots = safe_read_csv("position_lot_ledger.csv")
|
decisions = safe_read_csv("decision_log.csv")
|
case_summary = safe_read_csv("case_summary.csv")
|
|
all_case_ids = sorted(manifest.case_id.dropna().unique().tolist())
|
for case_id in all_case_ids:
|
write_case_image_board(case_id, manifest, lots, case_summary)
|
write_case_story_board(case_id, manifest, lots, decisions, case_summary)
|
write_root_story_board(manifest, lots, case_summary)
|
write_root_image_board(manifest, lots, case_summary)
|
|
|
if __name__ == "__main__":
|
main()
|
