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-STRICT-NOTE-FULL-RERUN-20260614-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
TZ = timezone(timedelta(hours=8))
|
DECISION_SOURCE_PREFIX = "CASE_ANALYSIS_ANALYST_MANUAL_BUY_POINT_CHART_REVIEW_EXTERNAL_DRAFT_BATCH"
|
VALID_ACTIONS = {"BUY", "REVIEW_HELD"}
|
|
|
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
|
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 require_nonblank(df: pd.DataFrame, columns: list[str]) -> None:
|
for col in columns:
|
if col not in df.columns:
|
raise RuntimeError(f"missing required column: {col}")
|
if df[col].isna().any() or df[col].astype(str).str.strip().eq("").any():
|
raise RuntimeError(f"blank required column: {col}")
|
|
|
def validate_source(decisions: pd.DataFrame, template: pd.DataFrame) -> None:
|
required = [
|
"external_decision_id",
|
"candidate_id",
|
"case_id",
|
"symbol",
|
"signal_trade_date",
|
"entry_trade_date",
|
"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",
|
]
|
require_nonblank(decisions, required)
|
if len(decisions) != len(template):
|
raise RuntimeError(f"manual decision row count mismatch: {len(decisions)} != {len(template)}")
|
if decisions["external_decision_id"].duplicated().any():
|
raise RuntimeError("duplicate external_decision_id")
|
if set(decisions["external_decision_id"]) != set(template["external_decision_id"]):
|
raise RuntimeError("manual decision external_decision_id set mismatch")
|
if not decisions["human_decision_action"].isin(VALID_ACTIONS).all():
|
raise RuntimeError("invalid manual action")
|
if not decisions["decision_source"].str.startswith(DECISION_SOURCE_PREFIX).all():
|
raise RuntimeError("unexpected decision_source")
|
decision_times = pd.to_datetime(decisions["decision_time"], utc=True, errors="raise")
|
now_utc = pd.Timestamp(datetime.now(TZ)).tz_convert("UTC")
|
if decision_times.gt(now_utc).any():
|
future_count = int(decision_times.gt(now_utc).sum())
|
raise RuntimeError(f"manual decision_time is in the future: {future_count}")
|
for row in decisions.itertuples(index=False):
|
chart = ROOT / row.review_input_chart_path
|
if not chart.exists() or sha256_file(chart) != row.review_input_chart_sha256:
|
raise RuntimeError(f"chart hash mismatch: {row.candidate_id}")
|
draft = ROOT / row.manual_draft_path
|
if not draft.exists() or sha256_file(draft) != row.manual_draft_sha256:
|
raise RuntimeError(f"manual draft hash mismatch: {row.external_decision_id}")
|
|
|
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 make_boundary_rows(held_rows: pd.DataFrame, scope: pd.DataFrame) -> list[dict]:
|
rows: list[dict] = []
|
for row in held_rows.itertuples(index=False):
|
rows.append(
|
{
|
"case_id": row.case_id,
|
"candidate_id": row.candidate_id,
|
"symbol": row.symbol,
|
"boundary_type": "BUY_POINT_REVIEW_HELD",
|
"boundary_reason": row.human_decision_reason_cn,
|
"external_decision_id": row.external_decision_id,
|
}
|
)
|
closed_scope = scope[scope["buy_point_replay_status"].eq("NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY")]
|
for row in closed_scope.itertuples(index=False):
|
rows.append(
|
{
|
"case_id": row.case_id,
|
"candidate_id": row.candidate_id,
|
"symbol": row.symbol,
|
"boundary_type": "NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY",
|
"boundary_reason": "market gate closed; no new BUY generated",
|
"external_decision_id": "",
|
}
|
)
|
gap_scope = scope[scope["buy_point_replay_status"].eq("BUY_POINT_MINUTE_DATA_GAP_HELD")]
|
for row in gap_scope.itertuples(index=False):
|
rows.append(
|
{
|
"case_id": row.case_id,
|
"candidate_id": row.candidate_id,
|
"symbol": row.symbol,
|
"boundary_type": "BUY_POINT_MINUTE_DATA_GAP_HELD",
|
"boundary_reason": "entry-day minute data gap; no manual BUY generated",
|
"external_decision_id": "",
|
}
|
)
|
return rows
|
|
|
def main() -> None:
|
decisions = pd.read_csv(ROOT / "manual_buy_decision_external_source_ledger.csv", encoding="utf-8-sig")
|
template = pd.read_csv(ROOT / "manual_buy_decision_external_template.csv", encoding="utf-8-sig")
|
candidates = pd.read_csv(ROOT / "strict_note_buy_point_review_candidate_ledger.csv", encoding="utf-8-sig")
|
scope = pd.read_csv(ROOT / "strict_note_buy_replay_scope.csv", encoding="utf-8-sig")
|
validate_source(decisions, template)
|
|
joined = candidates.merge(
|
decisions,
|
on=["case_id", "candidate_id", "symbol", "signal_trade_date", "entry_trade_date"],
|
how="left",
|
suffixes=("", "_decision"),
|
)
|
if joined["human_decision_action"].isna().any():
|
raise RuntimeError("candidate without manual decision")
|
|
buy_rows = joined[joined["human_decision_action"].eq("BUY")].copy()
|
held_rows = joined[joined["human_decision_action"].eq("REVIEW_HELD")].copy()
|
|
order_rows = []
|
lot_rows = []
|
for i, row in enumerate(buy_rows.itertuples(index=False), start=1):
|
order_id = f"ORD-{RUN_ID}-{i:05d}"
|
lot_id = f"LOT-{RUN_ID}-{i:05d}"
|
order_rows.append(
|
{
|
"order_id": order_id,
|
"case_id": row.case_id,
|
"candidate_id": row.candidate_id,
|
"external_decision_id": row.external_decision_id,
|
"symbol": row.symbol,
|
"trade_date": row.entry_trade_date,
|
"trade_time": "MANUAL_BUY_POINT_REVIEW_CONFIRMED",
|
"action": "BUY",
|
"price_source": "ENTRY_DAY_MANUAL_REVIEW_CLOSE_PROXY_PENDING_MINUTE_POINT",
|
"price": row.close_price,
|
"position_delta_pct": 0.04,
|
"tranche_index": 1,
|
"order_status": "STRICT_BUY_CONFIRMED_BY_EXTERNAL_MANUAL_DECISION",
|
"decision_source": row.decision_source,
|
"decision_reason_cn": row.human_decision_reason_cn,
|
"evidence_image_path": row.review_input_chart_path,
|
}
|
)
|
lot_rows.append(
|
{
|
"lot_id": lot_id,
|
"open_order_id": order_id,
|
"case_id": row.case_id,
|
"candidate_id": row.candidate_id,
|
"external_decision_id": row.external_decision_id,
|
"symbol": row.symbol,
|
"entry_trade_date": row.entry_trade_date,
|
"entry_price": row.close_price,
|
"position_pct": 0.04,
|
"lot_status": "OPEN_PENDING_STRICT_SELL_REPLAY",
|
"decision_source": row.decision_source,
|
}
|
)
|
|
orders = pd.DataFrame(order_rows)
|
lots = pd.DataFrame(lot_rows)
|
boundary = pd.DataFrame(make_boundary_rows(held_rows, scope))
|
write_csv(orders, "strict_note_buy_order_ledger.csv")
|
write_csv(lots, "strict_note_buy_lot_ledger.csv")
|
write_csv(boundary, "strict_note_buy_boundary_table.csv")
|
|
case_summary = (
|
scope.groupby("case_id")
|
.agg(selected_candidates=("candidate_id", "count"))
|
.reset_index()
|
.merge(orders.groupby("case_id").size().rename("buy_orders").reset_index(), on="case_id", how="left")
|
.merge(boundary.groupby("case_id").size().rename("boundary_rows").reset_index(), on="case_id", how="left")
|
)
|
case_summary["buy_orders"] = case_summary["buy_orders"].fillna(0).astype(int)
|
case_summary["boundary_rows"] = case_summary["boundary_rows"].fillna(0).astype(int)
|
case_summary["case_status"] = case_summary["buy_orders"].gt(0).map(
|
{True: "STRICT_BUY_CONFIRMED_PENDING_SELL_REPLAY", False: "NO_BUY_OR_BOUNDARY_ONLY"}
|
)
|
write_csv(case_summary, "strict_note_buy_case_summary.csv")
|
|
closed_scope = scope[scope["buy_point_replay_status"].eq("NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY")]
|
gap_scope = scope[scope["buy_point_replay_status"].eq("BUY_POINT_MINUTE_DATA_GAP_HELD")]
|
generated_at = datetime.now(TZ).isoformat(timespec="seconds")
|
items = [
|
("MANUAL_SOURCE_ROWS_MATCH_TEMPLATE", len(decisions) == len(template), f"decisions={len(decisions)}, template={len(template)}"),
|
("ALL_READY_CANDIDATES_DECIDED", len(joined) == len(decisions), f"ready={len(joined)}"),
|
("MARKET_CLOSED_NO_BUY", set(closed_scope["candidate_id"]).isdisjoint(set(orders["candidate_id"])) if not orders.empty else True, f"market_closed={len(closed_scope)}"),
|
("DATA_GAP_NO_BUY", set(gap_scope["candidate_id"]).isdisjoint(set(orders["candidate_id"])) if not orders.empty else True, f"data_gap={len(gap_scope)}"),
|
("BUY_ORDER_HAS_EXTERNAL_DECISION", orders["external_decision_id"].astype(str).str.len().gt(0).all() if not orders.empty else True, f"buy_orders={len(orders)}"),
|
("LOTS_MATCH_BUY_ORDERS", len(lots) == len(orders), f"lots={len(lots)}, orders={len(orders)}"),
|
("BOUNDARY_ROWS_COMPLETE", len(boundary) == len(held_rows) + len(closed_scope) + len(gap_scope), f"boundary={len(boundary)}"),
|
]
|
self_items = pd.DataFrame([{"item": k, "status": "PASS" if ok else "FAIL", "detail": detail} for k, ok, detail in items])
|
write_csv(self_items, "buy_apply_self_check_items.csv")
|
status = "PASS_FOR_STRICT_BUY_DECISION_APPLICATION_REVIEW_READY" if self_items["status"].eq("PASS").all() else "FAIL"
|
write_json(
|
{
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": status,
|
"pass_count": int(self_items["status"].eq("PASS").sum()),
|
"fail_count": int(self_items["status"].eq("FAIL").sum()),
|
},
|
"buy_apply_self_check.json",
|
)
|
write_json(
|
{
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": status,
|
"counts": {
|
"manual_decisions": int(len(decisions)),
|
"buy_orders": int(len(orders)),
|
"open_lots_pending_sell_replay": int(len(lots)),
|
"buy_review_held": int(len(held_rows)),
|
"market_gate_closed_boundary": int(len(closed_scope)),
|
"minute_data_gap_boundary": int(len(gap_scope)),
|
"boundary_rows_total": int(len(boundary)),
|
},
|
"boundary": "This package applies external BUY decisions only; sell/trend/rolling replay and returns are not generated yet.",
|
},
|
"buy_apply_summary.json",
|
)
|
(ROOT / "buy_apply_summary.md").write_text(
|
"# Strict Note BUY Decision Application Summary\n\n"
|
f"- generated_at: {generated_at}\n"
|
f"- manual decisions: {len(decisions)}\n"
|
f"- BUY orders: {len(orders)}\n"
|
f"- open lots pending sell replay: {len(lots)}\n"
|
f"- review-held among ready candidates: {len(held_rows)}\n"
|
f"- market-gate-closed boundary: {len(closed_scope)}\n"
|
f"- minute-data-gap boundary: {len(gap_scope)}\n"
|
f"- boundary rows total: {len(boundary)}\n\n"
|
"Boundary: this package applies external BUY decisions only. It does not generate SELL orders, returns, win rate, drawdown, or strategy-effectiveness conclusions.\n",
|
encoding="utf-8",
|
)
|
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()
|
