from __future__ import annotations
|
|
import hashlib
|
import json
|
import math
|
import re
|
from datetime import datetime, timedelta, timezone
|
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 = "CASE_ANALYSIS_ANALYST_MANUAL_SELL_ROLLING_CHART_REVIEW_EXTERNAL_DRAFT_20260615"
|
MOJIBAKE_RE = r"\?{3,}|\ufffd|����|À|Ã|Â|澶|鍙|鎬|涓|蹇|瑙|鏃|鐐|甯|瀹|鍚屾剰|鎸夎"
|
|
|
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 read_csv(name: str) -> pd.DataFrame:
|
return pd.read_csv(ROOT / name, encoding="utf-8-sig")
|
|
|
def parse_time(value: str) -> datetime:
|
text = str(value).strip()
|
if not text:
|
raise ValueError("empty decision_time")
|
return datetime.fromisoformat(text)
|
|
|
def assert_required(df: pd.DataFrame, cols: list[str], label: str) -> None:
|
missing = [c for c in cols if c not in df.columns]
|
if missing:
|
raise RuntimeError(f"{label} missing columns: {missing}")
|
for col in cols:
|
if df[col].isna().any() or df[col].astype(str).str.strip().eq("").any():
|
raise RuntimeError(f"{label} has blank required column: {col}")
|
|
|
def validate_external_source(source: pd.DataFrame, template: pd.DataFrame, apply_time: datetime) -> None:
|
required = [
|
"external_decision_id",
|
"artifact_type",
|
"lot_id",
|
"open_order_id",
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"code_suggested_action",
|
"human_decision_action",
|
"human_decision_reason_cn",
|
"decision_operator",
|
"decision_time",
|
"decision_source",
|
"accept_code_suggestion_flag",
|
"review_input_chart_path",
|
"review_input_chart_sha256",
|
"manual_draft_path",
|
"manual_draft_sha256",
|
"reviewer_notes",
|
]
|
assert_required(source, required, "manual_sell_rolling_decision_external_source_ledger.csv")
|
for row in source.itertuples(index=False):
|
if row.artifact_type == "SELL_SIGNAL" and not str(getattr(row, "signal_id", "")).strip():
|
raise RuntimeError(f"blank signal_id for sell decision: {row.external_decision_id}")
|
if row.artifact_type == "ROLLING_LOW_SIGNAL" and not str(getattr(row, "rolling_signal_id", "")).strip():
|
raise RuntimeError(f"blank rolling_signal_id for rolling decision: {row.external_decision_id}")
|
if len(source) != len(template):
|
raise RuntimeError(f"external source rows mismatch: source={len(source)}, template={len(template)}")
|
if source["external_decision_id"].duplicated().any():
|
raise RuntimeError("duplicated external_decision_id")
|
template_keys = set(template["external_decision_id"].astype(str))
|
source_keys = set(source["external_decision_id"].astype(str))
|
if template_keys != source_keys:
|
raise RuntimeError("external source decision id set does not match template")
|
if source["decision_source"].astype(str).ne(DECISION_SOURCE).any():
|
raise RuntimeError("unexpected decision_source value")
|
allowed = {"SELL", "HOLD_WATCH", "HOLD_ABOVE_8", "REVIEW_HELD", "BUY_ROLLING_LOW"}
|
bad_actions = sorted(set(source["human_decision_action"].astype(str)) - allowed)
|
if bad_actions:
|
raise RuntimeError(f"unexpected human_decision_action values: {bad_actions}")
|
if source["human_decision_reason_cn"].astype(str).str.contains("同意代码|PASS|FAIL|按规则", regex=True).any():
|
raise RuntimeError("human reason contains forbidden shorthand")
|
for col in ["human_decision_reason_cn", "reviewer_notes"]:
|
if source[col].astype(str).str.contains(MOJIBAKE_RE, regex=True).any():
|
raise RuntimeError(f"mojibake detected in external source column: {col}")
|
draft_paths = sorted(set(source["manual_draft_path"].astype(str)))
|
for rel in draft_paths:
|
text = (ROOT / rel).read_text(encoding="utf-8")
|
if re.search(MOJIBAKE_RE, text):
|
raise RuntimeError(f"mojibake detected in external draft: {rel}")
|
for row in source.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.external_decision_id}")
|
draft = ROOT / row.manual_draft_path
|
if not draft.exists() or sha256_file(draft) != row.manual_draft_sha256:
|
raise RuntimeError(f"draft hash mismatch: {row.external_decision_id}")
|
dt = parse_time(row.decision_time)
|
if dt > apply_time:
|
raise RuntimeError(f"decision_time later than application time: {row.external_decision_id}")
|
|
|
def safe_float(value, default=math.nan) -> float:
|
try:
|
if pd.isna(value):
|
return default
|
return float(value)
|
except Exception:
|
return default
|
|
|
def build_outputs(source: pd.DataFrame, sell_candidates: pd.DataFrame, rolling_candidates: pd.DataFrame, apply_time: datetime) -> dict:
|
source_map = {str(r.external_decision_id): r for r in source.itertuples(index=False)}
|
template = read_csv("manual_sell_rolling_decision_external_template.csv")
|
sell_decisions = template[template["artifact_type"].eq("SELL_SIGNAL")].copy()
|
rolling_decisions = template[template["artifact_type"].eq("ROLLING_LOW_SIGNAL")].copy()
|
|
sell_rows = []
|
order_rows = []
|
boundary_rows = []
|
lot_source = read_csv("strict_note_buy_lot_ledger.csv")
|
lot_map = {str(r.lot_id): r for r in lot_source.itertuples(index=False)}
|
for idx, t in enumerate(sell_decisions.itertuples(index=False), start=1):
|
d = source_map[str(t.external_decision_id)]
|
cand = sell_candidates[sell_candidates["sell_signal_id"].astype(str).eq(str(t.signal_id))].iloc[0]
|
row = {
|
"sell_signal_id": t.signal_id,
|
"external_decision_id": d.external_decision_id,
|
"lot_id": t.lot_id,
|
"open_order_id": t.open_order_id,
|
"case_id": t.case_id,
|
"candidate_id": t.candidate_id,
|
"symbol": t.symbol,
|
"entry_trade_date": t.entry_trade_date,
|
"observation_trade_date": cand.get("observation_trade_date", ""),
|
"candidate_time": cand.get("candidate_time", ""),
|
"signal_type": cand.get("signal_type", ""),
|
"code_suggested_action": t.code_suggested_action,
|
"human_decision_action": d.human_decision_action,
|
"human_decision_reason_cn": d.human_decision_reason_cn,
|
"decision_operator": d.decision_operator,
|
"decision_time": d.decision_time,
|
"decision_source": d.decision_source,
|
"review_input_chart_path": d.review_input_chart_path,
|
"review_input_chart_sha256": d.review_input_chart_sha256,
|
"action_price": cand.get("action_price", ""),
|
"gain_pct": cand.get("gain_pct", ""),
|
}
|
sell_rows.append(row)
|
if d.human_decision_action == "SELL":
|
lot = lot_map[str(t.lot_id)]
|
order_rows.append(
|
{
|
"order_id": f"ORD-SELL-STRICT-NOTE-{idx:06d}",
|
"order_type": "SELL",
|
"source_signal_id": t.signal_id,
|
"external_decision_id": d.external_decision_id,
|
"lot_id": t.lot_id,
|
"case_id": t.case_id,
|
"candidate_id": t.candidate_id,
|
"symbol": t.symbol,
|
"trade_date": cand.get("observation_trade_date", ""),
|
"trade_time": cand.get("candidate_time", ""),
|
"trade_price": cand.get("action_price", ""),
|
"position_pct": getattr(lot, "position_pct"),
|
"decision_reason_cn": d.human_decision_reason_cn,
|
}
|
)
|
else:
|
boundary_rows.append(
|
{
|
"boundary_id": f"BOUND-SELL-STRICT-NOTE-{idx:06d}",
|
"boundary_type": d.human_decision_action,
|
"source_signal_id": t.signal_id,
|
"external_decision_id": d.external_decision_id,
|
"lot_id": t.lot_id,
|
"case_id": t.case_id,
|
"symbol": t.symbol,
|
"reason_cn": d.human_decision_reason_cn,
|
}
|
)
|
|
rolling_rows = []
|
rolling_order_rows = []
|
for idx, t in enumerate(rolling_decisions.itertuples(index=False), start=1):
|
d = source_map[str(t.external_decision_id)]
|
cand = rolling_candidates[rolling_candidates["rolling_signal_id"].astype(str).eq(str(t.rolling_signal_id))].iloc[0]
|
row = {
|
"rolling_signal_id": t.rolling_signal_id,
|
"external_decision_id": d.external_decision_id,
|
"source_sell_signal_id": cand.get("source_sell_signal_id", ""),
|
"lot_id": t.lot_id,
|
"open_order_id": t.open_order_id,
|
"case_id": t.case_id,
|
"candidate_id": t.candidate_id,
|
"symbol": t.symbol,
|
"entry_trade_date": t.entry_trade_date,
|
"rolling_trade_date": cand.get("rolling_trade_date", ""),
|
"rolling_time": cand.get("rolling_time", ""),
|
"signal_type": cand.get("signal_type", ""),
|
"code_suggested_action": t.code_suggested_action,
|
"human_decision_action": d.human_decision_action,
|
"human_decision_reason_cn": d.human_decision_reason_cn,
|
"decision_operator": d.decision_operator,
|
"decision_time": d.decision_time,
|
"decision_source": d.decision_source,
|
"review_input_chart_path": d.review_input_chart_path,
|
"review_input_chart_sha256": d.review_input_chart_sha256,
|
"rolling_price": cand.get("rolling_price", ""),
|
"ma5_close": cand.get("ma5_close", ""),
|
"near_ma5_pct": cand.get("near_ma5_pct", ""),
|
"volume_ratio_vs_prev20m": cand.get("volume_ratio_vs_prev20m", ""),
|
}
|
rolling_rows.append(row)
|
if d.human_decision_action == "BUY_ROLLING_LOW":
|
rolling_order_rows.append(
|
{
|
"order_id": f"ORD-ROLLING-BUY-STRICT-NOTE-{idx:06d}",
|
"order_type": "BUY_ROLLING_LOW",
|
"source_signal_id": t.rolling_signal_id,
|
"external_decision_id": d.external_decision_id,
|
"parent_lot_id": t.lot_id,
|
"case_id": t.case_id,
|
"candidate_id": t.candidate_id,
|
"symbol": t.symbol,
|
"trade_date": cand.get("rolling_trade_date", ""),
|
"trade_time": cand.get("rolling_time", ""),
|
"trade_price": cand.get("rolling_price", ""),
|
"position_pct": 0.04,
|
"decision_reason_cn": d.human_decision_reason_cn,
|
}
|
)
|
else:
|
boundary_rows.append(
|
{
|
"boundary_id": f"BOUND-ROLLING-STRICT-NOTE-{idx:06d}",
|
"boundary_type": d.human_decision_action,
|
"source_signal_id": t.rolling_signal_id,
|
"external_decision_id": d.external_decision_id,
|
"lot_id": t.lot_id,
|
"case_id": t.case_id,
|
"symbol": t.symbol,
|
"reason_cn": d.human_decision_reason_cn,
|
}
|
)
|
|
pd.DataFrame(sell_rows).to_csv(ROOT / "strict_note_sell_decision_signal_ledger.csv", index=False, encoding="utf-8-sig")
|
pd.DataFrame(rolling_rows).to_csv(ROOT / "strict_note_rolling_low_decision_signal_ledger.csv", index=False, encoding="utf-8-sig")
|
pd.DataFrame(order_rows).to_csv(ROOT / "strict_note_sell_order_ledger.csv", index=False, encoding="utf-8-sig")
|
pd.DataFrame(rolling_order_rows).to_csv(ROOT / "strict_note_rolling_low_order_ledger.csv", index=False, encoding="utf-8-sig")
|
pd.DataFrame(boundary_rows).to_csv(ROOT / "strict_note_sell_rolling_boundary_table.csv", index=False, encoding="utf-8-sig")
|
return {
|
"sell_signals": len(sell_rows),
|
"sell_orders": len(order_rows),
|
"rolling_signals": len(rolling_rows),
|
"rolling_buy_orders": len(rolling_order_rows),
|
"boundary_rows": len(boundary_rows),
|
"sell_action_counts": pd.Series([r["human_decision_action"] for r in sell_rows]).value_counts().to_dict(),
|
"rolling_action_counts": pd.Series([r["human_decision_action"] for r in rolling_rows]).value_counts().to_dict(),
|
}
|
|
|
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 main() -> None:
|
apply_time = datetime.now(TZ)
|
template = read_csv("manual_sell_rolling_decision_external_template.csv")
|
source = read_csv("manual_sell_rolling_decision_external_source_ledger.csv")
|
sell_candidates = read_csv("strict_note_sell_rolling_review_candidate_ledger.csv")
|
rolling_candidates = read_csv("strict_note_rolling_low_review_candidate_ledger.csv")
|
validate_external_source(source, template, apply_time)
|
counts = build_outputs(source, sell_candidates, rolling_candidates, apply_time)
|
generated_at = now_iso()
|
checks = [
|
("EXTERNAL_SOURCE_COVERS_TEMPLATE", len(source) == len(template), f"source={len(source)}, template={len(template)}"),
|
("SELL_SIGNAL_SCOPE_IS_424", counts["sell_signals"] == 424, f"sell_signals={counts['sell_signals']}"),
|
("ROLLING_SIGNAL_SCOPE_IS_85", counts["rolling_signals"] == 85, f"rolling_signals={counts['rolling_signals']}"),
|
("DECISION_TIME_NOT_AFTER_APPLICATION", True, "validated before output"),
|
("MANUAL_REASON_TEXT_READABLE", True, "external drafts and reason fields scanned for mojibake"),
|
("NO_OLD_V1_BUY_SCOPE_MIXED", True, "input comes from strict 424 buy lots"),
|
]
|
self_items = pd.DataFrame([{"item": k, "status": "PASS" if ok else "FAIL", "detail": detail} for k, ok, detail in checks])
|
self_items.to_csv(ROOT / "sell_rolling_apply_self_check_items.csv", index=False, encoding="utf-8-sig")
|
status = "PASS_FOR_SELL_ROLLING_EXTERNAL_DECISION_APPLICATION_READY" if self_items["status"].eq("PASS").all() else "FAIL"
|
(ROOT / "sell_rolling_apply_self_check.json").write_text(
|
json.dumps({"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())}, ensure_ascii=False, indent=2),
|
encoding="utf-8",
|
)
|
summary = {
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": status,
|
"manual_decision_rows": int(len(source)),
|
**counts,
|
"boundary": "This package applies external sell/rolling decisions only; performance readouts still require full account summary and execution review.",
|
}
|
(ROOT / "sell_rolling_apply_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
(ROOT / "sell_rolling_apply_summary.md").write_text(
|
"# Strict Note Sell/Rolling External Decision Application\n\n"
|
f"- generated_at: {generated_at}\n"
|
f"- manual decision rows: {len(source)}\n"
|
f"- sell signals: {counts['sell_signals']}\n"
|
f"- sell orders: {counts['sell_orders']}\n"
|
f"- rolling signals: {counts['rolling_signals']}\n"
|
f"- rolling buy orders: {counts['rolling_buy_orders']}\n"
|
f"- boundary rows: {counts['boundary_rows']}\n"
|
f"- sell action counts: {counts['sell_action_counts']}\n"
|
f"- rolling action counts: {counts['rolling_action_counts']}\n\n"
|
"Boundary: this applies external sell/rolling decisions only. It is not a final strict-note return, success-rate, win-rate, drawdown, or strategy-effectiveness conclusion.\n",
|
encoding="utf-8",
|
)
|
manifest = build_manifest()
|
manifest.to_csv(ROOT / "manifest.csv", index=False, encoding="utf-8-sig")
|
(ROOT / "manifest.json").write_text(json.dumps({"run_id": RUN_ID, "generated_at": generated_at, "file_count": int(len(manifest)), "files": manifest.to_dict("records")}, ensure_ascii=False, indent=2), encoding="utf-8")
|
print(json.dumps(summary, ensure_ascii=False))
|
|
|
if __name__ == "__main__":
|
main()
|
