from __future__ import annotations import hashlib import json import os import re from pathlib import Path import pandas as pd import pymysql RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001" ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = ROOT.parents[2] 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 write_csv(df: pd.DataFrame, name: str) -> Path: path = ROOT / name df.to_csv(path, index=False, encoding="utf-8-sig") return path def main() -> None: cfg = json.loads((ROOT / "run_config.json").read_text(encoding="utf-8")) minute_start = cfg["data_sources"]["minute_price_source"]["actual_coverage_start"] minute_end = cfg["data_sources"]["minute_price_source"]["actual_coverage_end"] pull_start = "2022-10-01" pull_end = minute_end with get_conn() as conn: daily = pd.read_sql( """ SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount FROM a_share_daily_price WHERE trade_date BETWEEN %(start)s AND %(end)s ORDER BY symbol, trade_date """, conn, params={"start": pull_start, "end": pull_end}, ) breadth = pd.read_sql( """ SELECT trade_date, stock_count, up_count, flat_count, down_count, run_id, price_source_table FROM ts_market_breadth_daily_cache WHERE trade_date BETWEEN %(start)s AND %(end)s AND scope_type='ALL_A_SHARE' AND scope_value='ALL' ORDER BY trade_date """, conn, params={"start": "2023-01-01", "end": minute_end}, ) calendar = pd.read_sql( """ SELECT calendar_date, is_trading_day, trade_date_rank FROM a_share_trading_calendar WHERE calendar_date BETWEEN %(start)s AND %(end)s ORDER BY calendar_date """, conn, params={"start": "2023-01-01", "end": minute_end}, ) for col in ["trade_date"]: daily[col] = pd.to_datetime(daily[col]) breadth[col] = pd.to_datetime(breadth[col]) calendar["calendar_date"] = pd.to_datetime(calendar["calendar_date"]) price_cols = ["open_price", "high_price", "low_price", "close_price", "volume", "amount"] for col in price_cols: daily[col] = pd.to_numeric(daily[col], errors="coerce") daily = daily.sort_values(["symbol", "trade_date"]).reset_index(drop=True) grouped = daily.groupby("symbol", group_keys=False) daily["prev_close"] = grouped["close_price"].shift(1) daily["prev5_avg_volume"] = grouped["volume"].transform( lambda s: s.shift(1).rolling(5, min_periods=5).mean() ) daily["volume_ratio"] = daily["volume"] / daily["prev5_avg_volume"] daily["daily_return_pct"] = (daily["close_price"] / daily["prev_close"] - 1.0) * 100.0 daily["limit_up_event_flag"] = daily["prev_close"].gt(0) & (daily["high_price"] / daily["prev_close"] - 1.0).ge(0.095) daily["recent_limitup_30_flag"] = grouped["limit_up_event_flag"].transform( lambda s: s.rolling(30, min_periods=1).max().astype(bool) ) daily["upper_shadow_pct"] = ( (daily["high_price"] - daily[["open_price", "close_price"]].max(axis=1)) / daily["prev_close"] ) * 100.0 daily["body_pct"] = ((daily["close_price"] - daily["open_price"]).abs() / daily["prev_close"]) * 100.0 daily["prev60_high"] = grouped["high_price"].transform( lambda s: s.shift(1).rolling(60, min_periods=20).max() ) daily["prev60_low"] = grouped["low_price"].transform( lambda s: s.shift(1).rolling(60, min_periods=20).min() ) daily["flat60_range_pct"] = (daily["prev60_high"] / daily["prev60_low"] - 1.0) * 100.0 daily["flat60_flag"] = daily["flat60_range_pct"].le(35.0) # Previous-high reference volume: use the first occurrence of the 60-day rolling high. # This preserves the audited pilot candidate semantics while exposing the policy in ledgers. prev_high_volume = [] prev_high_ref_date = [] prev_high_ref_policy = [] last_limit_date = [] for _symbol, g in daily.groupby("symbol", sort=False): highs = g["high_price"].to_numpy() vols = g["volume"].to_numpy() dates = g["trade_date"].to_numpy() limit_flags = g["limit_up_event_flag"].to_numpy() for i in range(len(g)): start = max(0, i - 60) if i - start >= 20: window_highs = highs[start:i] max_pos = int(window_highs.argmax()) prev_high_volume.append(vols[start + max_pos]) prev_high_ref_date.append(pd.Timestamp(dates[start + max_pos])) prev_high_ref_policy.append("FIRST_PREVIOUS_HIGH_IN_60D_WINDOW") else: prev_high_volume.append(float("nan")) prev_high_ref_date.append(pd.NaT) prev_high_ref_policy.append("") lim_start = max(0, i - 29) recent_idx = [idx for idx in range(lim_start, i + 1) if limit_flags[idx]] last_limit_date.append(pd.Timestamp(dates[recent_idx[-1]]) if recent_idx else pd.NaT) daily["prev60_high_volume"] = prev_high_volume daily["prev60_high_ref_date"] = prev_high_ref_date daily["prev60_high_ref_policy"] = prev_high_ref_policy daily["last_limitup_date"] = last_limit_date daily["days_since_last_limitup"] = (daily["trade_date"] - daily["last_limitup_date"]).dt.days daily["touch_prev_high_flag"] = daily["prev60_high"].gt(0) & daily["high_price"].ge(daily["prev60_high"] * 0.995) daily["prev_high_volume_pass_flag"] = (~daily["touch_prev_high_flag"]) | daily["volume"].gt(daily["prev60_high_volume"]) trade_dates = sorted(pd.to_datetime(daily["trade_date"].drop_duplicates()).tolist()) entry_links = pd.DataFrame( { "signal_trade_date": trade_dates[:-1], "entry_trade_date": trade_dates[1:], } ) entry_links = entry_links[ (entry_links["entry_trade_date"] >= pd.Timestamp(minute_start)) & (entry_links["entry_trade_date"] <= pd.Timestamp(minute_end)) ].copy() breadth = breadth.rename(columns={"trade_date": "signal_trade_date"}) entry_links = entry_links.merge(breadth, on="signal_trade_date", how="left") entry_links["market_gate_open_flag"] = entry_links["up_count"].fillna(-1).ge(3000) entry_links["market_gate_status"] = entry_links["market_gate_open_flag"].map( {True: "MKT_GATE_OPEN_PREV_DAY_UP_3000", False: "NO_TRADE_MARKET_GATE_CLOSED"} ) signal_dates = set(entry_links["signal_trade_date"]) candidates = daily[daily["trade_date"].isin(signal_dates)].copy() candidates = candidates.merge( entry_links, left_on="trade_date", right_on="signal_trade_date", how="inner", suffixes=("", "_gate"), ) candidates = candidates[ candidates["prev_close"].gt(0) & candidates["prev5_avg_volume"].gt(0) & candidates["recent_limitup_30_flag"] & candidates["volume_ratio"].ge(cfg["candidate_rules"]["volume_ratio_threshold"]) & candidates["upper_shadow_pct"].gt(0) ].copy() candidates["strict_candidate_flag"] = candidates["prev_high_volume_pass_flag"] candidates["candidate_status"] = candidates["strict_candidate_flag"].map( {True: "PASS", False: "FAKE_BREAKOUT_RISK_REVIEW"} ) candidates["candidate_rank"] = ( candidates.sort_values( ["signal_trade_date", "upper_shadow_pct", "volume_ratio", "amount"], ascending=[True, False, False, False], ) .groupby("signal_trade_date") .cumcount() + 1 ) candidates = candidates[candidates["candidate_rank"].le(cfg["candidate_rules"]["upper_shadow_top_n"])].copy() candidates = candidates.sort_values(["entry_trade_date", "candidate_rank", "symbol"]).reset_index(drop=True) candidates["candidate_id"] = [ f"CAND-{RUN_ID}-{i + 1:05d}" for i in range(len(candidates)) ] out_cols = [ "candidate_id", "entry_trade_date", "signal_trade_date", "symbol", "candidate_rank", "candidate_status", "strict_candidate_flag", "market_gate_status", "market_gate_open_flag", "stock_count", "up_count", "flat_count", "down_count", "run_id", "price_source_table", "open_price", "high_price", "low_price", "close_price", "prev_close", "daily_return_pct", "volume", "amount", "prev5_avg_volume", "volume_ratio", "upper_shadow_pct", "body_pct", "recent_limitup_30_flag", "last_limitup_date", "days_since_last_limitup", "touch_prev_high_flag", "prev60_high", "prev60_high_volume", "prev60_high_ref_date", "prev60_high_ref_policy", "prev_high_volume_pass_flag", "flat60_flag", "flat60_range_pct", ] candidate_ledger = candidates[out_cols].copy() for col in ["entry_trade_date", "signal_trade_date", "last_limitup_date", "prev60_high_ref_date"]: candidate_ledger[col] = pd.to_datetime(candidate_ledger[col]).dt.strftime("%Y-%m-%d") entry_summary = entry_links[ [ "entry_trade_date", "signal_trade_date", "market_gate_status", "market_gate_open_flag", "stock_count", "up_count", "flat_count", "down_count", "run_id", "price_source_table", ] ].copy() entry_summary["entry_trade_date"] = entry_summary["entry_trade_date"].dt.strftime("%Y-%m-%d") entry_summary["signal_trade_date"] = entry_summary["signal_trade_date"].dt.strftime("%Y-%m-%d") candidate_counts = candidate_ledger.groupby("entry_trade_date").agg( candidate_count=("candidate_id", "count"), strict_candidate_count=("strict_candidate_flag", "sum"), review_candidate_count=("candidate_status", lambda s: int((s != "PASS").sum())), max_upper_shadow_pct=("upper_shadow_pct", "max"), max_volume_ratio=("volume_ratio", "max"), ).reset_index() summary = entry_summary.merge(candidate_counts, on="entry_trade_date", how="left") for col in ["candidate_count", "strict_candidate_count", "review_candidate_count"]: summary[col] = summary[col].fillna(0).astype(int) candidate_path = write_csv(candidate_ledger, "candidate_ledger.csv") summary_path = write_csv(summary, "candidate_date_summary.csv") open_dates = int(summary["market_gate_open_flag"].sum()) closed_dates = int((~summary["market_gate_open_flag"]).sum()) selected = [] def pick_one(label: str, frame: pd.DataFrame, sort_cols: list[str], ascending: list[bool]) -> None: nonlocal selected used_dates = {row["entry_trade_date"] for row in selected} pool = frame[~frame["entry_trade_date"].isin(used_dates)].copy() if pool.empty: return row = pool.sort_values(sort_cols, ascending=ascending).iloc[0].to_dict() row["selection_bucket"] = label selected.append(row) summary["entry_dt"] = pd.to_datetime(summary["entry_trade_date"]) open_with_candidates = summary[(summary["market_gate_open_flag"]) & (summary["strict_candidate_count"] > 0)] closed_with_candidates = summary[(~summary["market_gate_open_flag"]) & (summary["candidate_count"] > 0)] open_no_candidates = summary[(summary["market_gate_open_flag"]) & (summary["candidate_count"] == 0)] review_risk = summary[(summary["review_candidate_count"] > 0)] near_start = open_with_candidates.sort_values("entry_dt").head(25) near_end = open_with_candidates.sort_values("entry_dt", ascending=False).head(25) pick_one("OPEN_STRONG_CANDIDATE", open_with_candidates, ["strict_candidate_count", "max_upper_shadow_pct"], [False, False]) pick_one("OPEN_HIGH_VOLUME_RATIO", open_with_candidates, ["max_volume_ratio", "strict_candidate_count"], [False, False]) pick_one("MARKET_GATE_CLOSED_WITH_CANDIDATES", closed_with_candidates, ["candidate_count", "max_upper_shadow_pct"], [False, False]) pick_one("OPEN_NO_CANDIDATE", open_no_candidates, ["entry_dt"], [True]) pick_one("PREV_HIGH_REVIEW_RISK", review_risk, ["review_candidate_count", "candidate_count"], [False, False]) pick_one("MINUTE_COVERAGE_START_BOUNDARY", near_start, ["entry_dt"], [True]) pick_one("MINUTE_COVERAGE_END_BOUNDARY", near_end, ["entry_dt"], [False]) pick_one("MID_RANGE_NORMAL_OPEN", open_with_candidates, ["entry_dt"], [True]) case_rows = [] for i, row in enumerate(selected, 1): case_id = f"WUJI-PILOT-CASE-{i:02d}-{row['entry_trade_date'].replace('-', '')}" status = "SELECTED_FOR_PILOT_REPLAY" if not row["market_gate_open_flag"]: status = "SELECTED_NO_TRADE_MARKET_GATE_CLOSED" elif row["strict_candidate_count"] == 0: status = "SELECTED_NO_STRICT_CANDIDATE" case_rows.append( { "case_id": case_id, "entry_trade_date": row["entry_trade_date"], "signal_trade_date": row["signal_trade_date"], "selection_bucket": row["selection_bucket"], "case_status": status, "market_gate_status": row["market_gate_status"], "candidate_count": int(row["candidate_count"]), "strict_candidate_count": int(row["strict_candidate_count"]), "review_candidate_count": int(row["review_candidate_count"]), "up_count": "" if pd.isna(row["up_count"]) else int(row["up_count"]), "down_count": "" if pd.isna(row["down_count"]) else int(row["down_count"]), "selection_reason": ( "覆盖小样本分层：" + row["selection_bucket"] ), } ) case_index = pd.DataFrame(case_rows) case_path = write_csv(case_index, "case_index.csv") result = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": "2026-06-08T00:40:00+08:00", "stage": "CANDIDATE_POOL_GENERATED", "date_constraints": { "entry_start": minute_start, "entry_end": minute_end, "reason": "minute replay evidence required for buy/sell decision views", }, "input_rows": { "daily_rows": int(len(daily)), "breadth_rows": int(len(breadth)), "entry_dates": int(len(entry_links)), }, "candidate_counts": { "candidate_rows": int(len(candidate_ledger)), "candidate_entry_dates": int(candidate_ledger["entry_trade_date"].nunique()) if len(candidate_ledger) else 0, "strict_candidate_rows": int(candidate_ledger["strict_candidate_flag"].sum()) if len(candidate_ledger) else 0, "review_candidate_rows": int((candidate_ledger["candidate_status"] != "PASS").sum()) if len(candidate_ledger) else 0, "market_gate_open_entry_dates": open_dates, "market_gate_closed_entry_dates": closed_dates, }, "candidate_sort_policy": { "group_key": "signal_trade_date", "sort_keys": ["upper_shadow_pct", "volume_ratio", "amount"], "ascending": [False, False, False], "evidence_field": "amount", "top_n": cfg["candidate_rules"]["upper_shadow_top_n"], }, "previous_high_reference_volume_policy": { "policy_id": "FIRST_PREVIOUS_HIGH_IN_60D_WINDOW", "description": "When the signal day touches the previous 60-trading-day high, prev60_high_volume uses the first occurrence of the maximum high in the prior 60-trading-day window. This run preserves the audited pilot implementation and exposes prev60_high_ref_date / prev60_high_ref_policy for review.", "ledger_fields": ["prev60_high", "prev60_high_volume", "prev60_high_ref_date", "prev60_high_ref_policy"], }, "selected_cases": case_rows, "artifacts": { "candidate_ledger.csv": { "path": str(candidate_path.relative_to(ROOT)).replace("\\", "/"), "size": candidate_path.stat().st_size, "sha256": sha256_file(candidate_path), }, "candidate_date_summary.csv": { "path": str(summary_path.relative_to(ROOT)).replace("\\", "/"), "size": summary_path.stat().st_size, "sha256": sha256_file(summary_path), }, "case_index.csv": { "path": str(case_path.relative_to(ROOT)).replace("\\", "/"), "size": case_path.stat().st_size, "sha256": sha256_file(case_path), }, }, "limitations": [ "limit-up memory uses high/previous close >= 9.5% as a code proxy for first-pass candidate generation; later image/manual review must confirm semantics.", "flat60_flag uses a 60-day high/low range <= 35% only as a stratification signal, not a hard baseline buy rule.", "candidate_rank uses amount as the final same-day tie-breaker; candidate_ledger.csv now retains amount as sorting evidence.", "prev60_high_volume uses FIRST_PREVIOUS_HIGH_IN_60D_WINDOW and records prev60_high_ref_date for audit.", "candidate pool means observation pool only; it is not a buy list and produces no return conclusion.", ], "next_step": "Generate candidate daily K-line image package for selected pilot cases.", } (ROOT / "candidate_generation_summary.json").write_text( json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) summary_md = ROOT / "candidate_generation_summary.md" summary_md.write_text( "\n".join( [ "# candidate_generation_summary", "", f"run_id：`{RUN_ID}`", "阶段：`CANDIDATE_POOL_GENERATED`", "", "## 结果", "", f"- 入场日期范围：`{minute_start}` 至 `{minute_end}`", f"- 可评估 entry dates：{len(entry_links)}", f"- 候选行数：{len(candidate_ledger)}", f"- 有候选 entry dates：{candidate_ledger['entry_trade_date'].nunique() if len(candidate_ledger) else 0}", f"- strict candidate 行数：{int(candidate_ledger['strict_candidate_flag'].sum()) if len(candidate_ledger) else 0}", f"- review/risk candidate 行数：{int((candidate_ledger['candidate_status'] != 'PASS').sum()) if len(candidate_ledger) else 0}", f"- 市场闸门打开 entry dates：{open_dates}", f"- 市场闸门关闭 entry dates：{closed_dates}", f"- 已选择 pilot cases：{len(case_index)}", f"- 候选排名排序：`upper_shadow_pct desc -> volume_ratio desc -> amount desc`", f"- 前高参考量口径：`FIRST_PREVIOUS_HIGH_IN_60D_WINDOW`，账本保留 `prev60_high_ref_date`", "", "## 边界", "", "候选池只表示进入观察，不代表买入，不产生收益率、成功率、胜率或回撤结论。", "", "分钟线可回放边界为 `2023-03-24` 至 `2026-04-20`，后续小样本买卖图和回放不得超出该范围。", "", "## 下一步", "", "生成已选 pilot cases 的日 K 选股图片包和 `image_manifest.csv`。", "", ] ), encoding="utf-8", ) if __name__ == "__main__": main()