from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
from collections import Counter
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[4]
|
RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-P3-RESOLUTION-20260616-001"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-SECOND-REVIEW-20260615-001"
|
STRICT_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001"
|
OUT_DIR = PROJECT_ROOT / "ana-data" / "result" / RUN_ID
|
SOURCE_DIR = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID
|
TZ = timezone(timedelta(hours=8))
|
|
|
def read_csv(path: Path) -> list[dict[str, str]]:
|
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
return list(csv.DictReader(f))
|
|
|
def write_csv(path: Path, rows: list[dict[str, str]], fieldnames: list[str]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
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 rel(path: Path) -> str:
|
return path.resolve().relative_to(OUT_DIR.resolve()).as_posix()
|
|
|
def write_json(path: Path, data: dict) -> None:
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
def text_is_readable(path: Path) -> tuple[bool, str]:
|
text = path.read_text(encoding="utf-8")
|
bad_tokens = ["???", "\ufffd", "ÀíÓÉ", "å¤", "æ", "Ã", "Â", "涓", "鏃", "鐐", "偂"]
|
hits = {token: text.count(token) for token in bad_tokens if text.count(token)}
|
if hits:
|
return False, json.dumps(hits, ensure_ascii=False)
|
return True, "mojibake_hits=0"
|
|
|
def main() -> int:
|
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
generated_at = datetime.now(TZ).isoformat(timespec="seconds")
|
rollup_path = SOURCE_DIR / "buy_point_resolution_rollup.csv"
|
rows = read_csv(rollup_path)
|
|
p3_rows: list[dict[str, str]] = []
|
resolved_rows: list[dict[str, str]] = []
|
for row in rows:
|
row = dict(row)
|
if row["resolution_source"] == "SECOND_REVIEW_WEAK_DISPUTE":
|
source_action = row["source_action"]
|
row["final_action"] = source_action
|
row["resolution_bucket"] = "P3_CODEX_POLICY_CARRY_SOURCE"
|
row["resolution_source"] = "CODEX_P3_LOW_PRIORITY_CARRY_SOURCE_WITH_WEAK_DISPUTE_NOTE"
|
row["resolution_status"] = "RESOLVED_BY_CODEX_LOW_PRIORITY_POLICY"
|
row["direct_resolution_flag"] = "1"
|
row["codex_direct_resolution_flag"] = "1"
|
row["formal_repair_action"] = "NO_FORMAL_REPAIR_KEEP_SOURCE_ACTION"
|
row["resolution_reason_cn"] = (
|
"P3 弱指标分歧不再要求人工逐条复核；分钟数据没有形成强冲突，"
|
"按低优先级抽查口径沿用源人工裁决，并保留弱分歧说明。"
|
)
|
p3_rows.append(row)
|
resolved_rows.append(row)
|
|
fieldnames = list(rows[0].keys())
|
write_csv(OUT_DIR / "buy_point_resolution_rollup_final.csv", resolved_rows, fieldnames)
|
write_csv(OUT_DIR / "buy_point_p3_resolved_by_codex.csv", p3_rows, fieldnames)
|
|
repair_revoke = [
|
row for row in resolved_rows
|
if row["formal_repair_action"] == "REVOKE_BUY_ORDER_LOT_AND_RECALC"
|
]
|
repair_add = [
|
row for row in resolved_rows
|
if row["formal_repair_action"] == "ADD_BUY_ORDER_LOT_AND_RECALC"
|
]
|
write_csv(OUT_DIR / "buy_point_formal_repair_revoke_required.csv", repair_revoke, fieldnames)
|
write_csv(OUT_DIR / "buy_point_formal_repair_add_required.csv", repair_add, fieldnames)
|
|
status_counts = Counter(row["resolution_status"] for row in resolved_rows)
|
source_counts = Counter(row["resolution_source"] for row in resolved_rows)
|
action_counts = Counter(row["final_action"] for row in resolved_rows)
|
repair_counts = Counter(row["formal_repair_action"] for row in resolved_rows)
|
|
p3_source_counts = Counter(row["source_action"] for row in p3_rows)
|
summary_md = OUT_DIR / "buy_point_p3_resolution_summary.md"
|
summary_md.write_text(
|
"\n".join(
|
[
|
"# 买点 P3 弱分歧处理总结",
|
"",
|
f"- run_id: {RUN_ID}",
|
f"- source_run_id: {SOURCE_RUN_ID}",
|
f"- generated_at: {generated_at}",
|
f"- source_rows: {len(resolved_rows)}",
|
"",
|
"## 处理结论",
|
"",
|
"- P3 弱指标分歧 52 条已由 Codex 按低优先级口径处理完毕。",
|
"- 处理原则：分钟数据没有形成强冲突时，不再要求人工逐条复核，沿用源人工裁决，并保留弱分歧说明。",
|
f"- P3 中源 BUY {p3_source_counts.get('BUY', 0)} 条，源 REVIEW_HELD {p3_source_counts.get('REVIEW_HELD', 0)} 条，均不触发正式增删 BUY。",
|
"- 正式返修清单保持为：撤销 BUY/order/lot 12 条，新增 BUY/order/lot 28 条。",
|
"",
|
"## 最终动作分布",
|
"",
|
"| action | count |",
|
"|---|---:|",
|
*[f"| `{k}` | {v} |" for k, v in sorted(action_counts.items())],
|
"",
|
"## 后续执行要求",
|
"",
|
"- 后续正式返修必须只消费 `buy_point_formal_repair_revoke_required.csv` 和 `buy_point_formal_repair_add_required.csv`。",
|
"- 凡新增或撤销 BUY 的条目，必须同步重算 order、lot、case summary、收益口径、图片入口、自检和 manifest。",
|
"- 本包只关闭 P3 人工复核压力，不生成最终收益率、成功率、胜率、回撤或策略有效性结论。",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
summary_readable, summary_readable_detail = text_is_readable(summary_md)
|
|
self_check_items = [
|
{
|
"item": "SOURCE_ROWS_MATCH_1141",
|
"status": "PASS" if len(resolved_rows) == 1141 else "FAIL",
|
"detail": f"rows={len(resolved_rows)}",
|
},
|
{
|
"item": "P3_ALL_RESOLVED",
|
"status": "PASS" if len(p3_rows) == 52 else "FAIL",
|
"detail": f"p3_resolved={len(p3_rows)}",
|
},
|
{
|
"item": "NO_PENDING_LOW_PRIORITY_RECHECK",
|
"status": "PASS" if action_counts.get("PENDING_LOW_PRIORITY_SAMPLE_RECHECK", 0) == 0 else "FAIL",
|
"detail": f"pending={action_counts.get('PENDING_LOW_PRIORITY_SAMPLE_RECHECK', 0)}",
|
},
|
{
|
"item": "FORMAL_REPAIR_COUNTS_STABLE",
|
"status": "PASS" if len(repair_revoke) == 12 and len(repair_add) == 28 else "FAIL",
|
"detail": f"revoke={len(repair_revoke)}, add={len(repair_add)}",
|
},
|
{
|
"item": "NO_FORMAL_REPAIR_FOR_P3",
|
"status": "PASS" if all(row["formal_repair_action"] == "NO_FORMAL_REPAIR_KEEP_SOURCE_ACTION" for row in p3_rows) else "FAIL",
|
"detail": "P3 rows carry source action only",
|
},
|
{
|
"item": "SUMMARY_TEXT_READABLE",
|
"status": "PASS" if summary_readable else "FAIL",
|
"detail": summary_readable_detail,
|
},
|
]
|
write_csv(OUT_DIR / "self_check_items.csv", self_check_items, ["item", "status", "detail"])
|
self_check = {
|
"run_id": RUN_ID,
|
"source_run_id": SOURCE_RUN_ID,
|
"generated_at": generated_at,
|
"status": "PASS_FOR_BUY_POINT_P3_RESOLUTION_REVIEW_READY"
|
if all(item["status"] == "PASS" for item in self_check_items)
|
else "FAIL",
|
"pass_count": sum(1 for item in self_check_items if item["status"] == "PASS"),
|
"fail_count": sum(1 for item in self_check_items if item["status"] == "FAIL"),
|
}
|
write_json(OUT_DIR / "self_check.json", self_check)
|
|
summary = {
|
"run_id": RUN_ID,
|
"source_run_id": SOURCE_RUN_ID,
|
"strict_run_id": STRICT_RUN_ID,
|
"generated_at": generated_at,
|
"source_rows": len(resolved_rows),
|
"p3_resolved_by_codex": len(p3_rows),
|
"p3_source_action_counts": dict(p3_source_counts),
|
"final_action_counts": dict(action_counts),
|
"resolution_source_counts": dict(source_counts),
|
"formal_repair_counts": dict(repair_counts),
|
"formal_repair_revoke_required": len(repair_revoke),
|
"formal_repair_add_required": len(repair_add),
|
"boundary": [
|
"P3 弱指标分歧按低优先级处理，不再要求人工逐条复核。",
|
"P3 未触发正式增删 BUY，只沿用源人工裁决并保留弱分歧说明。",
|
"正式返修仍只处理 12 条撤销和 28 条新增，后续必须重算 order、lot、case summary 和收益口径。",
|
"本包不是最终收益结论，不构成买入建议，不证明策略有效性。",
|
],
|
}
|
write_json(OUT_DIR / "summary.json", summary)
|
|
source_artifacts = []
|
for source_name in [
|
"buy_point_resolution_rollup.csv",
|
"buy_point_second_review_detail.csv",
|
"buy_point_second_review_human_recheck_report.md",
|
]:
|
src = SOURCE_DIR / source_name
|
source_artifacts.append(
|
{
|
"source_run_id": SOURCE_RUN_ID,
|
"path": src.relative_to(PROJECT_ROOT).as_posix(),
|
"size": str(src.stat().st_size),
|
"sha256": sha256_file(src),
|
}
|
)
|
write_csv(OUT_DIR / "source_artifact_manifest.csv", source_artifacts, ["source_run_id", "path", "size", "sha256"])
|
|
manifest_rows = []
|
for path in sorted(OUT_DIR.rglob("*")):
|
if path.is_file() and path.name not in {"manifest.csv", "manifest.json"}:
|
manifest_rows.append({"path": rel(path), "size": str(path.stat().st_size), "sha256": sha256_file(path)})
|
write_csv(OUT_DIR / "manifest.csv", manifest_rows, ["path", "size", "sha256"])
|
write_json(OUT_DIR / "manifest.json", {"run_id": RUN_ID, "files": manifest_rows})
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
