from __future__ import annotations
|
|
import json
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-EXPAND-30-20260608-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
HELD_AUDIT_ID = "AUDIT-ANA-WUJI-EXPAND-30-20260608-EXEC-001"
|
ISSUE_ID = "ANA-ISSUE-WUJI-EXPAND-30-BOARD-STATUS-20260608-001"
|
REVIEW_STATUS = "扩样执行审核 HELD 后已返修，当前待复审"
|
|
|
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 read_json(name: str) -> dict:
|
path = ROOT / name
|
return json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
|
|
|
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 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 fmt_int(value: object) -> str:
|
if not has_value(value):
|
return "0"
|
return str(int(float(value)))
|
|
|
def fmt_pct(value: object) -> str:
|
if not has_value(value):
|
return ""
|
return f"{float(value):.4%}"
|
|
|
def rel_for_case(case_id: str, path: str) -> str:
|
return Path(path).relative_to(f"cases/{case_id}").as_posix()
|
|
|
def link_for_case(case_id: str, root_relative_path: object) -> str:
|
path = clean(root_relative_path)
|
if not path:
|
return ""
|
try:
|
return Path(path).relative_to(f"cases/{case_id}").as_posix()
|
except ValueError:
|
return Path("..", "..", path).as_posix()
|
|
|
def case_meta(case_id: str, case_index: pd.DataFrame) -> dict:
|
if case_index.empty:
|
return {}
|
rows = case_index[case_index.case_id == case_id]
|
return rows.iloc[0].to_dict() if not rows.empty else {}
|
|
|
def root_case_ids(manifest: pd.DataFrame, case_index: pd.DataFrame) -> list[str]:
|
ids: list[str] = []
|
if not case_index.empty and "case_id" in case_index.columns:
|
ids.extend(case_index.case_id.dropna().astype(str).tolist())
|
if not manifest.empty and "case_id" in manifest.columns:
|
ids.extend(manifest.case_id.dropna().astype(str).tolist())
|
return sorted(dict.fromkeys(ids))
|
|
|
def build_scope_lines(summary: dict, case_index: pd.DataFrame, lots: pd.DataFrame) -> list[str]:
|
case_count = int(len(case_index)) if not case_index.empty else int(summary.get("candidate_pool", {}).get("selected_case_days", 0))
|
anchor_count = (
|
int((case_index.expand_case_role == "ANCHOR_REUSED_FROM_AUDITED_7_CASE_PILOT").sum())
|
if not case_index.empty and "expand_case_role" in case_index.columns
|
else int(summary.get("candidate_pool", {}).get("anchor_case_days", 0))
|
)
|
new_count = (
|
int((case_index.expand_case_role == "NEW_EXPAND_30_CASE").sum())
|
if not case_index.empty and "expand_case_role" in case_index.columns
|
else int(summary.get("candidate_pool", {}).get("new_case_days", 0))
|
)
|
boundary_counts = lots[lots.lot_status != "CLOSED_BY_AI_SELL"].lot_status.value_counts().to_dict() if not lots.empty else {}
|
boundary_total = int(sum(boundary_counts.values()))
|
boundary_text = "；".join(f"{status} {count} 笔" for status, count in boundary_counts.items()) or "无"
|
return [
|
f"- 当前 run：`{RUN_ID}`",
|
"- 当前阶段：`EXPAND_30_EXECUTION_SELF_CHECK_DONE`",
|
f"- 审核状态：{REVIEW_STATUS}（关联审核 `{HELD_AUDIT_ID}`，问题 `{ISSUE_ID}`）",
|
f"- 范围：{case_count} 个案例日（{anchor_count} 个已审核锚点 + {new_count} 个新增分层案例日）",
|
f"- 边界 lot：{boundary_total} 笔保留（{boundary_text}）",
|
"- `RETURN_STAT_READY=false`",
|
"- 本入口仅用于当前 30 案例日扩样执行包复审；不得转写为完整 2023-2026 baseline 成功率、收益率、胜率、回撤或策略有效性结论。",
|
]
|
|
|
def write_case_image_board(
|
case_id: str,
|
manifest: pd.DataFrame,
|
lots: pd.DataFrame,
|
case_summary: pd.DataFrame,
|
case_index: pd.DataFrame,
|
) -> None:
|
case_dir = ROOT / "cases" / case_id
|
case_dir.mkdir(parents=True, exist_ok=True)
|
rows = manifest[manifest.case_id == case_id].copy() if not manifest.empty else pd.DataFrame()
|
summary = case_summary[case_summary.case_id == case_id] if not case_summary.empty else pd.DataFrame()
|
meta = case_meta(case_id, case_index)
|
|
lines = [
|
f"# {case_id} 图片审核板",
|
"",
|
"用途：给人工审核员按图复核本案例从选股、买入、卖点信号到卖出裁决的全链路。",
|
"",
|
"## 当前扩样执行包状态",
|
"",
|
f"- 当前 run：`{RUN_ID}`",
|
"- 当前阶段：`EXPAND_30_EXECUTION_SELF_CHECK_DONE`",
|
f"- 审核状态：{REVIEW_STATUS}",
|
f"- 入场日：{clean(meta.get('entry_trade_date'), '未记录')}；信号日：{clean(meta.get('signal_trade_date'), '未记录')}",
|
f"- 样本角色：{clean(meta.get('expand_case_role'), '未记录')}；分层：{clean(meta.get('selection_bucket'), '未记录')}",
|
f"- 市场闸门：{clean(meta.get('market_gate_status'), '未记录')}；案例状态：{clean(meta.get('case_status'), '未记录')}",
|
"- `RETURN_STAT_READY=false`",
|
"",
|
]
|
if not summary.empty:
|
s = summary.iloc[0]
|
lines.extend(
|
[
|
"## 案例读数",
|
"",
|
f"- 买入 lot：{fmt_int(s.buy_lot_count)}",
|
f"- 已闭合 lot：{fmt_int(s.closed_lot_count)}",
|
f"- 未闭合 / 边界 lot：{fmt_int(s.unresolved_lot_count)}",
|
f"- 闭合 lot 账户贡献合计：{fmt_pct(s.account_return_closed_lots)}（只用于账本复算，不是完整 baseline 结论）",
|
f"- 当前收益口径：{clean(s.return_boundary)}",
|
"",
|
]
|
)
|
else:
|
lines.extend(
|
[
|
"## 案例读数",
|
"",
|
"- 买入 lot：0",
|
"- 已闭合 lot：0",
|
"- 未闭合 / 边界 lot：0",
|
"- 说明：本案例日未生成 BUY，通常由市场闸门关闭或入场裁决未通过导致。",
|
"",
|
]
|
)
|
|
case_lots = lots[lots.case_id == case_id].copy() if not lots.empty else pd.DataFrame()
|
if not case_lots.empty:
|
lines.extend(["## 持仓状态", ""])
|
for _, lot in case_lots.iterrows():
|
exit_part = f"；卖出 {lot.exit_trade_date} {lot.exit_time} @ {float(lot.exit_price):.2f}" if has_value(lot.exit_price) else ""
|
lines.append(
|
f"- {lot.symbol}：`{lot.lot_status}`；买入 {lot.entry_trade_date} {lot.entry_time} @ {float(lot.entry_price):.2f}{exit_part}"
|
)
|
lines.append("")
|
|
for role, title in ROLE_TITLES.items():
|
group = rows[rows.chart_role == role].copy() if not rows.empty else pd.DataFrame()
|
if group.empty:
|
continue
|
lines.extend([f"## {title}", ""])
|
for _, row in group.iterrows():
|
rel = rel_for_case(case_id, row.path)
|
decision = clean(row.decision_time)
|
note = clean(row.note)
|
status = clean(row.status)
|
lines.extend(
|
[
|
f"### {row.symbol} {decision}".rstrip(),
|
"",
|
f"",
|
"",
|
f"- 图状态：`{status}`",
|
f"- 说明：{note}",
|
"",
|
]
|
)
|
|
lines.extend(
|
[
|
"## 审核边界",
|
"",
|
"- 本板是当前扩样执行包的图片第一入口；CSV 和 JSON 是反查材料。",
|
"- 当前扩样执行包处于 HELD 返修后待复审状态；执行复审通过前不得标记扩样执行通过。",
|
"- `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,
|
case_index: pd.DataFrame,
|
) -> None:
|
case_dir = ROOT / "cases" / case_id
|
case_dir.mkdir(parents=True, exist_ok=True)
|
summary = case_summary[case_summary.case_id == case_id] if not case_summary.empty else pd.DataFrame()
|
case_lots = lots[lots.case_id == case_id].copy() if not lots.empty else pd.DataFrame()
|
case_decisions = decisions[decisions.case_id == case_id].copy() if not decisions.empty else pd.DataFrame()
|
meta = case_meta(case_id, case_index)
|
lines = [
|
f"# {case_id} 一页式故事板",
|
"",
|
f"- 当前 run：`{RUN_ID}`",
|
"- 当前阶段：`EXPAND_30_EXECUTION_SELF_CHECK_DONE`",
|
f"- 审核状态：{REVIEW_STATUS}",
|
f"- 案例范围：30 个案例日受控扩样中的 1 个；样本角色：{clean(meta.get('expand_case_role'), '未记录')}",
|
"- 图片审核板：`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：{fmt_int(s.buy_lot_count)}",
|
f"- 已闭合 lot：{fmt_int(s.closed_lot_count)}",
|
f"- 未闭合 / 边界 lot：{fmt_int(s.unresolved_lot_count)}",
|
f"- 闭合 lot 账户贡献合计：{fmt_pct(s.account_return_closed_lots)}（只用于账本复算，不是最终收益）",
|
"- `RETURN_STAT_READY=false`",
|
"",
|
]
|
)
|
else:
|
lines.extend(["## 当前结论边界", "", "- 本案例日无 BUY，未进入收益统计候选。", "- `RETURN_STAT_READY=false`", ""])
|
|
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_link = link_for_case(case_id, row.evidence_image_path)
|
image = f"；图：[{Path(image_link).name}]({image_link})" if image_link else ""
|
lines.append(
|
f"- {row.decision_stage} / `{row.action_status}`：{row.symbol} {decision_time}{price}；{clean(row.decision_reason_cn)}{image}"
|
)
|
lines.append("")
|
|
if not case_lots.empty:
|
lines.extend(["## Lot 收口", ""])
|
for _, lot in case_lots.iterrows():
|
contribution = f"；账户贡献 {fmt_pct(lot.account_return_contribution_pct)}" if has_value(lot.account_return_contribution_pct) else ""
|
exit_part = f"；卖出 {lot.exit_trade_date} {lot.exit_time} @ {float(lot.exit_price):.2f}" if has_value(lot.exit_price) else ""
|
lines.append(
|
f"- {lot.trade_lot_id} / {lot.symbol}：`{lot.lot_status}`；买入 {lot.entry_trade_date} {lot.entry_time} @ {float(lot.entry_price):.2f}{exit_part}{contribution}"
|
)
|
lines.append("")
|
|
role_counts = manifest[manifest.case_id == case_id].chart_role.value_counts().to_dict() if not manifest.empty else {}
|
lines.extend(["## 图片清单概览", ""])
|
lines.extend([f"- {ROLE_TITLES.get(role, role)}：{count} 张" for role, count in role_counts.items()] or ["- 无图片"])
|
lines.extend(
|
[
|
"",
|
"## 审核提示",
|
"",
|
"- 先看 `case_image_board.md` 的图，再回查账本。",
|
"- 若看到日 K / 分钟线触发不一致或数据缺口，应按待审项处理，不得自行补收益。",
|
"- 当前扩样执行包仍待复审，不得据此引用完整 baseline 结论。",
|
"",
|
]
|
)
|
(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,
|
case_index: pd.DataFrame,
|
summary: dict,
|
) -> None:
|
lines = [
|
f"# {RUN_ID} 案例故事板总入口",
|
"",
|
"用途：给人工审核员从一个入口进入 30 个案例日受控扩样包；每个案例优先看图片审核板，再回查账本。",
|
"",
|
"## 总体边界",
|
"",
|
*build_scope_lines(summary, case_index, lots),
|
"",
|
"## 案例入口",
|
"",
|
"| case_id | 样本角色 | 入场日 | 市场闸门 | 买入lot | 已闭合 | 边界lot | 图片板 | 故事板 |",
|
"|---|---|---|---|---:|---:|---:|---|---|",
|
]
|
summary_by_case = case_summary.set_index("case_id") if not case_summary.empty else pd.DataFrame()
|
for _, row in case_index.sort_values("entry_trade_date").iterrows():
|
case_id = row.case_id
|
if not summary_by_case.empty and case_id in summary_by_case.index:
|
s = summary_by_case.loc[case_id]
|
buy_count = fmt_int(s.buy_lot_count)
|
closed_count = fmt_int(s.closed_lot_count)
|
unresolved_count = fmt_int(s.unresolved_lot_count)
|
else:
|
buy_count = closed_count = unresolved_count = "0"
|
lines.append(
|
f"| {case_id} | {row.expand_case_role} | {row.entry_trade_date} | {row.market_gate_status} | {buy_count} | {closed_count} | {unresolved_count} | [图片板](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,
|
case_index: pd.DataFrame,
|
summary: dict,
|
) -> None:
|
non_real_counts = lots[lots.lot_status != "CLOSED_BY_AI_SELL"].lot_status.value_counts() if not lots.empty else pd.Series(dtype=int)
|
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 图片审核入口",
|
"",
|
*build_scope_lines(summary, case_index, lots),
|
"",
|
"本入口用于人工审核图片链路。当前已生成候选日 K、买入 1 分钟复核、买入裁决、卖点日 K 信号、卖出 1 分钟裁决、订单账本和账户账本；本轮扩样执行审核已 HELD，当前为返修后待复审。",
|
"",
|
"## 总体边界",
|
"",
|
"- `RETURN_STAT_READY=false`",
|
f"- {non_real_summary}",
|
"- 根入口只做导航；逐图审核请进入各案例 `case_image_board.md`。",
|
"",
|
"## 案例入口",
|
"",
|
"| case_id | 样本角色 | 入场日 | 市场闸门 | 买入lot | 已闭合 | 边界lot | 图片板 | 故事板 |",
|
"|---|---|---|---|---:|---:|---:|---|---|",
|
]
|
summary_by_case = case_summary.set_index("case_id") if not case_summary.empty else pd.DataFrame()
|
for _, row in case_index.sort_values("entry_trade_date").iterrows():
|
case_id = row.case_id
|
if not summary_by_case.empty and case_id in summary_by_case.index:
|
s = summary_by_case.loc[case_id]
|
buy_count = fmt_int(s.buy_lot_count)
|
closed_count = fmt_int(s.closed_lot_count)
|
unresolved_count = fmt_int(s.unresolved_lot_count)
|
else:
|
buy_count = closed_count = unresolved_count = "0"
|
lines.append(
|
f"| {case_id} | {row.expand_case_role} | {row.entry_trade_date} | {row.market_gate_status} | {buy_count} | {closed_count} | {unresolved_count} | [图片板](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")
|
case_index = safe_read_csv("case_index.csv")
|
summary = read_json("summary.json")
|
|
all_case_ids = root_case_ids(manifest, case_index)
|
for case_id in all_case_ids:
|
write_case_image_board(case_id, manifest, lots, case_summary, case_index)
|
write_case_story_board(case_id, manifest, lots, decisions, case_summary, case_index)
|
write_root_story_board(manifest, lots, case_summary, case_index, summary)
|
write_root_image_board(manifest, lots, case_summary, case_index, summary)
|
|
|
if __name__ == "__main__":
|
main()
|
