from __future__ import annotations
|
|
import hashlib
|
import json
|
import os
|
import re
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
import pandas as pd
|
import pymysql
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md")
|
TZ = timezone(timedelta(hours=8))
|
|
|
def now_iso() -> str:
|
return datetime.now(TZ).isoformat(timespec="seconds")
|
|
|
def read_password() -> str:
|
env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD")
|
if env:
|
return env
|
text = LOCAL_DB_INDEX.read_text(encoding="utf-8")
|
match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE)
|
if not match:
|
raise RuntimeError("Unable to read local MySQL credential from approved local index.")
|
return match.group(1)
|
|
|
def get_conn():
|
return pymysql.connect(
|
host="127.0.0.1",
|
port=3306,
|
user="root",
|
password=read_password(),
|
database="tianxia",
|
charset="utf8mb4",
|
connect_timeout=5,
|
read_timeout=240,
|
)
|
|
|
def font(size: int):
|
for p in [
|
Path("C:/Windows/Fonts/msyh.ttc"),
|
Path("C:/Windows/Fonts/simhei.ttf"),
|
Path("C:/Windows/Fonts/simsun.ttc"),
|
]:
|
if p.exists():
|
return ImageFont.truetype(str(p), size)
|
return ImageFont.load_default()
|
|
|
FONT_TITLE = font(24)
|
FONT_MID = font(16)
|
FONT_SMALL = font(13)
|
|
|
def normalize_time(value) -> str:
|
text = str(value)
|
if "days" in text:
|
text = text.split()[-1]
|
if "." in text:
|
text = text.split(".")[0]
|
parts = text.split(":")
|
if len(parts) >= 3:
|
return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}"
|
return text
|
|
|
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
|
path.parent.mkdir(parents=True, exist_ok=True)
|
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 y_price(value: float, low: float, high: float, top: int, bottom: int) -> int:
|
if high <= low:
|
return (top + bottom) // 2
|
return bottom - int((value - low) / (high - low) * (bottom - top))
|
|
|
def draw_intraday_review_chart(cand: pd.Series, minute: pd.DataFrame, ma5: float | None, out_path: Path) -> None:
|
w, h = 1500, 860
|
img = Image.new("RGB", (w, h), "#fbfbf7")
|
d = ImageDraw.Draw(img)
|
d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
|
title = f"严格版买点人工复核图：{cand.symbol} {cand.entry_trade_date.strftime('%Y-%m-%d')}"
|
d.text((30, 22), title, fill="#111827", font=FONT_TITLE)
|
d.text((30, 56), f"case: {cand.case_id} candidate: {cand.candidate_id}", fill="#334155", font=FONT_SMALL)
|
|
plot_left, plot_top, plot_right, plot_bottom = 80, 105, 1080, 575
|
vol_top, vol_bottom = 625, 785
|
note_left, note_top = 1110, 110
|
d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8")
|
d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8")
|
|
m = minute.sort_values("trade_time").reset_index(drop=True)
|
open_ref = float(m.open_price.iloc[0])
|
refs = [open_ref, open_ref * 1.03, open_ref * 1.05, open_ref * 1.08]
|
if ma5 and ma5 > 0:
|
refs.append(float(ma5))
|
price_low = min(float(m.low_price.min()), min(refs)) * 0.998
|
price_high = max(float(m.high_price.max()), max(refs)) * 1.002
|
max_vol = max(float(m.volume.max()), 1.0)
|
n = len(m)
|
gap = (plot_right - plot_left) / max(n - 1, 1)
|
|
pts = []
|
for i, row in m.iterrows():
|
x = int(plot_left + gap * i)
|
y = y_price(float(row.close_price), price_low, price_high, plot_top, plot_bottom)
|
pts.append((x, y))
|
vh = int(float(row.volume) / max_vol * (vol_bottom - vol_top))
|
color = "#dc2626" if float(row.close_price) >= float(row.open_price) else "#16a34a"
|
d.line([x, vol_bottom, x, vol_bottom - vh], fill=color, width=2)
|
if i % max(1, n // 7) == 0:
|
d.text((x - 20, vol_bottom + 8), str(row.trade_time)[:5], fill="#64748b", font=FONT_SMALL)
|
if len(pts) > 1:
|
d.line(pts, fill="#2563eb", width=2)
|
|
for label, value, color in [
|
("开盘价", open_ref, "#0f172a"),
|
("+3%", open_ref * 1.03, "#f59e0b"),
|
("+5%", open_ref * 1.05, "#dc2626"),
|
("+8%", open_ref * 1.08, "#7c3aed"),
|
]:
|
yy = y_price(value, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, yy, plot_right, yy], fill=color, width=2)
|
d.text((plot_right + 8, yy - 8), f"{label} {value:.2f}", fill=color, font=FONT_SMALL)
|
if ma5 and ma5 > 0:
|
yy = y_price(float(ma5), price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, yy, plot_right, yy], fill="#059669", width=2)
|
d.text((plot_right + 8, yy - 8), f"日MA5 {float(ma5):.2f}", fill="#059669", font=FONT_SMALL)
|
|
d.rounded_rectangle([note_left, note_top, 1465, 785], radius=8, outline="#334155", fill="#ffffff")
|
notes = [
|
"人工裁决待填",
|
"本图只打包证据，不自动买入。",
|
"需人工/AI人工判断：",
|
"1. 底部承接是否强",
|
"2. 买点是否成立",
|
"3. 是否追高或假承接",
|
"",
|
"冻结硬筛证据：",
|
f"倍量：{float(cand.volume_ratio):.2f}",
|
f"回调：{float(cand.pullback_from_latest_limitup_close_pct):.2f}%",
|
f"长上影：{float(cand.upper_shadow_pct):.2f}%",
|
f"闸门：{cand.market_gate_status}",
|
"",
|
"最终动作不得由脚本代填。",
|
]
|
yy = note_top + 16
|
for i, line in enumerate(notes):
|
d.text((note_left + 16, yy), line, fill="#111827" if i == 0 else "#334155", font=FONT_TITLE if i == 0 else FONT_SMALL)
|
yy += 32 if i == 0 else (24 if line else 12)
|
d.text((30, 820), "decision_input_view：用于外部人工/AI人工裁决，不能反推成脚本自动 BUY。", fill="#334155", font=FONT_MID)
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
img.save(out_path)
|
|
|
def load_minute_and_ma5(eligible: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
|
if eligible.empty:
|
return pd.DataFrame(), pd.DataFrame()
|
with get_conn() as conn:
|
minute_parts = []
|
for trade_date, group in eligible.groupby(eligible.entry_trade_date.dt.strftime("%Y-%m-%d")):
|
symbols = sorted(group.symbol.unique().tolist())
|
ph = ",".join(["%s"] * len(symbols))
|
minute_parts.append(
|
pd.read_sql(
|
f"""
|
SELECT trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume
|
FROM a_share_minute_price
|
WHERE trade_date=%s AND symbol IN ({ph})
|
ORDER BY symbol, trade_date, trade_time
|
""",
|
conn,
|
params=[trade_date, *symbols],
|
)
|
)
|
minute = pd.concat(minute_parts, ignore_index=True) if minute_parts else pd.DataFrame()
|
symbols = sorted(eligible.symbol.unique().tolist())
|
ph = ",".join(["%s"] * len(symbols))
|
min_date = (eligible.signal_trade_date.min() - pd.Timedelta(days=20)).strftime("%Y-%m-%d")
|
max_date = eligible.signal_trade_date.max().strftime("%Y-%m-%d")
|
daily = pd.read_sql(
|
f"""
|
SELECT trade_date, symbol, close_price
|
FROM a_share_daily_price
|
WHERE symbol IN ({ph}) AND trade_date BETWEEN %s AND %s
|
ORDER BY symbol, trade_date
|
""",
|
conn,
|
params=[*symbols, min_date, max_date],
|
)
|
if not minute.empty:
|
minute["trade_date"] = pd.to_datetime(minute["trade_date"])
|
minute["trade_time"] = minute["trade_time"].map(normalize_time)
|
for col in ["open_price", "high_price", "low_price", "close_price", "volume"]:
|
minute[col] = pd.to_numeric(minute[col], errors="coerce")
|
if not daily.empty:
|
daily["trade_date"] = pd.to_datetime(daily["trade_date"])
|
daily["close_price"] = pd.to_numeric(daily["close_price"], errors="coerce")
|
daily["ma5"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(5, min_periods=5).mean())
|
return minute, daily
|
|
|
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:
|
selected = pd.read_csv(ROOT / "strict_note_selected_candidate_ledger.csv", encoding="utf-8-sig")
|
selected["entry_trade_date"] = pd.to_datetime(selected["entry_trade_date"])
|
selected["signal_trade_date"] = pd.to_datetime(selected["signal_trade_date"])
|
selected["eligible_for_buy_replay_flag"] = selected["market_gate_open_flag"].astype(str).str.lower().isin(["true", "1"])
|
selected["buy_replay_boundary_status"] = selected["eligible_for_buy_replay_flag"].map(
|
{True: "ELIGIBLE_FOR_BUY_POINT_MANUAL_REVIEW", False: "NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY"}
|
)
|
eligible = selected[selected["eligible_for_buy_replay_flag"]].copy()
|
boundary = selected[~selected["eligible_for_buy_replay_flag"]].copy()
|
|
minute, daily = load_minute_and_ma5(eligible)
|
chart_rows = []
|
missing_rows = []
|
template_rows = []
|
data_gap_candidate_ids = set()
|
for _, cand in eligible.iterrows():
|
m = minute[(minute["symbol"] == cand.symbol) & (minute["trade_date"] == cand.entry_trade_date)].copy()
|
ma_part = daily[(daily["symbol"] == cand.symbol) & (daily["trade_date"] <= cand.signal_trade_date)].tail(1)
|
ma5 = None if ma_part.empty or pd.isna(ma_part.iloc[0].ma5) else float(ma_part.iloc[0].ma5)
|
chart_path = Path("charts") / "buy_point_review" / cand.case_id / f"buy_point_review_{cand.candidate_id}.png"
|
if m.empty:
|
data_gap_candidate_ids.add(cand.candidate_id)
|
missing_rows.append(
|
{
|
"case_id": cand.case_id,
|
"candidate_id": cand.candidate_id,
|
"symbol": cand.symbol,
|
"entry_trade_date": cand.entry_trade_date.strftime("%Y-%m-%d"),
|
"missing_reason": "ENTRY_DAY_MINUTE_DATA_MISSING",
|
}
|
)
|
chart_rel = ""
|
chart_sha = ""
|
else:
|
draw_intraday_review_chart(cand, m, ma5, ROOT / chart_path)
|
chart_rel = chart_path.as_posix()
|
chart_sha = sha256_file(ROOT / chart_path)
|
chart_rows.append(
|
{
|
"case_id": cand.case_id,
|
"candidate_id": cand.candidate_id,
|
"symbol": cand.symbol,
|
"entry_trade_date": cand.entry_trade_date.strftime("%Y-%m-%d"),
|
"chart_role": "strict_note_buy_point_manual_review_input",
|
"path": chart_rel,
|
"sha256": chart_sha,
|
"status": "PASS",
|
}
|
)
|
if chart_rel:
|
template_rows.append(
|
{
|
"external_decision_id": f"EXT-BUY-{cand.candidate_id}",
|
"case_id": cand.case_id,
|
"candidate_id": cand.candidate_id,
|
"symbol": cand.symbol,
|
"signal_trade_date": cand.signal_trade_date.strftime("%Y-%m-%d"),
|
"entry_trade_date": cand.entry_trade_date.strftime("%Y-%m-%d"),
|
"code_suggested_action": "BUY_POINT_REVIEW_REQUIRED",
|
"human_decision_action": "",
|
"human_decision_reason_cn": "",
|
"decision_operator": "",
|
"decision_time": "",
|
"decision_source": "",
|
"review_input_chart_path": chart_rel,
|
"review_input_chart_sha256": chart_sha,
|
"accept_code_suggestion_flag": "",
|
"reviewer_notes": "",
|
}
|
)
|
|
selected["buy_point_replay_status"] = selected.apply(
|
lambda r: "NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY"
|
if not bool(r["eligible_for_buy_replay_flag"])
|
else (
|
"BUY_POINT_MINUTE_DATA_GAP_HELD"
|
if r["candidate_id"] in data_gap_candidate_ids
|
else "BUY_POINT_MANUAL_REVIEW_READY"
|
),
|
axis=1,
|
)
|
ready = selected[selected["buy_point_replay_status"].eq("BUY_POINT_MANUAL_REVIEW_READY")].copy()
|
data_gap = selected[selected["buy_point_replay_status"].eq("BUY_POINT_MINUTE_DATA_GAP_HELD")].copy()
|
|
write_csv(selected, "strict_note_buy_replay_scope.csv")
|
write_csv(ready, "strict_note_buy_point_review_candidate_ledger.csv")
|
write_csv(boundary, "strict_note_market_gate_boundary_ledger.csv")
|
write_csv(data_gap, "strict_note_buy_point_data_gap_boundary_ledger.csv")
|
write_csv(pd.DataFrame(template_rows), "manual_buy_decision_external_template.csv")
|
write_csv(pd.DataFrame(chart_rows), "chart_evidence_audit.csv")
|
write_csv(pd.DataFrame(missing_rows, columns=["case_id", "candidate_id", "symbol", "entry_trade_date", "missing_reason"]), "missing_chart_inputs.csv")
|
|
generated_at = now_iso()
|
items = [
|
("SCOPE_TOTAL_MATCHES_SELECTED", len(selected) == 3420, f"scope={len(selected)}"),
|
("READY_PLUS_BOUNDARIES_MATCH_SCOPE", len(ready) + len(data_gap) + len(boundary) == len(selected), f"ready={len(ready)}, data_gap={len(data_gap)}, market_closed={len(boundary)}"),
|
("MARKET_CLOSED_NOT_ELIGIBLE", boundary["eligible_for_buy_replay_flag"].eq(False).all(), f"boundary={len(boundary)}"),
|
("MANUAL_TEMPLATE_NO_FINAL_ACTION_PREFILL", pd.DataFrame(template_rows)["human_decision_action"].eq("").all(), "human_decision_action blank"),
|
("MANUAL_TEMPLATE_NO_REASON_PREFILL", pd.DataFrame(template_rows)["human_decision_reason_cn"].eq("").all(), "human_decision_reason_cn blank"),
|
("CHARTS_FOR_READY_CANDIDATES", len(chart_rows) == len(ready), f"charts={len(chart_rows)}, ready={len(ready)}"),
|
("MISSING_CHART_INPUTS_RECORDED_AS_HELD", len(missing_rows) == len(data_gap), f"missing={len(missing_rows)}, data_gap={len(data_gap)}"),
|
]
|
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_review_self_check_items.csv")
|
status = "PASS_FOR_BUY_POINT_MANUAL_REVIEW_PREP_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_review_self_check.json")
|
summary = {
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": status,
|
"counts": {
|
"selected_candidates": int(len(selected)),
|
"eligible_market_gate_open_before_data_gap": int(len(eligible)),
|
"buy_point_manual_review_ready": int(len(ready)),
|
"buy_point_minute_data_gap_held": int(len(data_gap)),
|
"market_gate_closed_boundary": int(len(boundary)),
|
"buy_point_review_charts": int(len(chart_rows)),
|
"missing_chart_inputs": int(len(missing_rows)),
|
},
|
"boundary": [
|
"No final BUY action is generated by this package.",
|
"manual_buy_decision_external_template.csv is intentionally blank for final human decision fields.",
|
"Market-gate-closed candidates are retained as boundary evidence only.",
|
],
|
}
|
write_json(summary, "buy_review_summary.json")
|
(ROOT / "buy_review_summary.md").write_text(
|
"# Strict Note Buy Point Manual Review Prep\n\n"
|
f"- generated_at: {generated_at}\n"
|
f"- selected candidates: {len(selected)}\n"
|
f"- market gate open before minute-data check: {len(eligible)}\n"
|
f"- buy point manual review ready: {len(ready)}\n"
|
f"- buy point minute data gap held: {len(data_gap)}\n"
|
f"- market gate closed boundary: {len(boundary)}\n"
|
f"- charts: {len(chart_rows)}\n"
|
f"- missing chart inputs: {len(missing_rows)}\n\n"
|
"Boundary: this package prepares external manual/AI-manual buy point review inputs only. It does not generate BUY orders or returns.\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()
|
