from __future__ import annotations
|
|
import hashlib
|
import json
|
import math
|
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(28)
|
FONT_MID = font(18)
|
FONT_SMALL = font(14)
|
|
|
def gate_label(raw: str) -> str:
|
return "开仓闸门打开" if raw == "MKT_GATE_OPEN_PREV_DAY_UP_3000" else "开仓闸门关闭"
|
|
|
def status_label(raw: str) -> str:
|
return "候选通过" if raw == "PASS" else "前高量能风险待复核"
|
|
|
def bool_label(raw) -> str:
|
return "是" if str(raw).lower() == "true" else "否"
|
|
|
def draw_wrapped(draw: ImageDraw.ImageDraw, xy: tuple[int, int], text: str, max_chars: int, fill: str, used_font, line_gap: int = 8) -> int:
|
x, y = xy
|
chunks = []
|
current = ""
|
for ch in text:
|
current += ch
|
if len(current) >= max_chars:
|
chunks.append(current)
|
current = ""
|
if current:
|
chunks.append(current)
|
for chunk in chunks:
|
draw.text((x, y), chunk, fill=fill, font=used_font)
|
y += used_font.size + line_gap if hasattr(used_font, "size") else 24
|
return y
|
|
|
def price_y(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_daily_chart(window: pd.DataFrame, cand: dict, out_path: Path) -> None:
|
w, h = 1500, 920
|
img = Image.new("RGB", (w, h), "#fbfbf7")
|
d = ImageDraw.Draw(img)
|
d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
|
|
title = f"选股日K图：{cand['symbol']} 入场日 {cand['entry_trade_date']}"
|
subtitle = (
|
f"信号日 {cand['signal_trade_date']}｜排名 {cand['candidate_rank']}｜"
|
f"量比 {float(cand['volume_ratio']):.2f}｜上影 {float(cand['upper_shadow_pct']):.2f}%｜"
|
f"闸门 {gate_label(cand['market_gate_status'])}"
|
)
|
d.text((32, 24), title, fill="#111827", font=FONT_TITLE)
|
d.text((32, 62), subtitle, fill="#334155", font=FONT_MID)
|
|
plot_left, plot_top, plot_right, plot_bottom = 80, 115, 1060, 610
|
vol_top, vol_bottom = 660, 820
|
note_left, note_top = 1090, 118
|
d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8")
|
d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8")
|
|
window = window.copy().reset_index(drop=True)
|
price_low = float(window["low_price"].min()) * 0.98
|
price_high = float(window["high_price"].max()) * 1.02
|
max_vol = max(float(window["volume"].max()), 1.0)
|
n = len(window)
|
gap = (plot_right - plot_left) / max(n, 1)
|
body_w = max(3, int(gap * 0.55))
|
|
# Grid and price labels.
|
for i in range(5):
|
p = price_low + (price_high - price_low) * i / 4
|
y = price_y(p, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y, plot_right, y], fill="#e2e8f0")
|
d.text((18, y - 9), f"{p:.2f}", fill="#64748b", font=FONT_SMALL)
|
|
ma_colors = {"ma5": "#2563eb", "ma20": "#f59e0b", "ma60": "#7c3aed"}
|
ma_points: dict[str, list[tuple[int, int]]] = {k: [] for k in ma_colors}
|
signal_date = str(cand["signal_trade_date"])
|
last_limit_date = str(cand.get("last_limitup_date") or "")
|
|
for i, row in window.iterrows():
|
cx = int(plot_left + gap * i + gap / 2)
|
op = float(row["open_price"])
|
hi = float(row["high_price"])
|
lo = float(row["low_price"])
|
cl = float(row["close_price"])
|
color = "#dc2626" if cl >= op else "#16a34a"
|
d.line([cx, price_y(lo, price_low, price_high, plot_top, plot_bottom), cx, price_y(hi, price_low, price_high, plot_top, plot_bottom)], fill=color, width=2)
|
y1 = price_y(op, price_low, price_high, plot_top, plot_bottom)
|
y2 = price_y(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)
|
trade_date = str(row["trade_date"])
|
if trade_date == signal_date:
|
d.line([cx, plot_top, cx, vol_bottom], fill="#0f172a", width=2)
|
d.text((cx - 32, plot_top - 24), "信号日", fill="#0f172a", font=FONT_SMALL)
|
if trade_date == last_limit_date:
|
d.ellipse([cx - 8, price_y(hi, price_low, price_high, plot_top, plot_bottom) - 22, cx + 8, price_y(hi, price_low, price_high, plot_top, plot_bottom) - 6], fill="#ef4444")
|
d.text((cx - 28, price_y(hi, price_low, price_high, plot_top, plot_bottom) - 48), "涨停记忆", fill="#ef4444", font=FONT_SMALL)
|
if i % max(1, n // 8) == 0:
|
d.text((cx - 28, vol_bottom + 8), trade_date[5:], fill="#64748b", font=FONT_SMALL)
|
for ma in ma_colors:
|
if pd.notna(row[ma]):
|
ma_points[ma].append((cx, price_y(float(row[ma]), price_low, price_high, plot_top, plot_bottom)))
|
|
for ma, pts in ma_points.items():
|
if len(pts) > 1:
|
d.line(pts, fill=ma_colors[ma], width=2)
|
legend_x = plot_left + 8
|
for ma, color in ma_colors.items():
|
d.text((legend_x, plot_bottom + 12), ma.upper(), fill=color, font=FONT_SMALL)
|
legend_x += 70
|
|
d.rounded_rectangle([note_left, note_top, 1460, 820], radius=8, outline="#334155", fill="#ffffff")
|
d.text((note_left + 18, note_top + 18), "候选判读", fill="#111827", font=FONT_TITLE)
|
notes = [
|
f"候选状态：{status_label(cand['candidate_status'])}",
|
f"市场闸门：{gate_label(cand['market_gate_status'])}",
|
f"上涨家数：{cand.get('up_count', '')}，下跌家数：{cand.get('down_count', '')}",
|
f"近30日涨停记忆：{cand.get('last_limitup_date', '')}",
|
f"量比：{float(cand['volume_ratio']):.2f}（阈值 1.70）",
|
f"长上影：{float(cand['upper_shadow_pct']):.2f}%",
|
f"触及前高：{bool_label(cand['touch_prev_high_flag'])}",
|
f"前高量能通过：{bool_label(cand['prev_high_volume_pass_flag'])}",
|
f"60日横盘标记：{bool_label(cand['flat60_flag'])}",
|
"",
|
"图片口径：decision_view",
|
"只展示信号日及以前日线。",
|
"本图只证明进入观察，",
|
"不代表已经买入。",
|
]
|
yy = note_top + 64
|
for line in notes:
|
if line:
|
yy = draw_wrapped(d, (note_left + 18, yy), line, 24, "#334155", FONT_SMALL, line_gap=5)
|
else:
|
yy += 16
|
|
footer = "无忌 baseline：候选池先看最近涨停记忆、放量、长上影、前高量能；买入还需后续 1 分钟 K 图和人工确认。"
|
d.text((32, 870), footer, fill="#334155", font=FONT_MID)
|
img.save(out_path)
|
|
|
def select_candidates(case_index: pd.DataFrame, candidate_ledger: pd.DataFrame) -> pd.DataFrame:
|
selected_rows = []
|
for _, case in case_index.iterrows():
|
rows = candidate_ledger[candidate_ledger["entry_trade_date"] == case["entry_trade_date"]].copy()
|
if rows.empty:
|
continue
|
if case["selection_bucket"] == "PREV_HIGH_REVIEW_RISK":
|
review = rows[rows["candidate_status"] != "PASS"].sort_values("candidate_rank").head(2)
|
strict = rows[rows["candidate_status"] == "PASS"].sort_values("candidate_rank").head(3)
|
chosen = pd.concat([strict, review], ignore_index=True).sort_values("candidate_rank").head(5)
|
else:
|
strict = rows[rows["candidate_status"] == "PASS"].sort_values("candidate_rank").head(5)
|
chosen = strict if len(strict) >= 5 else rows.sort_values("candidate_rank").head(5)
|
chosen = chosen.copy()
|
chosen["case_id"] = case["case_id"]
|
chosen["case_status"] = case["case_status"]
|
chosen["selection_bucket"] = case["selection_bucket"]
|
selected_rows.append(chosen)
|
return pd.concat(selected_rows, ignore_index=True) if selected_rows else pd.DataFrame()
|
|
|
def main() -> None:
|
candidate_ledger = pd.read_csv(ROOT / "candidate_ledger.csv", encoding="utf-8-sig")
|
case_index = pd.read_csv(ROOT / "case_index.csv", encoding="utf-8-sig")
|
selected = select_candidates(case_index, candidate_ledger)
|
selected.to_csv(ROOT / "selected_candidate_ledger.csv", index=False, encoding="utf-8-sig")
|
if selected.empty:
|
raise RuntimeError("No selected candidates available for image generation.")
|
|
selected["signal_trade_date"] = pd.to_datetime(selected["signal_trade_date"])
|
selected["entry_trade_date"] = pd.to_datetime(selected["entry_trade_date"])
|
symbols = sorted(selected["symbol"].unique().tolist())
|
min_date = (selected["signal_trade_date"].min() - pd.Timedelta(days=220)).strftime("%Y-%m-%d")
|
max_date = selected["signal_trade_date"].max().strftime("%Y-%m-%d")
|
placeholders = ",".join(["%s"] * len(symbols))
|
|
with get_conn() as conn:
|
daily = pd.read_sql(
|
f"""
|
SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount
|
FROM a_share_daily_price
|
WHERE trade_date BETWEEN %s AND %s
|
AND symbol IN ({placeholders})
|
ORDER BY symbol, trade_date
|
""",
|
conn,
|
params=[min_date, max_date, *symbols],
|
)
|
daily["trade_date"] = pd.to_datetime(daily["trade_date"])
|
for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
|
daily[col] = pd.to_numeric(daily[col], errors="coerce")
|
daily = daily.sort_values(["symbol", "trade_date"]).reset_index(drop=True)
|
grouped = daily.groupby("symbol", group_keys=False)
|
daily["ma5"] = grouped["close_price"].transform(lambda s: s.rolling(5, min_periods=1).mean())
|
daily["ma20"] = grouped["close_price"].transform(lambda s: s.rolling(20, min_periods=1).mean())
|
daily["ma60"] = grouped["close_price"].transform(lambda s: s.rolling(60, min_periods=1).mean())
|
|
manifest_rows = []
|
case_board_links = []
|
for _, case in case_index.iterrows():
|
case_id = case["case_id"]
|
case_dir = ROOT / "cases" / case_id
|
img_dir = case_dir / "img"
|
img_dir.mkdir(parents=True, exist_ok=True)
|
case_candidates = selected[selected["case_id"] == case_id].copy()
|
case_candidates.to_csv(case_dir / "candidate_ledger.csv", index=False, encoding="utf-8-sig")
|
case_manifest = []
|
board_lines = [
|
f"# {case_id} 图片审核板",
|
"",
|
f"入场日：`{case['entry_trade_date']}`",
|
f"信号日：`{case['signal_trade_date']}`",
|
f"样本分层：`{case['selection_bucket']}`",
|
f"当前状态：`{case['case_status']}`",
|
"",
|
"本板当前只完成选股日 K 图阶段；买入 1 分钟 K 图、持仓图、卖出图和结果汇总图尚未生成。",
|
"",
|
"## 1. 市场闸门",
|
"",
|
f"- 闸门状态：`{case['market_gate_status']}`",
|
f"- 上涨家数：`{case['up_count']}`",
|
f"- 下跌家数：`{case['down_count']}`",
|
"",
|
"## 2. 选股证据图",
|
"",
|
]
|
story_lines = [
|
f"# {case_id} 文字追溯板",
|
"",
|
"## 当前阶段",
|
"",
|
"`CANDIDATE_DAILY_IMAGE_PACKAGE_READY`",
|
"",
|
"当前只完成候选池和选股日 K 图；这些图只表示进入观察，不代表买入或收益结论。",
|
"",
|
]
|
for _, cand in case_candidates.iterrows():
|
signal_date = pd.Timestamp(cand["signal_trade_date"])
|
hist = daily[(daily["symbol"] == cand["symbol"]) & (daily["trade_date"] <= signal_date)].tail(100).copy()
|
hist["trade_date"] = hist["trade_date"].dt.strftime("%Y-%m-%d")
|
file_name = f"01_candidate_daily_100d_{cand['symbol'].replace('.', '_')}_{signal_date.strftime('%Y%m%d')}.png"
|
out_path = img_dir / file_name
|
cand_dict = cand.to_dict()
|
cand_dict["signal_trade_date"] = signal_date.strftime("%Y-%m-%d")
|
cand_dict["entry_trade_date"] = pd.Timestamp(cand["entry_trade_date"]).strftime("%Y-%m-%d")
|
draw_daily_chart(hist, cand_dict, out_path)
|
rel = out_path.relative_to(case_dir).as_posix()
|
root_rel = out_path.relative_to(ROOT).as_posix()
|
row = {
|
"case_id": case_id,
|
"symbol": cand["symbol"],
|
"trade_date": cand_dict["signal_trade_date"],
|
"event_id": cand["candidate_id"],
|
"chart_role": "candidate_daily_100d_decision_view",
|
"decision_time": f"{cand_dict['signal_trade_date']} close",
|
"path": root_rel,
|
"sha256": sha256_file(out_path),
|
"status": "PASS",
|
"note": "选股日K图，只展示信号日及以前数据；不代表买入。",
|
}
|
manifest_rows.append(row)
|
case_manifest.append(row)
|
board_lines.extend(
|
[
|
f"### {cand['symbol']} / 候选排名 {cand['candidate_rank']}",
|
"",
|
f"![{cand['symbol']}]({rel})",
|
"",
|
f"- 候选状态：`{cand['candidate_status']}`",
|
f"- 量比：`{float(cand['volume_ratio']):.2f}`；长上影：`{float(cand['upper_shadow_pct']):.2f}%`",
|
f"- 前高量能通过：`{cand['prev_high_volume_pass_flag']}`",
|
"",
|
]
|
)
|
(case_dir / "image_manifest.csv").write_text(
|
pd.DataFrame(case_manifest).to_csv(index=False), encoding="utf-8-sig"
|
)
|
(case_dir / "case_image_board.md").write_text("\n".join(board_lines) + "\n", encoding="utf-8")
|
(case_dir / "case_story_board.md").write_text("\n".join(story_lines) + "\n", encoding="utf-8")
|
case_files = [
|
"candidate_ledger.csv",
|
"image_manifest.csv",
|
"case_image_board.md",
|
"case_story_board.md",
|
]
|
case_manifest_json = {
|
"case_id": case_id,
|
"run_id": RUN_ID,
|
"stage": "CANDIDATE_DAILY_IMAGE_PACKAGE_READY",
|
"files": [
|
{
|
"path": name,
|
"size": (case_dir / name).stat().st_size,
|
"sha256": sha256_file(case_dir / name),
|
}
|
for name in case_files
|
],
|
"image_count": len(case_manifest),
|
}
|
(case_dir / "manifest.json").write_text(
|
json.dumps(case_manifest_json, ensure_ascii=False, indent=2) + "\n",
|
encoding="utf-8",
|
)
|
case_board_links.append(f"- [{case_id}](cases/{case_id}/case_image_board.md)")
|
|
image_manifest = pd.DataFrame(manifest_rows)
|
image_manifest.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig")
|
root_board = [
|
"# RUN 图片审核入口",
|
"",
|
f"run_id：`{RUN_ID}`",
|
"阶段：`CANDIDATE_DAILY_IMAGE_PACKAGE_READY`",
|
"",
|
"当前只完成候选池和选股日 K 图；买卖分时图、订单账本和收益复算尚未生成。",
|
"",
|
"## 案例入口",
|
"",
|
*case_board_links,
|
"",
|
]
|
(ROOT / "case_image_board.md").write_text("\n".join(root_board), encoding="utf-8")
|
(ROOT / "case_story_board.md").write_text(
|
"# RUN 文字追溯入口\n\n当前只完成候选池和选股日 K 图。后续进入买点审核后补充分时图、订单、持仓和收益复算。\n",
|
encoding="utf-8",
|
)
|
summary = {
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"generated_at": "2026-06-08T00:50:00+08:00",
|
"stage": "CANDIDATE_DAILY_IMAGE_PACKAGE_READY",
|
"case_count": int(case_index["case_id"].nunique()),
|
"selected_candidate_rows": int(len(selected)),
|
"image_count": int(len(image_manifest)),
|
"artifacts": {
|
"selected_candidate_ledger.csv": {
|
"size": (ROOT / "selected_candidate_ledger.csv").stat().st_size,
|
"sha256": sha256_file(ROOT / "selected_candidate_ledger.csv"),
|
},
|
"image_manifest.csv": {
|
"size": (ROOT / "image_manifest.csv").stat().st_size,
|
"sha256": sha256_file(ROOT / "image_manifest.csv"),
|
},
|
"case_image_board.md": {
|
"size": (ROOT / "case_image_board.md").stat().st_size,
|
"sha256": sha256_file(ROOT / "case_image_board.md"),
|
},
|
},
|
"next_step": "Generate entry 1-minute decision views and preliminary decision_log for selected pilot cases.",
|
}
|
(ROOT / "candidate_image_generation_summary.json").write_text(
|
json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
)
|
(ROOT / "candidate_image_generation_summary.md").write_text(
|
"\n".join(
|
[
|
"# candidate_image_generation_summary",
|
"",
|
f"run_id：`{RUN_ID}`",
|
"阶段：`CANDIDATE_DAILY_IMAGE_PACKAGE_READY`",
|
"",
|
f"- 案例数：{summary['case_count']}",
|
f"- 选中候选行：{summary['selected_candidate_rows']}",
|
f"- 生成选股日 K 图：{summary['image_count']}",
|
"",
|
"当前图片只用于选股审核，不产生买入或收益结论。",
|
"",
|
]
|
),
|
encoding="utf-8",
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
