from __future__ import annotations import hashlib import json import math import os import re from collections import Counter 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-STRICT-NOTE-FULL-RERUN-20260614-001" ROOT = Path(__file__).resolve().parents[1] LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md") TZ = timezone(timedelta(hours=8)) OBSERVATION_TRADING_DAYS = 10 GAIN_3 = 0.03 GAIN_5 = 0.05 GAIN_8 = 0.08 STOP_LOSS = -0.05 def now_iso() -> str: return datetime.now(TZ).isoformat(timespec="seconds") 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 normalize_date(value) -> str: return pd.to_datetime(value).strftime("%Y-%m-%d") def normalize_time(value) -> str: text = str(value) if " " 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}" if len(parts) == 2: return f"{int(parts[0]):02d}:{int(parts[1]):02d}:00" return text def read_mysql_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"密码：`([^`]+)`", text) 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_mysql_password(), database="tianxia", charset="utf8mb4", connect_timeout=5, read_timeout=240, write_timeout=240, ) def font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: for name in ["msyh.ttc", "simhei.ttf", "simsun.ttc"]: path = Path("C:/Windows/Fonts") / name if path.exists(): return ImageFont.truetype(str(path), size) return ImageFont.load_default() FONT_18 = font(18) FONT_22 = font(22) FONT_28 = font(28) def next_trade_date(trade_dates: list[str], trade_date: str) -> str: if trade_date not in trade_dates: return "" idx = trade_dates.index(trade_date) return trade_dates[idx + 1] if idx + 1 < len(trade_dates) else "" def window_dates(trade_dates: list[str], entry_date: str) -> list[str]: sellable = next_trade_date(trade_dates, entry_date) if not sellable or entry_date not in trade_dates: return [] start = trade_dates.index(sellable) end = min(len(trade_dates) - 1, trade_dates.index(entry_date) + OBSERVATION_TRADING_DAYS) return trade_dates[start : end + 1] def fetch_trade_calendar() -> list[str]: with get_conn() as conn: dates = pd.read_sql( """ SELECT DISTINCT trade_date FROM a_share_daily_price WHERE trade_date BETWEEN '2023-01-01' AND '2026-12-31' ORDER BY trade_date """, conn, ) return [normalize_date(v) for v in dates["trade_date"].tolist()] def chunked(values: list, size: int): for i in range(0, len(values), size): yield values[i : i + size] def fetch_market( lots: pd.DataFrame, trade_dates: list[str] ) -> tuple[dict[tuple[str, str], pd.DataFrame], dict[tuple[str, str], dict]]: minute_pairs: set[tuple[str, str]] = set() all_daily_dates: set[str] = set() symbols = sorted(lots["symbol"].unique()) for row in lots.itertuples(index=False): dates = window_dates(trade_dates, row.entry_trade_date) for d in dates: minute_pairs.add((row.symbol, d)) all_daily_dates.add(d) if row.entry_trade_date in trade_dates: idx = trade_dates.index(row.entry_trade_date) for d in trade_dates[max(0, idx - 5) : min(len(trade_dates), idx + OBSERVATION_TRADING_DAYS + 1)]: all_daily_dates.add(d) if not minute_pairs: return {}, {} minute_parts = [] pair_list = sorted(minute_pairs, key=lambda x: (x[1], x[0])) with get_conn() as conn: for pairs in chunked(pair_list, 600): placeholders = ",".join(["(%s,%s)"] * len(pairs)) params = [] for symbol, trade_date in pairs: params.extend([trade_date, symbol]) minute_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, symbol) IN ({placeholders}) ORDER BY symbol, trade_date, trade_time """, conn, params=params, ) ) ph = ",".join(["%s"] * len(symbols)) 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 ({ph}) AND trade_date BETWEEN %s AND %s ORDER BY symbol, trade_date """, conn, params=[*symbols, min(all_daily_dates), max(all_daily_dates)], ) minute = pd.concat(minute_parts, ignore_index=True) if minute_parts else pd.DataFrame() minute_lookup = {} if not minute.empty: minute["trade_date"] = minute["trade_date"].map(normalize_date) minute["trade_time"] = minute["trade_time"].map(normalize_time) for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]: minute[col] = pd.to_numeric(minute[col], errors="coerce") for (symbol, trade_date), group in minute.groupby(["symbol", "trade_date"]): minute_lookup[(symbol, trade_date)] = group.sort_values("trade_time").reset_index(drop=True) daily_lookup = {} if not daily.empty: daily["trade_date"] = daily["trade_date"].map(normalize_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_close"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(5, min_periods=3).mean()) daily_lookup = {(r["symbol"], r["trade_date"]): r.to_dict() for _, r in daily.iterrows()} return minute_lookup, daily_lookup def safe_float(value, default=math.nan) -> float: try: if pd.isna(value): return default return float(value) except Exception: return default def minutes_between(start: str, end: str) -> float: a = datetime.strptime(start, "%H:%M:%S") b = datetime.strptime(end, "%H:%M:%S") return (b - a).total_seconds() / 60 def candidate_row(row, d: str, t: str, signal_type: str, action: str, reason: str, price: float, gain: float) -> dict: return { "sell_signal_id": f"SELL-CAND-{row.lot_id}", "lot_id": row.lot_id, "open_order_id": row.open_order_id, "case_id": row.case_id, "candidate_id": row.candidate_id, "external_buy_decision_id": row.external_decision_id, "symbol": row.symbol, "entry_trade_date": row.entry_trade_date, "entry_price": row.entry_price, "position_pct": row.position_pct, "observation_trade_date": d, "candidate_time": t, "signal_type": signal_type, "code_suggested_action": action, "code_suggested_reason_cn": reason, "action_price": "" if math.isnan(price) else f"{price:.4f}", "gain_pct": "" if math.isnan(gain) else f"{gain:.8f}", "review_input_chart_path": "", "review_input_chart_sha256": "", } def make_candidate(row, trade_dates: list[str], minute_lookup: dict[tuple[str, str], pd.DataFrame]) -> dict: entry = safe_float(row.entry_price) first3: tuple[str, str] | None = None fast_watch = False above8 = False dates = window_dates(trade_dates, row.entry_trade_date) if not dates: return candidate_row(row, "", "", "SELL_REVIEW_DATA_GAP_HELD", "REVIEW_HELD", "交易日观察窗口缺失，保留待人工复核。", entry, 0.0) last_seen = None for d in dates: day = minute_lookup.get((row.symbol, d), pd.DataFrame()) if day.empty: return candidate_row(row, d, "", "SELL_REVIEW_DATA_GAP_HELD", "REVIEW_HELD", f"{d} 分钟线缺失，保留待人工复核。", entry, 0.0) for r in day.itertuples(index=False): price = safe_float(r.close_price) high_gain = safe_float(r.high_price) / entry - 1 low_gain = safe_float(r.low_price) / entry - 1 close_gain = price / entry - 1 last_seen = (d, r.trade_time, price, close_gain) if low_gain <= STOP_LOSS: return candidate_row(row, d, r.trade_time, "SELL_STOP_LOSS_5", "SELL", "总价跌破 -5% 止损线，代码建议卖出；最终动作需要外部人工确认。", price, close_gain) if first3 is None and high_gain >= GAIN_3: first3 = (d, r.trade_time) continue if first3 is None: continue fast = d == first3[0] and minutes_between(first3[1], r.trade_time) <= 10 if not fast_watch and high_gain >= GAIN_5 and fast: fast_watch = True continue if not fast_watch and d == first3[0] and minutes_between(first3[1], r.trade_time) > 10 and GAIN_3 <= close_gain < GAIN_5: return candidate_row(row, d, r.trade_time, "SELL_TREND_3_TO_5_GRADUAL", "SELL", "上涨进入 3%-5% 区间但不是快速冲过 5%，代码建议趋势止盈卖出；最终动作需要外部人工确认。", price, close_gain) if fast_watch and not above8 and high_gain >= GAIN_8: above8 = True continue if fast_watch and not above8 and low_gain < GAIN_5 and close_gain >= GAIN_3: return candidate_row(row, d, r.trade_time, "SELL_TREND_PULLBACK_3_TO_5", "SELL", "快速冲过 5% 后回落到 3%-5% 区间，代码建议卖出；最终动作需要外部人工确认。", price, close_gain) if above8 and last_seen: d, t, price, gain = last_seen return candidate_row(row, d, t, "TREND_ABOVE_8_HOLD", "HOLD_ABOVE_8", "快速冲过 5% 后曾超过 8%，代码建议强势持有观察；不卖理由需要图上展示并由外部人工确认。", price, gain) if fast_watch and last_seen: d, t, price, gain = last_seen return candidate_row(row, d, t, "TREND_FAST_BREAKOUT_5_WATCH", "HOLD_WATCH", "快速冲过 5%，代码建议观察不卖；不卖理由需要图上展示并由外部人工确认。", price, gain) if last_seen: d, t, price, gain = last_seen return candidate_row(row, d, t, "SELL_WINDOW_END_REVIEW_HELD", "REVIEW_HELD", "观察窗口结束仍未出现明确卖点，保留待人工复核。", price, gain) return candidate_row(row, "", "", "SELL_REVIEW_DATA_GAP_HELD", "REVIEW_HELD", "无有效分钟观察数据，保留待人工复核。", entry, 0.0) def wrap(text: str, n: int) -> list[str]: return [text[i : i + n] for i in range(0, len(text), n)] or [""] def draw_chart(signal: dict, minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict]) -> None: d = signal["observation_trade_date"] day = minute_lookup.get((signal["symbol"], d), pd.DataFrame()) out = ROOT / "charts" / "sell_rolling_review" / signal["case_id"] / f"{signal['sell_signal_id']}.png" out.parent.mkdir(parents=True, exist_ok=True) img = Image.new("RGB", (1500, 900), "#fbfaf6") draw = ImageDraw.Draw(img) draw.text((40, 24), f"严格版卖点/趋势候选：{signal['case_id']} {signal['symbol']} {d}", fill="#111111", font=FONT_28) left, top, right, bottom = 70, 110, 1020, 620 draw.rectangle((left, top, right, bottom), outline="#cccccc") prices = [safe_float(v) for v in day["close_price"].tolist()] if not day.empty else [] entry = safe_float(signal["entry_price"]) ma5 = safe_float(daily_lookup.get((signal["symbol"], d), {}).get("ma5_close")) levels = [entry, entry * 1.03, entry * 1.05, entry * 1.08] if not math.isnan(ma5): levels.append(ma5) all_prices = prices + levels lo, hi = min(all_prices), max(all_prices) pad = max((hi - lo) * 0.08, 0.01) lo -= pad hi += pad def x_at(i: int) -> float: if not prices or len(prices) == 1: return left return left + i * (right - left) / (len(prices) - 1) def y_at(p: float) -> float: return bottom - (p - lo) * (bottom - top) / (hi - lo) colors = [ (entry, "#111111", "买入价"), (entry * 1.03, "#2ca02c", "3%"), (entry * 1.05, "#ff7f0e", "5%"), (entry * 1.08, "#9467bd", "8%"), ] if not math.isnan(ma5): colors.append((ma5, "#8c564b", "日线MA5")) for price, color, label in colors: y = y_at(price) draw.line((left, y, right, y), fill=color, width=2) draw.text((right + 8, y - 10), f"{label} {price:.2f}", fill=color, font=FONT_18) if prices: pts = [(x_at(i), y_at(p)) for i, p in enumerate(prices)] draw.line(pts, fill="#1f77b4", width=2) if signal["candidate_time"] in set(day["trade_time"].tolist()): idx = day.index[day["trade_time"].eq(signal["candidate_time"])][0] x, y = x_at(int(idx)), y_at(safe_float(day.loc[idx, "close_price"])) draw.ellipse((x - 6, y - 6, x + 6, y + 6), fill="#d62728") draw.line((x, top, x, bottom), fill="#d62728", width=2) side_x = 1060 draw.text((side_x, 110), "外部人工裁决待填", fill="#111111", font=FONT_22) info = [ f"代码建议：{signal['code_suggested_action']}", f"信号：{signal['signal_type']}", f"时间：{signal['observation_trade_date']} {signal['candidate_time']}", f"收益候选：{signal['gain_pct']}", "理由：", ] y = 150 for line in info: draw.text((side_x, y), line, fill="#111111", font=FONT_18) y += 30 for part in wrap(signal["code_suggested_reason_cn"], 18): draw.text((side_x, y), part, fill="#111111", font=FONT_18) y += 28 draw.text((40, 825), "audit_view：本图只用于外部人工/AI人工复核，不是最终卖点裁决；不卖也必须在后续图上写清楚理由。", fill="#666666", font=FONT_18) img.save(out) rel = out.relative_to(ROOT).as_posix() signal["review_input_chart_path"] = rel signal["review_input_chart_sha256"] = sha256_file(out) def build_manifest() -> pd.DataFrame: rows = [] for path in sorted(ROOT.rglob("*")): if path.is_file() and path.name not in {"manifest.csv", "manifest.json"}: rows.append({"path": path.relative_to(ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path)}) return pd.DataFrame(rows) def main() -> None: lots = pd.read_csv(ROOT / "strict_note_buy_lot_ledger.csv", encoding="utf-8-sig") orders = pd.read_csv(ROOT / "strict_note_buy_order_ledger.csv", encoding="utf-8-sig") lots["entry_trade_date"] = lots["entry_trade_date"].map(normalize_date) lots = lots.merge(orders[["order_id", "evidence_image_path", "decision_reason_cn"]], left_on="open_order_id", right_on="order_id", how="left") if len(lots) != 424: raise RuntimeError(f"expected 424 strict BUY lots, got {len(lots)}") trade_dates = fetch_trade_calendar() minute_lookup, daily_lookup = fetch_market(lots, trade_dates) signals = [] for row in lots.itertuples(index=False): sig = make_candidate(row, trade_dates, minute_lookup) draw_chart(sig, minute_lookup, daily_lookup) signals.append(sig) signal_df = pd.DataFrame(signals) signal_df.to_csv(ROOT / "strict_note_sell_rolling_review_candidate_ledger.csv", index=False, encoding="utf-8-sig") template = signal_df.copy() template["external_decision_id"] = [f"EXT-SELL-STRICT-NOTE-{i:06d}" for i in range(1, len(template) + 1)] for col in [ "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "accept_code_suggestion_flag", "reviewer_notes", ]: template[col] = "" template.to_csv(ROOT / "manual_sell_rolling_decision_external_template.csv", index=False, encoding="utf-8-sig") chart_df = signal_df[["sell_signal_id", "case_id", "symbol", "review_input_chart_path", "review_input_chart_sha256"]].copy() chart_df["exists"] = chart_df["review_input_chart_path"].map(lambda p: (ROOT / p).exists()) chart_df.to_csv(ROOT / "sell_rolling_chart_evidence_audit.csv", index=False, encoding="utf-8-sig") counts = Counter(signal_df["code_suggested_action"]) generated_at = now_iso() checks = [ ("STRICT_BUY_LOT_SCOPE_IS_424", len(lots) == 424, f"lots={len(lots)}"), ("CANDIDATE_PER_BUY_LOT", len(signal_df) == len(lots), f"signals={len(signal_df)}"), ("TEMPLATE_FIELDS_BLANK", template["human_decision_action"].astype(str).str.strip().eq("").all(), "human fields blank"), ("CHARTS_EXIST", bool(chart_df["exists"].all()), f"charts={len(chart_df)}, missing={int((~chart_df['exists']).sum())}"), ("NO_STRICT_PERFORMANCE_READOUT", True, "prep package only; no return/success/win-rate generated"), ] self_items = pd.DataFrame([{"item": k, "status": "PASS" if ok else "FAIL", "detail": detail} for k, ok, detail in checks]) self_items.to_csv(ROOT / "sell_rolling_review_prep_self_check_items.csv", index=False, encoding="utf-8-sig") status = "PASS_FOR_SELL_ROLLING_MANUAL_REVIEW_PREP_READY" if self_items["status"].eq("PASS").all() else "FAIL" (ROOT / "sell_rolling_review_prep_self_check.json").write_text( json.dumps( { "run_id": RUN_ID, "generated_at": generated_at, "stage": status, "pass_count": int(self_items["status"].eq("PASS").sum()), "fail_count": int(self_items["status"].eq("FAIL").sum()), }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) summary = { "run_id": RUN_ID, "generated_at": generated_at, "stage": status, "strict_buy_lots_input": int(len(lots)), "sell_rolling_review_candidates": int(len(signal_df)), "code_suggested_action_counts": dict(counts), "boundary": "Prep only: final sell/hold/rolling actions require external manual decision source and execution review.", } (ROOT / "sell_rolling_review_prep_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") (ROOT / "sell_rolling_review_prep_summary.md").write_text( "# Strict Note Sell/Rolling Manual Review Prep\n\n" f"- generated_at: {generated_at}\n" f"- strict BUY lots input: {len(lots)}\n" f"- review candidates: {len(signal_df)}\n" f"- code suggested actions: {dict(counts)}\n\n" "Boundary: this is a manual-review prep package only. It does not generate final SELL orders, rolling BUY orders, return, success rate, win rate, drawdown, or strategy-effectiveness conclusions.\n", encoding="utf-8", ) manifest = build_manifest() manifest.to_csv(ROOT / "manifest.csv", index=False, encoding="utf-8-sig") (ROOT / "manifest.json").write_text( json.dumps({"run_id": RUN_ID, "generated_at": generated_at, "file_count": int(len(manifest)), "files": manifest.to_dict("records")}, ensure_ascii=False, indent=2), encoding="utf-8", ) print(json.dumps(summary, ensure_ascii=False)) if __name__ == "__main__": main()