from __future__ import annotations import csv import hashlib import importlib.util import json from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo import pandas as pd RUN_ID = "RUN-ANA-WUJI-RECENT-STRICT-CANDIDATES-20260611-001" ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = ROOT.parents[2] CHART_DIR = ROOT / "charts" SOURCE_SCRIPT = ( PROJECT_ROOT / "ana-data/result/RUN-ANA-WUJI-FULL-2023-2026-20260608-001/tools/generate_candidate_pool.py" ) def load_source_module(): spec = importlib.util.spec_from_file_location("wuji_candidate_source", SOURCE_SCRIPT) if spec is None or spec.loader is None: raise RuntimeError(f"Unable to load source script: {SOURCE_SCRIPT}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module 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 add_features(daily: pd.DataFrame) -> pd.DataFrame: 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["prev10_avg_volume"] = grouped["volume"].transform( lambda s: s.shift(1).rolling(10, min_periods=10).mean() ) daily["volume_ratio"] = daily["volume"] / daily["prev5_avg_volume"] daily["volume_ratio_ma10"] = daily["volume"] / daily["prev10_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["upper_shadow_pct"] = ( (daily["high_price"] - daily[["open_price", "close_price"]].max(axis=1)) / daily["prev_close"] ) * 100.0 day_range = daily["high_price"] - daily["low_price"] daily["upper_shadow_range_ratio"] = ( daily["high_price"] - daily[["open_price", "close_price"]].max(axis=1) ) / day_range.replace(0, pd.NA) 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 prior_limitup_30 = [] last_limitup_date = [] days_since_limitup = [] last_limitup_close = [] post_limitup_min_low = [] prev_high_volume = [] prev_high_ref_date = [] prev_high_ref_policy = [] for _symbol, group in daily.groupby("symbol", sort=False): highs = group["high_price"].to_numpy() lows = group["low_price"].to_numpy() closes = group["close_price"].to_numpy() vols = group["volume"].to_numpy() dates = group["trade_date"].to_numpy() limit_flags = group["limit_up_event_flag"].to_numpy() n = len(group) for i in range(n): start30 = max(0, i - 30) prior_limit_idx = [j for j in range(start30, i) if limit_flags[j]] if prior_limit_idx: last = prior_limit_idx[-1] prior_limitup_30.append(True) last_limitup_date.append(pd.Timestamp(dates[last])) days_since_limitup.append(i - last) last_limitup_close.append(closes[last]) post_limitup_min_low.append(float(lows[last + 1 : i + 1].min())) else: prior_limitup_30.append(False) last_limitup_date.append(pd.NaT) days_since_limitup.append(pd.NA) last_limitup_close.append(float("nan")) post_limitup_min_low.append(float("nan")) start60 = max(0, i - 60) if i - start60 >= 20: window_highs = highs[start60:i] max_pos = int(window_highs.argmax()) prev_high_volume.append(vols[start60 + max_pos]) prev_high_ref_date.append(pd.Timestamp(dates[start60 + 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("") daily["prior_limitup_30_flag"] = prior_limitup_30 daily["last_prior_limitup_date"] = last_limitup_date daily["days_since_prior_limitup"] = days_since_limitup daily["last_prior_limitup_close"] = last_limitup_close daily["post_limitup_min_low"] = post_limitup_min_low daily["pullback_from_last_limitup_close_pct"] = ( daily["post_limitup_min_low"] / daily["last_prior_limitup_close"] - 1.0 ) * 100.0 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["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"]) return daily def breadth_from_daily(daily: pd.DataFrame) -> pd.DataFrame: gate = daily[daily["prev_close"].gt(0)].copy() gate["up_flag"] = gate["close_price"].gt(gate["prev_close"]) gate["flat_flag"] = gate["close_price"].eq(gate["prev_close"]) gate["down_flag"] = gate["close_price"].lt(gate["prev_close"]) breadth = ( gate.groupby("trade_date") .agg( stock_count=("symbol", "count"), up_count=("up_flag", "sum"), flat_count=("flat_flag", "sum"), down_count=("down_flag", "sum"), ) .reset_index() .rename(columns={"trade_date": "signal_trade_date"}) ) breadth["market_gate_open_flag"] = breadth["up_count"].ge(3000) breadth["market_gate_status"] = breadth["market_gate_open_flag"].map( { True: "MKT_GATE_OPEN_SIGNAL_DAY_UP_3000_RECALC_FROM_DAILY", False: "NO_TRADE_MARKET_GATE_CLOSED_SIGNAL_DAY_RECALC_FROM_DAILY", } ) return breadth def build_strict_candidates( daily: pd.DataFrame, signal_date: pd.Timestamp, entry_date: pd.Timestamp | None, label: str, breadth: pd.DataFrame, ) -> pd.DataFrame: frame = daily[daily["trade_date"].eq(signal_date)].copy() frame = frame[~frame["symbol"].str.endswith(".BJ")].copy() frame = frame[ frame["prev_close"].gt(0) & frame["prev5_avg_volume"].gt(0) & frame["prior_limitup_30_flag"] & frame["volume_ratio"].ge(2.0) & frame["upper_shadow_pct"].ge(3.0) & frame["upper_shadow_range_ratio"].ge(0.40) & frame["pullback_from_last_limitup_close_pct"].le(-3.0) & frame["prev_high_volume_pass_flag"] ].copy() frame["signal_trade_date"] = signal_date frame["entry_trade_date"] = entry_date if entry_date is not None else pd.NaT frame = frame.merge(breadth, on="signal_trade_date", how="left") frame["scan_label"] = label frame["strict_candidate_status"] = "STRICT_CODE_PASS_SUPPORT_REVIEW_REQUIRED" frame["support_manual_decision"] = "SUPPORT_REVIEW_REQUIRED" frame["support_manual_reason_cn"] = "代码硬筛已满足：近30交易日prior涨停、涨停后回调至少3%、严格倍量、长上影、前高量能过滤；底部承接强弱需看图人工确认。" frame["candidate_rank"] = ( frame.sort_values( ["signal_trade_date", "upper_shadow_pct", "volume_ratio", "amount"], ascending=[True, False, False, False], ) .groupby("signal_trade_date") .cumcount() + 1 ) frame = frame.sort_values(["candidate_rank", "symbol"]).reset_index(drop=True) frame["candidate_id"] = [f"CAND-{RUN_ID}-{label}-{i + 1:04d}" for i in range(len(frame))] return frame def draw_chart(history: pd.DataFrame, row: pd.Series, output: Path) -> None: hist = history.tail(110).reset_index(drop=True) width = 1200 height = 680 left = 70 right = 30 top = 58 price_h = 430 vol_top = top + price_h + 35 vol_h = 120 n = max(len(hist), 1) slot = (width - left - right) / n candle_w = max(2.0, slot * 0.55) low = float(hist["low_price"].min()) high = float(hist["high_price"].max()) if high <= low: high = low + 1 pad = (high - low) * 0.05 low -= pad high += pad max_vol = float(hist["volume"].max() or 1) def x(i: int) -> float: return left + slot * (i + 0.5) def y_price(v: float) -> float: return top + (high - float(v)) / (high - low) * price_h def y_vol(v: float) -> float: return vol_top + vol_h - float(v) / max_vol * vol_h parts = [ f'', '', f'{row["symbol"]} strict candidate {pd.Timestamp(row["signal_trade_date"]).date()} vol={row["volume_ratio"]:.2f} upper={row["upper_shadow_pct"]:.2f}% pullback={row["pullback_from_last_limitup_close_pct"]:.2f}%', f'', f'', ] for frac in [0, 0.25, 0.5, 0.75, 1.0]: yy = top + frac * price_h price = high - frac * (high - low) parts.append(f'') parts.append(f'{price:.2f}') def ma_path(window: int, color: str) -> None: ma = hist["close_price"].rolling(window).mean() pts = [] for i, v in enumerate(ma): if pd.notna(v): pts.append(f"{x(i):.1f},{y_price(float(v)):.1f}") if len(pts) >= 2: parts.append(f'') ma_path(5, "#f0a000") ma_path(20, "#1f77b4") ma_path(60, "#9467bd") for i, r in hist.iterrows(): color = "#d62728" if r["close_price"] >= r["open_price"] else "#2ca02c" cx = x(i) parts.append( f'' ) y_open = y_price(r["open_price"]) y_close = y_price(r["close_price"]) rect_y = min(y_open, y_close) rect_h = max(1.0, abs(y_open - y_close)) parts.append( f'' ) vy = y_vol(r["volume"]) parts.append( f'' ) signal_idx = hist.index[hist["trade_date"].eq(row["signal_trade_date"])] if len(signal_idx): idx = int(signal_idx[0]) cx = x(idx) parts.append(f'') parts.append(f'') parts.append(f'signal') limit_date = row.get("last_prior_limitup_date") if pd.notna(limit_date): limit_idx = hist.index[hist["trade_date"].eq(pd.Timestamp(limit_date))] if len(limit_idx): cx = x(int(limit_idx[0])) parts.append(f'') parts.append(f'prior limit-up') tick_step = max(1, len(hist) // 8) for i in range(0, len(hist), tick_step): parts.append(f'{pd.Timestamp(hist.loc[i, "trade_date"]).strftime("%m-%d")}') parts.append('MA5') parts.append('MA20') parts.append('MA60') parts.append("") output.write_text("\n".join(parts), encoding="utf-8") def render_summary(summary: dict) -> str: return f"""# Wuji Recent Strict Candidate Scan run_id: `{summary["run_id"]}` generated_at: `{summary["generated_at"]}` This package is stricter than the broad non-BJ daily candidate list. It excludes all `.BJ` symbols and requires prior limit-up memory, pullback after that limit-up, strict double-volume, long upper shadow, and previous-high volume guard. ## Frozen Hard Filters - Exclude `.BJ`. - Prior limit-up within 30 trading days, excluding the signal day itself. - Pullback after the prior limit-up: min low after prior limit-up through signal day <= prior limit-up close - 3%. - Strict volume: `volume / previous 5 trading-day average volume >= 2.0`. - Long upper shadow: `upper_shadow_pct >= 3.0` and upper shadow accounts for at least 40% of daily range. - If touching previous 60-trading-day high, signal-day volume must be greater than the previous-high reference volume. ## Counts - Entry-ready strict code candidates: `{summary["counts"]["entry_ready_strict_code_rows"]}`. - Latest-signal strict code candidates: `{summary["counts"]["latest_signal_strict_code_rows"]}`. - Chart evidence files: `{summary["counts"]["chart_count"]}`. ## Boundary `support_manual_decision` is intentionally left as `SUPPORT_REVIEW_REQUIRED`: the phrase "bottom support must be strong" is an image/manual-review condition in the Wuji notes, not a pure hard-code condition. This package prepares the strict code candidates and chart evidence for manual support review and reviewer audit. """ def write_self_check(all_rows: pd.DataFrame, chart_rows: list[dict]) -> list[Path]: checks = [] def add(check_id: str, ok: bool, detail: str) -> None: checks.append({"check_id": check_id, "status": "PASS" if ok else "FAIL", "detail": detail}) add("NO_BJ_SYMBOL_RETAINED", int(all_rows["symbol"].str.endswith(".BJ").sum()) == 0, f"bj_rows={int(all_rows['symbol'].str.endswith('.BJ').sum())}") add("STRICT_VOLUME_GE_2", bool(all_rows["volume_ratio"].ge(2.0).all()), f"min_volume_ratio={all_rows['volume_ratio'].min() if len(all_rows) else 'NA'}") add("PRIOR_LIMITUP_30", bool(all_rows["prior_limitup_30_flag"].all()), f"rows={len(all_rows)}") add("LONG_UPPER_SHADOW", bool((all_rows["upper_shadow_pct"].ge(3.0) & all_rows["upper_shadow_range_ratio"].ge(0.40)).all()), f"rows={len(all_rows)}") add("PULLBACK_AFTER_LIMITUP", bool(all_rows["pullback_from_last_limitup_close_pct"].le(-3.0).all()), f"max_pullback_pct={all_rows['pullback_from_last_limitup_close_pct'].max() if len(all_rows) else 'NA'}") add("PREV_HIGH_VOLUME_GUARD", bool(all_rows["prev_high_volume_pass_flag"].all()), f"rows={len(all_rows)}") missing = [r["chart_path"] for r in chart_rows if not (ROOT / r["chart_path"]).exists()] add("CHARTS_EXIST", len(missing) == 0, f"missing={len(missing)}; charts={len(chart_rows)}") add("SUPPORT_REVIEW_NOT_AUTO_PASSED", set(all_rows["support_manual_decision"]) == {"SUPPORT_REVIEW_REQUIRED"} if len(all_rows) else True, "manual support review required") items_path = ROOT / "self_check_items.csv" with items_path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=["check_id", "status", "detail"]) writer.writeheader() writer.writerows(checks) json_path = ROOT / "self_check.json" json_path.write_text( json.dumps( { "schema_version": "1.0", "run_id": RUN_ID, "status": "PASS" if all(c["status"] == "PASS" for c in checks) else "FAIL", "items": checks, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) return [items_path, json_path] def write_manifest(files: list[Path]) -> Path: rows = [] for path in files: rows.append( { "path": path.relative_to(ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path), } ) manifest = ROOT / "manifest.json" manifest.write_text( json.dumps( { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(timespec="seconds"), "files": rows, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) return manifest def main() -> None: ROOT.mkdir(parents=True, exist_ok=True) CHART_DIR.mkdir(parents=True, exist_ok=True) source = load_source_module() generated_at = datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(timespec="seconds") with source.get_conn() as conn: latest_date = pd.read_sql( "SELECT MAX(trade_date) AS latest_trade_date FROM a_share_daily_price", conn )["latest_trade_date"].iloc[0] latest_date = pd.Timestamp(latest_date) pull_start = (latest_date - pd.Timedelta(days=320)).strftime("%Y-%m-%d") 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": latest_date.strftime("%Y-%m-%d")}, ) 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") daily = add_features(daily) breadth = breadth_from_daily(daily) trade_dates = sorted(daily["trade_date"].drop_duplicates()) latest_signal_date = pd.Timestamp(trade_dates[-1]) entry_signal_date = pd.Timestamp(trade_dates[-2]) entry_date = latest_signal_date entry_ready = build_strict_candidates( daily, entry_signal_date, entry_date, "ENTRY_READY_STRICT", breadth ) latest_signal = build_strict_candidates( daily, latest_signal_date, None, "LATEST_SIGNAL_STRICT", breadth ) all_rows = pd.concat([entry_ready, latest_signal], ignore_index=True) out_cols = [ "candidate_id", "scan_label", "entry_trade_date", "signal_trade_date", "symbol", "candidate_rank", "strict_candidate_status", "support_manual_decision", "support_manual_reason_cn", "market_gate_status", "market_gate_open_flag", "stock_count", "up_count", "flat_count", "down_count", "open_price", "high_price", "low_price", "close_price", "prev_close", "daily_return_pct", "volume", "amount", "prev5_avg_volume", "prev10_avg_volume", "volume_ratio", "volume_ratio_ma10", "upper_shadow_pct", "upper_shadow_range_ratio", "body_pct", "prior_limitup_30_flag", "last_prior_limitup_date", "days_since_prior_limitup", "last_prior_limitup_close", "post_limitup_min_low", "pullback_from_last_limitup_close_pct", "touch_prev_high_flag", "prev60_high", "prev60_high_volume", "prev60_high_ref_date", "prev60_high_ref_policy", "prev_high_volume_pass_flag", "flat60_range_pct", ] for df in [entry_ready, latest_signal, all_rows]: for col in out_cols: if col not in df.columns: df[col] = pd.NA chart_rows = [] for _, row in all_rows.iterrows(): hist = daily[ (daily["symbol"].eq(row["symbol"])) & (daily["trade_date"].le(pd.Timestamp(row["signal_trade_date"]))) ].copy() chart_path = CHART_DIR / f"{row['candidate_id']}_{row['symbol']}.svg" draw_chart(hist, row, chart_path) chart_rows.append( { "candidate_id": row["candidate_id"], "symbol": row["symbol"], "scan_label": row["scan_label"], "signal_trade_date": pd.Timestamp(row["signal_trade_date"]).strftime("%Y-%m-%d"), "chart_path": chart_path.relative_to(ROOT).as_posix(), "chart_exists": chart_path.exists(), "chart_sha256": sha256_file(chart_path) if chart_path.exists() else "", } ) chart_audit = pd.DataFrame(chart_rows) files = [ write_csv(all_rows[out_cols], "strict_candidate_ledger.csv"), write_csv(entry_ready[out_cols], "entry_ready_strict_candidates.csv"), write_csv(latest_signal[out_cols], "latest_signal_strict_candidates.csv"), write_csv(chart_audit, "chart_evidence_audit.csv"), ] summary = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": generated_at, "stage": "STRICT_DAILY_CANDIDATE_CODE_SCAN_SUPPORT_REVIEW_REQUIRED", "latest_daily_trade_date": latest_signal_date.strftime("%Y-%m-%d"), "rules": { "exclude": "*.BJ", "prior_limitup_30_flag": True, "volume_ratio_prev5_min": 2.0, "upper_shadow_pct_min": 3.0, "upper_shadow_range_ratio_min": 0.40, "pullback_from_last_limitup_close_pct_max": -3.0, "prev_high_volume_pass_flag": True, }, "counts": { "entry_ready_strict_code_rows": int(len(entry_ready)), "latest_signal_strict_code_rows": int(len(latest_signal)), "all_strict_code_rows": int(len(all_rows)), "chart_count": int(len(chart_rows)), "bj_rows_retained": int(all_rows["symbol"].str.endswith(".BJ").sum()) if len(all_rows) else 0, }, "market_gate": { "entry_ready": { "signal_trade_date": entry_signal_date.strftime("%Y-%m-%d"), "entry_trade_date": entry_date.strftime("%Y-%m-%d"), "market_gate_status": entry_ready["market_gate_status"].iloc[0] if len(entry_ready) else None, "up_count": int(entry_ready["up_count"].iloc[0]) if len(entry_ready) else None, }, "latest_signal": { "signal_trade_date": latest_signal_date.strftime("%Y-%m-%d"), "market_gate_status": latest_signal["market_gate_status"].iloc[0] if len(latest_signal) else None, "up_count": int(latest_signal["up_count"].iloc[0]) if len(latest_signal) else None, }, }, "boundary": [ "This is a strict daily code scan, not a buy recommendation.", "Bottom support strength is not auto-passed; support_manual_decision remains SUPPORT_REVIEW_REQUIRED.", "Minute-level buy point confirmation is not included.", ], } summary_json = ROOT / "summary.json" summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") summary_md = ROOT / "summary.md" summary_md.write_text(render_summary(summary), encoding="utf-8") files.extend([summary_json, summary_md]) files.extend([ROOT / r["chart_path"] for r in chart_rows]) files.extend(write_self_check(all_rows, chart_rows)) files.append(Path(__file__)) manifest = write_manifest(files) print( json.dumps( { "run_id": RUN_ID, "result_dir": ROOT.as_posix(), "entry_ready_strict_code_rows": len(entry_ready), "latest_signal_strict_code_rows": len(latest_signal), "all_strict_code_rows": len(all_rows), "chart_count": len(chart_rows), "manifest": manifest.as_posix(), }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()