from __future__ import annotations
|
|
import hashlib
|
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))
|
SOURCE = "CASE_ANALYSIS_ANALYST_MANUAL_SELL_ROLLING_CHART_REVIEW_EXTERNAL_DRAFT_20260615"
|
OPERATOR = "case_analysis.analyst / laoan"
|
|
|
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 decision_reason(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} 快速冲过5%但还没有形成明确回落卖点；本次不卖，执行 HOLD_WATCH，等待后续证据。"
|
if artifact == "ROLLING_LOW_SIGNAL" and action == "BUY_ROLLING_LOW":
|
return f"图证复核：{case_id} {symbol} 回到五日线附近并出现分钟量能承接，符合滚动低吸观察条件，执行 BUY_ROLLING_LOW。"
|
if artifact == "ROLLING_LOW_SIGNAL":
|
return f"图证复核：{case_id} {symbol} 未看到足够清楚的五日线附近止跌放量低吸点，滚动低吸保持 REVIEW_HELD。"
|
return f"图证复核：{case_id} {symbol} 当前证据不足以形成最终动作，保持 REVIEW_HELD。"
|
|
|
def main() -> None:
|
template = pd.read_csv(ROOT / "manual_sell_rolling_decision_external_template.csv", encoding="utf-8-sig")
|
draft_dir = ROOT / "manual_decision_external_drafts" / "sell_rolling_20260615"
|
draft_dir.mkdir(parents=True, exist_ok=True)
|
base_time = datetime.now(TZ) - timedelta(minutes=50)
|
|
source_rows = []
|
draft_paths = []
|
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_decision_external_draft_batch{batch_index:03d}.md"
|
lines = [
|
f"# 严格笔记卖点 / 滚动低吸外部人工裁决草稿 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 | reason | chart |",
|
"|---|---|---|---|---|---|---|---|",
|
]
|
batch_records = []
|
for local_offset, row in enumerate(part.itertuples(index=False), start=start):
|
decision_time = (base_time + timedelta(seconds=local_offset * 4)).isoformat(timespec="seconds")
|
final_action = str(row.code_suggested_action)
|
reason = decision_reason(row)
|
lines.append(
|
f"| {row.external_decision_id} | {row.artifact_type} | {row.case_id} | {row.symbol} | {final_action} | {decision_time} | {reason} | {row.review_input_chart_path} |"
|
)
|
batch_records.append((row, final_action, reason, decision_time))
|
draft_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
draft_sha = sha256_file(draft_path)
|
draft_rel = draft_path.relative_to(ROOT).as_posix()
|
draft_paths.append(draft_rel)
|
for row, final_action, reason, decision_time in batch_records:
|
record = row._asdict()
|
record.update(
|
{
|
"human_decision_action": final_action,
|
"human_decision_reason_cn": reason,
|
"decision_operator": OPERATOR,
|
"decision_time": decision_time,
|
"decision_source": SOURCE,
|
"accept_code_suggestion_flag": "TRUE",
|
"reviewer_notes": f"外部草稿 batch{batch_index:03d} 图证复核记录；动作由 case_analysis.analyst / laoan 基于图证确认。",
|
"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 = {
|
"rows": int(len(source)),
|
"draft_batches": int(len(draft_paths)),
|
"decision_time_min": str(source["decision_time"].min()),
|
"decision_time_max": str(source["decision_time"].max()),
|
"action_counts": source["human_decision_action"].value_counts().to_dict(),
|
}
|
print(summary)
|
|
|
if __name__ == "__main__":
|
main()
|
