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 short_id(value: object) -> str:
|
return str(value).replace(f"LOT-{RUN_ID}-", "LOT-")
|
|
|
def signal_label(signal_type: str) -> str:
|
return {
|
"STOP5_SELL_SIGNAL_CANDIDATE": "-5%止损候选",
|
"TREND_TAKE_PROFIT_SIGNAL_CANDIDATE": "趋势止盈候选",
|
"THREE_HIGH_NOT_RISING_SIGNAL_CANDIDATE": "三高不升候选",
|
"HOLD_REVIEW_CANDIDATE": "继续持仓复核",
|
}.get(signal_type, signal_type)
|
|
|
def wrap_text(text: object, max_chars: int = 22) -> list[str]:
|
lines: list[str] = []
|
for raw in str(text).splitlines():
|
item = raw.strip()
|
if not item:
|
lines.append("")
|
continue
|
while len(item) > max_chars:
|
lines.append(item[:max_chars])
|
item = item[max_chars:]
|
lines.append(item)
|
return lines
|
|
|
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_daily_exit_chart(df: pd.DataFrame, lot: pd.Series, signal: 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"持仓/卖点日K复核：{lot.symbol} 入场 {lot.entry_trade_date}"
|
d.text((32, 24), title, fill="#111827", font=FONT_TITLE)
|
d.text((32, 58), "当前为卖点信号材料，不直接生成 SELL；真实卖出需 AI/人工账本裁决。", fill="#334155", font=FONT_MID)
|
left, top, right, bottom = 80, 105, 1060, 590
|
vol_top, vol_bottom = 640, 790
|
note_left, note_top = 1090, 110
|
d.rectangle([left, top, right, bottom], outline="#94a3b8")
|
d.rectangle([left, vol_top, right, vol_bottom], outline="#94a3b8")
|
if df.empty:
|
d.text((left + 200, top + 180), "无日线数据", fill="#b91c1c", font=FONT_TITLE)
|
price_low, price_high = 0, 1
|
else:
|
price_low = min(float(df.low_price.min()), float(lot.entry_price) * 0.95) * 0.98
|
price_high = max(float(df.high_price.max()), float(lot.entry_price) * 1.05) * 1.02
|
max_vol = max(float(df.volume.max()), 1.0)
|
n = len(df)
|
gap = (right - left) / max(n, 1)
|
body_w = max(5, int(gap * 0.55))
|
for i, row in df.reset_index(drop=True).iterrows():
|
cx = int(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, top, bottom), cx, y_price(hi, price_low, price_high, top, bottom)], fill=color, width=2)
|
y1, y2 = y_price(op, price_low, price_high, top, bottom), y_price(cl, price_low, price_high, top, 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)
|
date_s = str(row.trade_date)
|
if date_s == str(lot.entry_trade_date):
|
d.line([cx, top, cx, vol_bottom], fill="#2563eb", width=2)
|
d.text((cx + 5, top + 8), "买入", fill="#2563eb", font=FONT_SMALL)
|
if date_s == signal.get("signal_trade_date", ""):
|
d.line([cx, top, cx, vol_bottom], fill="#b91c1c", width=2)
|
d.text((cx + 5, top + 30), "信号", fill="#b91c1c", font=FONT_SMALL)
|
if i % max(1, n // 6) == 0:
|
d.text((cx - 28, vol_bottom + 8), date_s[5:], fill="#64748b", font=FONT_SMALL)
|
for ref, label, color in [
|
(float(lot.entry_price), "买入价", "#2563eb"),
|
(float(lot.entry_price) * 0.95, "-5%止损", "#b91c1c"),
|
]:
|
yy = y_price(ref, price_low, price_high, top, bottom)
|
d.line([left, yy, right, yy], fill=color, width=2)
|
d.text((right - 98, yy - 8), f"{label} {ref:.2f}", fill=color, font=FONT_SMALL)
|
d.rounded_rectangle([note_left, note_top, 1460, 790], radius=8, outline="#334155", fill="#ffffff")
|
d.text((note_left + 18, note_top + 18), "卖点信号", fill="#111827", font=FONT_TITLE)
|
notes = [
|
f"lot：{short_id(lot.trade_lot_id)}",
|
f"入场：{lot.entry_trade_date} {lot.entry_time}",
|
f"买入价：{float(lot.entry_price):.2f}",
|
f"可卖日：{lot.sellable_from_trade_date}",
|
f"信号：{signal_label(signal['signal_type'])}",
|
f"信号日：{signal.get('signal_trade_date', '')}",
|
"说明：",
|
*wrap_text(signal["signal_note_cn"], 20),
|
"",
|
"本图只准备卖点材料，",
|
"不写真实 SELL。",
|
]
|
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
|
img.save(out_path)
|
|
|
def main() -> None:
|
lots = pd.read_csv(ROOT / "position_lot_ledger.csv", encoding="utf-8-sig")
|
if lots.empty:
|
raise RuntimeError("No open lots found.")
|
lots["entry_trade_date"] = pd.to_datetime(lots["entry_trade_date"])
|
lots["sellable_from_trade_date"] = pd.to_datetime(lots["sellable_from_trade_date"])
|
lots["entry_price"] = pd.to_numeric(lots["entry_price"], errors="coerce")
|
symbols = sorted(lots.symbol.unique().tolist())
|
min_date = (lots.entry_trade_date.min() - pd.Timedelta(days=90)).strftime("%Y-%m-%d")
|
max_date = (lots.entry_trade_date.max() + pd.Timedelta(days=25)).strftime("%Y-%m-%d")
|
sym_ph = ",".join(["%s"] * len(symbols))
|
with get_conn() as conn:
|
daily = pd.read_sql(
|
f"""
|
SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount
|
FROM a_share_daily_price
|
WHERE symbol IN ({sym_ph}) AND trade_date BETWEEN %s AND %s
|
ORDER BY symbol, trade_date
|
""",
|
conn,
|
params=[*symbols, min_date, max_date],
|
)
|
calendar = pd.read_sql(
|
"""
|
SELECT calendar_date
|
FROM a_share_trading_calendar
|
WHERE is_trading_day=1 AND calendar_date BETWEEN %s AND %s
|
ORDER BY calendar_date
|
""",
|
conn,
|
params=[min_date, max_date],
|
)
|
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")
|
calendar.calendar_date = pd.to_datetime(calendar.calendar_date)
|
trade_dates = list(calendar.calendar_date.dt.strftime("%Y-%m-%d"))
|
|
signal_rows = []
|
manifest_rows = []
|
for _, lot in lots.iterrows():
|
entry = lot.entry_trade_date.strftime("%Y-%m-%d")
|
sellable = lot.sellable_from_trade_date.strftime("%Y-%m-%d")
|
idx = trade_dates.index(entry)
|
obs_dates = trade_dates[idx : idx + 11]
|
obs = daily[(daily.symbol == lot.symbol) & (daily.trade_date.dt.strftime("%Y-%m-%d").isin(obs_dates))].copy()
|
sellable_obs = obs[obs.trade_date.dt.strftime("%Y-%m-%d") >= sellable].copy()
|
signal = {
|
"signal_type": "HOLD_REVIEW_CANDIDATE",
|
"signal_trade_date": sellable if not sellable_obs.empty else "",
|
"signal_note_cn": "观察窗口内未触发硬止损或趋势止盈代码信号，需人工继续复核卖点。",
|
"proxy_or_substitute_used_flag": False,
|
}
|
if not sellable_obs.empty:
|
stop = sellable_obs[sellable_obs.low_price.le(float(lot.entry_price) * 0.95)]
|
trend = sellable_obs[
|
sellable_obs.high_price.ge(float(lot.entry_price) * 1.05)
|
& sellable_obs.close_price.ge(sellable_obs.open_price)
|
]
|
three_high_signal = None
|
if len(sellable_obs) >= 3:
|
for i in range(2, len(sellable_obs)):
|
tri = sellable_obs.iloc[i - 2 : i + 1]
|
highs = list(tri.high_price)
|
if not (highs[0] < highs[1] < highs[2]):
|
three_high_signal = sellable_obs.iloc[i]
|
break
|
if not stop.empty:
|
r = stop.iloc[0]
|
signal = {
|
"signal_type": "STOP5_SELL_SIGNAL_CANDIDATE",
|
"signal_trade_date": r.trade_date.strftime("%Y-%m-%d"),
|
"signal_note_cn": "卖出候选：可卖日后触及单票-5%硬止损线，需AI/人工确认真实SELL。",
|
"proxy_or_substitute_used_flag": False,
|
}
|
elif not trend.empty:
|
r = trend.iloc[0]
|
signal = {
|
"signal_type": "TREND_TAKE_PROFIT_SIGNAL_CANDIDATE",
|
"signal_trade_date": r.trade_date.strftime("%Y-%m-%d"),
|
"signal_note_cn": "卖出候选：可卖日后出现相对买入价5%以上趋势性上涨，需AI/人工确认是否按原文止盈卖出。",
|
"proxy_or_substitute_used_flag": False,
|
}
|
elif three_high_signal is not None:
|
signal = {
|
"signal_type": "THREE_HIGH_NOT_RISING_SIGNAL_CANDIDATE",
|
"signal_trade_date": three_high_signal.trade_date.strftime("%Y-%m-%d"),
|
"signal_note_cn": "卖出候选：三日高点未逐步抬高，需AI/人工确认是否卖出。",
|
"proxy_or_substitute_used_flag": False,
|
}
|
case_dir = ROOT / "cases" / lot.case_id
|
img_dir = case_dir / "img"
|
img_dir.mkdir(parents=True, exist_ok=True)
|
out_path = img_dir / f"05_exit_daily_signal_{lot.symbol.replace('.', '_')}_{entry}.png"
|
window = daily[(daily.symbol == lot.symbol) & (daily.trade_date.dt.strftime("%Y-%m-%d") <= obs_dates[-1])].tail(80).copy()
|
window["trade_date"] = window.trade_date.dt.strftime("%Y-%m-%d")
|
lot_for_chart = lot.copy()
|
lot_for_chart.entry_trade_date = entry
|
lot_for_chart.sellable_from_trade_date = sellable
|
draw_daily_exit_chart(window, lot_for_chart, signal, out_path)
|
rel = out_path.relative_to(ROOT).as_posix()
|
signal_rows.append(
|
{
|
"trade_lot_id": lot.trade_lot_id,
|
"order_id": lot.order_id,
|
"case_id": lot.case_id,
|
"symbol": lot.symbol,
|
"entry_trade_date": entry,
|
"entry_time": lot.entry_time,
|
"entry_price": f"{float(lot.entry_price):.4f}",
|
"sellable_from_trade_date": sellable,
|
"signal_type": signal["signal_type"],
|
"signal_trade_date": signal.get("signal_trade_date", ""),
|
"signal_note_cn": signal["signal_note_cn"],
|
"manual_sell_decision": "PENDING_AI_MANUAL_DECISION",
|
"decision_chart_path": rel,
|
"t1_guard_passed_flag": signal.get("signal_trade_date", "") >= sellable if signal.get("signal_trade_date", "") else "",
|
"proxy_or_substitute_used_flag": signal["proxy_or_substitute_used_flag"],
|
}
|
)
|
manifest_rows.append(
|
{
|
"case_id": lot.case_id,
|
"symbol": lot.symbol,
|
"trade_date": signal.get("signal_trade_date", ""),
|
"event_id": f"{lot.trade_lot_id}_exit_signal",
|
"chart_role": "exit_daily_signal_review_view",
|
"decision_time": f"{signal.get('signal_trade_date', '')} close",
|
"path": rel,
|
"sha256": sha256_file(out_path),
|
"status": "PASS",
|
"note": "卖点/持仓日K复核图；不自动生成SELL。",
|
}
|
)
|
|
signals = pd.DataFrame(signal_rows)
|
signals.to_csv(ROOT / "sell_signal_candidates.csv", index=False, encoding="utf-8-sig")
|
root_manifest = pd.read_csv(ROOT / "image_manifest.csv", encoding="utf-8-sig")
|
combined = pd.concat([root_manifest, pd.DataFrame(manifest_rows)], ignore_index=True)
|
combined = combined.drop_duplicates(subset=["case_id", "symbol", "event_id", "chart_role"], keep="last")
|
combined.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig")
|
|
for case_id, group in pd.DataFrame(manifest_rows).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"
|
existing = re.sub(r"\n## 5\. 卖点 / 持仓日K信号图\n[\s\S]*$", "", existing.rstrip())
|
lines = [existing.rstrip(), "", "## 5. 卖点 / 持仓日K信号图", ""]
|
for _, row in group.iterrows():
|
rel = Path(row.path).relative_to(f"cases/{case_id}").as_posix()
|
lines.extend([f"### {row.symbol}", "", f"", "", "- 本图只提示卖点/持仓信号，不写真实 SELL。", ""])
|
board_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
combined[combined.case_id == case_id].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:45:00+08:00",
|
"stage": "SELL_SIGNAL_MATERIAL_READY",
|
"signal_counts": signals.signal_type.value_counts().to_dict(),
|
"signal_rows": len(signals),
|
"exit_signal_images": len(manifest_rows),
|
"artifacts": {
|
"sell_signal_candidates.csv": {
|
"size": (ROOT / "sell_signal_candidates.csv").stat().st_size,
|
"sha256": sha256_file(ROOT / "sell_signal_candidates.csv"),
|
},
|
"image_manifest.csv": {
|
"size": (ROOT / "image_manifest.csv").stat().st_size,
|
"sha256": sha256_file(ROOT / "image_manifest.csv"),
|
},
|
},
|
"boundary": "Signal material only; no SELL orders and no return statistics.",
|
"next_step": "AI/manual sell decision from sell_signal_candidates.csv and images.",
|
}
|
(ROOT / "exit_review_generation_summary.json").write_text(
|
json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
|
encoding="utf-8",
|
)
|
(ROOT / "exit_review_generation_summary.md").write_text(
|
"\n".join(
|
[
|
"# exit_review_generation_summary",
|
"",
|
f"run_id：`{RUN_ID}`",
|
"阶段：`SELL_SIGNAL_MATERIAL_READY`",
|
"",
|
f"- 卖点/持仓信号行：{len(signals)}",
|
f"- 卖点/持仓日K图：{len(manifest_rows)}",
|
"",
|
"当前只生成卖点材料，不产生 SELL 或收益结论。",
|
"",
|
]
|
),
|
encoding="utf-8",
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
