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()
|
