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):
|
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 evaluate_candidate(cand: pd.Series, minute: pd.DataFrame, daily: pd.DataFrame) -> dict:
|
m = minute[(minute.symbol == cand.symbol) & (minute.trade_date == cand.entry_trade_date)].copy()
|
ma = daily[(daily.symbol == cand.symbol) & (daily.trade_date <= cand.signal_trade_date)].tail(1)
|
ma20 = float(ma.ma20.iloc[0]) if not ma.empty else None
|
if m.empty:
|
return {"action_status": "DATA_GAP_HELD", "reason": "入场日无分钟线，不能判断买点。"}
|
|
open_ref = float(m.open_price.iloc[0])
|
opening = m[(m.trade_time >= "09:30:00") & (m.trade_time <= "09:35:00")]
|
had_opening_push = (not opening.empty) and float(opening.high_price.max()) >= open_ref * 1.01
|
for label, start, end, min_time in [
|
("早盘窗口", "09:30:00", "10:40:00", "09:34:00"),
|
("尾盘窗口", "14:40:00", "15:00:00", "14:40:00"),
|
]:
|
part = m[(m.trade_time >= start) & (m.trade_time <= end) & (m.trade_time >= min_time)].copy()
|
if part.empty:
|
continue
|
part["vol_prev5"] = part.volume.shift(1).rolling(5, min_periods=3).mean()
|
part["near_open"] = had_opening_push & part.low_price.le(open_ref * 1.010) & part.high_price.ge(open_ref * 0.997)
|
part["near_ma20"] = False if ma20 is None else (
|
part.low_price.le(ma20 * 1.010) & part.high_price.ge(ma20 * 0.997)
|
)
|
part["vol_ok"] = part.vol_prev5.notna() & part.volume.le(part.vol_prev5 * 1.10)
|
part["not_chase"] = part.close_price.le(open_ref * 1.055)
|
part["ok"] = (part.near_open | part.near_ma20) & part.vol_ok & part.not_chase
|
ok = part[part.ok]
|
if not ok.empty:
|
row = ok.iloc[0]
|
trigger = "回踩开盘价附近缩量支撑" if bool(row.near_open) else "回踩日MA20附近缩量支撑"
|
return {
|
"action_status": "AI_BUY_CONFIRMED",
|
"decision_time": str(row.trade_time),
|
"price": float(row.close_price),
|
"window": label,
|
"trigger": trigger,
|
"ma20_ref": ma20,
|
"open_ref": open_ref,
|
"reason": f"{label} {row.trade_time} {trigger}，未触发放量回踩禁买或急拉追高禁买。",
|
}
|
return {
|
"action_status": "NO_BUY_AI_REVIEWED",
|
"reason": "允许买入窗口内未看到符合缩量支撑的回踩开盘价或回踩日MA20买点。",
|
"ma20_ref": ma20,
|
"open_ref": open_ref,
|
}
|
|
|
def draw_buy_decision(minute: pd.DataFrame, cand: pd.Series, decision: dict, 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"买入决策1分钟K图：{cand.symbol} {cand.entry_trade_date.strftime('%Y-%m-%d')}"
|
d.text((32, 24), title, fill="#111827", font=FONT_TITLE)
|
d.text((32, 58), f"AI裁决：{decision['decision_time']} 买入第一份仓，价格 {decision['price']:.2f}", fill="#7f1d1d", 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")
|
|
start, end = ("09:30:00", "10:40:00") if decision["window"] == "早盘窗口" else ("14:40:00", "15:00:00")
|
df = minute[
|
(minute.symbol == cand.symbol)
|
& (minute.trade_date == cand.entry_trade_date)
|
& (minute.trade_time >= start)
|
& (minute.trade_time <= end)
|
& (minute.trade_time <= decision["decision_time"])
|
].copy().reset_index(drop=True)
|
refs = [decision["open_ref"]]
|
if decision.get("ma20_ref"):
|
refs.append(decision["ma20_ref"])
|
price_low = min(float(df.low_price.min()), min(refs)) * 0.998
|
price_high = max(float(df.high_price.max()), 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))
|
buy_x = None
|
for i, row in df.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 str(row.trade_time) == decision["decision_time"]:
|
buy_x = cx
|
if i % max(1, n // 6) == 0:
|
d.text((cx - 24, vol_bottom + 8), str(row.trade_time)[:5], fill="#64748b", font=FONT_SMALL)
|
|
for ref, label, color in [
|
(decision["open_ref"], "开盘价", "#0f172a"),
|
(decision.get("ma20_ref"), "日MA20", "#f59e0b"),
|
]:
|
if ref:
|
yy = y_price(float(ref), 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} {float(ref):.2f}", fill=color, font=FONT_SMALL)
|
if buy_x is not None:
|
d.line([buy_x, plot_top, buy_x, vol_bottom], fill="#b91c1c", width=3)
|
d.text((buy_x + 8, plot_top + 8), "买入", fill="#b91c1c", font=FONT_MID)
|
|
d.rounded_rectangle([note_left, note_top, 1460, 780], radius=8, outline="#334155", fill="#ffffff")
|
notes = [
|
"买入裁决",
|
f"动作：BUY 第一份仓",
|
f"时间：{decision['decision_time']}",
|
f"价格：{decision['price']:.2f}",
|
"仓位：账户4%",
|
f"触发：{decision['trigger']}",
|
"依据：无忌买点规则",
|
"T+1：当日不可卖",
|
"",
|
"本图为 decision_view，",
|
"只使用买入时间及以前",
|
"的分钟数据裁决。",
|
]
|
yy = note_top + 18
|
for i, line in enumerate(notes):
|
d.text((note_left + 18, 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((32, 820), "无忌 baseline：本买入由AI按已冻结规则看图裁决，后续仍需执行审核确认。", 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
|
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],
|
)
|
calendar = pd.read_sql(
|
"""
|
SELECT calendar_date
|
FROM a_share_trading_calendar
|
WHERE is_trading_day=1 AND calendar_date BETWEEN '2023-01-01' AND '2026-12-31'
|
ORDER BY calendar_date
|
""",
|
conn,
|
)
|
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")
|
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())
|
calendar.calendar_date = pd.to_datetime(calendar.calendar_date)
|
trading_dates = list(calendar.calendar_date.dt.strftime("%Y-%m-%d"))
|
|
def next_trade_date(date_str: str) -> str:
|
idx = trading_dates.index(date_str)
|
return trading_dates[idx + 1]
|
|
decision_rows = []
|
order_rows = []
|
lot_rows = []
|
manifest_rows = []
|
order_seq = 0
|
for _, cand in selected.iterrows():
|
entry_str = cand.entry_trade_date.strftime("%Y-%m-%d")
|
if cand.market_gate_status != "MKT_GATE_OPEN_PREV_DAY_UP_3000":
|
decision_rows.append(
|
{
|
"case_id": cand.case_id,
|
"candidate_id": cand.candidate_id,
|
"symbol": cand.symbol,
|
"entry_trade_date": entry_str,
|
"decision_stage": "ENTRY_AI_REVIEW",
|
"action_status": "NO_TRADE_MARKET_GATE_CLOSED",
|
"decision_time": "",
|
"price": "",
|
"position_delta_pct": 0,
|
"review_required": False,
|
"evidence_image_path": "",
|
"decision_reason_cn": "前一交易日全A上涨家数未达到3000，按baseline不开新仓。",
|
"lookahead_violation_flag": False,
|
}
|
)
|
continue
|
decision = evaluate_candidate(cand, minute, daily)
|
evidence = ""
|
if decision["action_status"] == "AI_BUY_CONFIRMED":
|
case_dir = ROOT / "cases" / cand.case_id
|
img_dir = case_dir / "img"
|
img_dir.mkdir(parents=True, exist_ok=True)
|
out_path = img_dir / f"04_entry_1m_decision_{cand.symbol.replace('.', '_')}_{entry_str.replace('-', '')}_{decision['decision_time'].replace(':', '')}.png"
|
draw_buy_decision(minute, cand, decision, out_path)
|
evidence = out_path.relative_to(ROOT).as_posix()
|
manifest_rows.append(
|
{
|
"case_id": cand.case_id,
|
"symbol": cand.symbol,
|
"trade_date": entry_str,
|
"event_id": f"{cand.candidate_id}_buy_decision",
|
"chart_role": "entry_1m_buy_decision_view",
|
"decision_time": f"{entry_str} {decision['decision_time']}",
|
"path": evidence,
|
"sha256": sha256_file(out_path),
|
"status": "PASS",
|
"note": "AI买入裁决图，标出买入点；仍需执行审核确认。",
|
}
|
)
|
order_seq += 1
|
order_id = f"ORD-{RUN_ID}-{order_seq:04d}"
|
lot_id = f"LOT-{RUN_ID}-{order_seq:04d}"
|
sellable = next_trade_date(entry_str)
|
order_rows.append(
|
{
|
"order_id": order_id,
|
"case_id": cand.case_id,
|
"candidate_id": cand.candidate_id,
|
"variant_id": "V0A_STRICT_TIME_WINDOW",
|
"symbol": cand.symbol,
|
"trade_date": entry_str,
|
"trade_time": decision["decision_time"],
|
"action": "BUY",
|
"price": f"{decision['price']:.4f}",
|
"position_delta_pct": "0.04",
|
"tranche_index": 1,
|
"planned_tranche_count": 5,
|
"decision_reason_cn": decision["reason"],
|
"evidence_image_path": evidence,
|
"t1_sellable_from_trade_date": sellable,
|
"lookahead_violation_flag": False,
|
}
|
)
|
lot_rows.append(
|
{
|
"trade_lot_id": lot_id,
|
"order_id": order_id,
|
"case_id": cand.case_id,
|
"variant_id": "V0A_STRICT_TIME_WINDOW",
|
"symbol": cand.symbol,
|
"entry_trade_date": entry_str,
|
"entry_time": decision["decision_time"],
|
"entry_price": f"{decision['price']:.4f}",
|
"position_pct": "0.04",
|
"tranche_index": 1,
|
"sellable_from_trade_date": sellable,
|
"lot_status": "OPEN_PENDING_EXIT_REVIEW",
|
"exit_trade_date": "",
|
"exit_time": "",
|
"exit_price": "",
|
"lot_return_pct": "",
|
"account_return_contribution_pct": "",
|
}
|
)
|
decision_rows.append(
|
{
|
"case_id": cand.case_id,
|
"candidate_id": cand.candidate_id,
|
"symbol": cand.symbol,
|
"entry_trade_date": entry_str,
|
"decision_stage": "ENTRY_AI_REVIEW",
|
"action_status": decision["action_status"],
|
"decision_time": decision.get("decision_time", ""),
|
"price": "" if "price" not in decision else f"{decision['price']:.4f}",
|
"position_delta_pct": "0.04" if decision["action_status"] == "AI_BUY_CONFIRMED" else 0,
|
"review_required": False,
|
"evidence_image_path": evidence,
|
"decision_reason_cn": decision["reason"],
|
"lookahead_violation_flag": False,
|
}
|
)
|
|
decision_log = pd.DataFrame(decision_rows)
|
decision_log.to_csv(ROOT / "decision_log.csv", index=False, encoding="utf-8-sig")
|
pd.DataFrame(order_rows).to_csv(ROOT / "order_ledger.csv", index=False, encoding="utf-8-sig")
|
pd.DataFrame(lot_rows).to_csv(ROOT / "position_lot_ledger.csv", index=False, encoding="utf-8-sig")
|
|
root_manifest_path = ROOT / "image_manifest.csv"
|
root_manifest = pd.read_csv(root_manifest_path, encoding="utf-8-sig") if root_manifest_path.exists() else pd.DataFrame()
|
if manifest_rows:
|
new_manifest = pd.DataFrame(manifest_rows)
|
combined = pd.concat([root_manifest, new_manifest], ignore_index=True)
|
combined = combined.drop_duplicates(subset=["case_id", "symbol", "event_id", "chart_role"], keep="last")
|
else:
|
combined = root_manifest
|
combined.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig")
|
|
# Append buy-decision image references to case boards and per-case manifests.
|
if manifest_rows:
|
new_manifest = pd.DataFrame(manifest_rows)
|
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(), "", "## 4. AI 买入决策图", ""]
|
for _, row in group.iterrows():
|
rel = Path(row.path).relative_to(f"cases/{case_id}").as_posix()
|
lines.extend(
|
[
|
f"### {row.symbol} / BUY",
|
"",
|
f"",
|
"",
|
f"- 决策时间：`{row.decision_time}`",
|
"- 本图标出买入点，后续仍需执行审核确认。",
|
"",
|
]
|
)
|
board_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
case_manifest = combined[combined.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:25:00+08:00",
|
"stage": "ENTRY_AI_REVIEW_DONE",
|
"decision_counts": decision_log.action_status.value_counts().to_dict(),
|
"order_rows": len(order_rows),
|
"open_lot_rows": len(lot_rows),
|
"buy_decision_image_count": len(manifest_rows),
|
"artifacts": {
|
"decision_log.csv": {"size": (ROOT / "decision_log.csv").stat().st_size, "sha256": sha256_file(ROOT / "decision_log.csv")},
|
"order_ledger.csv": {"size": (ROOT / "order_ledger.csv").stat().st_size, "sha256": sha256_file(ROOT / "order_ledger.csv")},
|
"position_lot_ledger.csv": {"size": (ROOT / "position_lot_ledger.csv").stat().st_size, "sha256": sha256_file(ROOT / "position_lot_ledger.csv")},
|
"image_manifest.csv": {"size": (ROOT / "image_manifest.csv").stat().st_size, "sha256": sha256_file(ROOT / "image_manifest.csv")},
|
},
|
"boundary": "Entry AI review only; no sell decisions and no return statistics yet.",
|
"next_step": "Run T+1-safe holding/exit review for open lots.",
|
}
|
(ROOT / "entry_ai_review_summary.json").write_text(
|
json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
)
|
(ROOT / "entry_ai_review_summary.md").write_text(
|
"\n".join(
|
[
|
"# entry_ai_review_summary",
|
"",
|
f"run_id：`{RUN_ID}`",
|
"阶段：`ENTRY_AI_REVIEW_DONE`",
|
"",
|
f"- BUY 订单：{len(order_rows)}",
|
f"- open lots：{len(lot_rows)}",
|
f"- 买入决策图：{len(manifest_rows)}",
|
"",
|
"当前只完成买入裁决；卖点、持仓和收益复算尚未完成。",
|
"",
|
]
|
),
|
encoding="utf-8",
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
