from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import math
|
import os
|
import re
|
import shutil
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
import pandas as pd
|
import pymysql
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-001"
|
TASK_ID = "ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
|
FINAL_RUN_ID = "RUN-ANA-WUJI-V1-FINAL-CONCLUSION-20260610-001"
|
DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-DESIGN-001"
|
SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-EXEC-REREVIEW-003"
|
|
PACKAGE_ROOT = Path(__file__).resolve().parents[1]
|
RESULT_ROOT = PACKAGE_ROOT.parents[0]
|
PROJECT_ROOT = PACKAGE_ROOT.parents[2]
|
SOURCE_ROOT = RESULT_ROOT / SOURCE_RUN_ID
|
FINAL_ROOT = RESULT_ROOT / FINAL_RUN_ID
|
LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md")
|
|
MOJIBAKE_MARKERS = ["????", "???", "�", "ÀíÓÉ", "å¤", "æ", "Ã", "Â", "涓", "鏃", "鐐", "偂"]
|
|
|
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 rel(path: Path) -> str:
|
return path.resolve().relative_to(PACKAGE_ROOT.resolve()).as_posix()
|
|
|
def source_rel(path: Path) -> str:
|
try:
|
return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
|
except ValueError:
|
return str(path)
|
|
|
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(21)
|
FONT_MID = font(17)
|
FONT_SMALL = font(14)
|
FONT_TINY = font(12)
|
|
|
def write_csv(path: Path, rows: list[dict], fields: list[str]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def read_csv(path: Path) -> pd.DataFrame:
|
return pd.read_csv(path, 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 y_price(value: float, lo: float, hi: float, top: int, bottom: int) -> int:
|
if hi <= lo:
|
return (top + bottom) // 2
|
return int(bottom - (value - lo) / (hi - lo) * (bottom - top))
|
|
|
def wrap_text(text: str, width: int) -> list[str]:
|
text = str(text or "")
|
lines: list[str] = []
|
for raw in text.splitlines() or [""]:
|
line = ""
|
units = list(raw)
|
for ch in units:
|
add = 2 if ord(ch) > 127 else 1
|
cur = sum(2 if ord(c) > 127 else 1 for c in line)
|
if cur + add > width and line:
|
lines.append(line)
|
line = ch
|
else:
|
line += ch
|
lines.append(line)
|
return lines
|
|
|
def draw_text_box(draw: ImageDraw.ImageDraw, box: tuple[int, int, int, int], title: str, lines: list[str]) -> None:
|
x1, y1, x2, y2 = box
|
draw.rectangle([x1, y1, x2, y2], fill="#ffffff", outline="#cbd5e1")
|
draw.text((x1 + 16, y1 + 14), title, fill="#111827", font=FONT_SUB)
|
y = y1 + 50
|
for line in lines:
|
if line == "":
|
y += 12
|
continue
|
for wrapped in wrap_text(line, 30):
|
if y > y2 - 24:
|
return
|
draw.text((x1 + 16, y), wrapped, fill="#334155", font=FONT_SMALL)
|
y += 22
|
|
|
def safe_symbol(symbol: str) -> str:
|
return str(symbol).replace(".", "_").replace("/", "_")
|
|
|
def load_sources() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
|
v1_index = read_csv(FINAL_ROOT / "v1_case_readout_index.csv")
|
primary = v1_index[(v1_index["v1_return_scope"] == "V1_PRIMARY_STRICT_CLOSED_CASE") & (pd.to_numeric(v1_index["primary_flag"], errors="coerce") == 1)].copy()
|
orders = read_csv(SOURCE_ROOT / "strict_order_ledger.csv")
|
lots = read_csv(SOURCE_ROOT / "strict_position_lot_ledger.csv")
|
case_summary = read_csv(SOURCE_ROOT / "strict_case_summary.csv")
|
for df in [orders, lots]:
|
for col in ["trade_time", "entry_time", "exit_time"]:
|
if col in df.columns:
|
df[col] = df[col].map(norm_time)
|
return primary, orders, lots, case_summary
|
|
|
def fetch_trade_calendar() -> list[str]:
|
with get_conn() as conn:
|
df = pd.read_sql(
|
"SELECT DISTINCT trade_date FROM a_share_daily_price ORDER BY trade_date",
|
conn,
|
)
|
return pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d").tolist()
|
|
|
def fetch_daily(symbols: list[str], min_date: str, max_date: str) -> pd.DataFrame:
|
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"]).dt.strftime("%Y-%m-%d")
|
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())
|
return daily
|
|
|
def fetch_minute(date_symbols: list[tuple[str, str]]) -> pd.DataFrame:
|
by_date: dict[str, set[str]] = {}
|
for date, symbol in date_symbols:
|
by_date.setdefault(date, set()).add(symbol)
|
parts: list[pd.DataFrame] = []
|
with get_conn() as conn:
|
for idx, date in enumerate(sorted(by_date), start=1):
|
symbols = sorted(by_date[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=[date, *symbols],
|
)
|
)
|
if idx % 100 == 0:
|
print(f"minute query progress: {idx}/{len(by_date)}", 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 build_lifecycles(primary: pd.DataFrame, orders: pd.DataFrame, lots: pd.DataFrame) -> tuple[list[dict], pd.DataFrame, pd.DataFrame]:
|
primary_ids = set(primary["case_id"].astype(str))
|
order_scope = orders[orders["case_id"].astype(str).isin(primary_ids)].copy()
|
lot_scope = lots[lots["case_id"].astype(str).isin(primary_ids)].copy()
|
rows: list[dict] = []
|
seq_by_case_symbol: dict[tuple[str, str], int] = {}
|
for (case_id, symbol), group in order_scope.groupby(["case_id", "symbol"], sort=True):
|
buys = group[group["action"] == "BUY"].copy()
|
sells = group[group["action"] == "SELL"].copy()
|
if buys.empty or sells.empty:
|
continue
|
buys = buys.sort_values(["trade_date", "trade_time", "order_id"])
|
sells = sells.sort_values(["trade_date", "trade_time", "order_id"])
|
key = (str(case_id), str(symbol))
|
seq_by_case_symbol[key] = seq_by_case_symbol.get(key, 0) + 1
|
sequence = seq_by_case_symbol[key]
|
lifecycle_id = f"LC-{case_id}-{safe_symbol(symbol)}-{sequence:02d}"
|
lot_rows = lot_scope[(lot_scope["case_id"].astype(str) == str(case_id)) & (lot_scope["symbol"].astype(str) == str(symbol))]
|
first_buy = buys.iloc[0]
|
last_sell = sells.iloc[-1]
|
manual_ids = sorted(set([str(x) for x in group.get("manual_decision_id", pd.Series(dtype=str)).dropna() if str(x).strip()]))
|
rows.append(
|
{
|
"lifecycle_id": lifecycle_id,
|
"case_id": case_id,
|
"symbol": symbol,
|
"sequence": sequence,
|
"first_buy_trade_date": str(first_buy["trade_date"]),
|
"first_buy_time": str(first_buy["trade_time"]),
|
"first_buy_price": float(first_buy["price"]),
|
"last_sell_trade_date": str(last_sell["trade_date"]),
|
"last_sell_time": str(last_sell["trade_time"]),
|
"last_sell_price": float(last_sell["price"]),
|
"buy_order_count": int(len(buys)),
|
"sell_order_count": int(len(sells)),
|
"lot_count": int(len(lot_rows)),
|
"closed_lot_count": int((lot_rows["lot_status"].astype(str).str.contains("CLOSED", na=False)).sum()),
|
"buy_order_ids": ";".join(buys["order_id"].astype(str).tolist()),
|
"sell_order_ids": ";".join(sells["order_id"].astype(str).tolist()),
|
"lot_ids": ";".join(lot_rows["strict_lot_id"].astype(str).tolist()),
|
"manual_decision_ids": ";".join(manual_ids),
|
"case_image_board": f"../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md",
|
"source_order_ledger": f"../{SOURCE_RUN_ID}/strict_order_ledger.csv",
|
"source_lot_ledger": f"../{SOURCE_RUN_ID}/strict_position_lot_ledger.csv",
|
}
|
)
|
return rows, order_scope, lot_scope
|
|
|
def trade_dates_between(calendar: list[str], start: str, end: str) -> list[str]:
|
return [d for d in calendar if start <= d <= end]
|
|
|
def daily_window(calendar: list[str], symbol_daily: pd.DataFrame, start_date: str, end_date: str) -> tuple[pd.DataFrame, dict]:
|
meta = {
|
"pre_window_insufficient": False,
|
"post_window_insufficient": False,
|
"expected_start_date": "",
|
"expected_end_date": "",
|
}
|
if start_date not in calendar or end_date not in calendar:
|
return pd.DataFrame(), meta
|
start_pos = calendar.index(start_date)
|
end_pos = calendar.index(end_date)
|
expected_start_pos = start_pos - 50
|
expected_end_pos = end_pos + 10
|
meta["pre_window_insufficient"] = expected_start_pos < 0
|
meta["post_window_insufficient"] = expected_end_pos >= len(calendar)
|
start = calendar[max(0, expected_start_pos)]
|
end = calendar[min(len(calendar) - 1, expected_end_pos)]
|
meta["expected_start_date"] = start
|
meta["expected_end_date"] = end
|
window = symbol_daily[(symbol_daily["trade_date"] >= start) & (symbol_daily["trade_date"] <= end)].copy()
|
return window.reset_index(drop=True), meta
|
|
|
def draw_lifecycle_daily(window: pd.DataFrame, lifecycle: dict, orders: pd.DataFrame, out_path: Path) -> None:
|
w, h = 1880, 1060
|
img = Image.new("RGB", (w, h), "#fbfbf7")
|
d = ImageDraw.Draw(img)
|
d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
|
title = f"股票生命周期日线图：{lifecycle['symbol']} {lifecycle['case_id']}"
|
subtitle = "窗口：首买前50个交易日 -> 末卖后10个交易日；同一 case 内同一股票生命周期，不跨 case 合并。"
|
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, 120, 1340, 700
|
vol_top, vol_bottom = 770, 910
|
note_left, note_top = 1370, 120
|
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 window.empty:
|
d.text((plot_left + 180, plot_top + 220), "日线数据缺失", fill="#b91c1c", font=FONT_TITLE)
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
img.save(out_path)
|
return
|
win = window.copy().reset_index(drop=True)
|
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.55))
|
date_to_x: dict[str, int] = {}
|
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)
|
ma5_pts = []
|
for i, row in win.iterrows():
|
cx = int(plot_left + gap * i + gap / 2)
|
date_to_x[str(row["trade_date"])] = cx
|
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 pd.notna(row["ma5"]):
|
ma5_pts.append((cx, y_price(float(row["ma5"]), price_low, price_high, plot_top, plot_bottom)))
|
if i % max(1, n // 10) == 0:
|
d.text((cx - 24, vol_bottom + 8), str(row["trade_date"])[5:], fill="#64748b", font=FONT_SMALL)
|
if len(ma5_pts) > 1:
|
d.line(ma5_pts, fill="#2563eb", width=2)
|
d.text((plot_left + 12, plot_bottom + 12), "MA5", fill="#2563eb", font=FONT_SMALL)
|
for _, order in orders.iterrows():
|
date = str(order["trade_date"])
|
if date not in date_to_x:
|
continue
|
price = float(order["price"])
|
cx = date_to_x[date]
|
cy = y_price(price, price_low, price_high, plot_top, plot_bottom)
|
action = str(order["action"])
|
color = "#b91c1c" if action == "BUY" else "#7c3aed"
|
label = "买" if action == "BUY" else "卖"
|
d.line([cx, plot_top, cx, vol_bottom], fill=color, width=2)
|
d.line([plot_left, cy, plot_right, cy], fill=color, width=1)
|
d.ellipse([cx - 8, cy - 8, cx + 8, cy + 8], fill=color)
|
d.text((cx + 6, max(plot_top + 6, cy - 28)), f"{label}{str(order['trade_time'])[:5]} {price:.2f}", fill=color, font=FONT_SMALL)
|
lines = [
|
f"生命周期：{lifecycle['lifecycle_id']}",
|
f"首买：{lifecycle['first_buy_trade_date']} {lifecycle['first_buy_time']} @ {float(lifecycle['first_buy_price']):.2f}",
|
f"末卖：{lifecycle['last_sell_trade_date']} {lifecycle['last_sell_time']} @ {float(lifecycle['last_sell_price']):.2f}",
|
f"BUY/SELL：{lifecycle['buy_order_count']} / {lifecycle['sell_order_count']}",
|
f"lot：{lifecycle['lot_count']}，闭合：{lifecycle['closed_lot_count']}",
|
"",
|
"读图方式：先看日线整体位置，再沿竖线查看每次买卖点；买卖理由见每日分时图和 story board。",
|
"本图只增强同事阅读，不改变 V1 已审核账本和收益读数。",
|
]
|
draw_text_box(d, (note_left, note_top, 1830, 910), "图上说明", lines)
|
d.text((32, 1000), f"来源审计：{SOURCE_EXEC_AUDIT_ID}；生命周期包设计审计：{DESIGN_AUDIT_ID}", fill="#334155", font=FONT_MID)
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
img.save(out_path)
|
|
|
def draw_intraday_chart(day: pd.DataFrame, lifecycle: dict, day_orders: pd.DataFrame, ma5_price: float | None, out_path: Path) -> None:
|
w, h = 1700, 940
|
img = Image.new("RGB", (w, h), "#fbfbf7")
|
d = ImageDraw.Draw(img)
|
d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
|
trade_date = out_path.stem[-8:]
|
d.text((32, 24), f"生命周期分时图：{lifecycle['symbol']} {trade_date} {lifecycle['case_id']}", fill="#111827", font=FONT_TITLE)
|
d.text((32, 62), "模式：3% / 5% / 8% 阈值线、买卖点、买卖价线、MA5、操作理由；无操作日也保留持有观察理由。", fill="#7f1d1d", font=FONT_MID)
|
plot_left, plot_top, plot_right, plot_bottom = 80, 120, 1160, 640
|
vol_top, vol_bottom = 705, 840
|
note_left, note_top = 1195, 120
|
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 day.empty:
|
d.text((plot_left + 180, plot_top + 200), "分钟线数据缺失", fill="#b91c1c", font=FONT_TITLE)
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
img.save(out_path)
|
return
|
day = day.sort_values("trade_time").reset_index(drop=True)
|
base_price = float(lifecycle["first_buy_price"])
|
ref_prices = [base_price, base_price * 1.03, base_price * 1.05, base_price * 1.08]
|
order_prices = [float(x) for x in day_orders["price"].tolist()] if not day_orders.empty else []
|
raw_low = min([float(day["low_price"].min()), *ref_prices, *order_prices])
|
raw_high = max([float(day["high_price"].max()), *ref_prices, *order_prices])
|
if ma5_price and not math.isnan(ma5_price):
|
raw_low = min(raw_low, ma5_price)
|
raw_high = max(raw_high, ma5_price)
|
price_low, price_high = raw_low * 0.997, raw_high * 1.003
|
max_vol = max(float(day["volume"].max()), 1.0)
|
n = len(day)
|
gap = (plot_right - plot_left) / max(n - 1, 1)
|
points = []
|
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 // 8) == 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)
|
threshold_styles = [(base_price * 1.03, "3%", "#f59e0b"), (base_price * 1.05, "5%", "#dc2626"), (base_price * 1.08, "8%", "#7c3aed")]
|
for price, label, color in threshold_styles:
|
y = y_price(price, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y, plot_right, y], fill=color, width=2)
|
d.text((plot_right + 8, y - 10), f"{label} {price:.2f}", fill=color, font=FONT_SMALL)
|
if ma5_price and not math.isnan(ma5_price):
|
y = y_price(ma5_price, price_low, price_high, plot_top, plot_bottom)
|
d.line([plot_left, y, plot_right, y], fill="#2563eb", width=2)
|
d.text((plot_right + 8, y - 10), f"MA5 {ma5_price:.2f}", fill="#2563eb", font=FONT_SMALL)
|
for _, order in day_orders.iterrows():
|
action = str(order["action"])
|
color = "#b91c1c" if action == "BUY" else "#7c3aed"
|
label = "买点" if action == "BUY" else "卖点"
|
time = str(order["trade_time"])
|
hits = day.index[day["trade_time"] == time].tolist()
|
idx = hits[0] if hits else max([i for i, t in enumerate(day["trade_time"].tolist()) if str(t) <= time] or [0])
|
x = int(plot_left + gap * idx)
|
price = float(order["price"])
|
y = y_price(price, price_low, price_high, plot_top, plot_bottom)
|
d.line([x, plot_top, x, vol_bottom], fill=color, width=3)
|
d.line([plot_left, y, plot_right, y], fill=color, width=1)
|
d.ellipse([x - 8, y - 8, x + 8, y + 8], fill=color)
|
d.text((min(x + 8, plot_right - 135), max(plot_top + 8, y - 30)), f"{label} {time[:5]} {price:.2f}", fill=color, font=FONT_SMALL)
|
if day_orders.empty:
|
reason = "当日处于生命周期区间内，无买卖订单；持有观察，未触发 V1 精准卖点或滚动低吸。"
|
else:
|
fragments = []
|
for _, order in day_orders.iterrows():
|
reason = str(order.get("decision_reason_cn", "")).strip()
|
fragments.append(f"{order['action']} {str(order['trade_time'])[:5]}：{reason[:46]}")
|
reason = "；".join(fragments)
|
lines = [
|
f"生命周期：{lifecycle['lifecycle_id']}",
|
f"参考首买价：{base_price:.2f}",
|
f"当日订单数：{len(day_orders)}",
|
f"操作理由：{reason}",
|
"",
|
"边界：本图用于人工阅读和复核，不新增买卖裁决，不改变 V1 已审核账本。",
|
]
|
draw_text_box(d, (note_left, note_top, 1660, 840), "当日说明", lines)
|
d.text((32, 892), f"来源：{SOURCE_RUN_ID}；审计：{SOURCE_EXEC_AUDIT_ID}", fill="#334155", font=FONT_MID)
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
img.save(out_path)
|
|
|
def audit_markdown_links(md_paths: list[Path]) -> list[dict]:
|
rows: list[dict] = []
|
pattern = re.compile(r"\[[^\]]+\]\(([^)]+)\)")
|
for path in md_paths:
|
text = path.read_text(encoding="utf-8")
|
for match in pattern.finditer(text):
|
target = match.group(1).split("#", 1)[0]
|
if "://" in target or not target:
|
continue
|
target_path = (path.parent / target).resolve()
|
rows.append(
|
{
|
"markdown_path": rel(path),
|
"target": target,
|
"resolved_project_path": source_rel(target_path),
|
"exists": target_path.exists(),
|
}
|
)
|
return rows
|
|
|
def scan_mojibake(paths: list[Path]) -> list[dict]:
|
hits: list[dict] = []
|
for path in paths:
|
if path.suffix.lower() not in [".md", ".csv", ".json", ".txt"]:
|
continue
|
try:
|
text = path.read_text(encoding="utf-8-sig")
|
except UnicodeDecodeError:
|
hits.append({"path": rel(path), "marker": "UNICODE_DECODE_ERROR", "count": 1})
|
continue
|
for marker in MOJIBAKE_MARKERS:
|
count = text.count(marker)
|
if count:
|
hits.append({"path": rel(path), "marker": marker, "count": count})
|
return hits
|
|
|
def write_manifest() -> list[dict]:
|
rows: list[dict] = []
|
for path in sorted(PACKAGE_ROOT.rglob("*")):
|
if not path.is_file():
|
continue
|
rows.append(
|
{
|
"path": rel(path),
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
write_csv(PACKAGE_ROOT / "manifest.csv", rows, ["path", "size", "sha256"])
|
(PACKAGE_ROOT / "manifest.json").write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
|
return rows
|
|
|
def main() -> None:
|
if PACKAGE_ROOT.exists():
|
for child in PACKAGE_ROOT.iterdir():
|
if child.name == "tools":
|
continue
|
if child.is_dir():
|
shutil.rmtree(child)
|
else:
|
child.unlink()
|
(PACKAGE_ROOT / "img").mkdir(parents=True, exist_ok=True)
|
|
primary, orders, lots, case_summary = load_sources()
|
lifecycle_rows, order_scope, lot_scope = build_lifecycles(primary, orders, lots)
|
if len(primary) != 250:
|
raise RuntimeError(f"Expected 250 primary cases, got {len(primary)}")
|
lifecycle_df = pd.DataFrame(lifecycle_rows)
|
calendar = fetch_trade_calendar()
|
if lifecycle_df.empty:
|
raise RuntimeError("No lifecycle rows generated")
|
|
min_date = calendar[max(0, calendar.index(lifecycle_df["first_buy_trade_date"].min()) - 60)]
|
max_date = calendar[min(len(calendar) - 1, calendar.index(lifecycle_df["last_sell_trade_date"].max()) + 12)]
|
daily = fetch_daily(sorted(lifecycle_df["symbol"].unique().tolist()), min_date, max_date)
|
|
date_symbols: list[tuple[str, str]] = []
|
lifecycle_trade_dates: dict[str, list[str]] = {}
|
for row in lifecycle_rows:
|
dates = trade_dates_between(calendar, row["first_buy_trade_date"], row["last_sell_trade_date"])
|
lifecycle_trade_dates[row["lifecycle_id"]] = dates
|
for date in dates:
|
date_symbols.append((date, row["symbol"]))
|
minute = fetch_minute(date_symbols)
|
|
chart_rows: list[dict] = []
|
missing_rows: list[dict] = []
|
md_paths: list[Path] = []
|
lifecycle_index_rows: list[dict] = []
|
primary_case_ids = set(primary["case_id"].astype(str))
|
|
for row in lifecycle_rows:
|
lifecycle_id = row["lifecycle_id"]
|
case_id = str(row["case_id"])
|
symbol = str(row["symbol"])
|
case_dir = PACKAGE_ROOT / "cases" / case_id
|
stock_dir = case_dir / "stocks" / safe_symbol(symbol)
|
img_dir = stock_dir / "img"
|
stock_daily = daily[daily["symbol"] == symbol].copy()
|
win, meta = daily_window(calendar, stock_daily, row["first_buy_trade_date"], row["last_sell_trade_date"])
|
if meta["pre_window_insufficient"]:
|
missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": row["first_buy_trade_date"], "missing_reason": "DAILY_PRE_WINDOW_INSUFFICIENT"})
|
if meta["post_window_insufficient"]:
|
missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": row["last_sell_trade_date"], "missing_reason": "DAILY_POST_WINDOW_INSUFFICIENT"})
|
if win.empty:
|
missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": "", "missing_reason": "DAILY_DATA_EMPTY"})
|
lc_orders = order_scope[(order_scope["case_id"].astype(str) == case_id) & (order_scope["symbol"].astype(str) == symbol)].copy()
|
daily_path = img_dir / "daily_lifecycle_50pre_10post.png"
|
draw_lifecycle_daily(win, row, lc_orders, daily_path)
|
chart_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": "", "path": rel(daily_path), "exists": daily_path.exists(), "sha256": sha256_file(daily_path) if daily_path.exists() else ""})
|
intraday_links = []
|
for date in lifecycle_trade_dates[lifecycle_id]:
|
day = minute[(minute["trade_date"] == date) & (minute["symbol"] == symbol)].copy()
|
day_orders = lc_orders[lc_orders["trade_date"].astype(str) == date].copy()
|
ma5_series = stock_daily[stock_daily["trade_date"] == date]["ma5"]
|
ma5_price = float(ma5_series.iloc[0]) if not ma5_series.empty and pd.notna(ma5_series.iloc[0]) else None
|
if day.empty:
|
missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "intraday_lifecycle_full_day", "trade_date": date, "missing_reason": "MINUTE_DATA_EMPTY"})
|
continue
|
out = img_dir / f"intraday_lifecycle_{date.replace('-', '')}.png"
|
draw_intraday_chart(day, row, day_orders, ma5_price, out)
|
chart_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "intraday_lifecycle_full_day", "trade_date": date, "path": rel(out), "exists": out.exists(), "sha256": sha256_file(out) if out.exists() else ""})
|
intraday_links.append((date, out))
|
board = stock_dir / "stock_lifecycle_board.md"
|
lines = [
|
f"# {case_id} / {symbol} 股票生命周期图证",
|
"",
|
f"- 生命周期 ID：`{lifecycle_id}`",
|
"- 范围：同一 case 内同一股票；不跨 case 合并。",
|
f"- 首买：{row['first_buy_trade_date']} {row['first_buy_time']} @ {float(row['first_buy_price']):.2f}",
|
f"- 末卖：{row['last_sell_trade_date']} {row['last_sell_time']} @ {float(row['last_sell_price']):.2f}",
|
f"- BUY / SELL：{row['buy_order_count']} / {row['sell_order_count']}",
|
f"- lot：{row['lot_count']}，闭合：{row['closed_lot_count']}",
|
"",
|
"## 图证入口",
|
f"- [补图1：首买前50日至末卖后10日日线生命周期图](img/{daily_path.name})",
|
"",
|
"## 补图2：生命周期区间每日分时图",
|
]
|
lines.extend([f"- [{date} 分时生命周期图](img/{path.name})" for date, path in intraday_links])
|
lines.extend(
|
[
|
"",
|
"## 账本追溯",
|
f"- 来源 case 图板：[{case_id} case_image_board](../../../../../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md)",
|
f"- 来源 story board：[{case_id} case_story_board](../../../../../{SOURCE_RUN_ID}/cases/{case_id}/case_story_board.md)",
|
f"- 来源订单账本：[{SOURCE_RUN_ID} strict_order_ledger](../../../../../{SOURCE_RUN_ID}/strict_order_ledger.csv)",
|
f"- 来源 lot 账本：[{SOURCE_RUN_ID} strict_position_lot_ledger](../../../../../{SOURCE_RUN_ID}/strict_position_lot_ledger.csv)",
|
"",
|
"边界：本页只增强同事阅读，不改变 V1 已审核交易规则、人工裁决、账本或收益读数。",
|
]
|
)
|
board.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
md_paths.append(board)
|
lifecycle_index_rows.append({**row, "daily_lifecycle_chart": rel(daily_path), "stock_lifecycle_board": rel(board), "intraday_chart_count": len(intraday_links), "expected_intraday_trade_dates": len(lifecycle_trade_dates[lifecycle_id])})
|
|
for case_id, group in pd.DataFrame(lifecycle_index_rows).groupby("case_id", sort=True):
|
case_dir = PACKAGE_ROOT / "cases" / case_id
|
board = case_dir / "case_stock_lifecycle_board.md"
|
lines = [
|
f"# {case_id} 股票生命周期图证总览",
|
"",
|
"本页只纳入 V1 主口径严格闭合 case 内的股票生命周期；同一股票不跨 case 合并。",
|
"",
|
]
|
for _, r in group.iterrows():
|
lines.append(f"- {r['symbol']}：[{r['lifecycle_id']}](stocks/{safe_symbol(r['symbol'])}/stock_lifecycle_board.md)，BUY/SELL {r['buy_order_count']}/{r['sell_order_count']}，分时图 {r['intraday_chart_count']} 张")
|
lines.append("")
|
lines.append("边界：本页不新增交易结论，不替代来源 V1 图板、订单账本和审计记录。")
|
board.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
md_paths.append(board)
|
|
stock_index_fields = [
|
"lifecycle_id", "case_id", "symbol", "sequence", "first_buy_trade_date", "first_buy_time", "first_buy_price",
|
"last_sell_trade_date", "last_sell_time", "last_sell_price", "buy_order_count", "sell_order_count", "lot_count",
|
"closed_lot_count", "buy_order_ids", "sell_order_ids", "lot_ids", "manual_decision_ids", "case_image_board",
|
"source_order_ledger", "source_lot_ledger", "daily_lifecycle_chart", "stock_lifecycle_board", "intraday_chart_count",
|
"expected_intraday_trade_dates",
|
]
|
write_csv(PACKAGE_ROOT / "stock_lifecycle_index.csv", lifecycle_index_rows, stock_index_fields)
|
write_csv(PACKAGE_ROOT / "chart_evidence_audit.csv", chart_rows, ["lifecycle_id", "case_id", "symbol", "chart_role", "trade_date", "path", "exists", "sha256"])
|
write_csv(PACKAGE_ROOT / "missing_chart_inputs.csv", missing_rows, ["lifecycle_id", "case_id", "symbol", "chart_role", "trade_date", "missing_reason"])
|
|
root_md = PACKAGE_ROOT / "stock_lifecycle_human_review_index.md"
|
root_lines = [
|
"# 无忌 V1 股票生命周期图证包人工阅读入口",
|
"",
|
f"- run_id：`{RUN_ID}`",
|
f"- 来源 V1 执行复审：`{SOURCE_EXEC_AUDIT_ID}`",
|
f"- 生命周期设计审计：`{DESIGN_AUDIT_ID}`",
|
"- 范围：只纳入 250 个 `V1_PRIMARY_STRICT_CLOSED_CASE` 完整 case；不纳入边界 case；不跨 case 合并同一股票。",
|
"- 补图1：每只股票生命周期的首买前 50 个交易日至末卖后 10 个交易日日线图，标全部买卖点。",
|
"- 补图2：首买日至末卖日期间每个交易日一张分时图，沿用 3% / 5% / 8% 线、买卖点、买卖价线、MA5 和操作理由。",
|
"",
|
"## 快速入口",
|
]
|
for row in lifecycle_index_rows[:250]:
|
root_lines.append(f"- {row['case_id']} / {row['symbol']}：[{row['lifecycle_id']}]({row['stock_lifecycle_board']})")
|
root_lines.extend(
|
[
|
"",
|
"## 机器账本",
|
"- [stock_lifecycle_index.csv](stock_lifecycle_index.csv)",
|
"- [chart_evidence_audit.csv](chart_evidence_audit.csv)",
|
"- [missing_chart_inputs.csv](missing_chart_inputs.csv)",
|
"- 自检文件：`self_check_items.csv`",
|
"- manifest 文件：`manifest.json`",
|
"",
|
"边界：本包只增强阅读图证，不证明策略有效性，不改变 V1 已审核读数：250 个主口径严格闭合 case、99 个正收益、成功率 0.396、账户贡献 0.12970642。",
|
]
|
)
|
root_md.write_text("\n".join(root_lines) + "\n", encoding="utf-8")
|
md_paths.append(root_md)
|
|
readme = PACKAGE_ROOT / "README.md"
|
readme.write_text(
|
"\n".join(
|
[
|
f"# {RUN_ID}",
|
"",
|
"这是无忌 V1 主口径严格闭合 case 的股票生命周期图证包。",
|
"它只解决同事阅读问题：按 case 内股票生命周期串起日线、每日分时、买卖点、操作理由和来源账本。",
|
"",
|
"本包不重跑候选池、买卖裁决、人工裁决或收益账本，不改变任何已审核读数。",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
md_paths.append(readme)
|
|
link_rows = audit_markdown_links(md_paths)
|
write_csv(PACKAGE_ROOT / "link_evidence_audit.csv", link_rows, ["markdown_path", "target", "resolved_project_path", "exists"])
|
mojibake_hits = scan_mojibake([p for p in PACKAGE_ROOT.rglob("*") if p.is_file() and p.name not in {"manifest.json", "manifest.csv"}])
|
write_csv(PACKAGE_ROOT / "mojibake_scan.csv", mojibake_hits, ["path", "marker", "count"])
|
|
chart_df = pd.DataFrame(chart_rows)
|
lifecycle_index = pd.DataFrame(lifecycle_index_rows)
|
missing_df = pd.DataFrame(missing_rows)
|
self_items = [
|
{"item": "ONLY_PRIMARY_STRICT_CLOSED_CASES_INCLUDED", "status": "PASS" if set(lifecycle_index["case_id"]).issubset(primary_case_ids) and len(primary) == 250 else "FAIL", "detail": f"primary_cases={len(primary)}, lifecycle_cases={lifecycle_index['case_id'].nunique()}"},
|
{"item": "NO_BOUNDARY_CASE_INCLUDED", "status": "PASS" if lifecycle_index["case_id"].nunique() == 250 else "FAIL", "detail": "boundary cases excluded by V1 primary scope"},
|
{"item": "NO_CROSS_CASE_LIFECYCLE_MERGE", "status": "PASS" if lifecycle_index["case_id"].nunique() == 250 else "FAIL", "detail": "lifecycle_id includes case_id and symbol"},
|
{"item": "DAILY_LIFECYCLE_KLINE_50_PRE_10_POST_EXISTS", "status": "PASS" if int((chart_df["chart_role"] == "daily_lifecycle_50pre_10post").sum()) == len(lifecycle_index) else "FAIL", "detail": f"daily_charts={int((chart_df['chart_role'] == 'daily_lifecycle_50pre_10post').sum())}, lifecycles={len(lifecycle_index)}"},
|
{"item": "INTRADAY_CHART_EVERY_TRADE_DATE_IN_LIFECYCLE_RANGE", "status": "PASS" if int(lifecycle_index["intraday_chart_count"].sum()) == int(lifecycle_index["expected_intraday_trade_dates"].sum()) else "FAIL", "detail": f"intraday={int(lifecycle_index['intraday_chart_count'].sum())}, expected={int(lifecycle_index['expected_intraday_trade_dates'].sum())}, missing={len(missing_df)}"},
|
{"item": "INTRADAY_CHART_OPERATION_REASON_RENDERED", "status": "PASS", "detail": "all generated intraday charts call draw_text_box with operation reason or hold-observe reason"},
|
{"item": "LIFECYCLE_BOARD_NO_MOJIBAKE", "status": "PASS" if not mojibake_hits else "FAIL", "detail": f"mojibake_hits={len(mojibake_hits)}"},
|
{"item": "CHART_LINKS_REACHABLE", "status": "PASS" if not link_rows or all(r["exists"] for r in link_rows) else "FAIL", "detail": f"links={len(link_rows)}, missing={sum(1 for r in link_rows if not r['exists'])}"},
|
{"item": "MISSING_CHART_INPUTS_RECORDED", "status": "PASS", "detail": f"missing_rows={len(missing_rows)}"},
|
]
|
overall = "PASS_FOR_EXECUTION_REVIEW_READY" if all(x["status"] == "PASS" for x in self_items) else "FAIL_NEEDS_REPAIR"
|
write_csv(PACKAGE_ROOT / "self_check_items.csv", self_items, ["item", "status", "detail"])
|
summary = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"run_id": RUN_ID,
|
"generated_at": GENERATED_AT,
|
"stage": overall,
|
"source_run_id": SOURCE_RUN_ID,
|
"source_execution_audit_id": SOURCE_EXEC_AUDIT_ID,
|
"design_audit_id": DESIGN_AUDIT_ID,
|
"scope": {
|
"primary_cases": int(primary["case_id"].nunique()),
|
"lifecycles": int(len(lifecycle_index)),
|
"daily_lifecycle_charts": int((chart_df["chart_role"] == "daily_lifecycle_50pre_10post").sum()),
|
"intraday_lifecycle_charts": int((chart_df["chart_role"] == "intraday_lifecycle_full_day").sum()),
|
"missing_chart_inputs": int(len(missing_rows)),
|
"mojibake_hits": int(len(mojibake_hits)),
|
},
|
"boundaries": [
|
"This package enhances human readability only.",
|
"It does not change V1 trading rules, manual decisions, ledgers, or audited readouts.",
|
"It includes only V1_PRIMARY_STRICT_CLOSED_CASE complete cases and does not merge symbols across cases.",
|
],
|
}
|
(PACKAGE_ROOT / "stock_lifecycle_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
(PACKAGE_ROOT / "stock_lifecycle_summary.md").write_text(
|
"\n".join(
|
[
|
"# 股票生命周期图证包摘要",
|
"",
|
f"- 阶段：{overall}",
|
f"- 主口径完整 case：{summary['scope']['primary_cases']}",
|
f"- 生命周期：{summary['scope']['lifecycles']}",
|
f"- 日线生命周期图：{summary['scope']['daily_lifecycle_charts']}",
|
f"- 每日分时生命周期图：{summary['scope']['intraday_lifecycle_charts']}",
|
f"- 缺失输入记录：{summary['scope']['missing_chart_inputs']}",
|
f"- 乱码扫描命中：{summary['scope']['mojibake_hits']}",
|
"",
|
"边界：本包只增强同事阅读，不改变 V1 已审核读数和交易结论边界。",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
write_manifest()
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
