1
2026-06-16 2d8cc2eb4b913c34d8317800458a85939de4da1e
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
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()