from __future__ import annotations
|
|
import hashlib
|
import json
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-SECOND-REVIEW-20260615-001"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001"
|
|
ROOT = Path(__file__).resolve().parents[1]
|
PROJECT_ROOT = ROOT.parents[2]
|
SOURCE_ROOT = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID
|
NOTE_PATH = PROJECT_ROOT / "ana-doc" / "wuji" / "profile" / "source_note" / "笔记精简版.md"
|
|
MINUTE_BASE = Path(r"E:\quant\2023_front_m")
|
DAILY_DIR = Path(r"E:\quant\a_share_daily_front_20230101_20260508_complete\daily")
|
TZ = timezone(timedelta(hours=8))
|
_MINUTE_FILE_CACHE: dict[str, tuple[pd.DataFrame, str]] = {}
|
_DAILY_FILE_CACHE: dict[str, tuple[pd.DataFrame, str]] = {}
|
|
|
def now_iso() -> str:
|
return datetime.now(TZ).isoformat(timespec="seconds")
|
|
|
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 write_csv(df: pd.DataFrame, name: str) -> Path:
|
path = ROOT / name
|
path.parent.mkdir(parents=True, exist_ok=True)
|
df.to_csv(path, index=False, encoding="utf-8-sig")
|
return path
|
|
|
def write_json(obj: dict, name: str) -> Path:
|
path = ROOT / name
|
path.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
|
return path
|
|
|
def minute_path(symbol: str) -> Path:
|
code, exchange = symbol.split(".")
|
return MINUTE_BASE / exchange / f"price_{code}.csv"
|
|
|
def load_entry_day_minute(symbol: str, trade_date: str) -> tuple[pd.DataFrame, str]:
|
if symbol not in _MINUTE_FILE_CACHE:
|
path = minute_path(symbol)
|
if not path.exists():
|
_MINUTE_FILE_CACHE[symbol] = (pd.DataFrame(), "LOCAL_MINUTE_FILE_MISSING")
|
else:
|
try:
|
df_all = pd.read_csv(path, dtype={"timetag": "string"})
|
df_all["trade_date_key"] = df_all["timetag"].str.slice(0, 8)
|
df_all["trade_time"] = df_all["timetag"].str.slice(9, 17)
|
for col in ["open", "high", "low", "close", "volumn", "amount"]:
|
df_all[col] = pd.to_numeric(df_all[col], errors="coerce")
|
df_all = df_all.dropna(subset=["open", "high", "low", "close"])
|
_MINUTE_FILE_CACHE[symbol] = (df_all, "LOCAL_MINUTE_FILE_LOADED")
|
except Exception as exc: # pragma: no cover - audit evidence path
|
_MINUTE_FILE_CACHE[symbol] = (pd.DataFrame(), f"LOCAL_MINUTE_READ_ERROR:{type(exc).__name__}")
|
cached, cache_status = _MINUTE_FILE_CACHE[symbol]
|
if cached.empty:
|
return pd.DataFrame(), cache_status
|
date_key = trade_date.replace("-", "")
|
df = cached[cached["trade_date_key"].eq(date_key)].copy()
|
if df.empty:
|
return pd.DataFrame(), "LOCAL_ENTRY_DAY_MINUTE_ROWS_MISSING"
|
df = df.sort_values("trade_time").reset_index(drop=True)
|
if df.empty:
|
return pd.DataFrame(), "LOCAL_ENTRY_DAY_MINUTE_ROWS_INVALID"
|
return df, "LOCAL_MINUTE_OK"
|
|
|
def load_signal_ma5(symbol: str, signal_date: str) -> tuple[float | None, str]:
|
if symbol not in _DAILY_FILE_CACHE:
|
path = DAILY_DIR / f"{symbol}.csv"
|
if not path.exists():
|
_DAILY_FILE_CACHE[symbol] = (pd.DataFrame(), "LOCAL_DAILY_FILE_MISSING")
|
else:
|
df_all = pd.read_csv(path, dtype={"trade_date": "string"})
|
if df_all.empty or "close" not in df_all.columns:
|
_DAILY_FILE_CACHE[symbol] = (pd.DataFrame(), "LOCAL_DAILY_INVALID")
|
else:
|
df_all["close"] = pd.to_numeric(df_all["close"], errors="coerce")
|
df_all = df_all.dropna(subset=["close"]).sort_values("trade_date")
|
_DAILY_FILE_CACHE[symbol] = (df_all, "LOCAL_DAILY_FILE_LOADED")
|
df, cache_status = _DAILY_FILE_CACHE[symbol]
|
if df.empty:
|
return None, cache_status
|
df = df[df["trade_date"] <= signal_date.replace("-", "")]
|
if len(df) < 5:
|
return None, "LOCAL_DAILY_MA5_INSUFFICIENT"
|
return float(df.tail(5)["close"].mean()), "LOCAL_DAILY_MA5_OK"
|
|
|
def pct(value: float, base: float) -> float:
|
return (value / base - 1.0) * 100.0 if base else 0.0
|
|
|
def time_mask(df: pd.DataFrame, start: str | None = None, end: str | None = None) -> pd.Series:
|
mask = pd.Series(True, index=df.index)
|
if start is not None:
|
mask &= df["trade_time"] >= start
|
if end is not None:
|
mask &= df["trade_time"] <= end
|
return mask
|
|
|
def max_ret_time(df: pd.DataFrame, open_ref: float) -> tuple[float | None, str]:
|
if df.empty:
|
return None, ""
|
idx = df["high"].idxmax()
|
row = df.loc[idx]
|
return pct(float(row["high"]), open_ref), str(row["trade_time"])
|
|
|
def min_ret_time(df: pd.DataFrame, open_ref: float) -> tuple[float | None, str]:
|
if df.empty:
|
return None, ""
|
idx = df["low"].idxmin()
|
row = df.loc[idx]
|
return pct(float(row["low"]), open_ref), str(row["trade_time"])
|
|
|
def streak_after_time(df: pd.DataFrame, condition: pd.Series, start: str) -> int:
|
part = df[df["trade_time"] >= start].copy()
|
if part.empty:
|
return 0
|
cond = condition.loc[part.index].tolist()
|
best = 0
|
cur = 0
|
for ok in cond:
|
if bool(ok):
|
cur += 1
|
best = max(best, cur)
|
else:
|
cur = 0
|
return int(best)
|
|
|
def review_metrics(row: pd.Series) -> dict:
|
symbol = row["symbol"]
|
entry_date = row["entry_trade_date"]
|
signal_date = row["signal_trade_date"]
|
minute, minute_status = load_entry_day_minute(symbol, entry_date)
|
ma5, ma5_status = load_signal_ma5(symbol, signal_date)
|
out: dict[str, object] = {
|
"local_minute_status": minute_status,
|
"local_daily_ma5_status": ma5_status,
|
"minute_rows": 0,
|
"ma5": ma5 if ma5 is not None else "",
|
}
|
if minute.empty:
|
return out
|
|
open_ref = float(minute.iloc[0]["open"])
|
first = minute.iloc[0]
|
early = minute[time_mask(minute, end="10:40:00")]
|
mid = minute[time_mask(minute, start="10:41:00", end="14:39:00")]
|
tail = minute[time_mask(minute, start="14:40:00")]
|
|
max_all, max_all_time = max_ret_time(minute, open_ref)
|
min_all, min_all_time = min_ret_time(minute, open_ref)
|
early_max, early_max_time = max_ret_time(early, open_ref)
|
early_min, early_min_time = min_ret_time(early, open_ref)
|
mid_max, mid_max_time = max_ret_time(mid, open_ref)
|
tail_max, tail_max_time = max_ret_time(tail, open_ref)
|
tail_min, tail_min_time = min_ret_time(tail, open_ref)
|
|
close_ret = pct(float(minute.iloc[-1]["close"]), open_ref)
|
first_high_ret = pct(float(first["high"]), open_ref)
|
first_low_ret = pct(float(first["low"]), open_ref)
|
first_close_ret = pct(float(first["close"]), open_ref)
|
above_open = minute["close"] >= open_ref
|
above_open_ratio = float(above_open.mean())
|
early_above_open_ratio = float((early["close"] >= open_ref).mean()) if not early.empty else 0.0
|
tail_above_open_ratio = float((tail["close"] >= open_ref).mean()) if not tail.empty else 0.0
|
above_3 = minute["close"] >= open_ref * 1.03
|
above_5 = minute["close"] >= open_ref * 1.05
|
tail_above_3_ratio = float((tail["close"] >= open_ref * 1.03).mean()) if not tail.empty else 0.0
|
early_above_3_ratio = float((early["close"] >= open_ref * 1.03).mean()) if not early.empty else 0.0
|
above_ma5_ratio = ""
|
early_above_ma5_ratio = ""
|
tail_above_ma5_ratio = ""
|
if ma5 is not None:
|
above_ma5_ratio = float((minute["close"] >= ma5).mean())
|
early_above_ma5_ratio = float((early["close"] >= ma5).mean()) if not early.empty else 0.0
|
tail_above_ma5_ratio = float((tail["close"] >= ma5).mean()) if not tail.empty else 0.0
|
|
strong_streak_3 = streak_after_time(minute, above_3, "09:30:00")
|
strong_streak_5 = streak_after_time(minute, above_5, "09:30:00")
|
mid_only_high = bool(
|
(mid_max is not None and mid_max >= 3.0)
|
and (early_max is None or early_max < 2.0)
|
and (tail_max is None or tail_max < 2.0)
|
)
|
|
out.update(
|
{
|
"minute_rows": int(len(minute)),
|
"open_ref": open_ref,
|
"first_high_ret_pct": first_high_ret,
|
"first_low_ret_pct": first_low_ret,
|
"first_close_ret_pct": first_close_ret,
|
"max_ret_pct": max_all,
|
"max_ret_time": max_all_time,
|
"min_ret_pct": min_all,
|
"min_ret_time": min_all_time,
|
"close_ret_pct": close_ret,
|
"above_open_ratio": above_open_ratio,
|
"early_above_open_ratio": early_above_open_ratio,
|
"tail_above_open_ratio": tail_above_open_ratio,
|
"early_max_ret_pct": early_max,
|
"early_max_ret_time": early_max_time,
|
"early_min_ret_pct": early_min,
|
"early_min_ret_time": early_min_time,
|
"mid_max_ret_pct": mid_max,
|
"mid_max_ret_time": mid_max_time,
|
"tail_max_ret_pct": tail_max,
|
"tail_max_ret_time": tail_max_time,
|
"tail_min_ret_pct": tail_min,
|
"tail_min_ret_time": tail_min_time,
|
"early_above_3_ratio": early_above_3_ratio,
|
"tail_above_3_ratio": tail_above_3_ratio,
|
"strong_streak_3_min": strong_streak_3,
|
"strong_streak_5_min": strong_streak_5,
|
"above_ma5_ratio": above_ma5_ratio,
|
"early_above_ma5_ratio": early_above_ma5_ratio,
|
"tail_above_ma5_ratio": tail_above_ma5_ratio,
|
"mid_only_high_flag": mid_only_high,
|
}
|
)
|
return out
|
|
|
def classify(row: pd.Series) -> tuple[str, str, str]:
|
action = str(row["human_decision_action"])
|
reason = str(row.get("human_decision_reason_cn", ""))
|
if row["local_minute_status"] != "LOCAL_MINUTE_OK":
|
return ("NO_DATA_RECHECK", "LOCAL_MINUTE_NOT_AVAILABLE", "本机分钟文件缺失，未形成独立数据异议。")
|
|
above = float(row["above_open_ratio"])
|
close_ret = float(row["close_ret_pct"])
|
max_ret = float(row["max_ret_pct"])
|
min_ret = float(row["min_ret_pct"])
|
early_max = float(row["early_max_ret_pct"])
|
tail_max = float(row["tail_max_ret_pct"])
|
early_above_open = float(row["early_above_open_ratio"])
|
tail_above_open = float(row["tail_above_open_ratio"])
|
early_above_3 = float(row["early_above_3_ratio"])
|
tail_above_3 = float(row["tail_above_3_ratio"])
|
first_low = float(row["first_low_ret_pct"])
|
first_high = float(row["first_high_ret_pct"])
|
mid_only_high = bool(row["mid_only_high_flag"])
|
|
strong_buy_shape = (
|
above >= 0.72
|
and close_ret >= 1.0
|
and max_ret >= 3.0
|
and min_ret >= -3.0
|
and (early_max >= 3.0 or tail_max >= 3.0 or early_above_3 >= 0.20 or tail_above_3 >= 0.40)
|
)
|
support_buy_shape = (
|
above >= 0.60
|
and close_ret >= 0.3
|
and max_ret >= 2.2
|
and min_ret >= -4.0
|
and not mid_only_high
|
)
|
weak_shape = (
|
(above < 0.40 and close_ret < 0.5)
|
or close_ret <= -1.2
|
or max_ret < 1.5
|
or (first_high >= 2.0 and close_ret < 0.0 and above < 0.50)
|
or (first_low <= -1.0 and early_above_open < 0.35 and tail_above_open < 0.50)
|
)
|
time_window_ambiguous = (
|
action == "BUY"
|
and ("午后" in reason)
|
and ("尾盘" not in reason)
|
and tail_max < 3.0
|
and early_max < 3.0
|
)
|
|
if action == "BUY":
|
if weak_shape:
|
return (
|
"STRONG_RECHECK",
|
"BUY_BUT_DATA_WEAK_OR_FALSE_SUPPORT",
|
"人工裁为 BUY，但分钟复算显示弱势、冲高回落或承接不足。",
|
)
|
if mid_only_high or time_window_ambiguous:
|
return (
|
"WEAK_RECHECK",
|
"BUY_TIME_WINDOW_OR_MIDDAY_STRENGTH_AMBIGUOUS",
|
"人工裁为 BUY，但强点主要不在 10:40 前/14:40 后，或理由只写午后而无精确买点。",
|
)
|
if not support_buy_shape:
|
return (
|
"WEAK_RECHECK",
|
"BUY_SUPPORT_SHAPE_NOT_STRONG_BY_METRICS",
|
"人工裁为 BUY，但量化指标只能支持弱承接，需要看原图确认。",
|
)
|
return ("AGREE", "BUY_SHAPE_ACCEPTABLE_BY_METRICS", "数据指标与 BUY 裁决基本一致。")
|
|
if action == "REVIEW_HELD":
|
if strong_buy_shape:
|
return (
|
"STRONG_RECHECK",
|
"HELD_BUT_DATA_STRONG_BUY_SHAPE",
|
"人工裁为 HELD，但分钟复算显示强势承接，建议重看图。",
|
)
|
if support_buy_shape:
|
return (
|
"WEAK_RECHECK",
|
"HELD_BUT_DATA_SUPPORT_BUY_SHAPE",
|
"人工裁为 HELD，但数据有一定买点形态，建议抽查。",
|
)
|
return ("AGREE", "HELD_SHAPE_WEAK_BY_METRICS", "数据指标与 HELD 裁决基本一致。")
|
|
return ("NO_DATA_RECHECK", "UNKNOWN_MANUAL_ACTION", "人工动作字段不是 BUY/REVIEW_HELD。")
|
|
|
def build_manifest() -> pd.DataFrame:
|
rows = []
|
for path in sorted(ROOT.rglob("*")):
|
if path.is_file() and path.name not in {"manifest.csv", "manifest.json"}:
|
rows.append(
|
{
|
"path": path.relative_to(ROOT).as_posix(),
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
return pd.DataFrame(rows)
|
|
|
def fmt_num(value: object, digits: int = 2) -> str:
|
try:
|
if value == "":
|
return ""
|
return f"{float(value):.{digits}f}"
|
except Exception:
|
return str(value)
|
|
|
def priority(row: pd.Series) -> tuple[int, str]:
|
issue = row["second_review_issue_code"]
|
if issue == "BUY_BUT_DATA_WEAK_OR_FALSE_SUPPORT":
|
return 1, "P1_BUY_DECISION_MAY_BE_WRONG"
|
if issue == "HELD_BUT_DATA_STRONG_BUY_SHAPE":
|
return 2, "P2_HELD_DECISION_WORTH_REVIEW"
|
if issue == "BUY_TIME_WINDOW_OR_MIDDAY_STRENGTH_AMBIGUOUS":
|
return 2, "P2_BUY_TIME_WINDOW_AMBIGUOUS"
|
if issue in {"BUY_SUPPORT_SHAPE_NOT_STRONG_BY_METRICS", "HELD_BUT_DATA_SUPPORT_BUY_SHAPE"}:
|
return 3, "P3_WEAK_METRIC_DISAGREEMENT"
|
return 9, "P9_NOT_PRIORITIZED"
|
|
|
def main() -> None:
|
ROOT.mkdir(parents=True, exist_ok=True)
|
manual = pd.read_csv(SOURCE_ROOT / "manual_buy_decision_external_source_ledger.csv", encoding="utf-8-sig")
|
candidates = pd.read_csv(SOURCE_ROOT / "strict_note_buy_point_review_candidate_ledger.csv", encoding="utf-8-sig")
|
keep_cols = [
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"market_group",
|
"signal_trade_date",
|
"entry_trade_date",
|
"candidate_rank",
|
"up_count",
|
"volume_ratio",
|
"pullback_from_latest_limitup_close_pct",
|
"upper_shadow_pct",
|
"market_gate_status",
|
]
|
merged = manual.merge(
|
candidates[[c for c in keep_cols if c in candidates.columns]],
|
on=["case_id", "candidate_id", "symbol", "signal_trade_date", "entry_trade_date"],
|
how="left",
|
suffixes=("", "_candidate"),
|
)
|
|
metric_rows = []
|
for _, row in merged.iterrows():
|
metric_rows.append(review_metrics(row))
|
metrics = pd.DataFrame(metric_rows)
|
detail = pd.concat([merged.reset_index(drop=True), metrics.reset_index(drop=True)], axis=1)
|
classified = detail.apply(classify, axis=1, result_type="expand")
|
detail["second_review_status"] = classified[0]
|
detail["second_review_issue_code"] = classified[1]
|
detail["second_review_reason_cn"] = classified[2]
|
priorities = detail.apply(priority, axis=1, result_type="expand")
|
detail["human_recheck_priority_num"] = priorities[0]
|
detail["human_recheck_priority"] = priorities[1]
|
detail["source_review_chart_abs_path"] = detail["review_input_chart_path"].map(
|
lambda p: str((SOURCE_ROOT / str(p)).resolve()) if isinstance(p, str) and p else ""
|
)
|
|
disputes = detail[detail["second_review_status"].isin(["STRONG_RECHECK", "WEAK_RECHECK"])].copy()
|
disputes = disputes.sort_values(
|
["human_recheck_priority_num", "entry_trade_date", "case_id", "candidate_id"],
|
ascending=[True, True, True, True],
|
)
|
strong = detail[detail["second_review_status"].eq("STRONG_RECHECK")].copy()
|
strong = strong.sort_values(
|
["human_recheck_priority_num", "entry_trade_date", "case_id", "candidate_id"],
|
ascending=[True, True, True, True],
|
)
|
|
cols_front = [
|
"second_review_status",
|
"second_review_issue_code",
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"entry_trade_date",
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"second_review_reason_cn",
|
"local_minute_status",
|
"minute_rows",
|
"first_high_ret_pct",
|
"first_low_ret_pct",
|
"max_ret_pct",
|
"max_ret_time",
|
"min_ret_pct",
|
"min_ret_time",
|
"close_ret_pct",
|
"above_open_ratio",
|
"early_max_ret_pct",
|
"early_max_ret_time",
|
"tail_max_ret_pct",
|
"tail_max_ret_time",
|
"mid_only_high_flag",
|
"source_review_chart_abs_path",
|
]
|
ordered = [c for c in cols_front if c in detail.columns] + [c for c in detail.columns if c not in cols_front]
|
|
write_csv(detail[ordered], "buy_point_second_review_detail.csv")
|
write_csv(disputes[ordered], "buy_point_second_review_recheck_list.csv")
|
write_csv(strong[ordered], "buy_point_second_review_strong_recheck.csv")
|
write_csv(disputes[ordered].head(80), "buy_point_second_review_human_top80.csv")
|
|
status_counts = detail["second_review_status"].value_counts(dropna=False).to_dict()
|
issue_counts = detail["second_review_issue_code"].value_counts(dropna=False).to_dict()
|
coverage_counts = detail["local_minute_status"].value_counts(dropna=False).to_dict()
|
action_status_counts = (
|
detail.groupby(["human_decision_action", "second_review_status"]).size().reset_index(name="count")
|
)
|
write_csv(action_status_counts, "buy_point_second_review_action_status_counts.csv")
|
|
generated_at = now_iso()
|
summary = {
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"source_run_id": SOURCE_RUN_ID,
|
"source_manual_decision_rows": int(len(manual)),
|
"detail_rows": int(len(detail)),
|
"status_counts": {str(k): int(v) for k, v in status_counts.items()},
|
"issue_counts": {str(k): int(v) for k, v in issue_counts.items()},
|
"local_minute_status_counts": {str(k): int(v) for k, v in coverage_counts.items()},
|
"boundary": [
|
"This is a second-pass data screen, not a replacement for human chart review.",
|
"Local MySQL was not used; local CSV coverage is incomplete versus the source package chart evidence.",
|
"The source BUY ledger does not contain exact buy minute; BUY timing disputes are therefore flagged for human recheck.",
|
],
|
"source_paths": {
|
"note": str(NOTE_PATH),
|
"source_package": str(SOURCE_ROOT),
|
"manual_decision_ledger": str(SOURCE_ROOT / "manual_buy_decision_external_source_ledger.csv"),
|
"candidate_ledger": str(SOURCE_ROOT / "strict_note_buy_point_review_candidate_ledger.csv"),
|
"minute_csv_base": str(MINUTE_BASE),
|
"daily_csv_dir": str(DAILY_DIR),
|
},
|
}
|
write_json(summary, "buy_point_second_review_summary.json")
|
(ROOT / "buy_point_second_review_summary.md").write_text(
|
"# 买点人工裁决二次数据复核\n\n"
|
f"- run_id: {RUN_ID}\n"
|
f"- generated_at: {generated_at}\n"
|
f"- source_run_id: {SOURCE_RUN_ID}\n"
|
f"- source manual decision rows: {len(manual)}\n"
|
f"- detail rows: {len(detail)}\n"
|
f"- local minute status counts: {coverage_counts}\n"
|
f"- second review status counts: {status_counts}\n"
|
f"- issue counts: {issue_counts}\n\n"
|
"## 边界\n\n"
|
"- 本复核是二次数据筛查，不替代最终人工看图裁决。\n"
|
"- 当前没有使用 MySQL；本机分钟 CSV 对源包 1141 条图证覆盖不完整。\n"
|
"- 源 BUY 账本没有精确买入分钟，涉及买入时间窗口的异议只能列为待人工重看。\n\n"
|
"## 输出\n\n"
|
"- `buy_point_second_review_detail.csv`: 全量二审明细。\n"
|
"- `buy_point_second_review_recheck_list.csv`: 强/弱异议清单。\n"
|
"- `buy_point_second_review_strong_recheck.csv`: 强异议清单。\n",
|
encoding="utf-8-sig",
|
)
|
|
p1 = disputes[disputes["human_recheck_priority"].eq("P1_BUY_DECISION_MAY_BE_WRONG")].copy()
|
p2 = disputes[disputes["human_recheck_priority"].str.startswith("P2_", na=False)].copy()
|
report_lines = [
|
"# 买点人工裁决二次复核给人工看的异议清单",
|
"",
|
f"- run_id: {RUN_ID}",
|
f"- generated_at: {generated_at}",
|
f"- source_run_id: {SOURCE_RUN_ID}",
|
f"- 本轮用本机分钟 CSV 独立复算覆盖：{int(detail['local_minute_status'].eq('LOCAL_MINUTE_OK').sum())} / {len(detail)}",
|
f"- P1：人工裁 BUY 但数据/图证偏弱，优先复核：{len(p1)}",
|
f"- P2：人工裁 HELD 但数据偏强，或 BUY 时间窗口不清，建议复核：{len(p2)}",
|
f"- P3：弱指标分歧，低优先级抽查：{len(disputes) - len(p1) - len(p2)}",
|
"",
|
"## P1 优先复核",
|
"",
|
"| case | candidate | symbol | date | close% | above_open | early_max% | tail_max% | 人工理由 | 图证 |",
|
"|---|---|---|---|---:|---:|---:|---:|---|---|",
|
]
|
for _, r in p1.iterrows():
|
report_lines.append(
|
"| "
|
+ " | ".join(
|
[
|
str(r["case_id"]),
|
str(r["candidate_id"]),
|
str(r["symbol"]),
|
str(r["entry_trade_date"]),
|
fmt_num(r["close_ret_pct"]),
|
fmt_num(r["above_open_ratio"]),
|
fmt_num(r["early_max_ret_pct"]),
|
fmt_num(r["tail_max_ret_pct"]),
|
str(r["human_decision_reason_cn"]).replace("|", "/"),
|
str(r["source_review_chart_abs_path"]).replace("|", "/"),
|
]
|
)
|
+ " |"
|
)
|
report_lines.extend(
|
[
|
"",
|
"## P2 复核说明",
|
"",
|
"- `HELD_BUT_DATA_STRONG_BUY_SHAPE`：数据上看有较强承接，但人工可能因为尾盘回落、波动大、买点不清而保守 HELD；这类不是直接判错，是建议重看图。",
|
"- `BUY_TIME_WINDOW_OR_MIDDAY_STRENGTH_AMBIGUOUS`：BUY 理由主要依赖午后/中段强度，但笔记的新仓买点窗口强调 10:40 前或 14:40 后，需人工确认是否合规。",
|
"",
|
"完整清单见 `buy_point_second_review_recheck_list.csv`；强异议见 `buy_point_second_review_strong_recheck.csv`。",
|
"",
|
"## 重要边界",
|
"",
|
"- 源 BUY 账本没有精确买入分钟，本报告不能替代最终成交点复核。",
|
"- 本机分钟 CSV 覆盖不完整；未覆盖项不列为数据异议。",
|
]
|
)
|
(ROOT / "buy_point_second_review_human_recheck_report.md").write_text(
|
"\n".join(report_lines) + "\n",
|
encoding="utf-8-sig",
|
)
|
|
manifest = build_manifest()
|
write_csv(manifest, "manifest.csv")
|
write_json(
|
{
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"file_count": int(len(manifest)),
|
"files": manifest.to_dict(orient="records"),
|
},
|
"manifest.json",
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
