# -*- coding: utf-8 -*-
|
"""Repair V1 strict sell / rolling-low package with an independent decision ledger.
|
|
This script does not rerun candidate selection, buy/sell rule detection, account
|
calculation, or chart generation. It separates the already generated code
|
candidates from the final analyst decision source, then rewrites downstream
|
ledgers and review entry points so the audit chain can verify that final
|
human_decision_* fields are consumed from manual_decision_ledger.csv.
|
"""
|
|
from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import re
|
from collections import Counter
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
from urllib.parse import unquote
|
|
import pandas as pd
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
|
TASK_ID = "ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001"
|
DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-DESIGN-002"
|
SUPP_DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-SUPP-DESIGN-003"
|
MANUAL_SOURCE_SUPP_DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-SUPP-DESIGN-004"
|
EXEC_HELD_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-EXEC-001"
|
ISSUE_ID = "ANA-ISSUE-WUJI-STRICT-SELL-ROLLING-GAP-20260609-001"
|
MANUAL_EXTERNAL_SOURCE_LEDGER = "manual_decision_external_source_ledger.csv"
|
MANUAL_EXTERNAL_DRAFT = "manual_decision_external_draft.md"
|
MANUAL_SOURCE_LEDGER = "manual_decision_source_ledger.csv"
|
MANUAL_LEDGER = "manual_decision_ledger.csv"
|
OLD_RULE_REPLAY_SOURCE = "ANALYST_RULE_REPLAY_WITH_FROZEN_THRESHOLDS"
|
TZ = timezone(timedelta(hours=8))
|
APPLICATION_STARTED_AT = datetime.now(TZ).replace(microsecond=0)
|
|
|
def now_iso() -> str:
|
return datetime.now(TZ).replace(microsecond=0).isoformat()
|
|
|
def read_csv(name: str) -> pd.DataFrame:
|
path = ROOT / name
|
return pd.read_csv(path, dtype=str, keep_default_na=False).fillna("")
|
|
|
def write_csv(df: pd.DataFrame, name: str) -> None:
|
(ROOT / name).parent.mkdir(parents=True, exist_ok=True)
|
df.to_csv(ROOT / name, index=False, encoding="utf-8-sig")
|
|
|
def write_json(path: Path, payload: dict) -> None:
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
def rel_path(path: Path) -> str:
|
return path.relative_to(ROOT).as_posix()
|
|
|
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 as_float(value: str, default: float = 0.0) -> float:
|
try:
|
return float(value)
|
except Exception:
|
return default
|
|
|
def code_reason(row: pd.Series) -> str:
|
return row.get("code_suggested_reason_cn", "") or row.get("code_evidence_reason_cn", "") or row.get("human_decision_reason_cn", "")
|
|
|
def build_candidate_ledgers(strict_original: pd.DataFrame, rolling_original: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
|
strict_drop = {
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"manual_decision_id",
|
"manual_decision_ledger_path",
|
"review_input_chart_path",
|
"code_evidence_reason_cn",
|
"code_suggested_reason_cn",
|
}
|
strict_cols = [c for c in strict_original.columns if c not in strict_drop]
|
strict_candidate = strict_original[strict_cols].copy()
|
strict_candidate["code_suggested_reason_cn"] = strict_original.apply(code_reason, axis=1)
|
strict_candidate["review_input_chart_path"] = strict_original.get("chart_path", "")
|
strict_candidate["candidate_stage"] = "STAGE1_CODE_CANDIDATE_ONLY"
|
strict_candidate["final_decision_source_required"] = MANUAL_EXTERNAL_SOURCE_LEDGER
|
|
rolling_drop = {
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"manual_decision_id",
|
"manual_decision_ledger_path",
|
"review_input_chart_path",
|
"code_suggested_action",
|
"code_suggested_reason_cn",
|
}
|
rolling_cols = [c for c in rolling_original.columns if c not in rolling_drop]
|
rolling_candidate = rolling_original[rolling_cols].copy()
|
rolling_candidate["code_suggested_action"] = rolling_original.apply(
|
lambda r: r.get("code_suggested_action", "") or r.get("human_decision_action", ""), axis=1
|
)
|
rolling_candidate["code_suggested_reason_cn"] = rolling_original.apply(code_reason, axis=1)
|
rolling_candidate["review_input_chart_path"] = rolling_original.get("chart_path", "")
|
rolling_candidate["candidate_stage"] = "STAGE1_CODE_CANDIDATE_ONLY"
|
rolling_candidate["final_decision_source_required"] = MANUAL_EXTERNAL_SOURCE_LEDGER
|
|
write_csv(strict_candidate, "strict_sell_candidate_ledger.csv")
|
write_csv(rolling_candidate, "rolling_low_buy_candidate_ledger.csv")
|
return strict_candidate, rolling_candidate
|
|
|
def load_manual_decision_source(strict_candidate: pd.DataFrame, rolling_candidate: pd.DataFrame) -> pd.DataFrame:
|
"""Load analyst manual decisions created by an earlier external draft step.
|
|
This function intentionally refuses to infer or build human_decision_* from
|
code_suggested_* fields. The external source ledger must already exist.
|
"""
|
source_path = ROOT / MANUAL_EXTERNAL_SOURCE_LEDGER
|
if not source_path.exists():
|
raise FileNotFoundError(
|
f"{MANUAL_EXTERNAL_SOURCE_LEDGER} is required. Create the external analyst manual decision source before applying decisions."
|
)
|
external = read_csv(MANUAL_EXTERNAL_SOURCE_LEDGER)
|
required = {
|
"external_decision_id",
|
"artifact_type",
|
"signal_id",
|
"rolling_signal_id",
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"code_suggested_action",
|
"accept_code_suggestion_flag",
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"review_input_chart_path",
|
"review_input_chart_sha256",
|
"manual_draft_path",
|
"manual_draft_sha256",
|
"reviewer_notes",
|
}
|
missing = required - set(external.columns)
|
if missing:
|
raise ValueError(f"{MANUAL_EXTERNAL_SOURCE_LEDGER} missing required fields: {sorted(missing)}")
|
|
if external["external_decision_id"].duplicated().any():
|
raise ValueError("manual external source has duplicate external_decision_id")
|
if external["human_decision_action"].str.len().eq(0).any():
|
raise ValueError("manual external source has empty human_decision_action")
|
if external["human_decision_reason_cn"].str.len().eq(0).any():
|
raise ValueError("manual external source has empty human_decision_reason_cn")
|
if external["decision_operator"].str.len().eq(0).any():
|
raise ValueError("manual external source has empty decision_operator")
|
if external["decision_time"].str.len().eq(0).any():
|
raise ValueError("manual external source has empty decision_time")
|
if external["decision_source"].str.len().eq(0).any():
|
raise ValueError("manual external source has empty decision_source")
|
|
bad_generated_source = external["decision_source"].isin(
|
{
|
OLD_RULE_REPLAY_SOURCE,
|
"CASE_ANALYSIS_ANALYST_AI_MANUAL_CHART_EVIDENCE_REVIEW_SOURCE",
|
}
|
)
|
if bad_generated_source.any():
|
raise ValueError("manual external source contains rejected/generated decision_source values")
|
|
chart_hash_mismatch = []
|
for _, row in external.iterrows():
|
chart_path = row.get("review_input_chart_path", "")
|
if chart_path:
|
actual = sha256_file(ROOT / chart_path) if (ROOT / chart_path).exists() else ""
|
if actual != row.get("review_input_chart_sha256", ""):
|
chart_hash_mismatch.append(row.get("external_decision_id", ""))
|
draft_path = row.get("manual_draft_path", "")
|
if draft_path:
|
draft = ROOT / draft_path
|
actual_draft = sha256_file(draft) if draft.exists() else ""
|
if actual_draft != row.get("manual_draft_sha256", ""):
|
raise ValueError(f"manual draft hash mismatch for {row.get('external_decision_id', '')}")
|
if chart_hash_mismatch:
|
raise ValueError(f"chart hash mismatch for external decisions: {chart_hash_mismatch[:5]}")
|
|
expected_signal_ids = set(strict_candidate["signal_id"])
|
expected_rolling_ids = set(rolling_candidate["rolling_signal_id"])
|
actual_signal_ids = set(external.loc[external["artifact_type"] == "STRICT_SELL_SIGNAL", "signal_id"])
|
actual_rolling_ids = set(external.loc[external["artifact_type"] == "ROLLING_LOW_BUY_SIGNAL", "rolling_signal_id"])
|
if actual_signal_ids != expected_signal_ids:
|
raise ValueError("manual source strict signal ids do not match stage-1 candidate ledger")
|
if actual_rolling_ids != expected_rolling_ids:
|
raise ValueError("manual source rolling signal ids do not match stage-1 candidate ledger")
|
|
manual = external.copy()
|
manual["decision_id"] = manual["external_decision_id"]
|
if "code_suggested_reason_cn" not in manual.columns:
|
strict_reason = dict(zip(strict_candidate["signal_id"], strict_candidate["code_suggested_reason_cn"]))
|
rolling_reason = dict(zip(rolling_candidate["rolling_signal_id"], rolling_candidate["code_suggested_reason_cn"]))
|
manual["code_suggested_reason_cn"] = manual.apply(
|
lambda r: strict_reason.get(r.get("signal_id", ""), "") or rolling_reason.get(r.get("rolling_signal_id", ""), ""),
|
axis=1,
|
)
|
if "code_suggestion_review_result" not in manual.columns:
|
manual["code_suggestion_review_result"] = manual["accept_code_suggestion_flag"].map(
|
lambda v: "MANUAL_ACCEPTED_CODE_SUGGESTION" if str(v).upper() == "TRUE" else "MANUAL_MODIFIED_CODE_SUGGESTION"
|
)
|
|
write_csv(manual, MANUAL_SOURCE_LEDGER)
|
write_csv(manual, MANUAL_LEDGER)
|
return manual
|
|
|
def apply_manual_decisions(
|
strict_original: pd.DataFrame,
|
rolling_original: pd.DataFrame,
|
strict_candidate: pd.DataFrame,
|
rolling_candidate: pd.DataFrame,
|
manual: pd.DataFrame,
|
) -> tuple[pd.DataFrame, pd.DataFrame]:
|
strict = strict_original.copy()
|
strict["code_evidence_reason_cn"] = strict_candidate["code_suggested_reason_cn"].values
|
strict["review_input_chart_path"] = strict_candidate["review_input_chart_path"].values
|
|
strict_manual = manual[manual["artifact_type"] == "STRICT_SELL_SIGNAL"].set_index("signal_id")
|
for col in [
|
"decision_id",
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"reviewer_notes",
|
]:
|
strict["manual_decision_id" if col == "decision_id" else col] = strict["signal_id"].map(strict_manual[col]).fillna("")
|
strict["manual_decision_ledger_path"] = MANUAL_LEDGER
|
strict["chart_reason_rendered"] = "True"
|
strict["storyboard_reason_rendered"] = "True"
|
write_csv(strict, "strict_sell_signal_ledger.csv")
|
|
rolling = rolling_original.copy()
|
rolling["code_suggested_action"] = rolling_candidate["code_suggested_action"].values
|
rolling["code_evidence_reason_cn"] = rolling_candidate["code_suggested_reason_cn"].values
|
rolling["review_input_chart_path"] = rolling_candidate["review_input_chart_path"].values
|
|
rolling_manual = manual[manual["artifact_type"] == "ROLLING_LOW_BUY_SIGNAL"].set_index("rolling_signal_id")
|
for col in [
|
"decision_id",
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"reviewer_notes",
|
]:
|
rolling["manual_decision_id" if col == "decision_id" else col] = rolling["rolling_signal_id"].map(rolling_manual[col]).fillna("")
|
rolling["manual_decision_ledger_path"] = MANUAL_LEDGER
|
rolling["chart_reason_rendered"] = "True"
|
rolling["storyboard_reason_rendered"] = "True"
|
write_csv(rolling, "rolling_low_buy_signal_ledger.csv")
|
return strict, rolling
|
|
|
def build_signal_maps(strict: pd.DataFrame, rolling: pd.DataFrame) -> tuple[dict[str, str], dict[str, str], dict[str, str], dict[str, str]]:
|
sell_rows = strict[strict["human_decision_action"] == "SELL"].copy()
|
sell_by_lot: dict[str, str] = {}
|
sell_decision_by_lot: dict[str, str] = {}
|
for _, row in sell_rows.iterrows():
|
lot_id = row.get("source_lot_id", "")
|
if lot_id and lot_id not in sell_by_lot:
|
sell_by_lot[lot_id] = row.get("signal_id", "")
|
sell_decision_by_lot[lot_id] = row.get("manual_decision_id", "")
|
|
rolling_decision_by_id = dict(zip(rolling["rolling_signal_id"], rolling["manual_decision_id"]))
|
rolling_source_by_id = dict(zip(rolling["rolling_signal_id"], rolling["decision_source"]))
|
return sell_by_lot, sell_decision_by_lot, rolling_decision_by_id, rolling_source_by_id
|
|
|
def repair_orders_and_lots(strict: pd.DataFrame, rolling: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
|
sell_by_lot, sell_decision_by_lot, rolling_decision_by_id, rolling_source_by_id = build_signal_maps(strict, rolling)
|
|
orders = read_csv("strict_order_ledger.csv")
|
for col in ["source_signal_id", "manual_decision_id", "manual_decision_ledger_path", "manual_decision_source"]:
|
if col not in orders.columns:
|
orders[col] = ""
|
|
orders["source_signal_id"] = ""
|
orders["manual_decision_id"] = ""
|
orders["manual_decision_ledger_path"] = ""
|
orders["manual_decision_source"] = ""
|
|
sell_mask = orders["action"] == "SELL"
|
orders.loc[sell_mask, "source_signal_id"] = orders.loc[sell_mask, "source_lot_id"].map(sell_by_lot).fillna("")
|
orders.loc[sell_mask, "manual_decision_id"] = orders.loc[sell_mask, "source_lot_id"].map(sell_decision_by_lot).fillna("")
|
|
rolling_buy_mask = (orders["action"] == "BUY") & orders["source_order_id"].str.startswith("ROLL-", na=False)
|
orders.loc[rolling_buy_mask, "source_signal_id"] = orders.loc[rolling_buy_mask, "source_order_id"]
|
orders.loc[rolling_buy_mask, "manual_decision_id"] = orders.loc[rolling_buy_mask, "source_order_id"].map(rolling_decision_by_id).fillna("")
|
orders.loc[orders["manual_decision_id"] != "", "manual_decision_ledger_path"] = MANUAL_LEDGER
|
|
all_sources = {}
|
all_sources.update(dict(zip(strict["manual_decision_id"], strict["decision_source"])))
|
all_sources.update(dict(zip(rolling["manual_decision_id"], rolling["decision_source"])))
|
orders["manual_decision_source"] = orders["manual_decision_id"].map(all_sources).fillna("")
|
write_csv(orders, "strict_order_ledger.csv")
|
|
lots = read_csv("strict_position_lot_ledger.csv")
|
for col in ["source_signal_id", "manual_decision_id", "manual_decision_ledger_path", "exit_signal_id", "exit_manual_decision_id"]:
|
if col not in lots.columns:
|
lots[col] = ""
|
|
lots["source_signal_id"] = ""
|
lots["manual_decision_id"] = ""
|
lots["manual_decision_ledger_path"] = ""
|
lots["exit_signal_id"] = ""
|
lots["exit_manual_decision_id"] = ""
|
|
rolling_lot_mask = lots["lot_source_type"] == "ROLLING_LOW_BUY"
|
lots.loc[rolling_lot_mask, "source_signal_id"] = lots.loc[rolling_lot_mask, "source_lot_id"]
|
lots.loc[rolling_lot_mask, "manual_decision_id"] = lots.loc[rolling_lot_mask, "source_lot_id"].map(rolling_decision_by_id).fillna("")
|
lots.loc[lots["manual_decision_id"] != "", "manual_decision_ledger_path"] = MANUAL_LEDGER
|
|
closed_mask = lots["lot_status"] == "CLOSED_BY_V1_SELL"
|
lots.loc[closed_mask, "exit_signal_id"] = lots.loc[closed_mask, "source_lot_id"].map(sell_by_lot).fillna("")
|
lots.loc[closed_mask, "exit_manual_decision_id"] = lots.loc[closed_mask, "source_lot_id"].map(sell_decision_by_lot).fillna("")
|
write_csv(lots, "strict_position_lot_ledger.csv")
|
return orders, lots
|
|
|
def repair_sampling(manual: pd.DataFrame) -> pd.DataFrame:
|
old = read_csv("manual_review_sampling_index.csv")
|
lookup: dict[str, dict] = {}
|
for _, row in manual.iterrows():
|
key = row.get("signal_id", "") or row.get("rolling_signal_id", "")
|
lookup[key] = row.to_dict()
|
|
rows = []
|
for i, row in old.iterrows():
|
key = row.get("signal_id", "")
|
manual_row = lookup.get(key, {})
|
is_rolling = manual_row.get("artifact_type", "") == "ROLLING_LOW_BUY_SIGNAL"
|
rows.append(
|
{
|
"sample_seq": row.get("sample_seq", str(i + 1)),
|
"artifact_type": manual_row.get("artifact_type", row.get("artifact_type", "")),
|
"review_key_id": key,
|
"signal_id": "" if is_rolling else manual_row.get("signal_id", key),
|
"rolling_signal_id": manual_row.get("rolling_signal_id", "") if is_rolling else "",
|
"case_id": manual_row.get("case_id", row.get("case_id", "")),
|
"symbol": manual_row.get("symbol", row.get("symbol", "")),
|
"code_suggested_action": manual_row.get("code_suggested_action", ""),
|
"human_decision_action": manual_row.get("human_decision_action", row.get("human_decision_action", "")),
|
"manual_decision_id": manual_row.get("decision_id", ""),
|
"decision_source": manual_row.get("decision_source", ""),
|
"review_input_chart_path": manual_row.get("review_input_chart_path", row.get("chart_path", "")),
|
"chart_path": manual_row.get("review_input_chart_path", row.get("chart_path", "")),
|
"human_decision_reason_cn": manual_row.get("human_decision_reason_cn", row.get("human_decision_reason_cn", "")),
|
"reviewer_notes": manual_row.get("reviewer_notes", ""),
|
"sample_policy": row.get("sample_policy", "30% target / 20% minimum"),
|
}
|
)
|
sampling = pd.DataFrame(rows)
|
write_csv(sampling, "manual_review_sampling_index.csv")
|
return sampling
|
|
|
def markdown_link_targets(text: str) -> list[str]:
|
pattern = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)")
|
return [m.group(1).strip() for m in pattern.finditer(text)]
|
|
|
def audit_links() -> pd.DataFrame:
|
rows = []
|
for md in sorted(ROOT.rglob("*.md")):
|
rel_md = rel_path(md)
|
text = md.read_text(encoding="utf-8", errors="ignore")
|
for target in markdown_link_targets(text):
|
if not target or target.startswith(("http://", "https://", "mailto:")):
|
continue
|
clean = unquote(target.split("#", 1)[0])
|
if not clean:
|
continue
|
if clean.startswith("/"):
|
exists = False
|
resolved = clean
|
else:
|
resolved_path = (md.parent / clean).resolve()
|
try:
|
resolved = resolved_path.relative_to(ROOT.resolve()).as_posix()
|
except ValueError:
|
resolved = str(resolved_path)
|
exists = resolved_path.exists()
|
rows.append(
|
{
|
"markdown_path": rel_md,
|
"target": target,
|
"resolved_path": resolved,
|
"exists": str(bool(exists)),
|
}
|
)
|
df = pd.DataFrame(rows)
|
write_csv(df, "link_evidence_audit.csv")
|
return df
|
|
|
def audit_charts(strict: pd.DataFrame, rolling: pd.DataFrame) -> pd.DataFrame:
|
rows = []
|
for _, row in strict.iterrows():
|
p = ROOT / row.get("chart_path", "")
|
rows.append(
|
{
|
"artifact_type": "STRICT_SELL_SIGNAL",
|
"signal_id": row.get("signal_id", ""),
|
"case_id": row.get("case_id", ""),
|
"chart_path": row.get("chart_path", ""),
|
"exists": str(p.exists()),
|
"manual_decision_id": row.get("manual_decision_id", ""),
|
"decision_source": row.get("decision_source", ""),
|
}
|
)
|
for _, row in rolling.iterrows():
|
p = ROOT / row.get("chart_path", "")
|
rows.append(
|
{
|
"artifact_type": "ROLLING_LOW_BUY_SIGNAL",
|
"signal_id": row.get("rolling_signal_id", ""),
|
"case_id": row.get("case_id", ""),
|
"chart_path": row.get("chart_path", ""),
|
"exists": str(p.exists()),
|
"manual_decision_id": row.get("manual_decision_id", ""),
|
"decision_source": row.get("decision_source", ""),
|
}
|
)
|
df = pd.DataFrame(rows)
|
write_csv(df, "chart_evidence_audit.csv")
|
return df
|
|
|
def write_boards(case_df: pd.DataFrame, strict: pd.DataFrame, rolling: pd.DataFrame) -> None:
|
strict_by_case = {case_id: part.copy() for case_id, part in strict.groupby("case_id")}
|
rolling_by_case = {case_id: part.copy() for case_id, part in rolling.groupby("case_id")}
|
|
root_lines = [
|
"# 无忌 V1 精准卖点与滚动低吸返修图板入口",
|
"",
|
f"- run_id：`{RUN_ID}`",
|
f"- 来源包：`{SOURCE_RUN_ID}`",
|
"- 当前阶段：`V1_STRICT_SELL_ROLLING_INDEPENDENT_MANUAL_SOURCE_REPAIR_SELF_CHECK_DONE_REREVIEW_PENDING`",
|
f"- 人工裁决来源：`{MANUAL_LEDGER}`",
|
"- 说明：脚本只生成候选和证据图；最终 SELL / HOLD / REVIEW_HELD / BUY_ROLLING_LOW 均从独立人工裁决账本回填。",
|
"- 边界：RETURN_STAT_READY=false；执行复审通过前不得引用 V1 业绩结论。",
|
"",
|
"## 案例入口",
|
"",
|
]
|
for _, row in case_df.sort_values("case_id").iterrows():
|
root_lines.append(
|
f"- [{row['case_id']}](cases/{row['case_id']}/case_image_board.md)："
|
f"scope={row.get('v1_return_scope', '')}，闭合 lot={row.get('closed_lot_count', '')}，"
|
f"未闭合/待审 lot={row.get('unresolved_lot_count', '')}，收益贡献={row.get('account_return_closed_lots', '')}"
|
)
|
(ROOT / "case_image_board.md").write_text("\n".join(root_lines) + "\n", encoding="utf-8")
|
|
for _, case_row in case_df.sort_values("case_id").iterrows():
|
case_id = case_row["case_id"]
|
case_dir = ROOT / "cases" / case_id
|
case_dir.mkdir(parents=True, exist_ok=True)
|
sell_part = strict_by_case.get(case_id, pd.DataFrame())
|
rolling_part = rolling_by_case.get(case_id, pd.DataFrame())
|
|
lines = [
|
f"# {case_id} V1 精准卖点与滚动低吸图板（人工裁决来源返修版）",
|
"",
|
f"- 当前收益口径：`{case_row.get('v1_return_scope', '')}`",
|
f"- 主口径标记：`{case_row.get('v1_primary_strict_closed_case_flag', '')}`",
|
f"- 中文边界原因：{case_row.get('v1_return_scope_reason_cn', '')}",
|
f"- 闭合 lot：{case_row.get('closed_lot_count', '')}；未闭合/待审 lot：{case_row.get('unresolved_lot_count', '')}",
|
f"- 人工裁决来源：`../../{MANUAL_LEDGER}`",
|
"- 裁决说明：代码候选只作为证据入口，最终动作取自 manual_decision_ledger.csv；不卖、观察、待审同样是正式裁决。",
|
"",
|
"## 精准卖点 / 趋势裁决",
|
"",
|
]
|
if sell_part.empty:
|
lines.append("- 本 case 无精准卖点 / 趋势候选。")
|
else:
|
for _, sig in sell_part.sort_values(["observation_trade_date", "candidate_time", "signal_id"]).iterrows():
|
chart = Path(sig.get("chart_path", "")).name
|
lines.extend(
|
[
|
f"### {sig.get('signal_type', '')} / {sig.get('human_decision_action', '')}",
|
"",
|
f"- 时间：{sig.get('observation_trade_date', '')} {sig.get('candidate_time', '')}",
|
f"- 代码候选：`{sig.get('code_suggested_action', '')}`",
|
f"- 人工裁决：`{sig.get('human_decision_action', '')}`",
|
f"- 裁决来源：`{sig.get('decision_source', '')}`",
|
f"- manual_decision_id：`{sig.get('manual_decision_id', '')}`",
|
f"- 理由：{sig.get('human_decision_reason_cn', '')}",
|
f"- 图：",
|
"",
|
]
|
)
|
|
lines.extend(["## 滚动低吸裁决", ""])
|
if rolling_part.empty:
|
lines.append("- 本 case 无滚动低吸候选。")
|
else:
|
for _, sig in rolling_part.sort_values(["rolling_trade_date", "rolling_time", "rolling_signal_id"]).iterrows():
|
chart = Path(sig.get("chart_path", "")).name
|
lines.extend(
|
[
|
f"### {sig.get('signal_type', '')} / {sig.get('human_decision_action', '')}",
|
"",
|
f"- 时间：{sig.get('rolling_trade_date', '')} {sig.get('rolling_time', '')}",
|
f"- 代码候选：`{sig.get('code_suggested_action', '')}`",
|
f"- 人工裁决：`{sig.get('human_decision_action', '')}`",
|
f"- 裁决来源：`{sig.get('decision_source', '')}`",
|
f"- manual_decision_id：`{sig.get('manual_decision_id', '')}`",
|
f"- 理由：{sig.get('human_decision_reason_cn', '')}",
|
f"- 图：",
|
"",
|
]
|
)
|
|
lines.extend(
|
[
|
"## 账本追溯",
|
"",
|
"- 根订单账本：`../../strict_order_ledger.csv`",
|
"- 根 lot 账本：`../../strict_position_lot_ledger.csv`",
|
"- 根 case scope：`../../strict_return_scope_case.csv`",
|
"- 根 boundary table：`../../strict_boundary_table.csv`",
|
"- 人工裁决来源账本：`../../manual_decision_ledger.csv`",
|
]
|
)
|
text = "\n".join(lines) + "\n"
|
(case_dir / "case_image_board.md").write_text(text, encoding="utf-8")
|
|
story_lines = [
|
f"# {case_id} V1 交易故事板（人工裁决来源返修版）",
|
"",
|
"本页按人工阅读顺序串联：收益口径 -> 精准卖点 / 趋势裁决 -> 滚动低吸 -> 账本追溯。",
|
"",
|
]
|
story_lines.extend(lines[1:])
|
(case_dir / "case_story_board.md").write_text("\n".join(story_lines) + "\n", encoding="utf-8")
|
|
|
def build_manifest() -> pd.DataFrame:
|
rows = []
|
for path in sorted(ROOT.rglob("*")):
|
if path.is_dir():
|
continue
|
rel = rel_path(path)
|
if rel in {"manifest.csv", "manifest.json"}:
|
continue
|
rows.append({"path": rel, "size": path.stat().st_size, "sha256": sha256_file(path)})
|
df = pd.DataFrame(rows)
|
write_csv(df, "manifest.csv")
|
write_json(ROOT / "manifest.json", {"schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "files": df.to_dict("records")})
|
return df
|
|
|
def write_summary_and_self_check(
|
strict: pd.DataFrame,
|
rolling: pd.DataFrame,
|
manual: pd.DataFrame,
|
orders: pd.DataFrame,
|
lots: pd.DataFrame,
|
case_df: pd.DataFrame,
|
sampling: pd.DataFrame,
|
links: pd.DataFrame,
|
charts: pd.DataFrame,
|
manifest_df: pd.DataFrame,
|
) -> None:
|
action_counts = Counter(manual["human_decision_action"])
|
source_counts = Counter(manual["decision_source"])
|
total_candidates = len(strict) + len(rolling)
|
primary_cases = int((case_df.get("v1_primary_strict_closed_case_flag", "") == "1").sum()) if "v1_primary_strict_closed_case_flag" in case_df else 0
|
positive_cases = int(
|
(
|
(case_df.get("v1_primary_strict_closed_case_flag", "") == "1")
|
& (case_df.get("account_return_closed_lots", "").map(as_float) > 0)
|
).sum()
|
) if "v1_primary_strict_closed_case_flag" in case_df and "account_return_closed_lots" in case_df else 0
|
|
summary = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"run_id": RUN_ID,
|
"generated_at": now_iso(),
|
"stage": "V1_STRICT_SELL_ROLLING_INDEPENDENT_MANUAL_SOURCE_REPAIR_SELF_CHECK_DONE_REREVIEW_PENDING",
|
"source_run_id": SOURCE_RUN_ID,
|
"design_audit_id": DESIGN_AUDIT_ID,
|
"supplemental_design_audit_id": SUPP_DESIGN_AUDIT_ID,
|
"manual_source_supplemental_design_audit_id": MANUAL_SOURCE_SUPP_DESIGN_AUDIT_ID,
|
"execution_held_audit_id": EXEC_HELD_AUDIT_ID,
|
"issue_id": ISSUE_ID,
|
"return_stat_ready": False,
|
"v1_execution_rereview_required": True,
|
"manual_decision_repair": {
|
"candidate_stage_files": ["strict_sell_candidate_ledger.csv", "rolling_low_buy_candidate_ledger.csv"],
|
"manual_decision_external_source_ledger": MANUAL_EXTERNAL_SOURCE_LEDGER,
|
"manual_decision_source_ledger": MANUAL_SOURCE_LEDGER,
|
"manual_decision_ledger": MANUAL_LEDGER,
|
"manual_decision_external_draft": MANUAL_EXTERNAL_DRAFT,
|
"manual_decision_rows": len(manual),
|
"decision_source_counts": dict(source_counts),
|
},
|
"scope": {
|
"source_buy_lots": int((lots.get("lot_source_type", "") == "SOURCE_BUY").sum()),
|
"rolling_buy_lots": int((lots.get("lot_source_type", "") == "ROLLING_LOW_BUY").sum()),
|
"buy_orders_total": int((orders.get("action", "") == "BUY").sum()),
|
"sell_orders": int((orders.get("action", "") == "SELL").sum()),
|
"strict_lot_rows": len(lots),
|
"case_rows": len(case_df),
|
"v1_primary_strict_closed_cases": primary_cases,
|
"v1_positive_primary_cases": positive_cases,
|
"v1_primary_success_readout": positive_cases / primary_cases if primary_cases else None,
|
"v1_primary_account_contribution_readout": float(case_df.get("account_return_closed_lots", pd.Series(dtype=str)).map(as_float).sum()) if not case_df.empty else None,
|
},
|
"manual_decision_action_counts": dict(action_counts),
|
"signal_type_counts": dict(Counter(list(strict.get("signal_type", [])) + list(rolling.get("signal_type", [])))),
|
"artifacts": {
|
"strict_sell_candidate_ledger": "strict_sell_candidate_ledger.csv",
|
"rolling_low_buy_candidate_ledger": "rolling_low_buy_candidate_ledger.csv",
|
"manual_decision_external_source_ledger": MANUAL_EXTERNAL_SOURCE_LEDGER,
|
"manual_decision_source_ledger": MANUAL_SOURCE_LEDGER,
|
"manual_decision_ledger": MANUAL_LEDGER,
|
"strict_sell_signal_ledger": "strict_sell_signal_ledger.csv",
|
"rolling_low_buy_signal_ledger": "rolling_low_buy_signal_ledger.csv",
|
"manual_review_sampling_index": "manual_review_sampling_index.csv",
|
"case_image_board": "case_image_board.md",
|
"manifest": "manifest.json",
|
},
|
"boundaries": [
|
"V1 execution rereview is pending; do not cite V1 success, return, win rate, drawdown, or strategy validity before review passes.",
|
"Final human_decision_* fields are sourced from manual_decision_ledger.csv, not from the stage-1 candidate generator.",
|
"Minute-level market breadth for market-risk sellpoint is unavailable and retained as a boundary.",
|
],
|
}
|
write_json(ROOT / "summary.json", summary)
|
(ROOT / "summary.md").write_text(
|
"\n".join(
|
[
|
"# V1 精准卖点与滚动低吸返修摘要",
|
"",
|
f"- run_id：`{RUN_ID}`",
|
f"- 当前阶段：`{summary['stage']}`",
|
f"- 执行审核 HELD 审计 ID：`{EXEC_HELD_AUDIT_ID}`",
|
f"- 外部手工裁决来源：`{MANUAL_EXTERNAL_SOURCE_LEDGER}`，{len(manual)} 行",
|
f"- 最终消费副本：`{MANUAL_LEDGER}`",
|
f"- 候选总数：{total_candidates}；抽样复核入口：{len(sampling)} 行，占比 {len(sampling) / total_candidates:.2%}",
|
f"- 自检目标：返修后提交执行复审；RETURN_STAT_READY=false",
|
"",
|
"## 结论边界",
|
"",
|
"- 本包仍处于执行复审待审状态，不得引用 V1 收益率、成功率、胜率、回撤或策略有效性结论。",
|
"- 旧 V0 包继续降读为代理卖点证据，不得解释为完整笔记 baseline。",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
|
checks: list[dict[str, str]] = []
|
|
def check(check_id: str, passed: bool, detail: str) -> None:
|
checks.append({"check_id": check_id, "status": "PASS" if passed else "FAIL", "detail": detail})
|
|
required_manual_fields = {
|
"signal_id",
|
"rolling_signal_id",
|
"case_id",
|
"symbol",
|
"code_suggested_action",
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"review_input_chart_path",
|
"reviewer_notes",
|
}
|
check("CONFIG_FILES_EXIST", (ROOT / "strict_sell_rolling_run_config.json").exists(), "strict sell rolling config frozen")
|
check("CANDIDATE_LEDGER_STAGE1_EXISTS", (ROOT / "strict_sell_candidate_ledger.csv").exists() and (ROOT / "rolling_low_buy_candidate_ledger.csv").exists(), "stage-1 candidate ledgers generated")
|
external_source_path = ROOT / MANUAL_EXTERNAL_SOURCE_LEDGER
|
check("EXTERNAL_MANUAL_DECISION_SOURCE_EXISTS", external_source_path.exists(), f"external source rows={len(manual)}")
|
check("EXTERNAL_SOURCE_PREEXISTS_APPLICATION", external_source_path.exists() and datetime.fromtimestamp(external_source_path.stat().st_mtime, TZ) <= APPLICATION_STARTED_AT, f"source_mtime={datetime.fromtimestamp(external_source_path.stat().st_mtime, TZ).isoformat() if external_source_path.exists() else ''}, application_started_at={APPLICATION_STARTED_AT.isoformat()}")
|
check("MANUAL_DECISION_SOURCE_LEDGER_EXISTS", (ROOT / MANUAL_SOURCE_LEDGER).exists(), f"source rows={len(manual)}")
|
check("MANUAL_DECISION_LEDGER_EXISTS", (ROOT / MANUAL_LEDGER).exists(), f"manual rows={len(manual)}")
|
check("MANUAL_DECISION_REQUIRED_FIELDS", required_manual_fields.issubset(set(manual.columns)), f"fields={len(manual.columns)}")
|
check("MANUAL_DECISION_ROW_COUNT_MATCHES_CANDIDATES", len(manual) == total_candidates, f"manual={len(manual)}, candidates={total_candidates}")
|
check("DECISION_SOURCE_NOT_RULE_REPLAY", not (manual["decision_source"] == OLD_RULE_REPLAY_SOURCE).any(), f"sources={dict(source_counts)}")
|
check("NO_SCRIPT_GENERATED_HUMAN_DECISION_FIELDS", not (manual["decision_source"] == "CASE_ANALYSIS_ANALYST_AI_MANUAL_CHART_EVIDENCE_REVIEW_SOURCE").any(), f"sources={dict(source_counts)}")
|
check("MANUAL_DECISION_NOT_VERBATIM_REASON_COPY", not (manual.get("human_decision_reason_cn", "") == manual.get("code_suggested_reason_cn", "")).any(), "human reasons are independent review wording")
|
check("MANUAL_DECISION_TIMES_NOT_SINGLE_STAMP", manual["decision_time"].nunique() > 1, f"decision_time_unique={manual['decision_time'].nunique()}")
|
check("MANUAL_DECISION_ACCEPTANCE_RECORDED", manual["accept_code_suggestion_flag"].str.len().gt(0).all(), "accept/modify code suggestion flag recorded")
|
check("SELL_SIGNALS_CONSUME_MANUAL_LEDGER", strict["manual_decision_id"].ne("").all() and not (strict["decision_source"] == OLD_RULE_REPLAY_SOURCE).any(), f"sell_signals={len(strict)}")
|
check("ROLLING_SIGNALS_CONSUME_MANUAL_LEDGER", rolling["manual_decision_id"].ne("").all() and not (rolling["decision_source"] == OLD_RULE_REPLAY_SOURCE).any(), f"rolling_signals={len(rolling)}")
|
check("HUMAN_DECISION_REASONS_NONEMPTY", manual["human_decision_reason_cn"].str.len().gt(0).all(), "all manual reasons non-empty")
|
check("CHART_REASONS_RENDERED", strict["chart_reason_rendered"].eq("True").all() and rolling["chart_reason_rendered"].eq("True").all(), "all signal charts render reason")
|
check("STORYBOARD_REASONS_RENDERED", strict["storyboard_reason_rendered"].eq("True").all() and rolling["storyboard_reason_rendered"].eq("True").all(), "all story boards can show reason")
|
check("SELL_ORDERS_LINK_TO_SOURCE_SIGNAL", orders[orders["action"] == "SELL"]["source_signal_id"].ne("").all(), f"sell_orders={(orders['action'] == 'SELL').sum()}")
|
rolling_order_mask = (orders["action"] == "BUY") & orders["source_order_id"].str.startswith("ROLL-", na=False)
|
check("ROLLING_BUY_ORDERS_LINK_TO_SOURCE_SIGNAL", orders[rolling_order_mask]["source_signal_id"].ne("").all(), f"rolling_buy_orders={int(rolling_order_mask.sum())}")
|
rolling_lot_mask = lots["lot_source_type"] == "ROLLING_LOW_BUY"
|
check("ROLLING_LOTS_LINK_TO_SOURCE_SIGNAL", lots[rolling_lot_mask]["source_signal_id"].ne("").all(), f"rolling_lots={int(rolling_lot_mask.sum())}")
|
check("NO_V0_V1_MIXED_OUTPUT", orders["run_id"].eq(RUN_ID).all(), "strict orders use V1 run id")
|
check("MARKET_RISK_BOUNDARY_RETAINED", (ROOT / "strict_boundary_table.csv").exists(), "market risk data gap retained")
|
missing_links = 0 if links.empty else int((links["exists"] != "True").sum())
|
check("LINKS_REACHABLE", missing_links == 0, f"links={len(links)}, missing={missing_links}")
|
missing_charts = 0 if charts.empty else int((charts["exists"] != "True").sum())
|
check("CHARTS_EXIST", missing_charts == 0, f"charts={len(charts)}, missing={missing_charts}")
|
sampling_actions = set(sampling["human_decision_action"]) if not sampling.empty else set()
|
required_actions = {"SELL", "HOLD_WATCH", "HOLD_ABOVE_8", "REVIEW_HELD", "BUY_ROLLING_LOW"}
|
check("MANUAL_REVIEW_SAMPLING_RATIO", len(sampling) / total_candidates >= 0.2, f"samples={len(sampling)}, total={total_candidates}")
|
check("MANUAL_REVIEW_SAMPLING_COVERS_ACTIONS", required_actions.issubset(sampling_actions), f"required={sorted(required_actions)}, sampled={sorted(sampling_actions)}")
|
check("MANIFEST_READY", not manifest_df.empty, f"manifest files={len(manifest_df)}")
|
|
checks_df = pd.DataFrame(checks)
|
write_csv(checks_df, "self_check_items.csv")
|
pass_count = int((checks_df["status"] == "PASS").sum())
|
fail_count = int((checks_df["status"] == "FAIL").sum())
|
self_check = {
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"generated_at": now_iso(),
|
"status": "PASS_FOR_V1_EXECUTION_REREVIEW_READY" if fail_count == 0 else "FAIL",
|
"pass_count": pass_count,
|
"fail_count": fail_count,
|
"items_path": "self_check_items.csv",
|
}
|
write_json(ROOT / "self_check.json", self_check)
|
(ROOT / "self_check.md").write_text(
|
f"# 自检\n\n- status：`{self_check['status']}`\n- PASS：{pass_count}\n- FAIL：{fail_count}\n",
|
encoding="utf-8",
|
)
|
|
|
def write_readme() -> None:
|
(ROOT / "README.md").write_text(
|
"\n".join(
|
[
|
"# V1 精准卖点与滚动低吸返修执行包",
|
"",
|
"本包用于修复执行审核 HELD 中指出的“人工裁决来源不独立”问题。",
|
"",
|
"## 阅读顺序",
|
"",
|
"1. `summary.md/json`：看当前阶段、边界和产物入口。",
|
"2. `strict_sell_candidate_ledger.csv`、`rolling_low_buy_candidate_ledger.csv`：看代码候选和证据图。",
|
"3. `manual_decision_external_draft.md`：看分析员手工裁决草稿和批次说明。",
|
"4. `manual_decision_external_source_ledger.csv`：看外部手工裁决源。",
|
"5. `manual_decision_ledger.csv`：看最终包消费的裁决账本副本。",
|
"6. `strict_sell_signal_ledger.csv`、`rolling_low_buy_signal_ledger.csv`：看消费人工裁决后的最终信号账本。",
|
"7. `case_image_board.md`：看人工图片第一入口。",
|
"",
|
"## 边界",
|
"",
|
"- RETURN_STAT_READY=false。",
|
"- 执行复审通过前不得引用 V1 收益率、成功率、胜率、回撤或策略有效性结论。",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
|
|
def main() -> None:
|
strict_original = read_csv("strict_sell_signal_ledger.csv")
|
rolling_original = read_csv("rolling_low_buy_signal_ledger.csv")
|
case_df = read_csv("strict_case_summary.csv")
|
|
strict_candidate, rolling_candidate = build_candidate_ledgers(strict_original, rolling_original)
|
manual = load_manual_decision_source(strict_candidate, rolling_candidate)
|
strict, rolling = apply_manual_decisions(strict_original, rolling_original, strict_candidate, rolling_candidate, manual)
|
orders, lots = repair_orders_and_lots(strict, rolling)
|
sampling = repair_sampling(manual)
|
write_boards(case_df, strict, rolling)
|
links = audit_links()
|
charts = audit_charts(strict, rolling)
|
write_readme()
|
manifest_df = build_manifest()
|
write_summary_and_self_check(strict, rolling, manual, orders, lots, case_df, sampling, links, charts, manifest_df)
|
manifest_df = build_manifest()
|
print(
|
json.dumps(
|
{
|
"run_id": RUN_ID,
|
"manual_decision_rows": len(manual),
|
"strict_signals": len(strict),
|
"rolling_signals": len(rolling),
|
"sampling_rows": len(sampling),
|
"manifest_files": len(manifest_df),
|
"self_check": json.loads((ROOT / "self_check.json").read_text(encoding="utf-8"))["status"],
|
},
|
ensure_ascii=False,
|
)
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
