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
|
NEAR_MA5_PCT = 0.015
|
ROLLING_VOLUME_RATIO = 1.5
|
ROLLING_START_TIME = "10:40:00"
|
ROLLING_END_TIME = "14:40:00"
|
|
|
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 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 window_dates(trade_dates: list[str], entry_date: str, start_date: str) -> list[str]:
|
if entry_date not in trade_dates or start_date not in trade_dates:
|
return []
|
start = trade_dates.index(start_date)
|
end = min(len(trade_dates) - 1, trade_dates.index(entry_date) + OBSERVATION_TRADING_DAYS)
|
return trade_dates[start : end + 1]
|
|
|
def chunked(values: list, size: int):
|
for i in range(0, len(values), size):
|
yield values[i : i + size]
|
|
|
def fetch_market(signals: 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(signals["symbol"].unique())
|
for row in signals.itertuples(index=False):
|
start_date = row.observation_trade_date if row.observation_trade_date else row.entry_trade_date
|
dates = window_dates(trade_dates, row.entry_trade_date, start_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 = []
|
with get_conn() as conn:
|
pair_list = sorted(minute_pairs, key=lambda x: (x[1], x[0]))
|
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 in_rolling_time(time_text: str) -> bool:
|
return ROLLING_START_TIME <= time_text <= ROLLING_END_TIME
|
|
|
def rolling_candidate(row, minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict], trade_dates: list[str]) -> dict:
|
start_date = row.observation_trade_date if row.observation_trade_date else row.entry_trade_date
|
dates = window_dates(trade_dates, row.entry_trade_date, start_date)
|
base = {
|
"rolling_signal_id": f"ROLL-CAND-{row.lot_id}",
|
"source_sell_signal_id": row.sell_signal_id,
|
"lot_id": row.lot_id,
|
"open_order_id": row.open_order_id,
|
"case_id": row.case_id,
|
"candidate_id": row.candidate_id,
|
"symbol": row.symbol,
|
"entry_trade_date": row.entry_trade_date,
|
"rolling_trade_date": "",
|
"rolling_time": "",
|
"signal_type": "ROLLING_LOW_BUY_REVIEW_HELD",
|
"code_suggested_action": "REVIEW_HELD",
|
"code_suggested_reason_cn": "趋势观察后未找到五日线附近止跌放量的滚动低吸确认点，保留待人工复核。",
|
"rolling_price": "",
|
"ma5_close": "",
|
"near_ma5_pct": "",
|
"volume_ratio_vs_prev20m": "",
|
"review_input_chart_path": "",
|
"review_input_chart_sha256": "",
|
}
|
for d in dates:
|
day = minute_lookup.get((row.symbol, d), pd.DataFrame())
|
daily = daily_lookup.get((row.symbol, d), {})
|
ma5 = safe_float(daily.get("ma5_close"))
|
if day.empty or math.isnan(ma5):
|
continue
|
day = day.copy()
|
day["prev20_volume_avg"] = day["volume"].rolling(20, min_periods=5).mean().shift(1)
|
for r in day.itertuples(index=False):
|
if not in_rolling_time(r.trade_time):
|
continue
|
price = safe_float(r.close_price)
|
near = abs(price - ma5) / ma5 if ma5 else math.nan
|
vol_avg = safe_float(getattr(r, "prev20_volume_avg", math.nan))
|
vol_ratio = safe_float(r.volume) / vol_avg if vol_avg and not math.isnan(vol_avg) else math.nan
|
if not math.isnan(near) and not math.isnan(vol_ratio) and near <= NEAR_MA5_PCT and vol_ratio >= ROLLING_VOLUME_RATIO:
|
base.update(
|
{
|
"rolling_trade_date": d,
|
"rolling_time": r.trade_time,
|
"signal_type": "ADD_ROLLING_LOW_BUY",
|
"code_suggested_action": "BUY_ROLLING_LOW",
|
"code_suggested_reason_cn": f"趋势观察后回到五日线附近，距 MA5 {near:.2%}，分钟量能相对前 20 分钟均量放大 {vol_ratio:.2f} 倍，代码建议进入滚动低吸人工确认。",
|
"rolling_price": f"{price:.4f}",
|
"ma5_close": f"{ma5:.4f}",
|
"near_ma5_pct": f"{near:.8f}",
|
"volume_ratio_vs_prev20m": f"{vol_ratio:.4f}",
|
}
|
)
|
return base
|
return base
|
|
|
def wrap(text: str, n: int) -> list[str]:
|
return [text[i : i + n] for i in range(0, len(text), n)] or [""]
|
|
|
def draw_rolling_chart(signal: dict, minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict]) -> None:
|
d = signal["rolling_trade_date"] or signal["entry_trade_date"]
|
day = minute_lookup.get((signal["symbol"], d), pd.DataFrame())
|
out = ROOT / "charts" / "rolling_low_review" / signal["case_id"] / f"{signal['rolling_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 [safe_float(signal.get("rolling_price"), 1.0)]
|
ma5 = safe_float(signal.get("ma5_close"))
|
levels = prices + ([] if math.isnan(ma5) else [ma5])
|
lo, hi = min(levels), max(levels)
|
pad = max((hi - lo) * 0.08, 0.01)
|
lo -= pad
|
hi += pad
|
|
def x_at(i: int) -> float:
|
return left if len(prices) <= 1 else left + i * (right - left) / (len(prices) - 1)
|
|
def y_at(price: float) -> float:
|
return bottom - (price - lo) * (bottom - top) / (hi - lo)
|
|
if prices and not day.empty:
|
pts = [(x_at(i), y_at(p)) for i, p in enumerate(prices)]
|
draw.line(pts, fill="#1f77b4", width=2)
|
if not math.isnan(ma5):
|
y = y_at(ma5)
|
draw.line((left, y, right, y), fill="#8c564b", width=2)
|
draw.text((right + 8, y - 10), f"日线MA5 {ma5:.2f}", fill="#8c564b", font=FONT_18)
|
if not day.empty and signal["rolling_time"] in set(day["trade_time"].tolist()):
|
idx = day.index[day["trade_time"].eq(signal["rolling_time"])][0]
|
price = safe_float(day.loc[idx, "close_price"])
|
x, y = x_at(int(idx)), y_at(price)
|
draw.ellipse((x - 7, y - 7, x + 7, y + 7), fill="#d62728")
|
draw.line((x, top, x, bottom), fill="#d62728", width=2)
|
draw.text((x + 8, y - 28), f"低吸候选 {price:.2f}", fill="#d62728", font=FONT_18)
|
|
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['rolling_trade_date']} {signal['rolling_time']}",
|
f"MA5：{signal['ma5_close']}",
|
"理由：",
|
]
|
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人工确认滚动低吸，不生成最终 BUY；不买也必须写清理由。", 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:
|
sell_df = pd.read_csv(ROOT / "strict_note_sell_rolling_review_candidate_ledger.csv", encoding="utf-8-sig")
|
hold_df = sell_df[sell_df["code_suggested_action"].isin(["HOLD_WATCH", "HOLD_ABOVE_8"])].copy()
|
trade_dates = fetch_trade_calendar()
|
minute_lookup, daily_lookup = fetch_market(hold_df, trade_dates) if not hold_df.empty else ({}, {})
|
|
rolling_rows = []
|
for row in hold_df.itertuples(index=False):
|
sig = rolling_candidate(row, minute_lookup, daily_lookup, trade_dates)
|
draw_rolling_chart(sig, minute_lookup, daily_lookup)
|
rolling_rows.append(sig)
|
rolling_df = pd.DataFrame(rolling_rows)
|
rolling_df.to_csv(ROOT / "strict_note_rolling_low_review_candidate_ledger.csv", index=False, encoding="utf-8-sig")
|
|
sell_template = sell_df.copy()
|
sell_template.insert(0, "artifact_type", "SELL_SIGNAL")
|
sell_template["signal_id"] = sell_template["sell_signal_id"]
|
sell_template["rolling_signal_id"] = ""
|
|
rolling_template = rolling_df.copy()
|
rolling_template.insert(0, "artifact_type", "ROLLING_LOW_SIGNAL")
|
rolling_template["signal_id"] = ""
|
if not rolling_template.empty:
|
rolling_template["sell_signal_id"] = ""
|
|
common_cols = [
|
"artifact_type",
|
"signal_id",
|
"rolling_signal_id",
|
"sell_signal_id",
|
"lot_id",
|
"open_order_id",
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"entry_trade_date",
|
"signal_type",
|
"code_suggested_action",
|
"code_suggested_reason_cn",
|
"review_input_chart_path",
|
"review_input_chart_sha256",
|
]
|
template = pd.concat([sell_template, rolling_template], ignore_index=True, sort=False)
|
for col in common_cols:
|
if col not in template.columns:
|
template[col] = ""
|
template = template[common_cols].copy()
|
template["external_decision_id"] = [f"EXT-SELLROLL-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_rows = []
|
for df, artifact, id_col in [(sell_df, "SELL_SIGNAL", "sell_signal_id"), (rolling_df, "ROLLING_LOW_SIGNAL", "rolling_signal_id")]:
|
for r in df.itertuples(index=False):
|
chart_rows.append(
|
{
|
"artifact_type": artifact,
|
"signal_id": getattr(r, id_col),
|
"case_id": r.case_id,
|
"symbol": r.symbol,
|
"review_input_chart_path": r.review_input_chart_path,
|
"review_input_chart_sha256": r.review_input_chart_sha256,
|
"exists": (ROOT / r.review_input_chart_path).exists(),
|
}
|
)
|
chart_df = pd.DataFrame(chart_rows)
|
chart_df.to_csv(ROOT / "sell_rolling_chart_evidence_audit.csv", index=False, encoding="utf-8-sig")
|
|
generated_at = now_iso()
|
sell_counts = Counter(sell_df["code_suggested_action"])
|
rolling_counts = Counter(rolling_df["code_suggested_action"]) if not rolling_df.empty else Counter()
|
checks = [
|
("STRICT_SELL_SIGNAL_SCOPE_IS_424", len(sell_df) == 424, f"sell_signals={len(sell_df)}"),
|
("ROLLING_CANDIDATES_COVER_HOLD_SIGNALS", len(rolling_df) == len(hold_df), f"rolling={len(rolling_df)}, hold_signals={len(hold_df)}"),
|
("UNIFIED_TEMPLATE_COVERS_SELL_AND_ROLLING", len(template) == len(sell_df) + len(rolling_df), f"template={len(template)}"),
|
("TEMPLATE_FINAL_HUMAN_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": 424,
|
"sell_review_candidates": int(len(sell_df)),
|
"rolling_low_review_candidates": int(len(rolling_df)),
|
"manual_decision_template_rows": int(len(template)),
|
"sell_code_suggested_action_counts": dict(sell_counts),
|
"rolling_code_suggested_action_counts": dict(rolling_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"
|
"- strict BUY lots input: 424\n"
|
f"- sell review candidates: {len(sell_df)}\n"
|
f"- rolling low review candidates: {len(rolling_df)}\n"
|
f"- manual decision template rows: {len(template)}\n"
|
f"- sell code suggested actions: {dict(sell_counts)}\n"
|
f"- rolling code suggested actions: {dict(rolling_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()
|
