from __future__ import annotations
|
|
import hashlib
|
import json
|
import os
|
import re
|
from pathlib import Path
|
|
import pandas as pd
|
import pymysql
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
LOCAL_DB_INDEX = Path(
|
r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md"
|
)
|
|
|
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=120,
|
)
|
|
|
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 font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
|
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(26)
|
FONT_MID = font(17)
|
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 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_chart(df: pd.DataFrame, cand: dict, window_name: str, out_path: Path, ma20_ref: float | None) -> 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"买点复核1分钟K图：{cand['symbol']} {cand['entry_trade_date']} {window_name}"
|
d.text((32, 24), title, fill="#111827", font=FONT_TITLE)
|
d.text((32, 58), "当前为人工复核材料，不自动判定 BUY；买点必须由后续人工/AI 看图确认。", fill="#334155", font=FONT_MID)
|
|
plot_left, plot_top, plot_right, plot_bottom = 80, 105, 1060, 575
|
vol_top, vol_bottom = 625, 780
|
note_left, note_top = 1090, 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")
|
|
if df.empty:
|
d.text((plot_left + 160, plot_top + 180), "该窗口无分钟线数据", fill="#b91c1c", font=FONT_TITLE)
|
price_low, price_high = 0.0, 1.0
|
else:
|
price_low = float(df["low_price"].min()) * 0.998
|
price_high = float(df["high_price"].max()) * 1.002
|
refs = [float(df["open_price"].iloc[0])]
|
if ma20_ref:
|
refs.append(ma20_ref)
|
price_low = min(price_low, min(refs) * 0.998)
|
price_high = max(price_high, max(refs) * 1.002)
|
max_vol = max(float(df["volume"].max()), 1.0)
|
n = len(df)
|
gap = (plot_right - plot_left) / max(n, 1)
|
body_w = max(3, int(gap * 0.55))
|
for i, row in df.reset_index(drop=True).iterrows():
|
cx = int(plot_left + gap * i + gap / 2)
|
op, hi, lo, cl = [float(row[c]) for c in ["open_price", "high_price", "low_price", "close_price"]]
|
color = "#dc2626" if cl >= op else "#16a34a"
|
d.line([cx, y_price(lo, price_low, price_high, plot_top, plot_bottom), cx, y_price(hi, price_low, price_high, plot_top, plot_bottom)], fill=color, width=2)
|
y1, y2 = y_price(op, price_low, price_high, plot_top, plot_bottom), y_price(cl, price_low, price_high, plot_top, plot_bottom)
|
d.rectangle([cx - body_w // 2, min(y1, y2), cx + body_w // 2, max(y1, y2)], fill=color, outline=color)
|
vh = int(float(row["volume"]) / max_vol * (vol_bottom - vol_top))
|
d.rectangle([cx - body_w // 2, vol_bottom - vh, cx + body_w // 2, vol_bottom], fill=color, outline=color)
|
if i % max(1, n // 6) == 0:
|
d.text((cx - 24, vol_bottom + 8), str(row["trade_time"])[:5], fill="#64748b", font=FONT_SMALL)
|
|
open_ref = float(df["open_price"].iloc[0])
|
y_open = y_price(open_ref, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y_open, plot_right, y_open], fill="#0f172a", width=2)
|
d.text((plot_right + 8, y_open - 8), f"开盘价 {open_ref:.2f}", fill="#0f172a", font=FONT_SMALL)
|
if ma20_ref:
|
y_ma20 = y_price(ma20_ref, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y_ma20, plot_right, y_ma20], fill="#f59e0b", width=2)
|
d.text((plot_right + 8, y_ma20 - 8), f"日MA20 {ma20_ref:.2f}", fill="#b45309", font=FONT_SMALL)
|
|
for i in range(5):
|
p = price_low + (price_high - price_low) * i / 4
|
y = y_price(p, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y, plot_right, y], fill="#e2e8f0")
|
d.text((18, y - 8), f"{p:.2f}", fill="#64748b", font=FONT_SMALL)
|
|
d.rounded_rectangle([note_left, note_top, 1460, 780], radius=8, outline="#334155", fill="#ffffff")
|
d.text((note_left + 18, note_top + 18), "买点复核提示", fill="#111827", font=FONT_TITLE)
|
notes = [
|
f"候选排名：{cand['candidate_rank']}",
|
f"候选状态：{cand['candidate_status']}",
|
f"窗口：{window_name}",
|
"允许买点：10:40前或14:40后",
|
"看图重点：",
|
"1. 回踩20日均线附近",
|
"2. 回踩开盘价附近",
|
"3. 缩量有支撑",
|
"4. 放量回踩均线禁买",
|
"5. 急拉放量不追",
|
"",
|
"当前动作：ENTRY_REVIEW_PENDING",
|
"不写 BUY，不计收益。",
|
]
|
yy = note_top + 58
|
for line in notes:
|
d.text((note_left + 18, yy), line, fill="#334155", font=FONT_SMALL)
|
yy += 24 if line else 12
|
d.text((32, 820), "无忌 baseline：买点由1分钟K图和人工确认裁决；本图只准备证据，不产生交易结论。", fill="#334155", font=FONT_MID)
|
img.save(out_path)
|
|
|
def main() -> None:
|
selected = pd.read_csv(ROOT / "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"])
|
open_selected = selected[selected["market_gate_status"] == "MKT_GATE_OPEN_PREV_DAY_UP_3000"].copy()
|
|
symbols = sorted(open_selected["symbol"].unique().tolist())
|
dates = sorted(open_selected["entry_trade_date"].dt.strftime("%Y-%m-%d").unique().tolist())
|
min_signal = (open_selected["signal_trade_date"].min() - pd.Timedelta(days=80)).strftime("%Y-%m-%d")
|
max_signal = open_selected["signal_trade_date"].max().strftime("%Y-%m-%d")
|
|
minute = pd.DataFrame()
|
daily = pd.DataFrame()
|
if symbols and dates:
|
sym_ph = ",".join(["%s"] * len(symbols))
|
date_ph = ",".join(["%s"] * len(dates))
|
with get_conn() as conn:
|
minute = pd.read_sql(
|
f"""
|
SELECT trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume, amount
|
FROM a_share_minute_price
|
WHERE symbol IN ({sym_ph}) AND trade_date IN ({date_ph})
|
ORDER BY symbol, trade_date, trade_time
|
""",
|
conn,
|
params=[*symbols, *dates],
|
)
|
daily = pd.read_sql(
|
f"""
|
SELECT trade_date, symbol, close_price
|
FROM a_share_daily_price
|
WHERE symbol IN ({sym_ph}) AND trade_date BETWEEN %s AND %s
|
ORDER BY symbol, trade_date
|
""",
|
conn,
|
params=[*symbols, min_signal, max_signal],
|
)
|
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", "amount"]:
|
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["ma20"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(20, min_periods=1).mean())
|
|
manifest_rows = []
|
decision_rows = []
|
morning_count = 0
|
late_count = 0
|
data_gap_count = 0
|
|
root_manifest_path = ROOT / "image_manifest.csv"
|
if root_manifest_path.exists():
|
root_manifest = pd.read_csv(root_manifest_path, encoding="utf-8-sig")
|
else:
|
root_manifest = pd.DataFrame()
|
|
for _, cand in selected.iterrows():
|
case_id = cand["case_id"]
|
case_dir = ROOT / "cases" / case_id
|
img_dir = case_dir / "img"
|
img_dir.mkdir(parents=True, exist_ok=True)
|
entry_str = cand["entry_trade_date"].strftime("%Y-%m-%d")
|
signal_str = cand["signal_trade_date"].strftime("%Y-%m-%d")
|
if cand["market_gate_status"] != "MKT_GATE_OPEN_PREV_DAY_UP_3000":
|
decision_rows.append(
|
{
|
"case_id": case_id,
|
"candidate_id": cand["candidate_id"],
|
"symbol": cand["symbol"],
|
"entry_trade_date": entry_str,
|
"decision_stage": "ENTRY_GATE",
|
"action_status": "NO_TRADE_MARKET_GATE_CLOSED",
|
"review_required": False,
|
"evidence_image_path": "",
|
"decision_reason_cn": "前一交易日全A上涨家数未达到3000，按baseline不开新仓。",
|
"lookahead_violation_flag": False,
|
}
|
)
|
continue
|
|
ma20_rows = daily[(daily["symbol"] == cand["symbol"]) & (daily["trade_date"] <= cand["signal_trade_date"])].tail(1)
|
ma20_ref = float(ma20_rows["ma20"].iloc[0]) if not ma20_rows.empty else None
|
m = minute[(minute["symbol"] == cand["symbol"]) & (minute["trade_date"] == cand["entry_trade_date"])].copy()
|
windows = [
|
("早盘窗口", "morning", "09:30:00", "10:40:00"),
|
("尾盘窗口", "late", "14:40:00", "15:00:00"),
|
]
|
candidate_paths = []
|
missing_windows = []
|
for label, key, start, end in windows:
|
part = m[(m["trade_time"] >= start) & (m["trade_time"] <= end)].copy()
|
file_name = f"03_entry_1m_{key}_review_{cand['symbol'].replace('.', '_')}_{entry_str.replace('-', '')}.png"
|
out_path = img_dir / file_name
|
cand_dict = cand.to_dict()
|
cand_dict["entry_trade_date"] = entry_str
|
cand_dict["signal_trade_date"] = signal_str
|
draw_chart(part, cand_dict, label, out_path, ma20_ref)
|
if key == "morning":
|
morning_count += 1
|
else:
|
late_count += 1
|
if part.empty:
|
data_gap_count += 1
|
missing_windows.append(label)
|
rel = out_path.relative_to(ROOT).as_posix()
|
candidate_paths.append(rel)
|
manifest_rows.append(
|
{
|
"case_id": case_id,
|
"symbol": cand["symbol"],
|
"trade_date": entry_str,
|
"event_id": f"{cand['candidate_id']}_{key}",
|
"chart_role": f"entry_1m_{key}_review_view",
|
"decision_time": f"{entry_str} {end}",
|
"path": rel,
|
"sha256": sha256_file(out_path),
|
"status": "PASS" if not part.empty else "DATA_GAP_HELD",
|
"note": "1分钟买点复核图；不自动判定BUY。",
|
}
|
)
|
decision_rows.append(
|
{
|
"case_id": case_id,
|
"candidate_id": cand["candidate_id"],
|
"symbol": cand["symbol"],
|
"entry_trade_date": entry_str,
|
"decision_stage": "ENTRY_REVIEW",
|
"action_status": "ENTRY_REVIEW_PENDING" if not missing_windows else "ENTRY_REVIEW_DATA_GAP",
|
"review_required": True,
|
"evidence_image_path": ";".join(candidate_paths),
|
"decision_reason_cn": (
|
"已生成1分钟买点复核图，需人工/AI按缩量支撑、回踩均线/开盘价、急拉放量禁追等规则裁决。"
|
if not missing_windows
|
else "部分买点窗口缺少分钟线：" + "、".join(missing_windows)
|
),
|
"lookahead_violation_flag": False,
|
}
|
)
|
|
new_manifest = pd.DataFrame(manifest_rows)
|
if not root_manifest.empty:
|
combined_manifest = pd.concat([root_manifest, new_manifest], ignore_index=True)
|
combined_manifest = combined_manifest.drop_duplicates(subset=["case_id", "symbol", "event_id", "chart_role"], keep="last")
|
else:
|
combined_manifest = new_manifest
|
combined_manifest.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig")
|
|
decision_log = pd.DataFrame(decision_rows)
|
decision_log.to_csv(ROOT / "decision_log.csv", index=False, encoding="utf-8-sig")
|
|
# Append entry section to case boards.
|
for case_id, group in new_manifest.groupby("case_id"):
|
case_dir = ROOT / "cases" / case_id
|
board_path = case_dir / "case_image_board.md"
|
existing = board_path.read_text(encoding="utf-8") if board_path.exists() else f"# {case_id} 图片审核板\n"
|
lines = [existing.rstrip(), "", "## 3. 买点 1分钟K 复核图", ""]
|
for _, row in group.iterrows():
|
rel = Path(row["path"]).relative_to(f"cases/{case_id}").as_posix()
|
lines.extend(
|
[
|
f"### {row['symbol']} / {row['chart_role']}",
|
"",
|
f"![{row['symbol']}]({rel})",
|
"",
|
f"- 状态：`{row['status']}`",
|
"- 本图只用于买点复核，不自动写 BUY。",
|
"",
|
]
|
)
|
board_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
case_manifest = combined_manifest[combined_manifest["case_id"] == case_id]
|
case_manifest.to_csv(case_dir / "image_manifest.csv", index=False, encoding="utf-8-sig")
|
|
summary = {
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"generated_at": "2026-06-08T01:05:00+08:00",
|
"stage": "ENTRY_1M_REVIEW_IMAGES_READY",
|
"decision_rows": int(len(decision_log)),
|
"entry_review_images": int(len(new_manifest)),
|
"morning_images": morning_count,
|
"late_images": late_count,
|
"data_gap_windows": data_gap_count,
|
"no_trade_market_gate_closed_rows": int((decision_log["action_status"] == "NO_TRADE_MARKET_GATE_CLOSED").sum()),
|
"entry_review_pending_rows": int((decision_log["action_status"] == "ENTRY_REVIEW_PENDING").sum()),
|
"entry_review_data_gap_rows": int((decision_log["action_status"] == "ENTRY_REVIEW_DATA_GAP").sum()),
|
"artifacts": {
|
"decision_log.csv": {
|
"size": (ROOT / "decision_log.csv").stat().st_size,
|
"sha256": sha256_file(ROOT / "decision_log.csv"),
|
},
|
"image_manifest.csv": {
|
"size": (ROOT / "image_manifest.csv").stat().st_size,
|
"sha256": sha256_file(ROOT / "image_manifest.csv"),
|
},
|
},
|
"next_step": "Perform manual/AI entry review from 1-minute K charts. Do not create order_ledger until BUY/NO_BUY decisions are confirmed.",
|
}
|
(ROOT / "entry_review_generation_summary.json").write_text(
|
json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
|
encoding="utf-8",
|
)
|
(ROOT / "entry_review_generation_summary.md").write_text(
|
"\n".join(
|
[
|
"# entry_review_generation_summary",
|
"",
|
f"run_id：`{RUN_ID}`",
|
"阶段：`ENTRY_1M_REVIEW_IMAGES_READY`",
|
"",
|
f"- decision rows：{summary['decision_rows']}",
|
f"- 1分钟买点复核图：{summary['entry_review_images']}",
|
f"- 数据缺口窗口：{summary['data_gap_windows']}",
|
f"- 闸门关闭不交易行：{summary['no_trade_market_gate_closed_rows']}",
|
f"- 待人工/AI买点复核行：{summary['entry_review_pending_rows']}",
|
"",
|
"当前不产生订单、不产生收益结论。",
|
"",
|
]
|
),
|
encoding="utf-8",
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
