from __future__ import annotations
|
|
import hashlib
|
import json
|
import sys
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
import record_p2_resolution as record # noqa: E402
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-SECOND-REVIEW-20260615-001"
|
TZ = timezone(timedelta(hours=8))
|
|
ROOT = Path(__file__).resolve().parents[1]
|
PACKET_ROOT = ROOT / "p2_step_review_packets"
|
|
HARD_PULLBACK_PCTPT = 5.0
|
STRONG_HIGH_PCT = 5.0
|
WEAK_CLOSE_PCT = 2.0
|
WEAK_TAIL_MAX_PCT = 2.0
|
|
|
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, path: Path) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
df.to_csv(path, index=False, encoding="utf-8-sig")
|
|
|
def write_text(path: Path, text: str) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(text, encoding="utf-8-sig")
|
|
|
def num(value: object) -> float | None:
|
try:
|
value = float(value)
|
except Exception:
|
return None
|
if pd.isna(value):
|
return None
|
return value
|
|
|
def fmt(value: object, digits: int = 2) -> str:
|
value = num(value)
|
if value is None:
|
return ""
|
return f"{value:.{digits}f}"
|
|
|
def classify(row: pd.Series) -> tuple[str, str, str]:
|
max_ret = num(row.get("max_ret_pct"))
|
close_ret = num(row.get("close_ret_pct"))
|
tail_max = num(row.get("tail_max_ret_pct"))
|
if max_ret is None or close_ret is None:
|
return "DATA_INSUFFICIENT", "P2_PULLBACK_DATA_INSUFFICIENT", "缺少盘中高点或收盘涨幅，继续待审。"
|
|
pullback = max_ret - close_ret
|
tail_pullback = max_ret - tail_max if tail_max is not None else None
|
|
hard_pullback = pullback >= HARD_PULLBACK_PCTPT
|
weak_tail_fade = (
|
max_ret >= STRONG_HIGH_PCT
|
and close_ret <= WEAK_CLOSE_PCT
|
and tail_max is not None
|
and tail_max <= WEAK_TAIL_MAX_PCT
|
)
|
|
if hard_pullback or weak_tail_fade:
|
reason = (
|
f"按 P2 批量口径：盘中最高 {fmt(max_ret)}%，收盘 {fmt(close_ret)}%，"
|
f"高点回落 {fmt(pullback)} 个百分点"
|
)
|
if tail_pullback is not None:
|
reason += f"，尾段最高较高点回落 {fmt(tail_pullback)} 个百分点"
|
reason += "，属于冲高后大幅回落，标记 REVIEW_HELD。"
|
return "REVIEW_HELD", "P2_BIG_INTRADAY_PULLBACK", reason
|
|
reason = (
|
f"按 P2 批量口径：盘中最高 {fmt(max_ret)}%，收盘 {fmt(close_ret)}%，"
|
f"高点回落 {fmt(pullback)} 个百分点"
|
)
|
if tail_max is not None:
|
reason += f"，14:40 后最高 {fmt(tail_max)}%"
|
reason += "，未触发大幅回落阈值，标记 BUY。"
|
return "BUY", "P2_NO_BIG_INTRADAY_PULLBACK", reason
|
|
|
def build_decisions() -> pd.DataFrame:
|
index = pd.read_csv(PACKET_ROOT / "p2_step_review_index.csv", dtype=str, encoding="utf-8-sig").fillna("")
|
recheck = pd.read_csv(ROOT / "buy_point_second_review_recheck_list.csv", dtype=str, encoding="utf-8-sig").fillna("")
|
recheck = recheck[recheck["human_recheck_priority"].astype(str).str.startswith("P2_", na=False)].copy()
|
|
metric_cols = [
|
"candidate_id",
|
"max_ret_pct",
|
"max_ret_time",
|
"close_ret_pct",
|
"tail_min_ret_pct",
|
"tail_max_ret_pct",
|
"tail_max_ret_time",
|
"above_open_ratio",
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"second_review_issue_code",
|
"second_review_reason_cn",
|
]
|
merged = index.merge(recheck[metric_cols], on="candidate_id", how="left", suffixes=("", "_detail"))
|
rows = []
|
for _, row in merged.iterrows():
|
final_action, rule_code, reason = classify(row)
|
max_ret = num(row.get("max_ret_pct"))
|
close_ret = num(row.get("close_ret_pct"))
|
tail_max = num(row.get("tail_max_ret_pct"))
|
rows.append(
|
{
|
"packet_order": row["order"],
|
"case_id": row["case_id"],
|
"candidate_id": row["candidate_id"],
|
"symbol": row["symbol"],
|
"entry_trade_date": row["entry_trade_date"],
|
"original_action": row["original_action"],
|
"final_action": final_action,
|
"rule_code": rule_code,
|
"rule_reason_cn": reason,
|
"max_ret_pct": row.get("max_ret_pct", ""),
|
"max_ret_time": row.get("max_ret_time", ""),
|
"close_ret_pct": row.get("close_ret_pct", ""),
|
"tail_max_ret_pct": row.get("tail_max_ret_pct", ""),
|
"tail_max_ret_time": row.get("tail_max_ret_time", ""),
|
"tail_min_ret_pct": row.get("tail_min_ret_pct", ""),
|
"pullback_from_day_high_pctpt": "" if max_ret is None or close_ret is None else max_ret - close_ret,
|
"tail_pullback_from_day_high_pctpt": "" if max_ret is None or tail_max is None else max_ret - tail_max,
|
"above_open_ratio": row.get("above_open_ratio", ""),
|
"packet_path": row["packet_path"],
|
}
|
)
|
decisions = pd.DataFrame(rows)
|
decisions["packet_order_num"] = pd.to_numeric(decisions["packet_order"], errors="coerce")
|
decisions = decisions.sort_values("packet_order_num").drop(columns=["packet_order_num"])
|
return decisions
|
|
|
def apply_decisions(decisions: pd.DataFrame) -> None:
|
index = pd.read_csv(PACKET_ROOT / "p2_step_review_index.csv", dtype=str, encoding="utf-8-sig").fillna("")
|
by_order = {int(row["order"]): row for _, row in index.iterrows()}
|
|
for _, decision in decisions.iterrows():
|
order = int(decision["packet_order"])
|
row = by_order[order]
|
final_action = decision["final_action"]
|
reason = decision["rule_reason_cn"]
|
record.update_packet(row, final_action, reason)
|
record.update_ledger(row, final_action, reason)
|
record.update_rollup(row, final_action, reason)
|
|
rollup = pd.read_csv(record.ROLLUP_PATH, dtype=str, encoding="utf-8-sig").fillna("")
|
record.update_p2_summary()
|
record.update_resolution_summary(rollup)
|
record.update_repair_files(rollup)
|
|
|
def write_summary(decisions: pd.DataFrame) -> None:
|
counts = decisions["final_action"].value_counts().sort_index()
|
rule_counts = decisions["rule_code"].value_counts().sort_index()
|
lines = [
|
"# P2 冲高回落批量裁决说明",
|
"",
|
f"- updated_at: {now_iso()}",
|
"- 口径：冲高大幅回落标记 REVIEW_HELD；未触发大幅回落阈值标记 BUY。",
|
f"- 大幅回落阈值 1：盘中最高涨幅 - 收盘涨幅 >= {HARD_PULLBACK_PCTPT:.1f} 个百分点。",
|
f"- 大幅回落阈值 2：盘中最高涨幅 >= {STRONG_HIGH_PCT:.1f}% 且收盘 <= {WEAK_CLOSE_PCT:.1f}% 且 14:40 后最高 <= {WEAK_TAIL_MAX_PCT:.1f}%。",
|
"",
|
"## 最终动作分布",
|
"",
|
"| final_action | count |",
|
"|---|---:|",
|
]
|
for action, count in counts.items():
|
lines.append(f"| `{action}` | {int(count)} |")
|
lines += ["", "## 规则命中分布", "", "| rule_code | count |", "|---|---:|"]
|
for code, count in rule_counts.items():
|
lines.append(f"| `{code}` | {int(count)} |")
|
lines += [
|
"",
|
"## 明细",
|
"",
|
"| # | case | symbol | date | final | max% | close% | pullback | reason |",
|
"|---:|---|---|---|---|---:|---:|---:|---|",
|
]
|
for _, row in decisions.iterrows():
|
lines.append(
|
f"| {row['packet_order']} | {row['case_id']} | {row['symbol']} | {row['entry_trade_date']} | "
|
f"{row['final_action']} | {fmt(row['max_ret_pct'])} | {fmt(row['close_ret_pct'])} | "
|
f"{fmt(row['pullback_from_day_high_pctpt'])} | {row['rule_code']} |"
|
)
|
write_text(PACKET_ROOT / "p2_pullback_rule_summary.md", "\n".join(lines) + "\n")
|
|
|
def update_manifest() -> None:
|
rows = []
|
for path in sorted(ROOT.rglob("*")):
|
if path.is_file() and "__pycache__" not in path.parts:
|
rows.append({"path": path.relative_to(ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path)})
|
write_csv(pd.DataFrame(rows), ROOT / "manifest.csv")
|
(ROOT / "manifest.json").write_text(
|
json.dumps({"run_id": RUN_ID, "updated_at": now_iso(), "file_count": len(rows), "files": rows}, ensure_ascii=False, indent=2),
|
encoding="utf-8",
|
)
|
|
|
def main() -> None:
|
decisions = build_decisions()
|
write_csv(decisions, PACKET_ROOT / "p2_pullback_rule_decision.csv")
|
write_summary(decisions)
|
apply_decisions(decisions)
|
update_manifest()
|
|
|
if __name__ == "__main__":
|
main()
|
