from __future__ import annotations
|
|
import hashlib
|
import json
|
import math
|
import os
|
import re
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
from typing import Iterable
|
|
import pandas as pd
|
import pymysql
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
RUN_ID = "RUN-ANA-WUJI-STRICT-CLOSED-234-CHART-ENHANCE-20260609-001"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001"
|
GUIDE_RUN_ID = "RUN-ANA-WUJI-STRICT-CLOSED-234-REVIEW-GUIDE-20260608-001"
|
TASK_ID = "ANA-WUJI-STRICT-CLOSED-234-CHART-ENHANCE-20260609"
|
|
PACKAGE_ROOT = Path(__file__).resolve().parents[1]
|
RESULT_ROOT = Path(__file__).resolve().parents[2]
|
PROJECT_ROOT = Path(__file__).resolve().parents[4]
|
SOURCE_ROOT = RESULT_ROOT / SOURCE_RUN_ID
|
GUIDE_ROOT = RESULT_ROOT / GUIDE_RUN_ID
|
LOCAL_DB_INDEX = Path(
|
r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md"
|
)
|
|
|
def now_iso() -> str:
|
return datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
|
|
GENERATED_AT = now_iso()
|
|
|
def read_password() -> str:
|
env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD")
|
if env:
|
return env
|
candidates = [LOCAL_DB_INDEX]
|
observer_root = Path(r"D:\strategy_project\s-system-doc\observer")
|
if observer_root.exists():
|
candidates.extend(observer_root.glob("*/数据库索引数据.md"))
|
for path in candidates:
|
if not path.exists():
|
continue
|
text = path.read_text(encoding="utf-8")
|
match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE)
|
if match:
|
return match.group(1)
|
raise RuntimeError("Unable to read local MySQL credential from approved local index.")
|
|
|
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=180,
|
)
|
|
|
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(28)
|
FONT_SUB = font(20)
|
FONT_MID = font(17)
|
FONT_SMALL = font(14)
|
FONT_TINY = font(12)
|
|
|
def read_csv(name: str) -> pd.DataFrame:
|
return pd.read_csv(SOURCE_ROOT / name, encoding="utf-8-sig")
|
|
|
def norm_time(value) -> str:
|
if pd.isna(value):
|
return ""
|
if hasattr(value, "total_seconds"):
|
seconds = int(value.total_seconds())
|
h, rem = divmod(seconds, 3600)
|
m, s = divmod(rem, 60)
|
return f"{h:02d}:{m:02d}:{s:02d}"
|
text = str(value).strip()
|
if "days" in text and " " in text:
|
text = text.split()[-1]
|
if re.fullmatch(r"\d{2}:\d{2}$", text):
|
return f"{text}:00"
|
if re.fullmatch(r"\d{2}:\d{2}:\d{2}(\.\d+)?", text):
|
return text[:8]
|
return text
|
|
|
def to_bool_int(series: pd.Series) -> pd.Series:
|
return pd.to_numeric(series, errors="coerce").fillna(0).astype(int)
|
|
|
def y_price(value: float, low: float, high: float, top: int, bottom: int) -> int:
|
if not math.isfinite(value) or high <= low:
|
return (top + bottom) // 2
|
return bottom - int((value - low) / (high - low) * (bottom - top))
|
|
|
def wrap_text(text: str, max_chars: int) -> list[str]:
|
text = "" if pd.isna(text) else str(text)
|
lines: list[str] = []
|
current = ""
|
for ch in text:
|
current += ch
|
if len(current) >= max_chars:
|
lines.append(current)
|
current = ""
|
if current:
|
lines.append(current)
|
return lines or [""]
|
|
|
def pct_text(value) -> str:
|
try:
|
return f"{float(value):.2%}"
|
except Exception:
|
return ""
|
|
|
def money_text(value) -> str:
|
try:
|
return f"{float(value):.6f}"
|
except Exception:
|
return ""
|
|
|
def gate_cn(value: str) -> str:
|
return "市场闸门打开" if value == "MKT_GATE_OPEN_PREV_DAY_UP_3000" else "市场闸门关闭"
|
|
|
def status_cn(value: str) -> str:
|
if value == "PASS":
|
return "候选通过"
|
if value == "FAKE_BREAKOUT_RISK_REVIEW":
|
return "前高假突破风险复核"
|
return str(value)
|
|
|
def draw_text_box(
|
draw: ImageDraw.ImageDraw,
|
box: tuple[int, int, int, int],
|
title: str,
|
lines: Iterable[str],
|
) -> None:
|
x1, y1, x2, y2 = box
|
draw.rounded_rectangle([x1, y1, x2, y2], radius=8, outline="#334155", fill="#ffffff")
|
draw.text((x1 + 18, y1 + 16), title, fill="#111827", font=FONT_TITLE)
|
y = y1 + 58
|
for line in lines:
|
if not line:
|
y += 10
|
continue
|
for wrapped in wrap_text(line, 25):
|
if y > y2 - 22:
|
return
|
draw.text((x1 + 18, y), wrapped, fill="#334155", font=FONT_SMALL)
|
y += 23
|
|
|
def fetch_daily(selected: pd.DataFrame) -> pd.DataFrame:
|
symbols = sorted(selected["symbol"].dropna().unique().tolist())
|
min_date = (pd.to_datetime(selected["signal_trade_date"]).min() - pd.Timedelta(days=230)).strftime("%Y-%m-%d")
|
max_date = (pd.to_datetime(selected["signal_trade_date"]).max() + pd.Timedelta(days=45)).strftime("%Y-%m-%d")
|
parts: list[pd.DataFrame] = []
|
with get_conn() as conn:
|
for i in range(0, len(symbols), 180):
|
chunk = symbols[i : i + 180]
|
ph = ",".join(["%s"] * len(chunk))
|
parts.append(
|
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 ({ph})
|
ORDER BY symbol, trade_date
|
""",
|
conn,
|
params=[min_date, max_date, *chunk],
|
)
|
)
|
daily = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame()
|
if daily.empty:
|
return daily
|
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)
|
daily["ma5"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(5, min_periods=1).mean())
|
daily["ma20"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(20, min_periods=1).mean())
|
daily["ma60"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(60, min_periods=1).mean())
|
return daily
|
|
|
def fetch_minute(orders: pd.DataFrame) -> pd.DataFrame:
|
date_symbol_map: dict[str, set[str]] = {}
|
for _, row in orders.iterrows():
|
date_symbol_map.setdefault(str(row["trade_date"]), set()).add(str(row["symbol"]))
|
parts: list[pd.DataFrame] = []
|
with get_conn() as conn:
|
for idx, trade_date in enumerate(sorted(date_symbol_map), start=1):
|
symbols = sorted(date_symbol_map[trade_date])
|
ph = ",".join(["%s"] * len(symbols))
|
parts.append(
|
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 trade_date = %s AND symbol IN ({ph})
|
ORDER BY symbol, trade_date, trade_time
|
""",
|
conn,
|
params=[trade_date, *symbols],
|
)
|
)
|
if idx % 100 == 0:
|
print(f"minute query progress: {idx}/{len(date_symbol_map)} dates", flush=True)
|
minute = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame()
|
if minute.empty:
|
return minute
|
minute["trade_date"] = pd.to_datetime(minute["trade_date"]).dt.strftime("%Y-%m-%d")
|
minute["trade_time"] = minute["trade_time"].map(norm_time)
|
for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
|
minute[col] = pd.to_numeric(minute[col], errors="coerce")
|
return minute
|
|
|
def daily_window(daily: pd.DataFrame, symbol: str, signal_date: str) -> pd.DataFrame:
|
sd = pd.Timestamp(signal_date)
|
rows = daily[daily["symbol"] == symbol].sort_values("trade_date").reset_index(drop=True)
|
hits = rows.index[rows["trade_date"] == sd].tolist()
|
if not hits:
|
return pd.DataFrame()
|
pos = hits[0]
|
start = max(0, pos - 90)
|
end = min(len(rows), pos + 11)
|
return rows.iloc[start:end].copy().reset_index(drop=True)
|
|
|
def draw_daily_enhanced_chart(window: pd.DataFrame, cand: pd.Series, out_path: Path) -> None:
|
w, h = 1680, 980
|
img = Image.new("RGB", (w, h), "#fbfbf7")
|
d = ImageDraw.Draw(img)
|
d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
|
|
signal_date = str(cand["signal_trade_date"])
|
entry_date = str(cand["entry_trade_date"])
|
title = f"选股日K图增强审阅：{cand['symbol']} 信号日 {signal_date}"
|
subtitle = "窗口：信号日前90个交易日 + 信号日后10个交易日；信号后部分只作 audit_view，不参与当时选股。"
|
d.text((32, 22), title, fill="#111827", font=FONT_TITLE)
|
d.text((32, 62), subtitle, fill="#7f1d1d", font=FONT_MID)
|
|
plot_left, plot_top, plot_right, plot_bottom = 80, 118, 1190, 640
|
vol_top, vol_bottom = 700, 850
|
note_left, note_top = 1220, 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")
|
|
win = window.copy().reset_index(drop=True)
|
win["trade_date_str"] = win["trade_date"].dt.strftime("%Y-%m-%d")
|
price_low = float(win["low_price"].min()) * 0.98
|
price_high = float(win["high_price"].max()) * 1.02
|
max_vol = max(float(win["volume"].max()), 1.0)
|
n = len(win)
|
gap = (plot_right - plot_left) / max(n, 1)
|
body_w = max(3, int(gap * 0.58))
|
signal_positions = win.index[win["trade_date_str"] == signal_date].tolist()
|
entry_positions = win.index[win["trade_date_str"] == entry_date].tolist()
|
signal_pos = signal_positions[0] if signal_positions else None
|
|
if signal_pos is not None and signal_pos + 1 < n:
|
x1 = int(plot_left + gap * (signal_pos + 1))
|
d.rectangle([x1, plot_top, plot_right, vol_bottom], fill="#fff7ed")
|
d.text((x1 + 8, plot_top + 8), "信号后10日审阅区", fill="#9a3412", font=FONT_SMALL)
|
|
for i in range(5):
|
price = price_low + (price_high - price_low) * i / 4
|
y = y_price(price, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y, plot_right, y], fill="#e2e8f0")
|
d.text((18, y - 8), f"{price:.2f}", fill="#64748b", font=FONT_SMALL)
|
|
ma_points = {"ma5": [], "ma20": [], "ma60": []}
|
ma_colors = {"ma5": "#2563eb", "ma20": "#f59e0b", "ma60": "#7c3aed"}
|
for i, row in win.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, 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 = y_price(op, price_low, price_high, plot_top, plot_bottom)
|
y2 = 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)
|
for ma in ma_points:
|
if pd.notna(row[ma]):
|
ma_points[ma].append((cx, y_price(float(row[ma]), price_low, price_high, plot_top, plot_bottom)))
|
if row["trade_date_str"] == signal_date:
|
d.line([cx, plot_top, cx, vol_bottom], fill="#0f172a", width=3)
|
d.text((cx + 5, plot_top + 30), "信号日", fill="#0f172a", font=FONT_SMALL)
|
if row["trade_date_str"] == entry_date:
|
d.line([cx, plot_top, cx, vol_bottom], fill="#b91c1c", width=2)
|
d.text((cx + 5, plot_top + 55), "买入日", fill="#b91c1c", font=FONT_SMALL)
|
if i % max(1, n // 9) == 0:
|
d.text((cx - 24, vol_bottom + 8), row["trade_date_str"][5:], fill="#64748b", font=FONT_SMALL)
|
|
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 += 75
|
|
lines = [
|
f"候选排名：{cand['candidate_rank']}",
|
f"候选状态：{status_cn(cand['candidate_status'])}",
|
f"市场闸门：{gate_cn(cand['market_gate_status'])}",
|
f"上涨/下跌家数：{cand.get('up_count', '')}/{cand.get('down_count', '')}",
|
f"量比：{float(cand['volume_ratio']):.2f}，长上影：{float(cand['upper_shadow_pct']):.2f}%",
|
f"近30日涨停：{cand.get('last_limitup_date', '')}",
|
f"前高引用：{cand.get('prev60_high_ref_date', '')}",
|
f"前高量能通过：{cand.get('prev_high_volume_pass_flag', '')}",
|
"",
|
"重要边界：信号日后10天仅帮助同事看后续走势，不能反推当时是否入池、买入或卖出。",
|
"原决策仍以已审核 decision_view、订单账本和 scope 表为准。",
|
]
|
draw_text_box(d, (note_left, note_top, 1640, 850), "图上说明", lines)
|
footer = f"来源：{SOURCE_RUN_ID}；本图为增强 audit_view，不新增交易结论。"
|
d.text((32, 916), footer, fill="#334155", font=FONT_MID)
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
img.save(out_path)
|
|
|
def find_event_index(df: pd.DataFrame, time_str: str) -> int:
|
if df.empty:
|
return 0
|
hits = df.index[df["trade_time"] == time_str].tolist()
|
if hits:
|
return hits[0]
|
before = df.index[df["trade_time"] <= time_str].tolist()
|
if before:
|
return before[-1]
|
return 0
|
|
|
def draw_minute_line_chart(df: pd.DataFrame, order: pd.Series, action: str, out_path: Path) -> None:
|
w, h = 1540, 860
|
img = Image.new("RGB", (w, h), "#fbfbf7")
|
d = ImageDraw.Draw(img)
|
d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
|
action_cn = "买入" if action == "BUY" else "卖出"
|
point_cn = "买点" if action == "BUY" else "卖点"
|
color = "#b91c1c" if action == "BUY" else "#7c3aed"
|
title = f"{action_cn}日整日分钟行情折线图：{order['symbol']} {order['trade_date']}"
|
d.text((32, 24), title, fill="#111827", font=FONT_TITLE)
|
d.text((32, 60), f"标记：{point_cn} {order['trade_time']} @ {float(order['price']):.2f}；整日走势只作人工审阅，不改写原裁决。", fill=color, font=FONT_MID)
|
|
plot_left, plot_top, plot_right, plot_bottom = 80, 110, 1080, 590
|
vol_top, vol_bottom = 645, 790
|
note_left, note_top = 1115, 112
|
d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8")
|
d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8")
|
|
day = df.copy().sort_values("trade_time").reset_index(drop=True)
|
event_price = float(order["price"])
|
if day.empty:
|
d.text((plot_left + 80, plot_top + 180), "分钟线数据缺失", fill="#b91c1c", font=FONT_TITLE)
|
img.save(out_path)
|
return
|
price_low = min(float(day["low_price"].min()), event_price) * 0.998
|
price_high = max(float(day["high_price"].max()), event_price) * 1.002
|
max_vol = max(float(day["volume"].max()), 1.0)
|
n = len(day)
|
gap = (plot_right - plot_left) / max(n - 1, 1)
|
points: list[tuple[int, int]] = []
|
for i, row in day.iterrows():
|
cx = int(plot_left + gap * i)
|
cy = y_price(float(row["close_price"]), price_low, price_high, plot_top, plot_bottom)
|
points.append((cx, cy))
|
vh = int(float(row["volume"]) / max_vol * (vol_bottom - vol_top))
|
d.line([cx, vol_bottom, cx, vol_bottom - vh], fill="#cbd5e1", width=1)
|
if i % max(1, n // 7) == 0:
|
d.text((cx - 20, vol_bottom + 8), str(row["trade_time"])[:5], fill="#64748b", font=FONT_SMALL)
|
if len(points) > 1:
|
d.line(points, fill="#0f766e", width=3)
|
|
for i in range(5):
|
price = price_low + (price_high - price_low) * i / 4
|
y = y_price(price, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y, plot_right, y], fill="#e2e8f0")
|
d.text((18, y - 8), f"{price:.2f}", fill="#64748b", font=FONT_SMALL)
|
|
event_idx = find_event_index(day, str(order["trade_time"]))
|
event_x = int(plot_left + gap * event_idx)
|
event_y = y_price(event_price, price_low, price_high, plot_top, plot_bottom)
|
d.line([event_x, plot_top, event_x, vol_bottom], fill=color, width=3)
|
d.ellipse([event_x - 8, event_y - 8, event_x + 8, event_y + 8], fill=color)
|
d.text((min(event_x + 8, plot_right - 120), max(plot_top + 8, event_y - 42)), f"{point_cn} {order['trade_time'][:5]}", fill=color, font=FONT_SUB)
|
|
open_price = float(day.iloc[0]["open_price"])
|
open_y = y_price(open_price, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, open_y, plot_right, open_y], fill="#334155", width=1)
|
d.text((plot_right - 95, open_y - 16), f"开盘 {open_price:.2f}", fill="#334155", font=FONT_SMALL)
|
|
lines = [
|
f"动作：{action_cn}",
|
f"订单：{order['order_id']}",
|
f"时间：{order['trade_time']}",
|
f"价格：{float(order['price']):.2f}",
|
f"仓位变化：{float(order['position_delta_pct']):.2%}",
|
f"证据来源：{order.get('evidence_image_path', '')}",
|
"",
|
"本图新增整日折线视角，目的是让同事看清买卖点处于全天什么位置。",
|
"原始买卖裁决仍以来源包的1分钟K图、订单账本和已审核scope为准。",
|
"本图不新增收益率或策略有效性结论。",
|
]
|
draw_text_box(d, (note_left, note_top, 1500, 790), f"{action_cn}说明", lines)
|
d.text((32, 820), f"来源：{SOURCE_RUN_ID}；增强包：{RUN_ID}", fill="#334155", font=FONT_MID)
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
img.save(out_path)
|
|
|
def local_link_targets(markdown_path: Path) -> list[dict]:
|
text = markdown_path.read_text(encoding="utf-8")
|
targets = []
|
for match in re.finditer(r"!?\[[^\]]*\]\(([^)]+)\)", text):
|
raw = match.group(1).strip()
|
if not raw or raw.startswith("#") or "://" in raw:
|
continue
|
no_anchor = raw.split("#", 1)[0]
|
if not no_anchor:
|
continue
|
target = (markdown_path.parent / no_anchor).resolve()
|
targets.append(
|
{
|
"markdown_path": markdown_path.relative_to(PACKAGE_ROOT).as_posix(),
|
"link": raw,
|
"target_exists": target.exists(),
|
"target_path": str(target),
|
}
|
)
|
return targets
|
|
|
def write_manifest() -> None:
|
files = []
|
for path in sorted(PACKAGE_ROOT.rglob("*")):
|
if not path.is_file():
|
continue
|
if path.name == "manifest.json":
|
continue
|
rel = path.relative_to(PACKAGE_ROOT).as_posix()
|
files.append(
|
{
|
"path": rel,
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
manifest = {
|
"run_id": RUN_ID,
|
"task_id": TASK_ID,
|
"generated_at": GENERATED_AT,
|
"source_run_id": SOURCE_RUN_ID,
|
"source_guide_run_id": GUIDE_RUN_ID,
|
"manifest_self_hash_excluded": True,
|
"file_count": len(files),
|
"files": files,
|
}
|
(PACKAGE_ROOT / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
def write_source_manifest() -> None:
|
source_files = [
|
SOURCE_ROOT / "full_return_stat_case_scope.csv",
|
SOURCE_ROOT / "full_selected_candidate_ledger.csv",
|
SOURCE_ROOT / "order_ledger.csv",
|
SOURCE_ROOT / "position_lot_ledger.csv",
|
SOURCE_ROOT / "case_summary.csv",
|
SOURCE_ROOT / "full_return_stat_summary.json",
|
SOURCE_ROOT / "manifest.json",
|
GUIDE_ROOT / "strict_closed_234_case_index.csv",
|
GUIDE_ROOT / "strict_closed_234_file_map.csv",
|
GUIDE_ROOT / "manifest.json",
|
]
|
rows = []
|
for path in source_files:
|
rows.append(
|
{
|
"source_path": path.relative_to(PROJECT_ROOT).as_posix(),
|
"exists": path.exists(),
|
"size": path.stat().st_size if path.exists() else "",
|
"sha256": sha256_file(path) if path.exists() else "",
|
}
|
)
|
pd.DataFrame(rows).to_csv(PACKAGE_ROOT / "source_artifact_manifest.csv", index=False, encoding="utf-8-sig")
|
|
|
def build_case_board(
|
case_id: str,
|
case_scope: pd.Series,
|
case_summary: pd.Series | None,
|
candidate_rows: pd.DataFrame,
|
buy_rows: pd.DataFrame,
|
sell_rows: pd.DataFrame,
|
chart_rows: list[dict],
|
) -> None:
|
case_dir = PACKAGE_ROOT / "cases" / case_id
|
source_case_dir = SOURCE_ROOT / "cases" / case_id
|
lines = [
|
f"# {case_id} 增强图片审核板",
|
"",
|
f"- 增强包：`{RUN_ID}`",
|
f"- 来源全量包：`{SOURCE_RUN_ID}`",
|
f"- 来源原图片板：[打开原 case_image_board.md](../../../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md)",
|
f"- 当前收益口径：`PRIMARY_STRICT_CLOSED_CASE`",
|
f"- RETURN_STAT_READY：`false`",
|
"",
|
"本板是给同事看的图片增强入口，不重跑候选池、买卖裁决或账本，不新增收益结论。",
|
"选股日K图包含信号日后10个交易日的事后审阅区；该区域只能帮助复盘，不能反推当时选股或买卖决策。",
|
"",
|
"## 1. 选股日K图：信号日前90天 + 后10天",
|
"",
|
]
|
for _, cand in candidate_rows.sort_values("candidate_rank").iterrows():
|
rows = [r for r in chart_rows if r["case_id"] == case_id and r["chart_role"] == "candidate_daily_90pre_10post_audit_view" and r["symbol"] == cand["symbol"]]
|
if rows:
|
rel = Path(rows[0]["path"]).relative_to(f"cases/{case_id}").as_posix()
|
lines.extend(
|
[
|
f"### 候选排名 {int(cand['candidate_rank'])}：{cand['symbol']}",
|
"",
|
f"![{cand['symbol']}]({rel})",
|
"",
|
f"- 信号日：`{cand['signal_trade_date']}`；买入日：`{cand['entry_trade_date']}`",
|
f"- 候选状态：`{cand['candidate_status']}`；量比 `{float(cand['volume_ratio']):.2f}`；长上影 `{float(cand['upper_shadow_pct']):.2f}%`",
|
"",
|
]
|
)
|
lines.extend(["## 2. 买入日整日分钟行情折线图", ""])
|
if buy_rows.empty:
|
lines.append("- 本案例无 BUY 订单。")
|
for _, order in buy_rows.sort_values(["trade_date", "trade_time", "symbol"]).iterrows():
|
rows = [r for r in chart_rows if r["event_id"] == order["order_id"] and r["chart_role"] == "entry_full_day_minute_line_audit_view"]
|
if rows:
|
rel = Path(rows[0]["path"]).relative_to(f"cases/{case_id}").as_posix()
|
lines.extend(
|
[
|
f"### 买点：{order['symbol']} {order['trade_date']} {order['trade_time']}",
|
"",
|
f"![买点 {order['symbol']}]({rel})",
|
"",
|
f"- 买入价格：`{float(order['price']):.4f}`；仓位变化：`{float(order['position_delta_pct']):.2%}`",
|
f"- 来源订单：`{order['order_id']}`",
|
"",
|
]
|
)
|
lines.extend(["## 3. 卖出日整日分钟行情折线图", ""])
|
if sell_rows.empty:
|
lines.append("- 本案例无 SELL 订单。")
|
for _, order in sell_rows.sort_values(["trade_date", "trade_time", "symbol"]).iterrows():
|
rows = [r for r in chart_rows if r["event_id"] == order["order_id"] and r["chart_role"] == "exit_full_day_minute_line_audit_view"]
|
if rows:
|
rel = Path(rows[0]["path"]).relative_to(f"cases/{case_id}").as_posix()
|
lines.extend(
|
[
|
f"### 卖点：{order['symbol']} {order['trade_date']} {order['trade_time']}",
|
"",
|
f"![卖点 {order['symbol']}]({rel})",
|
"",
|
f"- 卖出价格：`{float(order['price']):.4f}`；仓位变化：`{float(order['position_delta_pct']):.2%}`",
|
f"- 来源订单：`{order['order_id']}`；来源 lot：`{order.get('source_lot_id', '')}`",
|
"",
|
]
|
)
|
ret = "" if case_summary is None else money_text(case_summary.get("account_return_closed_lots", ""))
|
success = "" if case_scope is None else case_scope.get("primary_case_success_flag", "")
|
lines.extend(
|
[
|
"## 4. 账本和收益口径追溯",
|
"",
|
f"- 主口径成功标记：`{success}`",
|
f"- 闭合 lot 账户贡献合计：`{ret}`",
|
f"- 来源 case scope：`{SOURCE_RUN_ID}/full_return_stat_case_scope.csv`",
|
f"- 来源订单账本：`{SOURCE_RUN_ID}/order_ledger.csv`",
|
f"- 来源 lot 账本：`{SOURCE_RUN_ID}/position_lot_ledger.csv`",
|
f"- 来源账户流水：`{SOURCE_RUN_ID}/daily_account_ledger.csv`",
|
"",
|
"禁止过读：本 case 属于 234 个严格闭合主口径子集；不得把该子集直接说成完整 743 个 entry date 的无边界表现。",
|
]
|
)
|
(case_dir / "case_image_board.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
def build_root_docs(case_index: pd.DataFrame, chart_rows: list[dict]) -> None:
|
readme = [
|
"# 无忌交易系统 234 个严格闭合案例增强图片包",
|
"",
|
f"- 生成时间：`{GENERATED_AT}`",
|
f"- 增强包：`{RUN_ID}`",
|
f"- 来源全量包：`{SOURCE_RUN_ID}`",
|
f"- 来源说明包：`{GUIDE_RUN_ID}`",
|
"",
|
"## 这个包解决什么问题",
|
"",
|
"同事反馈原 `case_image_board.md` 还需要三类更方便的图：",
|
"",
|
"1. 选股日K图按信号日前90个交易日 + 信号日后10个交易日展示。",
|
"2. 增加买入日整日分钟行情折线图，并清楚标记买点。",
|
"3. 增加卖出日整日分钟行情折线图，并清楚标记卖点。",
|
"",
|
"本包只增强图片阅读入口，不重跑候选池、买卖裁决或账本，不新增收益结论。",
|
"信号日后的10个交易日和整日分钟折线图均属于 `audit_view`，用于人工复核，不得反推当时决策。",
|
"",
|
"## 怎么看",
|
"",
|
"1. 先打开本目录 `case_image_board.md`，从234个 case 列表进入单个 case。",
|
"2. 单个 case 先看增强选股日K图，再看买入日整日分钟折线图，最后看卖出日整日分钟折线图。",
|
"3. 需要核对数字时回到来源全量包的 `order_ledger.csv`、`position_lot_ledger.csv`、`daily_account_ledger.csv` 和 `full_return_stat_case_scope.csv`。",
|
"",
|
"## 关键边界",
|
"",
|
"- 当前范围仍是 `PRIMARY_STRICT_CLOSED_CASE` 的 234 个严格闭合 case。",
|
"- `RETURN_STAT_READY=false` 保留。",
|
"- 不得把 234 个 case 写成完整 743 个 entry date 的无边界收益或成功率。",
|
"- 不得声称无忌 baseline 策略有效性已经被证明。",
|
]
|
(PACKAGE_ROOT / "README.md").write_text("\n".join(readme) + "\n", encoding="utf-8")
|
|
lines = [
|
"# 234 个严格闭合案例增强图片总入口",
|
"",
|
f"- 增强包：`{RUN_ID}`",
|
f"- 来源全量包：`{SOURCE_RUN_ID}`",
|
f"- 增强图数量：`{len(chart_rows)}`",
|
"",
|
"## 入口列表",
|
"",
|
"| case_id | 入场日 | 主口径收益贡献 | 买入图 | 卖出图 | 图片板 |",
|
"|---|---:|---:|---:|---:|---|",
|
]
|
for _, row in case_index.sort_values("entry_trade_date").iterrows():
|
lines.append(
|
f"| `{row['case_id']}` | {row['entry_trade_date']} | {money_text(row['account_return_closed_lots'])} | {int(row['buy_chart_count'])} | {int(row['sell_chart_count'])} | [打开](cases/{row['case_id']}/case_image_board.md) |"
|
)
|
(PACKAGE_ROOT / "case_image_board.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
def main() -> None:
|
PACKAGE_ROOT.mkdir(parents=True, exist_ok=True)
|
(PACKAGE_ROOT / "cases").mkdir(parents=True, exist_ok=True)
|
write_source_manifest()
|
|
scope = read_csv("full_return_stat_case_scope.csv")
|
strict = scope[
|
(scope["case_scope_status"] == "PRIMARY_STRICT_CLOSED_CASE")
|
& (to_bool_int(scope["primary_strict_closed_case_flag"]) == 1)
|
].copy()
|
strict_ids = set(strict["case_id"])
|
selected = read_csv("full_selected_candidate_ledger.csv")
|
selected = selected[selected["case_id"].isin(strict_ids)].copy()
|
selected["signal_trade_date"] = pd.to_datetime(selected["signal_trade_date"]).dt.strftime("%Y-%m-%d")
|
selected["entry_trade_date"] = pd.to_datetime(selected["entry_trade_date"]).dt.strftime("%Y-%m-%d")
|
selected["candidate_rank"] = pd.to_numeric(selected["candidate_rank"], errors="coerce")
|
orders = read_csv("order_ledger.csv")
|
orders = orders[orders["case_id"].isin(strict_ids) & orders["action"].isin(["BUY", "SELL"])].copy()
|
orders["trade_date"] = pd.to_datetime(orders["trade_date"]).dt.strftime("%Y-%m-%d")
|
orders["trade_time"] = orders["trade_time"].map(norm_time)
|
orders["price"] = pd.to_numeric(orders["price"], errors="coerce")
|
orders["position_delta_pct"] = pd.to_numeric(orders["position_delta_pct"], errors="coerce")
|
case_summary = read_csv("case_summary.csv")
|
case_summary = case_summary[case_summary["case_id"].isin(strict_ids)].copy()
|
|
print(f"strict cases={len(strict)}, selected={len(selected)}, orders={len(orders)}", flush=True)
|
daily = fetch_daily(selected)
|
minute = fetch_minute(orders)
|
print(f"daily rows={len(daily)}, minute rows={len(minute)}", flush=True)
|
|
chart_rows: list[dict] = []
|
missing_rows: list[dict] = []
|
case_index_rows: list[dict] = []
|
|
for seq, (_, case_scope) in enumerate(strict.sort_values("entry_trade_date").iterrows(), start=1):
|
case_id = case_scope["case_id"]
|
case_dir = PACKAGE_ROOT / "cases" / case_id
|
img_dir = case_dir / "img"
|
img_dir.mkdir(parents=True, exist_ok=True)
|
candidate_rows = selected[selected["case_id"] == case_id].copy()
|
buy_rows = orders[(orders["case_id"] == case_id) & (orders["action"] == "BUY")].copy()
|
sell_rows = orders[(orders["case_id"] == case_id) & (orders["action"] == "SELL")].copy()
|
summary_rows = case_summary[case_summary["case_id"] == case_id]
|
summary_row = summary_rows.iloc[0] if not summary_rows.empty else None
|
|
for _, cand in candidate_rows.sort_values("candidate_rank").iterrows():
|
window = daily_window(daily, str(cand["symbol"]), str(cand["signal_trade_date"]))
|
out = img_dir / f"01_candidate_daily_90pre_10post_rank{int(cand['candidate_rank']):02d}_{str(cand['symbol']).replace('.', '_')}_{str(cand['signal_trade_date']).replace('-', '')}.png"
|
if window.empty:
|
missing_rows.append({"case_id": case_id, "symbol": cand["symbol"], "kind": "daily", "date": cand["signal_trade_date"]})
|
continue
|
draw_daily_enhanced_chart(window, cand, out)
|
chart_rows.append(
|
{
|
"case_id": case_id,
|
"symbol": cand["symbol"],
|
"event_id": cand["candidate_id"],
|
"chart_role": "candidate_daily_90pre_10post_audit_view",
|
"action": "CANDIDATE",
|
"trade_date": cand["signal_trade_date"],
|
"trade_time": "close",
|
"path": out.relative_to(PACKAGE_ROOT).as_posix(),
|
"source_order_id": "",
|
"source_candidate_id": cand["candidate_id"],
|
"status": "PASS",
|
"note": "信号日前90个交易日 + 信号日后10个交易日；信号后区间仅作audit_view。",
|
"size": out.stat().st_size,
|
"sha256": sha256_file(out),
|
}
|
)
|
|
for _, order in pd.concat([buy_rows, sell_rows], ignore_index=True).sort_values(["trade_date", "trade_time", "symbol", "order_id"]).iterrows():
|
action = str(order["action"])
|
role = "entry_full_day_minute_line_audit_view" if action == "BUY" else "exit_full_day_minute_line_audit_view"
|
prefix = "07_entry_full_day_minute_line" if action == "BUY" else "08_exit_full_day_minute_line"
|
day = minute[
|
(minute["symbol"] == order["symbol"])
|
& (minute["trade_date"] == order["trade_date"])
|
].copy()
|
safe_order = re.sub(r"[^A-Za-z0-9]+", "_", str(order["order_id"]))[-36:]
|
out = img_dir / f"{prefix}_{str(order['symbol']).replace('.', '_')}_{str(order['trade_date']).replace('-', '')}_{str(order['trade_time']).replace(':', '')}_{safe_order}.png"
|
if day.empty:
|
missing_rows.append({"case_id": case_id, "symbol": order["symbol"], "kind": action.lower(), "date": order["trade_date"]})
|
continue
|
draw_minute_line_chart(day, order, action, out)
|
chart_rows.append(
|
{
|
"case_id": case_id,
|
"symbol": order["symbol"],
|
"event_id": order["order_id"],
|
"chart_role": role,
|
"action": action,
|
"trade_date": order["trade_date"],
|
"trade_time": order["trade_time"],
|
"path": out.relative_to(PACKAGE_ROOT).as_posix(),
|
"source_order_id": order["order_id"],
|
"source_candidate_id": order.get("candidate_id", ""),
|
"status": "PASS",
|
"note": "整日分钟行情折线图；仅作audit_view，标记来源订单买卖点。",
|
"size": out.stat().st_size,
|
"sha256": sha256_file(out),
|
}
|
)
|
|
build_case_board(case_id, case_scope, summary_row, candidate_rows, buy_rows, sell_rows, chart_rows)
|
case_chart_rows = [r for r in chart_rows if r["case_id"] == case_id]
|
(case_dir / "image_manifest.csv").write_text(
|
pd.DataFrame(case_chart_rows).to_csv(index=False),
|
encoding="utf-8-sig",
|
)
|
case_index_rows.append(
|
{
|
"case_id": case_id,
|
"entry_trade_date": case_scope["entry_trade_date"],
|
"signal_trade_date": case_scope["signal_trade_date"],
|
"account_return_closed_lots": case_scope["account_return_closed_lots"],
|
"primary_case_success_flag": case_scope["primary_case_success_flag"],
|
"candidate_chart_count": len([r for r in case_chart_rows if r["chart_role"] == "candidate_daily_90pre_10post_audit_view"]),
|
"buy_chart_count": len([r for r in case_chart_rows if r["chart_role"] == "entry_full_day_minute_line_audit_view"]),
|
"sell_chart_count": len([r for r in case_chart_rows if r["chart_role"] == "exit_full_day_minute_line_audit_view"]),
|
"enhanced_case_image_board": f"cases/{case_id}/case_image_board.md",
|
"source_case_image_board": f"{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md",
|
}
|
)
|
if seq % 25 == 0:
|
print(f"case progress: {seq}/{len(strict)}", flush=True)
|
|
chart_df = pd.DataFrame(chart_rows)
|
chart_df.to_csv(PACKAGE_ROOT / "chart_evidence_audit.csv", index=False, encoding="utf-8-sig")
|
missing_df = pd.DataFrame(missing_rows)
|
missing_df.to_csv(PACKAGE_ROOT / "missing_chart_inputs.csv", index=False, encoding="utf-8-sig")
|
case_index = pd.DataFrame(case_index_rows).sort_values("entry_trade_date")
|
case_index.to_csv(PACKAGE_ROOT / "enhanced_case_index.csv", index=False, encoding="utf-8-sig")
|
build_root_docs(case_index, chart_rows)
|
|
link_rows = []
|
for md in sorted(PACKAGE_ROOT.rglob("*.md")):
|
link_rows.extend(local_link_targets(md))
|
link_df = pd.DataFrame(link_rows)
|
link_df.to_csv(PACKAGE_ROOT / "link_evidence_audit.csv", index=False, encoding="utf-8-sig")
|
|
expected_daily = len(strict) * 5
|
expected_buy = len(orders[orders["action"] == "BUY"])
|
expected_sell = len(orders[orders["action"] == "SELL"])
|
self_checks = [
|
("SOURCE_FULL_RUN_EXISTS", SOURCE_ROOT.exists(), str(SOURCE_ROOT)),
|
("SOURCE_REVIEW_GUIDE_EXISTS", GUIDE_ROOT.exists(), str(GUIDE_ROOT)),
|
("STRICT_CASE_COUNT_234", len(strict) == 234, f"strict_cases={len(strict)}"),
|
("SELECTED_CANDIDATE_ROWS_1170", len(selected) == expected_daily, f"selected={len(selected)}, expected={expected_daily}"),
|
("DAILY_AUDIT_CHARTS_COMPLETE", len(chart_df[chart_df["chart_role"] == "candidate_daily_90pre_10post_audit_view"]) == expected_daily, f"actual={len(chart_df[chart_df['chart_role'] == 'candidate_daily_90pre_10post_audit_view'])}, expected={expected_daily}"),
|
("BUY_FULL_DAY_LINE_CHARTS_COMPLETE", len(chart_df[chart_df["chart_role"] == "entry_full_day_minute_line_audit_view"]) == expected_buy, f"actual={len(chart_df[chart_df['chart_role'] == 'entry_full_day_minute_line_audit_view'])}, expected={expected_buy}"),
|
("SELL_FULL_DAY_LINE_CHARTS_COMPLETE", len(chart_df[chart_df["chart_role"] == "exit_full_day_minute_line_audit_view"]) == expected_sell, f"actual={len(chart_df[chart_df['chart_role'] == 'exit_full_day_minute_line_audit_view'])}, expected={expected_sell}"),
|
("CASE_IMAGE_BOARDS_234", len(list((PACKAGE_ROOT / "cases").glob("*/case_image_board.md"))) == 234, f"boards={len(list((PACKAGE_ROOT / 'cases').glob('*/case_image_board.md')))}"),
|
("MISSING_CHART_INPUTS_ZERO", missing_df.empty, f"missing={len(missing_df)}"),
|
("MARKDOWN_LOCAL_LINKS_REACHABLE", bool(link_df.empty or link_df["target_exists"].all()), f"links={len(link_df)}, missing={0 if link_df.empty else int((~link_df['target_exists']).sum())}"),
|
("RETURN_STAT_READY_FALSE_PRESERVED", True, "enhancement package does not change source return_stat_ready=false"),
|
]
|
self_df = pd.DataFrame(
|
[
|
{
|
"check_id": check_id,
|
"status": "PASS" if passed else "FAIL",
|
"detail": detail,
|
}
|
for check_id, passed, detail in self_checks
|
]
|
)
|
self_df.to_csv(PACKAGE_ROOT / "self_check_items.csv", index=False, encoding="utf-8-sig")
|
fail_count = int((self_df["status"] != "PASS").sum())
|
self_json = {
|
"run_id": RUN_ID,
|
"task_id": TASK_ID,
|
"generated_at": GENERATED_AT,
|
"source_run_id": SOURCE_RUN_ID,
|
"status": "PASS_FOR_STRICT_CLOSED_234_CHART_ENHANCEMENT_REVIEW_READY" if fail_count == 0 else "FAIL",
|
"pass_count": int((self_df["status"] == "PASS").sum()),
|
"fail_count": fail_count,
|
"strict_case_count": len(strict),
|
"daily_audit_chart_count": int((chart_df["chart_role"] == "candidate_daily_90pre_10post_audit_view").sum()),
|
"buy_full_day_line_chart_count": int((chart_df["chart_role"] == "entry_full_day_minute_line_audit_view").sum()),
|
"sell_full_day_line_chart_count": int((chart_df["chart_role"] == "exit_full_day_minute_line_audit_view").sum()),
|
"return_stat_ready": False,
|
"full_baseline_conclusion_allowed": False,
|
}
|
(PACKAGE_ROOT / "self_check.json").write_text(json.dumps(self_json, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
(PACKAGE_ROOT / "self_check.md").write_text(
|
"\n".join(
|
[
|
"# 自检结果",
|
"",
|
f"- 状态：`{self_json['status']}`",
|
f"- PASS：`{self_json['pass_count']}`",
|
f"- FAIL：`{self_json['fail_count']}`",
|
f"- 选股增强日K图：`{self_json['daily_audit_chart_count']}`",
|
f"- 买入日整日分钟折线图：`{self_json['buy_full_day_line_chart_count']}`",
|
f"- 卖出日整日分钟折线图：`{self_json['sell_full_day_line_chart_count']}`",
|
"",
|
"本包只增强人工图片入口，不改变来源全量包的收益口径和 `RETURN_STAT_READY=false`。",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
write_manifest()
|
print(json.dumps(self_json, ensure_ascii=False, indent=2), flush=True)
|
|
|
if __name__ == "__main__":
|
main()
|
