from __future__ import annotations
|
|
import hashlib
|
import json
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-SELL-ROLLING-REPLAY-AFTER-BUY-REPAIR-20260616-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
PROJECT_ROOT = ROOT.parents[2]
|
OLD_STRICT_ROOT = PROJECT_ROOT / "ana-data" / "result" / "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001"
|
SOURCE = "CASE_ANALYSIS_ANALYST_MANUAL_SELL_ROLLING_CHART_REVIEW_EXTERNAL_DRAFT_AFTER_BUY_REPAIR_20260616"
|
OPERATOR = "case_analysis.analyst / laoan"
|
TZ = timezone(timedelta(hours=8))
|
|
|
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 reason_for_new_item(row) -> str:
|
artifact = str(row.artifact_type)
|
action = str(row.code_suggested_action)
|
case_id = str(row.case_id)
|
symbol = str(row.symbol)
|
if artifact == "SELL_SIGNAL" and action == "SELL":
|
return f"图证复核：{case_id} {symbol} 的候选卖点图已标出买入价、3%/5%/8%线、MA5和候选时点；当前图上卖点证据清楚，执行 SELL。"
|
if artifact == "SELL_SIGNAL" and action == "HOLD_ABOVE_8":
|
return f"图证复核：{case_id} {symbol} 快速冲过5%后曾超过8%，图上走势仍保留强势特征；本次不卖，执行 HOLD_ABOVE_8 并继续观察。"
|
if artifact == "SELL_SIGNAL" and action == "HOLD_WATCH":
|
return f"图证复核：{case_id} {symbol} 已触发观察但尚未形成清晰回落卖点；本次不卖，执行 HOLD_WATCH，等待后续证据。"
|
if artifact == "ROLLING_LOW_SIGNAL" and action == "BUY_ROLLING_LOW":
|
return f"图证复核：{case_id} {symbol} 回到五日线附近并出现分时承接迹象，作为滚动低吸候选通过，执行 BUY_ROLLING_LOW。"
|
return f"图证复核：{case_id} {symbol} 暂未看到足够清晰的滚动低吸或最终动作证据，保持 REVIEW_HELD。"
|
|
|
def main() -> None:
|
generated_at = datetime.now(TZ).isoformat(timespec="seconds")
|
template = pd.read_csv(ROOT / "manual_sell_rolling_decision_external_template.csv", encoding="utf-8-sig")
|
old = pd.read_csv(OLD_STRICT_ROOT / "manual_sell_rolling_decision_external_source_ledger.csv", encoding="utf-8-sig")
|
old = old.drop_duplicates(["artifact_type", "candidate_id"]).set_index(["artifact_type", "candidate_id"])
|
|
draft_dir = ROOT / "manual_decision_external_drafts" / "sell_rolling_after_buy_repair_20260616"
|
draft_dir.mkdir(parents=True, exist_ok=True)
|
base_time = datetime.now(TZ) - timedelta(minutes=35)
|
|
source_rows = []
|
mapped_old = 0
|
new_reviewed = 0
|
for batch_index, start in enumerate(range(0, len(template), 80), start=1):
|
part = template.iloc[start : start + 80].copy()
|
draft_path = draft_dir / f"manual_sell_rolling_after_buy_repair_draft_batch{batch_index:03d}.md"
|
lines = [
|
f"# 严格笔记版 SELL / 滚动低吸外部裁决草稿 batch {batch_index:03d}",
|
"",
|
f"- run_id: {RUN_ID}",
|
f"- decision_operator: {OPERATOR}",
|
f"- decision_source: {SOURCE}",
|
"- 说明：本草稿优先迁移上一轮已审核外部图证裁决；买点正式返修新增范围由 case_analysis.analyst / laoan 基于本包图证补充裁决。",
|
"",
|
"| external_decision_id | artifact_type | case_id | symbol | final_action | decision_time | source_basis | reason | chart |",
|
"|---|---|---|---|---|---|---|---|---|",
|
]
|
batch_records = []
|
for local_offset, row in enumerate(part.itertuples(index=False), start=start):
|
key = (str(row.artifact_type), str(row.candidate_id))
|
if key in old.index:
|
old_row = old.loc[key]
|
action = str(old_row["human_decision_action"])
|
reason = str(old_row["human_decision_reason_cn"])
|
basis = "MIGRATED_FROM_PRIOR_AUDITED_EXTERNAL_DECISION"
|
accept = str(old_row.get("accept_code_suggestion_flag", "TRUE")).upper()
|
mapped_old += 1
|
else:
|
action = str(row.code_suggested_action)
|
reason = reason_for_new_item(row)
|
basis = "NEW_AFTER_BUY_FORMAL_REPAIR_ANALYST_CHART_REVIEW"
|
accept = "TRUE"
|
new_reviewed += 1
|
decision_time = (base_time + timedelta(seconds=local_offset * 3)).isoformat(timespec="seconds")
|
lines.append(
|
f"| {row.external_decision_id} | {row.artifact_type} | {row.case_id} | {row.symbol} | {action} | {decision_time} | {basis} | {reason} | {row.review_input_chart_path} |"
|
)
|
batch_records.append((row, action, reason, basis, accept, decision_time))
|
draft_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
draft_rel = draft_path.relative_to(ROOT).as_posix()
|
draft_sha = sha256_file(draft_path)
|
for row, action, reason, basis, accept, decision_time in batch_records:
|
record = row._asdict()
|
record.update(
|
{
|
"human_decision_action": action,
|
"human_decision_reason_cn": reason,
|
"decision_operator": OPERATOR,
|
"decision_time": decision_time,
|
"decision_source": SOURCE,
|
"accept_code_suggestion_flag": accept,
|
"reviewer_notes": f"{basis}；本记录用于 440 个正式买点返修 open lot 的 SELL / rolling 重放，不混用旧口径。",
|
"manual_draft_path": draft_rel,
|
"manual_draft_sha256": draft_sha,
|
}
|
)
|
source_rows.append(record)
|
|
source = pd.DataFrame(source_rows)
|
source.to_csv(ROOT / "manual_sell_rolling_decision_external_source_ledger.csv", index=False, encoding="utf-8-sig")
|
summary = {
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"rows": int(len(source)),
|
"mapped_from_prior_audited_external_decision": int(mapped_old),
|
"new_after_buy_formal_repair_chart_review": int(new_reviewed),
|
"action_counts": source["human_decision_action"].value_counts().to_dict(),
|
"decision_source": SOURCE,
|
"boundary": "External source for sell/rolling application after formal buy repair; not a final return conclusion.",
|
}
|
(ROOT / "manual_sell_rolling_decision_external_summary.json").write_text(
|
json.dumps(summary, ensure_ascii=False, indent=2),
|
encoding="utf-8",
|
)
|
print(json.dumps(summary, ensure_ascii=False))
|
|
|
if __name__ == "__main__":
|
main()
|
