# -*- coding: utf-8 -*-
|
"""Create only a blank manual-decision review template.
|
|
This script is intentionally not allowed to generate final human_decision_*,
|
decision_time, or decision_source fields. The V1 external manual decision source
|
must be filled by case_analysis.analyst / laoan outside this script, then applied
|
with tools/apply_manual_decision_repair.py.
|
"""
|
|
from __future__ import annotations
|
|
import csv
|
import json
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
WORKBOOK = "manual_decision_review_workbook.csv"
|
TEMPLATE = "manual_decision_external_source_template.csv"
|
|
|
def read_csv(name: str) -> pd.DataFrame:
|
return pd.read_csv(ROOT / name, dtype=str, keep_default_na=False).fillna("")
|
|
|
def code_reason(row: pd.Series) -> str:
|
return row.get("code_evidence_reason_cn", "") or row.get("code_suggested_reason_cn", "") or row.get("human_decision_reason_cn", "")
|
|
|
def build_workbook() -> pd.DataFrame:
|
strict = read_csv("strict_sell_candidate_ledger.csv")
|
rolling = read_csv("rolling_low_buy_candidate_ledger.csv")
|
rows: list[dict[str, str]] = []
|
|
for _, row in strict.iterrows():
|
rows.append(
|
{
|
"artifact_type": "STRICT_SELL_SIGNAL",
|
"signal_id": row.get("signal_id", ""),
|
"rolling_signal_id": "",
|
"case_id": row.get("case_id", ""),
|
"candidate_id": row.get("candidate_id", ""),
|
"symbol": row.get("symbol", ""),
|
"signal_type": row.get("signal_type", ""),
|
"trade_date": row.get("observation_trade_date", ""),
|
"trade_time": row.get("candidate_time", ""),
|
"code_suggested_action": row.get("code_suggested_action", ""),
|
"code_suggested_reason_cn": code_reason(row),
|
"review_input_chart_path": row.get("review_input_chart_path", "") or row.get("chart_path", ""),
|
"numeric_context": json.dumps(
|
{
|
"gain_pct": row.get("gain_pct", ""),
|
"action_price": row.get("action_price", ""),
|
"entry_support_price": row.get("entry_support_price", ""),
|
"trend_first3_time": row.get("trend_first3_time", ""),
|
"fast_breakout_minutes": row.get("fast_breakout_minutes", ""),
|
},
|
ensure_ascii=False,
|
),
|
}
|
)
|
|
for _, row in rolling.iterrows():
|
rows.append(
|
{
|
"artifact_type": "ROLLING_LOW_BUY_SIGNAL",
|
"signal_id": "",
|
"rolling_signal_id": row.get("rolling_signal_id", ""),
|
"case_id": row.get("case_id", ""),
|
"candidate_id": row.get("candidate_id", ""),
|
"symbol": row.get("symbol", ""),
|
"signal_type": row.get("signal_type", ""),
|
"trade_date": row.get("rolling_trade_date", ""),
|
"trade_time": row.get("rolling_time", ""),
|
"code_suggested_action": row.get("code_suggested_action", ""),
|
"code_suggested_reason_cn": code_reason(row),
|
"review_input_chart_path": row.get("review_input_chart_path", "") or row.get("chart_path", ""),
|
"numeric_context": json.dumps(
|
{
|
"rolling_price": row.get("rolling_price", ""),
|
"ma5_close": row.get("ma5_close", ""),
|
"near_ma5_pct": row.get("near_ma5_pct", ""),
|
"volume_ratio_vs_prev20m": row.get("volume_ratio_vs_prev20m", ""),
|
},
|
ensure_ascii=False,
|
),
|
}
|
)
|
|
workbook = pd.DataFrame(rows)
|
workbook.to_csv(ROOT / WORKBOOK, index=False, encoding="utf-8-sig")
|
return workbook
|
|
|
def build_blank_template(workbook: pd.DataFrame) -> pd.DataFrame:
|
fieldnames = [
|
"external_decision_id",
|
"artifact_type",
|
"signal_id",
|
"rolling_signal_id",
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"code_suggested_action",
|
"accept_code_suggestion_flag",
|
"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",
|
"reviewer_notes",
|
]
|
rows = []
|
for seq, (_, row) in enumerate(workbook.iterrows(), start=1):
|
rows.append(
|
{
|
"external_decision_id": f"EXT-MANUAL-RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001-{seq:06d}",
|
"artifact_type": row.get("artifact_type", ""),
|
"signal_id": row.get("signal_id", ""),
|
"rolling_signal_id": row.get("rolling_signal_id", ""),
|
"case_id": row.get("case_id", ""),
|
"candidate_id": row.get("candidate_id", ""),
|
"symbol": row.get("symbol", ""),
|
"code_suggested_action": row.get("code_suggested_action", ""),
|
"accept_code_suggestion_flag": "",
|
"human_decision_action": "",
|
"human_decision_reason_cn": "",
|
"decision_operator": "",
|
"decision_time": "",
|
"decision_source": "",
|
"review_input_chart_path": row.get("review_input_chart_path", ""),
|
"review_input_chart_sha256": "",
|
"manual_draft_path": "",
|
"manual_draft_sha256": "",
|
"reviewer_notes": "",
|
}
|
)
|
template = pd.DataFrame(rows, columns=fieldnames)
|
template.to_csv(ROOT / TEMPLATE, index=False, encoding="utf-8-sig", quoting=csv.QUOTE_MINIMAL)
|
return template
|
|
|
def main() -> None:
|
workbook = build_workbook()
|
template = build_blank_template(workbook)
|
print(
|
json.dumps(
|
{
|
"workbook": WORKBOOK,
|
"template": TEMPLATE,
|
"rows": len(template),
|
"note": "Template only. Fill manual_decision_external_source_ledger.csv manually before running apply_manual_decision_repair.py.",
|
},
|
ensure_ascii=False,
|
)
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
